feat(evaluation): independent retrieved-learning sampling; retire retrieval-log capture

reflexio/models/api_schema/domain/entities.py

@@ -123,8 +123,6 @@
     "ShareLink",
     "AdminInvalidateCacheRequest",
     "AdminInvalidateCacheResponse",
-    "PlaybookRetrievalLogItem",
-    "PlaybookRetrievalLog",
     "LineageEvent",
     "LineageContext",
     "RecordRef",
@@ -489,86 +487,6 @@ class RetrievedLearningEvaluationResult(BaseModel):
     created_at: int = Field(default_factory=lambda: int(datetime.now(UTC).timestamp()))
 
 
-class PlaybookRetrievalLogItem(BaseModel):
-    """One retrieved playbook plus the serve-time attribution snapshot.
-
-    ``agent_playbook_id`` remains as a nullable legacy compatibility field for
-    old agent-only rows. New rows should prefer ``target_kind`` + ``target_id``.
-    """
-
-    retrieval_log_item_id: int = 0
-    retrieval_log_id: int = 0
-    ordinal: int
-    agent_playbook_id: int | None = Field(default=None, gt=0)
-    target_kind: Literal["agent_playbook", "user_playbook"] | None = None
-    target_id: int | None = Field(default=None, gt=0)
-    target_title: str = ""
-    target_content: str = ""
-    target_trigger: str | None = None
-    source_user_playbook_ids: list[int] = Field(default_factory=list)
-    source_interaction_ids_by_user_playbook_id: dict[str, list[int]] = Field(
-        default_factory=dict
-    )
-    source_window_snapshot_mode: Literal["final_response", "live_lookup_compat"] = (
-        "live_lookup_compat"
-    )
-
-    @model_validator(mode="after")
-    def backfill_legacy_target_fields(self) -> Self:
-        if self.target_kind is None:
-            if self.agent_playbook_id is None:
-                raise ValueError(
-                    "PlaybookRetrievalLogItem requires target_kind/target_id or"
-                    " legacy agent_playbook_id"
-                )
-            self.target_kind = "agent_playbook"
-            self.target_id = self.agent_playbook_id
-            return self
-
-        if self.target_id is None:
-            if self.target_kind != "agent_playbook" or self.agent_playbook_id is None:
-                raise ValueError(
-                    "PlaybookRetrievalLogItem.target_id is required for explicit"
-                    " targets"
-                )
-            self.target_id = self.agent_playbook_id
-
-        if self.target_kind == "user_playbook":
-            if self.agent_playbook_id is not None:
-                raise ValueError(
-                    "user_playbook retrieval-log items cannot carry agent_playbook_id"
-                )
-            return self
-
-        if self.agent_playbook_id is None:
-            self.agent_playbook_id = self.target_id
-        elif self.agent_playbook_id != self.target_id:
-            raise ValueError(
-                "agent_playbook retrieval-log item agent_playbook_id must match"
-                " target_id"
-            )
-        return self
-
-
-class PlaybookRetrievalLog(BaseModel):
-    """A retrieval-log header with ordered item rows.
-
-    Used by retrieval-capture consumers to correlate retrieval decisions with
-    downstream outcomes. ``retrieval_log_id`` is assigned by the storage layer;
-    ``shown_items`` stores ids and serve-time attribution snapshots only.
-    """
-
-    retrieval_log_id: int = 0
-    request_id: str
-    session_id: str
-    interaction_id: int | None = None
-    user_id: str
-    query: str | None = None
-    agent_version: str | None = None
-    shown_items: list[PlaybookRetrievalLogItem] = Field(default_factory=list)
-    created_at: int = 0
-
-
 class LineageEvent(BaseModel):
     """Append-only, content-free provenance record. NEVER carries content/PII.
 

reflexio/models/api_schema/domain/governance.py

@@ -15,6 +15,10 @@
     "request",
     "session",
     "agent_success_evaluation_result",
+    # READ-COMPAT ONLY — do not remove. The retrieval-capture subsystem is gone
+    # and no writer emits this entity type any more, but historical audit_events
+    # rows still carry it; dropping the literal would make Pydantic reject those
+    # rows on read.
     "playbook_retrieval_log",
     "org",
 ]

reflexio/models/config_schema.py

@@ -513,6 +513,18 @@ class AgentSuccessConfig(_ExtractorWindowOverrideCompatMixin, BaseModel):
         le=1.0,
         description="Fraction of sessions to evaluate automatically.",
     )
+    retrieved_learning_sampling_rate: float | None = Field(
+        default=None,
+        ge=0.0,
+        le=1.0,
+        description=(
+            "Fraction of sessions to judge for retrieved-learning relevance and"
+            " impact. None inherits sampling_rate. Set this independently when"
+            " the retrieved-learning verdicts feed a downstream consumer (e.g."
+            " the offline playbook tuner) that needs denser coverage than the"
+            " session-success judge."
+        ),
+    )
     window_size_override: int | None = Field(default=None, gt=0)
     stride_size_override: int | None = Field(default=None, gt=0)
 

reflexio/server/extensions.py

@@ -16,10 +16,6 @@
 
 from fastapi import APIRouter
 
-from reflexio.server.services.unified_search_service import (
-    RetrievalCaptureHook,
-    configure_retrieval_capture_hook,
-)
 from reflexio.server.tracing import Tracer, configure_tracer
 from reflexio.server.usage_metrics import (
     UsageEventRecorder,
@@ -113,14 +109,6 @@ def set_usage_recorder(self, recorder: UsageEventRecorder | None) -> None:
         """
         configure_usage_event_recorder(recorder)
 
