feat(experiments): shadow mode — observe-only candidate fan-out (Step 5.7b)

TRACKER.md

@@ -16,10 +16,11 @@
 | **Last updated** | 2026-06-07 |
 | **Current phase** | Phase 5 — Eval & Observability (**6 / 7 steps**) |
 | **Overall** | **63 / 84 steps** — Phases 0–4 complete |
-| **Next action** | **Step 5.7b — Shadow mode** — fan out N% of live queries to a candidate retriever (observe-only) and feed the A/B tracker. Final Phase-5 step, delivered in slices 5.7a–d. |
+| **Next action** | **Step 5.7c — A/B routing** — deterministic variant assignment that actually *serves* the candidate to a fraction of users (the first slice that can change a response). Final Phase-5 step, slices 5.7a–d. |
 
 **Recently shipped**
 
+- **5.7b** ✅ Shadow mode — observe-only candidate fan-out (`ShadowRunner`, background task) feeding the A/B tracker — [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145)
 - **5.7a** ✅ A/B analyzer + experiment tracker + `GET /v1/status/experiments` dashboard — [#143](https://github.com/officialCodeWork/AgentContextOS/pull/143)
 - **5.6** ✅ Status & Metrics GUI (full build) — drift/feedback/cost cards, query-trace viewer, regression bisector, Grafana dashboards, cross-links — [#137–#142](https://github.com/officialCodeWork/AgentContextOS/pull/142)
 - **5.5** ✅ Drift monitors — `rag-drift`, five PSI / mean-drop monitors — [#136](https://github.com/officialCodeWork/AgentContextOS/pull/136)
@@ -523,7 +524,7 @@
 | 5.6f | — Cross-links + close-out | ✅ | [#142](https://github.com/officialCodeWork/AgentContextOS/pull/142) |
 | 5.7 | A/B testing & shadow mode | 🚧 | — |
 | 5.7a | — A/B analyzer + tracker + dashboard | ✅ | [#143](https://github.com/officialCodeWork/AgentContextOS/pull/143) |
-| 5.7b | — Shadow mode | ⏳ | — |
+| 5.7b | — Shadow mode | ✅ | [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145) |
 | 5.7c | — A/B routing | ⏳ | — |
 | 5.7d | — Console + close-out | ⏳ | — |
 
@@ -612,9 +613,17 @@
 - `cfg.experiments` (opt-in, since A/B routing can change responses); wired inert by default; **observe-only** — no query-path change yet (shadow / routing are 5.7b/c); `ragctl experiments`; 17 tests
 - [ADR-0032](docs/adr/ADR-0032-ab-testing-shadow-mode.md), [reference/experiments.md](docs/reference/experiments.md)
 
-#### 5.7b — Shadow mode ⏳
+#### 5.7b — Shadow mode ✅ [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145)
 
-- **Next up.** Fan out N% of live queries to a candidate retriever (observe-only) and feed the A/B tracker
+- New `ShadowRunner` (`rag_gateway.experiments`) runs a **candidate** retriever alongside the served **control** path on a sampled fraction of live queries — observe-only — and feeds both variants' `outcome_metric` (mean retrieval score) into the 5.7a `ABExperimentTracker`
+- **Never delays the response:** scheduled as a FastAPI `BackgroundTask` (runs after the response is sent); **degrade-open** (`experiment.shadow_failed`); skipped on a retrieval-cache hit
+- **Deterministic** `request_id`-hash sampling; candidate = any `SupportsRoute` (config builds a `RetrievalRouter` differing only in `shadow_candidate` RRF weights; production injects one via `build_app(shadow_runner=…)`); same `read_chunk` PDP — no coverage-linter entry
+- `cfg.experiments` gains `shadow_enabled` / `shadow_sample_rate` / `shadow_experiment` / `shadow_candidate`; doubly opt-in (`enabled` **and** `shadow_enabled`); `ragctl shadow`; 27 tests
+- [ADR-0032](docs/adr/ADR-0032-ab-testing-shadow-mode.md), [reference/experiments.md](docs/reference/experiments.md), [architecture/ab-shadow-mode.md](docs/architecture/ab-shadow-mode.md)
+
+#### 5.7c — A/B routing ⏳
+
+- **Next up.** Deterministic variant assignment that serves the candidate to a fraction of users (the first slice that can change a response) and tags it
 
 ## Phase 6 — Governance & Tenancy (Weeks 28–34) ⏳
 

apps/gateway/src/rag_gateway/app.py

@@ -105,7 +105,9 @@
     FallbackChain,
     FallbackConfig,
     HybridRetriever,
+    HybridWeights,
     RetrievalRouter,
+    RouterConfig,
     SemanticEmbeddingCache,
 )
 from rag_webhooks import build_default_dispatcher, ingest_completed_event
@@ -166,13 +168,18 @@ def build_default_understanding() -> QueryUnderstandingPipeline:
 def build_default_retrieval_router(
     *,
     breaker_registry: BreakerRegistry | None = None,
+    base_weights: HybridWeights | None = None,
 ) -> RetrievalRouter:
     """Noop-backed HybridRetriever + RetrievalRouter.
 
     Wires all three retrieval surfaces so the router can fire any
     shape's preferred path; production wiring substitutes real
     pgvector / Elasticsearch / Neo4j backends.
 
+    ``base_weights`` overrides the per-source RRF weights (before shape bias) —
+    used to build a *candidate* router for shadow mode (Step 5.7b) that differs
+    from the control only in its fusion balance.
+
     When ``breaker_registry`` is supplied (Step 4.4), each backend is wrapped
     in a circuit breaker drawn from the registry, so a failing backend is
     isolated (open breaker → ``CircuitOpenError`` → ``HybridRetriever`` drops
@@ -192,9 +199,11 @@ def build_default_retrieval_router(
         keyword_backend=keyword,
         graph_backend=graph,
     )
+    config = RouterConfig(base_weights=base_weights) if base_weights is not None else None
     return RetrievalRouter(
         hybrid=hybrid,
         embedder=NoopEmbedder(dimension=8),
+        config=config,
     )
 
 
@@ -356,6 +365,7 @@ def build_app(
     drift_registry: Any | None = None,
     cost_tracker: Any | None = None,
     experiment_tracker: Any | None = None,
+    shadow_runner: Any | None = None,
     enable_cors: bool = True,
     default_tenant_id: TenantId | None = None,
 ) -> FastAPI:
@@ -548,6 +558,13 @@ def build_app(
     # by ``GET /v1/status/experiments``.  ``None`` in the plain ``build_app``.
     app.state.experiment_tracker = experiment_tracker
 
+    # Shadow-mode fan-out (Step 5.7b) — runs a candidate retriever observe-only
+    # on a sample of live queries and feeds the experiment tracker, from a
+    # background task so the served response is never delayed.  ``None`` (inert)
+    # in the plain ``build_app``; the config-driven wiring builds it from
+    # ``cfg.experiments`` when ``shadow_enabled``.
+    app.state.shadow_runner = shadow_runner
+
     # Outbound webhooks (Step 3.9) — the subscription registry + the delivery
     # dispatcher behind ``/v1/webhooks/subscriptions`` and the
     # ``ingest.completed`` emission.  Defaults: in-memory store + an HTTP

apps/gateway/src/rag_gateway/experiments.py

@@ -0,0 +1,163 @@
+"""Shadow-mode A/B fan-out (Step 5.7b).
+
+On a sampled fraction of live queries, run a **candidate** retriever alongside
+the served (**control**) path — *observe-only*, never affecting the response —
+and feed both variants' outcome metric into the
+:class:`~rag_observability.experiments.ABExperimentTracker`, which the dashboard
+(``GET /v1/status/experiments``) compares with the pure
+``analyze_ab_experiment`` (lift + confidence interval, Step 5.7a).
+
+Design (mirrors the Phase-5 observe-only features — drift, cost):
+
+* **Never alters the request.** The candidate retrieval is scheduled as a
+  FastAPI ``BackgroundTask``, so it runs *after* the response is sent — the
+  served path's latency is untouched.
+* **Degrade-open.** Any failure in the candidate retrieval is logged
+  (``experiment.shadow_failed``) and swallowed; the response was already sent.
+* **Deterministic sampling.** Whether a query is shadowed is a pure hash of its
+  ``request_id`` against the sample rate, so the decision is reproducible and
+  testable (no RNG in the hot path).
+
+The candidate is any :class:`~rag_retrieval.router.SupportsRoute` (a second
+:class:`~rag_retrieval.router.RetrievalRouter` with different fusion weights is
+the config-driven default; production injects one over the real backends).
+A/B *routing* — actually serving the candidate to a fraction of users — is the
+next slice (5.7c); this slice is strictly observe-only.
+
+See [docs/reference/experiments.md](../../../../docs/reference/experiments.md)
+and [docs/architecture/ab-shadow-mode.md](../../../../docs/architecture/ab-shadow-mode.md).
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+
+from rag_core import get_logger
+from rag_core.types import RequestContext
+from rag_observability.experiments import ABExperimentTracker
+
+_log = get_logger(__name__)
+
+__all__ = ["ShadowRunner", "outcome_metric"]
+
+# Largest value of an 8-hex-digit digest prefix — the denominator that maps a
+# hash into the half-open unit interval ``[0, 1)`` for sampling.
+_HASH_DENOM = 0xFFFFFFFF
+
+
+def outcome_metric(chunk_refs: list[Any]) -> float:
+    """Scalar retrieval-quality outcome for a result set: the mean ref score.
+
+    Higher is better; an empty result set scores ``0.0``.  This is the same
+    retrieval-score signal the Step 5.5 drift monitors observe — a backend-only
+    proxy that needs no served answer (the candidate is never shown to a user).
+    Richer outcomes (latency, recall proxies, per-shape buckets) are future work.
+    """
+    scores = [float(s) for r in chunk_refs if (s := getattr(r, "score", None)) is not None]
+    if not scores:
+        return 0.0
+    return sum(scores) / len(scores)
+
+
+def _sampled(request_id: str, sample_rate: float) -> bool:
+    """Deterministic in ``request_id`` — hash → ``[0, 1)`` vs ``sample_rate``."""
+    if sample_rate <= 0.0:
+        return False
+    if sample_rate >= 1.0:
+        return True
+    digest = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:8]
+    return (int(digest, 16) / _HASH_DENOM) < sample_rate
+
+
+class ShadowRunner:
+    """Runs a candidate retriever in shadow and feeds the experiment tracker.
+
+    Holds a candidate :class:`~rag_retrieval.router.SupportsRoute` (the
+    alternative retrieval config), the :class:`ABExperimentTracker` to feed, the
+    sample rate, and the experiment + variant labels.  :meth:`should_sample`
+    decides inclusion; :meth:`run` does the observe-only candidate retrieval and
+    records both variants' :func:`outcome_metric`.
+
+    Stateless beyond its dependencies, so one instance is shared across requests
+    (the tracker it feeds is itself thread-safe).
+    """
+
+    def __init__(
+        self,
+        *,
+        candidate_router: Any,  # SupportsRoute — Any to match the gateway's deps style.
+        tracker: ABExperimentTracker,
+        sample_rate: float = 0.1,
+        experiment: str = "shadow",
+        control_variant: str = "control",
+        candidate_variant: str = "candidate",
+    ) -> None:
+        self._candidate = candidate_router
+        self._tracker = tracker
+        self._sample_rate = max(0.0, min(1.0, sample_rate))
+        self._experiment = experiment
+        self._control = control_variant
+        self._candidate_variant = candidate_variant
+
+    @property
+    def experiment(self) -> str:
+        return self._experiment
+
+    def should_sample(self, request_id: str) -> bool:
+        """Whether this ``request_id`` is in the shadow sample (deterministic)."""
+        return _sampled(request_id, self._sample_rate)
+
+    async def run(
+        self,
+        ctx: RequestContext,
+        *,
+        text: str,
+        expansion_terms: dict[str, list[str]],
+        hyde_vector: list[float] | None,
+        corpus_ids: list[Any] | None,
+        top_k: int,
+        control_refs: list[Any],
+    ) -> None:
+        """Observe-only: retrieve with the candidate, record both outcome metrics.
+
+        ``control_refs`` is the result set the served path already produced, so
+        the control's metric needs no re-retrieval; only the candidate runs here.
+        Degrade-open — any failure logs ``experiment.shadow_failed`` and returns.
+        """
+        try:
+            _decision, candidate_refs = await self._candidate.route(
+                ctx,
+                text=text,
+                expansion_terms=expansion_terms,
+                hyde_vector=hyde_vector,
+                corpus_ids=corpus_ids,
+                top_k=top_k,
+            )
+            control_value = outcome_metric(control_refs)
+            candidate_value = outcome_metric(candidate_refs)
+            self._tracker.observe(self._experiment, self._control, control_value)
+            self._tracker.observe(self._experiment, self._candidate_variant, candidate_value)
+            _log.info(
+                "experiment.shadow_complete",
+                extra={
+                    "event_kind": "experiment.shadow_complete",
+                    "tenant_id": str(ctx.tenant_id),
+                    "experiment": self._experiment,
+                    "control_value": control_value,
+                    "candidate_value": candidate_value,
+                    "candidate_results_n": len(candidate_refs),
+                },
+            )
+        except (KeyboardInterrupt, SystemExit):
+            raise
+        except Exception as exc:
+            _log.warning(
+                "experiment.shadow_failed",
+                extra={
+                    "event_kind": "experiment.shadow_failed",
+                    "tenant_id": str(ctx.tenant_id),
+                    "experiment": self._experiment,
+                    "error": repr(exc),
+                },
+            )