diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index eba9b744..64444237 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/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. diff --git a/reflexio/models/api_schema/domain/governance.py b/reflexio/models/api_schema/domain/governance.py index db1c5d5b..63c37858 100644 --- a/reflexio/models/api_schema/domain/governance.py +++ b/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", ] diff --git a/reflexio/models/config_schema.py b/reflexio/models/config_schema.py index c6189fad..8831dfaf 100644 --- a/reflexio/models/config_schema.py +++ b/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) diff --git a/reflexio/server/extensions.py b/reflexio/server/extensions.py index becb7ce3..00704939 100644 --- a/reflexio/server/extensions.py +++ b/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. diff --git a/reflexio/server/services/agent_success_evaluation/runner.py b/reflexio/server/services/agent_success_evaluation/runner.py index ef736651..9ebbdf0f 100644 --- a/reflexio/server/services/agent_success_evaluation/runner.py +++ b/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, diff --git a/reflexio/server/services/agent_success_evaluation/sampling.py b/reflexio/server/services/agent_success_evaluation/sampling.py new file mode 100644 index 00000000..8defa920 --- /dev/null +++ b/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), + ) diff --git a/reflexio/server/services/generation_service.py b/reflexio/server/services/generation_service.py index 17cca51c..103271be 100644 --- a/reflexio/server/services/generation_service.py +++ b/reflexio/server/services/generation_service.py @@ -1,7 +1,6 @@ from __future__ import annotations import contextvars -import hashlib import logging import os import threading @@ -28,6 +27,10 @@ from reflexio.server.services.agent_success_evaluation.runner import ( run_group_evaluation, ) +from reflexio.server.services.agent_success_evaluation.sampling import ( + samples_agent_success, + samples_retrieved_learning, +) from reflexio.server.services.agent_success_evaluation.scheduler import ( GroupEvaluationScheduler, ) @@ -72,6 +75,20 @@ GENERATION_SERVICE_TIMEOUT_SECONDS = 600 _STALL_WARNING_PREFIX = "Reflexio learning is paused" +# Retrieved-learning statuses that committed NOTHING and have no other retry +# trigger. These are the only ones worth re-running. +# +# "degraded" is deliberately EXCLUDED. It is an APPLIED, fingerprint-fenced +# commit: the rows are persisted and only the chunks whose judge failed carry +# NULL impact. Retrying it re-executes EVERY relevance + impact chunk for the +# session and delete/re-inserts rows that are already committed — and a +# deterministically degrading chunk (an over-length learning, a content-filter +# refusal) degrades again on every attempt, so the whole bill buys nothing. +# It is re-judged fresh (new generation) on the next scheduled run and +# self-heals to "complete" once the transient failure clears. +_RETRIABLE_RETRIEVED_STATUSES = frozenset({"failed", "pending"}) +_MAX_RETRIEVED_LEARNING_RETRIES = 3 + # ── Durable-learning same-user guard (F4) ── # Service prefix for the per-user in-progress lock that serializes concurrent # durable-learning jobs for one user (compute → persist across the fence). @@ -108,15 +125,6 @@ def _retention_cleanup_interval_seconds() -> float: _retention_cleanup_lock = threading.Lock() -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 _org_in_durable_allowlist(org_id: str | None) -> bool: """Whether ``org_id`` may use the durable learning queue. @@ -1148,9 +1156,10 @@ def _schedule_group_evaluation_if_needed( source (str | None): Optional source label. """ session_id = new_request.session_id - if not self._should_sample_group_evaluation( + run_agent_success, run_retrieved_learning = self._sampled_evaluation_families( user_id=user_id, session_id=session_id - ): + ) + if not (run_agent_success or run_retrieved_learning): logger.info( "Skipping group evaluation scheduling for unsampled session=%s user=%s", session_id, @@ -1169,9 +1178,12 @@ def make_callback( _src: str | None, _rc: RequestContext, _llm: LiteLLMClient, + _run_agent_success: bool, + _run_retrieved_learning: bool, + _attempt: int, ) -> Callable[[], None]: def callback() -> None: - run_group_evaluation( + outcome = run_group_evaluation( org_id=_org_id, user_id=_user_id, session_id=_sid, @@ -1179,7 +1191,47 @@ def callback() -> None: source=_src, request_context=_rc, llm_client=_llm, + run_agent_success=_run_agent_success, + run_retrieved_learning=_run_retrieved_learning, ) + # Retry sweep: a retrieved-learning run that ends non-terminal + # (degraded / failed / pending) is NEVER retried on its own — + # nothing re-triggers it unless the session receives more + # traffic, so a session that degrades once would stay invisible + # to downstream consumers forever. Re-arm the session on the + # scheduler, bounded, and only for the agent-success family's + # sibling: regen jobs and the on-demand grade route drive their + # own retries. + if ( + _run_retrieved_learning + and outcome.retrieved_learning_status + in _RETRIABLE_RETRIEVED_STATUSES + and _attempt < _MAX_RETRIEVED_LEARNING_RETRIES + ): + logger.info( + "event=retrieved_learning_retry_scheduled session=%s" + " status=%s attempt=%d", + _sid, + outcome.retrieved_learning_status, + _attempt + 1, + ) + scheduler.schedule( + key, + make_callback( + _org_id, + _user_id, + _sid, + _av, + _src, + _rc, + _llm, + # The success judge already ran (or was skipped); do + # not pay for it again on a retrieved-only retry. + False, + True, + _attempt + 1, + ), + ) return callback @@ -1193,24 +1245,35 @@ def callback() -> None: source, self.request_context, self.client, + run_agent_success, + run_retrieved_learning, + 0, ), ) - def _should_sample_group_evaluation(self, *, user_id: str, session_id: str) -> bool: - config = self.configurator.get_config() - agent_success_config = getattr(config, "agent_success_config", None) - if agent_success_config is None: - return False + def _sampled_evaluation_families( + self, *, user_id: str, session_id: str + ) -> tuple[bool, bool]: + """Which judge families this session is sampled for. - sampling_rate = agent_success_config.sampling_rate - if sampling_rate >= 1.0: - return True - if sampling_rate <= 0.0: - return False + The two families sample independently (see + ``agent_success_evaluation.sampling``). We schedule when EITHER admits + the session and pass both flags to the runner, so a session sampled only + for retrieved-learning never pays the session-success judge. + Returns: + tuple[bool, bool]: ``(run_agent_success, run_retrieved_learning)``. + """ + config = self.configurator.get_config() + agent_success_config = getattr(config, "agent_success_config", None) + scope = { + "org_id": self.org_id, + "user_id": user_id, + "session_id": session_id, + } return ( - _stable_group_sampling_fraction(self.org_id, user_id, session_id) - < sampling_rate + samples_agent_success(agent_success_config, **scope), + samples_retrieved_learning(agent_success_config, **scope), ) def _maybe_run_reflection( diff --git a/reflexio/server/services/storage/retention.py b/reflexio/server/services/storage/retention.py index 302163b0..48fa6fef 100644 --- a/reflexio/server/services/storage/retention.py +++ b/reflexio/server/services/storage/retention.py @@ -102,6 +102,12 @@ class RetentionTarget: "created_at", ("event_id",), ), + # RETIRED WRITER, LIVE PII. Nothing writes playbook_retrieval_logs any more + # (the retrieval-capture subsystem is gone), but the table still exists and + # still holds user_id/session_id until a later release DROPs it — so it must + # keep being trimmed. Every backend gates on ``_retention_table_exists`` + # before counting/deleting, so this target no-ops cleanly once the table is + # dropped and an old task never issues a raw DELETE against a missing table. RetentionTarget( "playbook_retrieval_logs", "playbook_retrieval_logs", @@ -149,6 +155,9 @@ class CascadeRef: "playbook_optimization_candidates": ( CascadeRef("playbook_optimization_evaluations", "candidate_id"), ), + # Retired writer, live PII — see the RETENTION_TARGETS note above. The + # cascade only runs when the parent target yielded keys, which requires the + # parent table to exist, so a dropped pair no-ops without a raw DELETE. "playbook_retrieval_logs": ( CascadeRef("playbook_retrieval_log_items", "retrieval_log_id"), ), diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py b/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py index 5cbee228..1cd5d601 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py @@ -414,7 +414,10 @@ def load_bounded_retrieved_learning_snapshot( @SQLiteStorageBase.handle_exceptions def get_matching_retrieved_learning_terminal_state( - self, user_id: str, session_id: str, session_fingerprint: str + self, + user_id: str, + session_id: str, + session_fingerprint: str, ) -> dict[str, Any] | None: state_key = build_retrieved_learning_state_key(user_id, session_id) with self._lock: @@ -593,6 +596,26 @@ def get_retrieved_learning_evaluation_results( rows = self._fetchall(sql, params) return [_row_to_retrieved_learning_result(r) for r in rows] + @SQLiteStorageBase.handle_exceptions + def get_retrieved_learning_evaluation_results_in_window( + self, + from_ts: int, + to_ts: int, + *, + agent_version: str | None = None, + ) -> list[RetrievedLearningEvaluationResult]: + sql = ( + "SELECT * FROM retrieved_learning_evaluation " + "WHERE created_at >= ? AND created_at <= ?" + ) + params: list[Any] = [from_ts, to_ts] + if agent_version is not None: + sql += " AND agent_version = ?" + params.append(agent_version) + sql += " ORDER BY created_at ASC, result_id ASC" + rows = self._fetchall(sql, params) + return [_row_to_retrieved_learning_result(row) for row in rows] + def _current_epoch(self) -> int: return int(datetime.now(UTC).timestamp()) diff --git a/reflexio/server/services/storage/storage_base/__init__.py b/reflexio/server/services/storage/storage_base/__init__.py index 53a260db..59d04bd5 100644 --- a/reflexio/server/services/storage/storage_base/__init__.py +++ b/reflexio/server/services/storage/storage_base/__init__.py @@ -28,7 +28,6 @@ from ._lineage import EntityType, LineageEventMixin from ._operations import OperationMixin from ._requests import RequestMixin -from ._retrieval_log import RetrievalLogMixin from ._shadow_verdicts import ShadowVerdictsMixin from ._share_links import ShareLinkMixin from ._stall_state import StallStateMixin @@ -62,7 +61,6 @@ class BaseStorage( PlaybookSourceLinkageMixin, OptimizationJobStoreMixin, AgentEvaluationResultStoreMixin, - RetrievalLogMixin, AuditEventStoreMixin, PurgeOperationStoreMixin, SubjectBarrierMixin, @@ -303,7 +301,6 @@ def learning_jobs_columns(self) -> list[str]: "InteractionStoreMixin", "ProfileSearchMixin", "ProfileStoreMixin", - "RetrievalLogMixin", "PriorAnswerMatch", "RunToolDependencyKind", "RunToolDependencyRecord", diff --git a/reflexio/server/services/storage/storage_base/_retrieval_log.py b/reflexio/server/services/storage/storage_base/_retrieval_log.py deleted file mode 100644 index 5773e962..00000000 --- a/reflexio/server/services/storage/storage_base/_retrieval_log.py +++ /dev/null @@ -1,53 +0,0 @@ -from reflexio.models.api_schema.domain.entities import PlaybookRetrievalLog - - -class RetrievalLogMixin: - """Mixin for playbook retrieval log storage methods. - - These methods are optional retrieval-capture storage hooks. They are - intentionally concrete (not ``@abstractmethod``) so concrete storage classes - remain instantiable; each raises ``NotImplementedError`` until a backend - provides a real implementation. Stored item rows may use either the legacy - ``agent_playbook_id`` field or the explicit ``target_kind`` / ``target_id`` - pair; readers should preserve backward compatibility for both. - """ - - def save_playbook_retrieval_log(self, log: PlaybookRetrievalLog) -> int: - """Persist a retrieval log entry and return its assigned id. - - Args: - log (PlaybookRetrievalLog): The log entry to save. ``retrieval_log_id`` - may be 0; the storage layer assigns a real id on insert. - - Returns: - int: The assigned ``retrieval_log_id``. - """ - raise NotImplementedError - - def get_playbook_retrieval_logs( - self, - *, - session_id: str | None = None, - request_id: str | None = None, - interaction_id: int | None = None, - user_id: str | None = None, - start_time: int | None = None, - end_time: int | None = None, - ) -> list[PlaybookRetrievalLog]: - """Retrieve playbook retrieval log entries, optionally filtered. - - Args: - session_id (str | None): Filter to logs for this session. If None, - no session filter is applied. - request_id (str | None): Filter to logs for this request. If None, - no request filter is applied. - interaction_id (int | None): Filter to logs for this interaction. - user_id (str | None): Filter to logs for this user. - start_time (int | None): Inclusive lower bound on ``created_at``. - end_time (int | None): Inclusive upper bound on ``created_at``. - - Returns: - list[PlaybookRetrievalLog]: Matching log entries, ordered by - ``created_at`` ascending. - """ - raise NotImplementedError diff --git a/reflexio/server/services/storage/storage_base/playbook/_eval_results.py b/reflexio/server/services/storage/storage_base/playbook/_eval_results.py index 7d568c03..cf608222 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_eval_results.py +++ b/reflexio/server/services/storage/storage_base/playbook/_eval_results.py @@ -189,14 +189,17 @@ def load_bounded_retrieved_learning_snapshot( @abstractmethod def get_matching_retrieved_learning_terminal_state( - self, user_id: str, session_id: str, session_fingerprint: str + self, + user_id: str, + session_id: str, + session_fingerprint: str, ) -> dict[str, Any] | None: """Return the session's terminal evaluation state if still fresh. Implementations must use a writer transaction and recompute the live - session fingerprint in that transaction. A match requires terminal - status and equality among the supplied, persisted, and live - fingerprints. + session fingerprint in that transaction. A match requires a status in + ``TERMINAL_RETRIEVED_STATUSES`` and equality among the supplied, + persisted, and live fingerprints. Args: user_id (str): Session owner. @@ -298,3 +301,29 @@ def get_retrieved_learning_evaluation_results( list[RetrievedLearningEvaluationResult]: Matching rows. """ raise NotImplementedError + + @abstractmethod + def get_retrieved_learning_evaluation_results_in_window( + self, + from_ts: int, + to_ts: int, + *, + agent_version: str | None = None, + ) -> list[RetrievedLearningEvaluationResult]: + """Read every verdict whose ``created_at`` falls in a closed window. + + Unlike :meth:`get_retrieved_learning_evaluation_results`, this read is + exhaustive: it has no ``limit`` and must not truncate. Consumers treat + the result as the complete set of verdicts for the window, so a silently + short read would drop real signal rather than merely paginate it. + + Args: + from_ts (int): Inclusive lower bound on ``created_at``. + to_ts (int): Inclusive upper bound on ``created_at``. + agent_version (str, optional): Restrict to one agent version. + + Returns: + list[RetrievedLearningEvaluationResult]: Every matching row, + ordered by ``created_at ASC, result_id ASC``. + """ + raise NotImplementedError diff --git a/reflexio/server/services/storage/storage_base/retrieved_learning_state.py b/reflexio/server/services/storage/storage_base/retrieved_learning_state.py index d08fbac5..976aa95d 100644 --- a/reflexio/server/services/storage/storage_base/retrieved_learning_state.py +++ b/reflexio/server/services/storage/storage_base/retrieved_learning_state.py @@ -27,8 +27,12 @@ RETRIEVED_LEARNING_EVALUATION_VERSION = 2 # Statuses persisted in _operation_state. "complete" and "not_applicable" are -# terminal (the fast path may short-circuit on them); "degraded" and "failed" -# are retried by the next scheduled or forced run. +# terminal (the fast path may short-circuit on them, and consumers such as the +# offline tuner read TERMINAL only). "degraded" is an APPLIED commit with a +# partial row set; it is not terminal and is re-judged fresh on the next +# scheduled run, self-healing to "complete" once the transient failure clears. +# "failed" and "pending" committed nothing and are retried by the next scheduled +# or forced run. type RetrievedLearningPersistedStatus = Literal[ "pending", "in_progress", "complete", "degraded", "failed", "not_applicable" ] diff --git a/reflexio/server/services/unified_search_service.py b/reflexio/server/services/unified_search_service.py index 54496281..ac15483f 100644 --- a/reflexio/server/services/unified_search_service.py +++ b/reflexio/server/services/unified_search_service.py @@ -110,20 +110,6 @@ _embedding_cache: OrderedDict[tuple[str, int, str, str], tuple[float, list[float]]] = ( OrderedDict() ) -RetrievalCaptureHook = Callable[ - [UnifiedSearchRequest, UnifiedSearchResponse, BaseStorage, str], None -] -_retrieval_capture_hook: RetrievalCaptureHook | None = None - - -def configure_retrieval_capture_hook(hook: RetrievalCaptureHook | None) -> None: - """Register an optional final-response retrieval capture hook. - - Deployments that capture retrieval logs install the hook; OSS leaves it - unset so unified search behavior is unchanged by default. - """ - global _retrieval_capture_hook - _retrieval_capture_hook = hook def run_unified_search( @@ -310,34 +296,9 @@ def run_unified_search( ) if session_id: session_seen_cache.record(org_id, session_id, _served_entity_keys(response)) - _maybe_capture_final_response( - request=request, - response=response, - storage=storage, - org_id=org_id, - ) return response -def _maybe_capture_final_response( - *, - request: UnifiedSearchRequest, - response: UnifiedSearchResponse, - storage: BaseStorage, - org_id: str, -) -> None: - hook = _retrieval_capture_hook - if hook is None: - return - try: - hook(request, response, storage, org_id) - except Exception: - logger.warning( - "Unified search retrieval capture hook failed", - exc_info=True, - ) - - def _run_phase_a( query: str, storage: BaseStorage, diff --git a/tests/server/services/agent_success_evaluation/test_sampling.py b/tests/server/services/agent_success_evaluation/test_sampling.py new file mode 100644 index 00000000..4d9f9fc5 --- /dev/null +++ b/tests/server/services/agent_success_evaluation/test_sampling.py @@ -0,0 +1,100 @@ +"""Per-family sampling: agent-success and retrieved-learning sample independently.""" + +from __future__ import annotations + +from reflexio.models.config_schema import AgentSuccessConfig +from reflexio.server.services.agent_success_evaluation.sampling import ( + samples_agent_success, + samples_retrieved_learning, + stable_group_sampling_fraction, +) + +SCOPE = {"org_id": "org-1", "user_id": "user-1", "session_id": "sess-1"} + + +def _config(sampling_rate: float, retrieved: float | None = None) -> AgentSuccessConfig: + return AgentSuccessConfig( + success_definition_prompt="Did the agent succeed?", + sampling_rate=sampling_rate, + retrieved_learning_sampling_rate=retrieved, + ) + + +def test_the_field_default_is_none_when_never_supplied() -> None: + """Pin the DEFAULT, by never passing the field. + + The earlier version of this test called a helper that passed + `retrieved_learning_sampling_rate=None` explicitly, so it asserted its own + argument rather than the field default. It therefore stayed green while the + default silently drifted to 0.1 — which would have doubled the eval LLM bill + for every org that never opted in. Construct the config WITHOUT the field or + this test proves nothing. + """ + config = AgentSuccessConfig(success_definition_prompt="Did the agent succeed?") + + assert config.retrieved_learning_sampling_rate is None + + +def test_an_org_that_never_opted_in_keeps_its_previous_behavior() -> None: + """No opt-in => retrieved-learning samples exactly where agent-success does.""" + for rate in (0.0, 0.05, 1.0): + config = AgentSuccessConfig( + success_definition_prompt="Did the agent succeed?", + sampling_rate=rate, + ) + assert samples_retrieved_learning(config, **SCOPE) == samples_agent_success( + config, **SCOPE + ) + + +def test_retrieved_learning_can_sample_denser_than_agent_success() -> None: + """The whole point: dense tuner coverage without paying the success judge.""" + config = _config(0.0, retrieved=1.0) + + assert samples_agent_success(config, **SCOPE) is False + assert samples_retrieved_learning(config, **SCOPE) is True + + +def test_retrieved_learning_can_sample_sparser_than_agent_success() -> None: + config = _config(1.0, retrieved=0.0) + + assert samples_agent_success(config, **SCOPE) is True + assert samples_retrieved_learning(config, **SCOPE) is False + + +def test_zero_rate_never_samples_and_full_rate_always_samples() -> None: + assert samples_agent_success(_config(0.0), **SCOPE) is False + assert samples_agent_success(_config(1.0), **SCOPE) is True + assert samples_retrieved_learning(_config(1.0, retrieved=0.0), **SCOPE) is False + assert samples_retrieved_learning(_config(0.0, retrieved=1.0), **SCOPE) is True + + +def test_missing_agent_success_config_samples_neither() -> None: + assert samples_agent_success(None, **SCOPE) is False + assert samples_retrieved_learning(None, **SCOPE) is False + + +def test_sampling_is_stable_and_scoped_to_the_session() -> None: + """Both families must agree on WHICH sessions are in a given fraction.""" + first = stable_group_sampling_fraction("org-1", "user-1", "sess-1") + assert first == stable_group_sampling_fraction("org-1", "user-1", "sess-1") + assert 0.0 <= first < 1.0 + assert first != stable_group_sampling_fraction("org-1", "user-1", "sess-2") + assert first != stable_group_sampling_fraction("org-2", "user-1", "sess-1") + + +def test_a_denser_retrieved_rate_is_a_superset_of_the_agent_success_sample() -> None: + """Nesting matters: raising only the retrieved rate must not DROP sessions + that the success judge would have sampled, because both families key off the + same stable fraction.""" + config = _config(0.1, retrieved=0.9) + sampled_success = 0 + both = 0 + for n in range(500): + scope = {"org_id": "org-1", "user_id": "u", "session_id": f"s-{n}"} + if samples_agent_success(config, **scope): + sampled_success += 1 + if samples_retrieved_learning(config, **scope): + both += 1 + assert sampled_success > 0 + assert both == sampled_success diff --git a/tests/server/services/storage/test_storage_contract_retrieval_logs.py b/tests/server/services/storage/test_storage_contract_retrieval_logs.py deleted file mode 100644 index e0872da7..00000000 --- a/tests/server/services/storage/test_storage_contract_retrieval_logs.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Contract tests for playbook retrieval-log scaffolding in OSS storage.""" - -import pytest - -from reflexio.models.api_schema.domain import ( - PlaybookRetrievalLog, - PlaybookRetrievalLogItem, -) -from reflexio.server.services.storage.storage_base import BaseStorage - -pytestmark = pytest.mark.integration - - -def test_retrieval_log_models_expose_header_and_item_defaults() -> None: - item = PlaybookRetrievalLogItem(ordinal=0, agent_playbook_id=101) - log = PlaybookRetrievalLog(request_id="req-1", session_id="sess-1", user_id="u1") - - assert item.retrieval_log_item_id == 0 - assert item.retrieval_log_id == 0 - assert item.source_user_playbook_ids == [] - assert item.source_interaction_ids_by_user_playbook_id == {} - - assert log.retrieval_log_id == 0 - assert log.interaction_id is None - assert log.query is None - assert log.agent_version is None - assert log.shown_items == [] - assert log.created_at == 0 - - -def test_sqlite_retrieval_log_methods_remain_unimplemented( - storage: BaseStorage, -) -> None: - log = PlaybookRetrievalLog( - request_id="req-1", - session_id="sess-1", - user_id="u1", - shown_items=[ - PlaybookRetrievalLogItem( - ordinal=0, - agent_playbook_id=101, - source_user_playbook_ids=[11], - source_interaction_ids_by_user_playbook_id={"11": [201]}, - ) - ], - ) - - with pytest.raises(NotImplementedError): - storage.save_playbook_retrieval_log(log) - - with pytest.raises(NotImplementedError): - storage.get_playbook_retrieval_logs(user_id="u1") diff --git a/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py b/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py index 76154644..c9d242ea 100644 --- a/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py +++ b/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py @@ -688,3 +688,124 @@ def begin() -> None: for t in threads: t.join() assert sorted(generations) == [1, 2] + + +def _seed_many_profiles(storage, count: int) -> list[RetrievedLearning]: + """Seed ``count`` eligible profiles; return refs attaching all of them.""" + storage.add_user_profile( + USER, + [ + UserProfile( + profile_id=f"win-prof-{n}", + user_id=USER, + content=f"fact {n}", + last_modified_timestamp=1, + generated_from_request_id="r1", + ) + for n in range(count) + ], + ) + return [ + RetrievedLearning(kind="profile", learning_id=f"win-prof-{n}") + for n in range(count) + ] + + +def _commit(storage, results: list[RetrievedLearningEvaluationResult]) -> None: + snapshot = storage.load_bounded_retrieved_learning_snapshot(USER, SESSION) + generation = storage.begin_retrieved_learning_evaluation_run(USER, SESSION) + storage.replace_retrieved_learning_evaluation_results( + USER, + SESSION, + generation, + session_fingerprint(snapshot), + "complete", + {}, + results, + ) + + +def test_window_read_is_empty_when_no_verdicts_exist(storage) -> None: + _seed_session(storage) + assert ( + storage.get_retrieved_learning_evaluation_results_in_window(0, 2_000_000_000) + == [] + ) + + +@pytest.mark.parametrize("count", [100, 101, 250]) +def test_window_read_is_exhaustive_past_the_paged_readers_cap(storage, count) -> None: + """The paged reader caps at limit=100 with no cursor; this one must not. + + The offline tuner treats the window read as the COMPLETE set of verdicts, so + a silently short read drops real playbook signal rather than merely + paginating it. + """ + refs = _seed_many_profiles(storage, count) + _seed_session(storage, refs=refs) + _commit( + storage, [_result(storage, "profile", f"win-prof-{n}") for n in range(count)] + ) + + assert ( + len(storage.get_retrieved_learning_evaluation_results(session_id=SESSION)) + == 100 + ) + + rows = storage.get_retrieved_learning_evaluation_results_in_window(0, 2_000_000_000) + assert len(rows) == count + assert {row.learning_id for row in rows} == {f"win-prof-{n}" for n in range(count)} + + +def test_window_read_bounds_are_inclusive_and_exclude_outside(storage) -> None: + refs = _seed_many_profiles(storage, 2) + _seed_session(storage, refs=refs) + _commit( + storage, + [ + _result(storage, "profile", "win-prof-0"), + _result(storage, "profile", "win-prof-1"), + ], + ) + + (sample,) = storage.get_retrieved_learning_evaluation_results_in_window( + 0, 2_000_000_000 + )[:1] + at = sample.created_at + + assert len(storage.get_retrieved_learning_evaluation_results_in_window(at, at)) == 2 + assert ( + storage.get_retrieved_learning_evaluation_results_in_window(at + 1, at + 100) + == [] + ) + assert ( + storage.get_retrieved_learning_evaluation_results_in_window(at - 100, at - 1) + == [] + ) + + +def test_window_read_orders_ascending_and_filters_agent_version(storage) -> None: + refs = _seed_many_profiles(storage, 3) + _seed_session(storage, refs=refs) + _commit(storage, [_result(storage, "profile", f"win-prof-{n}") for n in range(3)]) + + rows = storage.get_retrieved_learning_evaluation_results_in_window(0, 2_000_000_000) + keys = [(row.created_at, row.result_id) for row in rows] + assert keys == sorted(keys), ( + "window read must be ordered created_at ASC, result_id ASC" + ) + + assert ( + len( + storage.get_retrieved_learning_evaluation_results_in_window( + 0, 2_000_000_000, agent_version="v1" + ) + ) + == 3 + ) + assert ( + storage.get_retrieved_learning_evaluation_results_in_window( + 0, 2_000_000_000, agent_version="does-not-exist" + ) + == [] + ) diff --git a/tests/server/services/test_generation_service_scheduling.py b/tests/server/services/test_generation_service_scheduling.py index d01d0473..135c9d93 100644 --- a/tests/server/services/test_generation_service_scheduling.py +++ b/tests/server/services/test_generation_service_scheduling.py @@ -289,3 +289,216 @@ def test_learning_stall_path_calls_post_publish_helper( storage.add_request.assert_called_once() storage.add_user_interactions_bulk.assert_called_once() post_publish.assert_called_once() + + +def _set_rates( + service: GenerationService, *, success: float, retrieved: float | None +) -> None: + cast(Any, service.configurator.get_config).return_value = Config( + storage_config=StorageConfigSQLite(), + agent_success_config=AgentSuccessConfig( + success_definition_prompt="Evaluate whether the agent succeeded.", + sampling_rate=success, + retrieved_learning_sampling_rate=retrieved, + ), + ) + + +def _schedule_and_capture( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> MagicMock: + """Schedule, then run the queued callback with run_group_evaluation stubbed.""" + scheduler = MagicMock() + monkeypatch.setattr( + "reflexio.server.services.generation_service.GroupEvaluationScheduler.get_instance", + lambda: scheduler, + ) + runner = MagicMock(name="run_group_evaluation") + monkeypatch.setattr( + "reflexio.server.services.generation_service.run_group_evaluation", runner + ) + + service._schedule_group_evaluation_if_needed( + new_request=MagicMock(session_id="sess_split"), + user_id="user_test", + agent_version="v_test", + source=None, + ) + if scheduler.schedule.call_count: + scheduler.schedule.call_args[0][1]() # invoke the queued callback + return runner + + +def test_retrieved_only_sampling_schedules_but_skips_the_success_judge( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> None: + """The core coverage guarantee: dense tuner signal without the success bill. + + A session sampled ONLY for retrieved-learning must still be scheduled, and + the runner must be told to skip the session-success judge. + """ + _set_rates(service, success=0.0, retrieved=1.0) + + runner = _schedule_and_capture(service, monkeypatch) + + runner.assert_called_once() + kwargs = runner.call_args.kwargs + assert kwargs["run_agent_success"] is False + assert kwargs["run_retrieved_learning"] is True + + +def test_success_only_sampling_skips_the_retrieved_learning_judge( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> None: + _set_rates(service, success=1.0, retrieved=0.0) + + runner = _schedule_and_capture(service, monkeypatch) + + runner.assert_called_once() + kwargs = runner.call_args.kwargs + assert kwargs["run_agent_success"] is True + assert kwargs["run_retrieved_learning"] is False + + +def test_neither_family_sampled_does_not_schedule_at_all( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> None: + _set_rates(service, success=0.0, retrieved=0.0) + + runner = _schedule_and_capture(service, monkeypatch) + + runner.assert_not_called() + + +def test_unset_retrieved_rate_inherits_success_rate( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> None: + """Orgs that have not opted in keep exactly their previous behavior.""" + _set_rates(service, success=1.0, retrieved=None) + + runner = _schedule_and_capture(service, monkeypatch) + + kwargs = runner.call_args.kwargs + assert kwargs["run_agent_success"] is True + assert kwargs["run_retrieved_learning"] is True + + +def _drain_scheduler(scheduler: MagicMock, *, max_rounds: int = 10) -> int: + """Run queued callbacks until the scheduler stops re-arming. Returns rounds.""" + rounds = 0 + seen = 0 + while scheduler.schedule.call_count > seen and rounds < max_rounds: + seen = scheduler.schedule.call_count + scheduler.schedule.call_args_list[seen - 1][0][1]() + rounds += 1 + return rounds + + +def _run_with_outcome( + service: GenerationService, + monkeypatch: pytest.MonkeyPatch, + retrieved_status: str, +) -> tuple[MagicMock, MagicMock]: + scheduler = MagicMock() + monkeypatch.setattr( + "reflexio.server.services.generation_service.GroupEvaluationScheduler.get_instance", + lambda: scheduler, + ) + runner = MagicMock( + name="run_group_evaluation", + return_value=SimpleNamespace( + agent_success_status="complete", + retrieved_learning_status=retrieved_status, + retrieved_learning_fingerprint=None, + ), + ) + monkeypatch.setattr( + "reflexio.server.services.generation_service.run_group_evaluation", runner + ) + _set_rates(service, success=1.0, retrieved=1.0) + + service._schedule_group_evaluation_if_needed( + new_request=MagicMock(session_id="sess_retry"), + user_id="user_test", + agent_version="v_test", + source=None, + ) + _drain_scheduler(scheduler) + return scheduler, runner + + +@pytest.mark.parametrize("status", ["failed", "pending"]) +def test_a_run_that_committed_nothing_is_retried( + service: GenerationService, monkeypatch: pytest.MonkeyPatch, status: str +) -> None: + """`failed` and `pending` persisted NO rows, so nothing else re-triggers them + unless the session happens to get more traffic. These are the retriable ones.""" + _, runner = _run_with_outcome(service, monkeypatch, status) + + # 1 initial + 3 bounded retries. + assert runner.call_count == 1 + 3 + + retries = runner.call_args_list[1:] + for call in retries: + # A retry must not re-pay the session-success judge. + assert call.kwargs["run_agent_success"] is False + assert call.kwargs["run_retrieved_learning"] is True + + +def test_degraded_is_never_retried( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> None: + """`degraded` is an APPLIED, fingerprint-fenced commit — its rows are already + persisted, and only the chunks whose judge failed carry NULL impact. + + Retrying it would re-execute EVERY relevance + impact chunk for the session + and delete/re-insert rows that are already committed. A deterministically + degrading chunk (an over-length learning, a content-filter refusal) degrades + again on every attempt, so a bounded 3-retry sweep would burn 4x the judge + bill on that slice and buy nothing. It is re-judged fresh on the next + scheduled run and self-heals to "complete" once the transient failure clears. + """ + _, runner = _run_with_outcome(service, monkeypatch, "degraded") + + assert runner.call_count == 1 + + +@pytest.mark.parametrize("status", ["complete", "not_applicable"]) +def test_terminal_retrieved_learning_is_not_retried( + service: GenerationService, monkeypatch: pytest.MonkeyPatch, status: str +) -> None: + _, runner = _run_with_outcome(service, monkeypatch, status) + + assert runner.call_count == 1 + + +def test_retrieved_learning_retry_does_not_fire_when_family_not_sampled( + service: GenerationService, monkeypatch: pytest.MonkeyPatch +) -> None: + """A session not admitted for retrieved-learning must never retry it.""" + scheduler = MagicMock() + monkeypatch.setattr( + "reflexio.server.services.generation_service.GroupEvaluationScheduler.get_instance", + lambda: scheduler, + ) + runner = MagicMock( + return_value=SimpleNamespace( + agent_success_status="complete", + retrieved_learning_status="skipped", + retrieved_learning_fingerprint=None, + ), + ) + monkeypatch.setattr( + "reflexio.server.services.generation_service.run_group_evaluation", runner + ) + _set_rates(service, success=1.0, retrieved=0.0) + + service._schedule_group_evaluation_if_needed( + new_request=MagicMock(session_id="sess_norl"), + user_id="user_test", + agent_version="v_test", + source=None, + ) + _drain_scheduler(scheduler) + + assert runner.call_count == 1 diff --git a/tests/server/services/test_unified_search_service.py b/tests/server/services/test_unified_search_service.py index 62d92a5c..2e704bfd 100644 --- a/tests/server/services/test_unified_search_service.py +++ b/tests/server/services/test_unified_search_service.py @@ -27,7 +27,6 @@ from reflexio.server.services.storage.storage_base import BaseStorage from reflexio.server.services.unified_search_service import ( _search_agent_playbooks_via_storage, - configure_retrieval_capture_hook, run_unified_search, ) @@ -191,86 +190,6 @@ def test_reformulated_query_none_when_unchanged(self, _reformulator_cls): self.assertTrue(result.success) self.assertIsNone(result.reformulated_query) - @patch("reflexio.server.services.unified_search_service.QueryReformulator") - def test_capture_hook_receives_final_shaped_response( - self, _reformulator_cls - ) -> None: - """The capture seam must see post-floor, post-suppression results only.""" - _reformulator_cls.return_value.rewrite.return_value = ReformulationResult( - standalone_query="same query" - ) - storage = _mock_storage() - - pre_floor_agent = _agent_playbook(10, PlaybookStatus.PENDING) - object.__setattr__( - pre_floor_agent, "_source_user_playbook_ids", frozenset({101}) - ) - kept_user_playbook = UserPlaybook( - user_playbook_id=102, - user_id="user-1", - agent_version="claude-code", - request_id="req-102", - playbook_name="pb", - content="kept", - ) - suppressed_user_playbook = UserPlaybook( - user_playbook_id=101, - user_id="user-1", - agent_version="claude-code", - request_id="req-101", - playbook_name="pb", - content="suppressed", - ) - captured = [] - - with ( - patch( - "reflexio.server.services.unified_search_service._run_phase_b", - return_value=( - [], - [pre_floor_agent, _agent_playbook(11, PlaybookStatus.APPROVED)], - [suppressed_user_playbook, kept_user_playbook], - ), - ), - patch( - "reflexio.server.services.unified_search_service._apply_floors", - return_value=( - [], - [pre_floor_agent], - [suppressed_user_playbook, kept_user_playbook], - ), - ), - ): - configure_retrieval_capture_hook( - lambda request, response, _storage, org_id: captured.append( - ( - request.request_id, - org_id, - [pb.agent_playbook_id for pb in response.agent_playbooks], - [pb.user_playbook_id for pb in response.user_playbooks], - ) - ) - ) - try: - result = run_unified_search( - request=UnifiedSearchRequest( - query="same query", - user_id="user-1", - request_id="req-1", - session_id="sess-1", - ), - org_id="test-org", - storage=storage, - llm_client=MagicMock(), - prompt_manager=MagicMock(), - retrieval_floor=RetrievalFloorConfig(enabled=True, pool_size=5), - ) - finally: - configure_retrieval_capture_hook(None) - - self.assertTrue(result.success) - self.assertEqual(captured, [("req-1", "test-org", [10], [102])]) - @patch("reflexio.server.services.unified_search_service.QueryReformulator") def test_recency_uses_pool_and_combined_score_after_phase_b( self, _reformulator_cls diff --git a/tests/server/services/test_unified_search_session_dedup.py b/tests/server/services/test_unified_search_session_dedup.py index 26406047..9c9fba65 100644 --- a/tests/server/services/test_unified_search_session_dedup.py +++ b/tests/server/services/test_unified_search_session_dedup.py @@ -20,7 +20,6 @@ from reflexio.server.services.pre_retrieval import ReformulationResult from reflexio.server.services.retrieval.session_dedup import session_seen_cache from reflexio.server.services.unified_search_service import ( - configure_retrieval_capture_hook, run_unified_search, ) @@ -107,7 +106,6 @@ def _search(storage: MagicMock, session_id: str | None) -> tuple: class TestUnifiedSearchSessionDedup(unittest.TestCase): def setUp(self): session_seen_cache.clear() - configure_retrieval_capture_hook(None) def tearDown(self): session_seen_cache.clear()