-    def set_retrieval_capture(self, hook: RetrievalCaptureHook | None) -> None:
-        """Install (or clear) the process-global retrieval-capture hook.
-
-        Args:
-            hook (RetrievalCaptureHook | None): Hook callable, or None to disable.
-        """
-        configure_retrieval_capture_hook(hook)
-
 
 @dataclass(frozen=True)
 class AppContext:
@@ -152,7 +140,7 @@ def install_services(self) -> None:  # noqa: B027
         """Register runtime-service providers (process-global). Default: none."""
 
     def install_hooks(self, hooks: HookRegistry) -> None:  # noqa: B027
-        """Install pipeline hooks (tracer/usage-recorder/retrieval-capture). Default: none.
+        """Install pipeline hooks (tracer / usage-recorder). Default: none.
 
         Args:
             hooks (HookRegistry): Registry to install hooks through.

reflexio/server/services/agent_success_evaluation/runner.py

@@ -102,6 +102,8 @@ def run_group_evaluation(
     llm_client: LiteLLMClient,
     *,
     force_regenerate: bool = False,
+    run_agent_success: bool = True,
+    run_retrieved_learning: bool = True,
 ) -> GroupEvaluationOutcome:
     """Run agent success evaluation for an entire session.
 
@@ -129,6 +131,12 @@ def run_group_evaluation(
         force_regenerate: When True, bypass the already-evaluated short-circuit
             and the completeness delay gate so the regenerate worker can
             re-evaluate sessions of any age regardless of prior state.
+        run_agent_success: Whether this session was admitted for the
+            session-success judge. The publish scheduler samples the two
+            families independently (see ``sampling.py``) and passes the result;
+            direct callers leave it True.
+        run_retrieved_learning: Whether this session was admitted for the
+            retrieved-learning relevance/impact judge.
 
     Returns:
         GroupEvaluationOutcome: Per-family statuses for this invocation.
@@ -160,7 +168,24 @@ def run_group_evaluation(
             )
             return GroupEvaluationOutcome("skipped", "skipped")
 
-    # 3. Check if agent success is already evaluated — skipped in
+    # 3. Per-family admission. The scheduler samples the two families
+    # independently and tells us which ones this session was admitted for, so a
+    # session sampled only for retrieved-learning never pays the session-success
+    # judge. Direct callers (regen jobs, the on-demand grade route) leave both
+    # flags at their default and run both families, as before.
+    if not run_agent_success:
+        return _finish_with_retrieved_evaluation(
+            "skipped",
+            user_id=user_id,
+            session_id=session_id,
+            agent_version=agent_version,
+            request_context=request_context,
+            llm_client=llm_client,
+            force_regenerate=force_regenerate,
+            run_retrieved_learning=run_retrieved_learning,
+        )
+
+    # 4. Check if agent success is already evaluated — skipped in
     # force_regenerate mode so the regenerate worker can re-evaluate a session
     # that's already been marked. Retrieved-learning evaluation still runs
     # below: its completion is independent of the agent-success marker.
@@ -187,6 +212,7 @@ def run_group_evaluation(
             request_context=request_context,
             llm_client=llm_client,
             force_regenerate=force_regenerate,
+            run_retrieved_learning=run_retrieved_learning,
         )
 
     # 4. Fetch interactions for all requests
@@ -318,6 +344,7 @@ def run_group_evaluation(
         request_context=request_context,
         llm_client=llm_client,
         force_regenerate=force_regenerate,
+        run_retrieved_learning=run_retrieved_learning,
     )
 
 
@@ -330,8 +357,11 @@ def _finish_with_retrieved_evaluation(
     request_context: RequestContext,
     llm_client: LiteLLMClient,
     force_regenerate: bool,
+    run_retrieved_learning: bool = True,
 ) -> GroupEvaluationOutcome:
     """Run the retrieved-learning phase best-effort and build the outcome."""
+    if not run_retrieved_learning:
+        return GroupEvaluationOutcome(agent_success_status, "skipped")
     try:
         retrieved_status, fingerprint = _run_retrieved_learning_evaluation(
             user_id=user_id,

reflexio/server/services/agent_success_evaluation/sampling.py

@@ -0,0 +1,79 @@
+"""Per-family sampling for group evaluation.
+
+A scheduled group evaluation runs two independent judge families over the same
+session: **agent success** (one session-level verdict) and **retrieved learning**
+(one relevance/impact verdict per served learning). They are sampled separately,
+because their consumers need different coverage: the retrieved-learning verdicts
+feed the offline playbook tuner, which needs enough sessions per playbook to
+clear its evidence floors, while the session-success judge is a product signal
+that is fine at a low rate.
+
+This module is imported by exactly ONE caller: the publish scheduler gate in
+`generation_service`. It samples both families once, admits the session if either
+says yes, and passes the two booleans to the runner. The runner never samples —
+it is TOLD. That is stronger than sharing a helper: with a single computation
+site, the two families cannot disagree because there is nothing to disagree with.
+
+Direct callers of the runner (regen jobs, the on-demand grade route) leave both
+flags at their default and run both families, as before. Sampling is a scheduling
+decision, not a runner one.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+
+
+def stable_group_sampling_fraction(org_id: str, user_id: str, session_id: str) -> float:
+    """Return a deterministic [0, 1) sample value for one session."""
+    key = f"{org_id}\0{user_id}\0{session_id}".encode()
+    digest = hashlib.sha256(key).digest()
+    return int.from_bytes(digest[:8], "big") / 2**64
+
+
+def _samples(rate: float, fraction: float) -> bool:
+    if rate >= 1.0:
+        return True
+    if rate <= 0.0:
+        return False
+    return fraction < rate
+
+
+def samples_agent_success(
+    agent_success_config: Any | None,
+    *,
+    org_id: str,
+    user_id: str,
+    session_id: str,
+) -> bool:
+    """Whether this session is sampled for the session-success judge."""
+    if agent_success_config is None:
+        return False
+    return _samples(
+        float(agent_success_config.sampling_rate),
+        stable_group_sampling_fraction(org_id, user_id, session_id),
+    )
+
+
+def samples_retrieved_learning(
+    agent_success_config: Any | None,
+    *,
+    org_id: str,
+    user_id: str,
+    session_id: str,
+) -> bool:
+    """Whether this session is sampled for the retrieved-learning judge.
+
+    ``retrieved_learning_sampling_rate=None`` inherits ``sampling_rate``, so an
+    org that has not opted in keeps exactly its previous behavior.
+    """
+    if agent_success_config is None:
+        return False
+    rate = getattr(agent_success_config, "retrieved_learning_sampling_rate", None)
+    if rate is None:
+        rate = agent_success_config.sampling_rate
+    return _samples(
+        float(rate),
+        stable_group_sampling_fraction(org_id, user_id, session_id),
+    )