diff --git a/reflexio/client/client.py b/reflexio/client/client.py index a775c6389..58399412e 100644 --- a/reflexio/client/client.py +++ b/reflexio/client/client.py @@ -63,6 +63,7 @@ IS_TEST_ENV = os.environ.get("IS_TEST_ENV", "false").strip() == "true" BACKEND_URL = "http://127.0.0.1:8000" if IS_TEST_ENV else "https://www.reflexio.ai/" +REVIEW_USER_PLAYBOOKS_TIMEOUT_SECONDS = 600 from reflexio.models.api_schema.domain.entities import ( UpgradeProfilesRequest, @@ -1630,7 +1631,11 @@ def review_user_playbooks( """Re-review current user playbooks created within a bounded window. The default ``report_only=True`` runs inline and returns one decision per - selected playbook. + selected playbook. Each playbook is reviewed from its full finalized + extraction-run window plus any cited consolidation evidence, independent + of the current extractor window size or source filter. Rows with missing + or invalid persisted provenance are skipped. Only cited interactions are + reviewer evidence; other run-window interactions provide chronology. With ``report_only=False`` the server accepts the run and applies it in the background, so the response carries only ``run_id`` — not @@ -1650,6 +1655,11 @@ def review_user_playbooks( "POST", "/api/review_user_playbooks", json=request.model_dump(mode="json"), + timeout=( + max(self.timeout, REVIEW_USER_PLAYBOOKS_TIMEOUT_SECONDS) + if report_only + else self.timeout + ), ) return ReviewUserPlaybooksResponse(**response) diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index ce2ba4713..75a0cd985 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -1737,9 +1737,10 @@ class ReviewUserPlaybookEdit(BaseModel): class ReviewUserPlaybookResult(BaseModel): """One persisted user playbook's re-review outcome. - ``skip`` means the row could not be reviewed at all (its cited evidence is - no longer reconstructable), not that the reviewer chose to leave it alone — - that is ``accept``. A skipped row is never written to. + ``skip`` means the row could not be reviewed at all because its finalized + generation-window or cited-evidence provenance is absent, missing, or + invalid, not that the reviewer chose to leave it alone — that is ``accept``. + A skipped row is never written to. """ user_playbook_id: int = Field(gt=0) diff --git a/reflexio/server/middleware.py b/reflexio/server/middleware.py index aeafa6968..86331792b 100644 --- a/reflexio/server/middleware.py +++ b/reflexio/server/middleware.py @@ -25,6 +25,7 @@ SYNC_REQUEST_TIMEOUT_SECONDS = ( 600 # Longer timeout for synchronous processing (wait_for_response=true) ) +SYNC_REQUEST_PATHS = frozenset({"/api/review_user_playbooks"}) SUSPICIOUS_USER_AGENTS = ["bot", "crawler", "spider", "scraper", "curl", "wget"] ALLOWED_EMPTY_UA_PATHS = ["/health", "/"] # Paths that allow empty user agents DEFAULT_MAX_BODY_BYTES = 10 * 1024 * 1024 @@ -128,7 +129,10 @@ async def dispatch( # Use longer timeout for synchronous processing requests timeout = REQUEST_TIMEOUT_SECONDS - if request.query_params.get("wait_for_response", "").lower() == "true": + if ( + request.url.path in SYNC_REQUEST_PATHS + or request.query_params.get("wait_for_response", "").lower() == "true" + ): timeout = SYNC_REQUEST_TIMEOUT_SECONDS try: diff --git a/reflexio/server/services/playbook/README.md b/reflexio/server/services/playbook/README.md index 14fe7248c..1be0275f4 100644 --- a/reflexio/server/services/playbook/README.md +++ b/reflexio/server/services/playbook/README.md @@ -66,8 +66,19 @@ do not use this reviewer. ### Persisted Review (`review_service.py`) `POST /api/review_user_playbooks` selects the newest current user playbooks in -an inclusive creation-time window, capped by `top_k`. A row whose cited evidence -can no longer be reconstructed yields a `skip` decision and the run continues. +an inclusive creation-time window, capped by `top_k`. A row whose original +generation window or cited evidence can no longer be reconstructed yields a +`skip` decision and the run continues. + +Manual review reconstructs context from the full interaction window persisted on +the row's finalized playbook-extraction run, plus any extra cited interactions +retained through consolidation. It never substitutes the current extractor +window or silently falls back to the smaller cited-evidence subset. Automatic +post-generation review continues to use the generation call's configured window. +Only the playbook row's cited interaction IDs become candidate evidence units; +the rest of the generation window is ancillary chronology. Evidence spans are +rebuilt from those exact stored interactions instead of requiring every cited +span to remain duplicated in the row's bounded combined `source_span`. Report mode runs inline and returns `accept`, `edit`, `reject`, and `skip` decisions without writes. Apply mode runs in the **background** (one LLM call diff --git a/reflexio/server/services/playbook/components/reviewer.py b/reflexio/server/services/playbook/components/reviewer.py index fcde32a93..4db67ef9f 100644 --- a/reflexio/server/services/playbook/components/reviewer.py +++ b/reflexio/server/services/playbook/components/reviewer.py @@ -41,6 +41,10 @@ } +class PlaybookCandidateEvidenceError(ValueError): + """A persisted candidate's cited evidence cannot be reconstructed.""" + + class CandidateRevision(BaseModel): """Complete replacement fields for one narrowly revised candidate.""" @@ -253,42 +257,21 @@ def _candidate_evidence_units( units_by_candidate: dict[str, list[CandidateEvidenceUnit]] = {} for candidate_index, candidate in enumerate(candidates, start=1): candidate_id = f"C{candidate_index}" - combined_span = candidate.source_span or "" units: list[CandidateEvidenceUnit] = [] for interaction_id in candidate.source_interaction_ids: mapped_units = by_interaction_id.get(interaction_id) if not mapped_units: - raise ValueError( + raise PlaybookCandidateEvidenceError( f"Reviewer cannot map {candidate_id} evidence to a local turn" ) - matching_units = [ - unit - for unit in mapped_units - if unit.source_span - and ( - unit.source_span in combined_span - or combined_span in unit.source_span - ) - ] - if not matching_units: - span_fragments = [ - fragment.strip() - for fragment in combined_span.split("\n\n") - if fragment.strip() - ] - matching_units = [ - unit - for unit in mapped_units - if any( - fragment == unit.source_span or fragment in unit.source_span - for fragment in span_fragments - ) - ] - if not matching_units: - raise ValueError( - f"Reviewer cannot resolve {candidate_id} validated source span" - ) - for mapped_unit in matching_units: + # Exact persisted interaction IDs are the durable evidence + # identity. Rebuild their bounded visible spans from storage; + # do not require every span to survive in the playbook's + # combined ``source_span``, which may be truncated or may have + # been assembled from multiple rows during consolidation. + for mapped_unit in mapped_units: + if not mapped_unit.source_span: + continue if any( unit.turn_ref == mapped_unit.turn_ref and unit.source_span == mapped_unit.source_span @@ -304,7 +287,9 @@ def _candidate_evidence_units( ) ) if not units: - raise ValueError(f"Reviewer received {candidate_id} without evidence") + raise PlaybookCandidateEvidenceError( + f"Reviewer received {candidate_id} without evidence" + ) units_by_candidate[candidate_id] = units return units_by_candidate diff --git a/reflexio/server/services/playbook/review_service.py b/reflexio/server/services/playbook/review_service.py index df5883d7e..c20009d22 100644 --- a/reflexio/server/services/playbook/review_service.py +++ b/reflexio/server/services/playbook/review_service.py @@ -10,6 +10,7 @@ from typing import Literal from reflexio.models.api_schema.domain.entities import ( + Interaction, LineageContext, ReviewUserPlaybookEdit, ReviewUserPlaybookResult, @@ -20,14 +21,12 @@ from reflexio.models.api_schema.internal_schema import RequestInteractionDataModel from reflexio.server.api_endpoints.request_context import RequestContext from reflexio.server.llm.litellm_client import LiteLLMClient -from reflexio.server.services.extractor_interaction_utils import ( - get_extractor_window_params, -) from reflexio.server.services.playbook.components.consolidator import ( PlaybookConsolidator, ) from reflexio.server.services.playbook.components.reviewer import ( CandidateReviewDecision, + PlaybookCandidateEvidenceError, PlaybookCandidateReviewer, ) from reflexio.server.services.playbook.playbook_edit_apply import apply_playbook_edit @@ -55,7 +54,7 @@ class _PlaybookReviewRaceError(RuntimeError): class _PlaybookNotReviewableError(ValueError): - """A selected row's cited evidence can no longer be reconstructed.""" + """A selected row's generation window or cited evidence is unavailable.""" @dataclass @@ -82,39 +81,98 @@ def __init__( def _review_window( self, - playbooks: list[UserPlaybook], - *, - window_size: int, + playbook: UserPlaybook, ) -> list[RequestInteractionDataModel]: - first = playbooks[0] - if not first.user_id: + if not playbook.user_id: + raise _PlaybookNotReviewableError( + f"User playbook {playbook.user_playbook_id} has no owning user_id" + ) + if not is_evidence_validated(playbook): + raise _PlaybookNotReviewableError( + f"User playbook {playbook.user_playbook_id} lacks validated evidence" + ) + + generation_run = self.storage.get_latest_finalized_agent_run_for_request( + org_id=self.request_context.org_id, + extractor_kind="playbook", + user_id=playbook.user_id, + request_id=playbook.request_id, + ) + if generation_run is None or not generation_run.binding.source_interaction_ids: raise _PlaybookNotReviewableError( - f"User playbook {first.user_playbook_id} has no owning user_id" + f"User playbook {playbook.user_playbook_id} has no complete " + "generation-window provenance" + ) + + # The extraction run records the complete prompt window. The playbook row + # stores only the evidence cited by that candidate and may also retain + # cited IDs from older rows consumed by consolidation. Review needs both: + # full original chronology plus every persisted evidence unit. + source_ids = list( + dict.fromkeys( + [ + *generation_run.binding.source_interaction_ids, + *playbook.source_interaction_ids, + ] ) - source_filter = [first.source] if first.source else None - window, _ = self.storage.get_last_k_interactions_grouped( - user_id=first.user_id, - k=window_size, - sources=source_filter, - end_time=max(playbook.created_at for playbook in playbooks), ) - window_interaction_ids = { - interaction.interaction_id - for request_model in window - for interaction in request_model.interactions + interactions_by_id = { + interaction.interaction_id: interaction + for interaction in self.storage.get_interactions_by_ids(source_ids) } - for playbook in playbooks: - if not is_evidence_validated(playbook): + missing_interaction_ids = [ + interaction_id + for interaction_id in source_ids + if interaction_id not in interactions_by_id + ] + if missing_interaction_ids: + raise _PlaybookNotReviewableError( + f"User playbook {playbook.user_playbook_id} is missing persisted " + f"generation-window interactions: {missing_interaction_ids}" + ) + + interactions_by_request: dict[str, list[Interaction]] = {} + for interaction in interactions_by_id.values(): + if interaction.user_id != playbook.user_id: + raise _PlaybookNotReviewableError( + f"User playbook {playbook.user_playbook_id} has source interaction " + f"{interaction.interaction_id} owned by another user" + ) + interactions_by_request.setdefault(interaction.request_id, []).append( + interaction + ) + + review_window: list[RequestInteractionDataModel] = [] + for request_id, interactions in interactions_by_request.items(): + request = self.storage.get_request(request_id) + if request is None: raise _PlaybookNotReviewableError( - f"User playbook {playbook.user_playbook_id} lacks validated evidence" + f"User playbook {playbook.user_playbook_id} is missing persisted " + f"source request {request_id}" ) - missing = set(playbook.source_interaction_ids) - window_interaction_ids - if missing: + if request.user_id != playbook.user_id: raise _PlaybookNotReviewableError( - f"User playbook {playbook.user_playbook_id} review window is " - f"missing source interactions: {sorted(missing)}" + f"User playbook {playbook.user_playbook_id} has source request " + f"{request_id} owned by another user" + ) + review_window.append( + RequestInteractionDataModel( + session_id=request.session_id, + request=request, + interactions=sorted( + interactions, + key=lambda item: (item.created_at, item.interaction_id), + ), ) - return window + ) + + review_window.sort( + key=lambda group: ( + group.interactions[0].created_at, + group.interactions[0].interaction_id, + ) + ) + return review_window def _existing_playbooks( self, @@ -187,11 +245,6 @@ def _review_each( playbook_config = root_config.user_playbook_extractor_config if playbook_config is None: raise ValueError("User playbook extraction is not configured") - window_size, _ = get_extractor_window_params( - playbook_config, - root_config.window_size, - root_config.stride_size, - ) reviewer = PlaybookCandidateReviewer( request_context=self.request_context, llm_client=self.client, @@ -214,7 +267,7 @@ def _review_each( for playbook in selected: candidates = [playbook] try: - window = self._review_window(candidates, window_size=window_size) + window = self._review_window(playbook) except _PlaybookNotReviewableError as exc: # Selection is "newest current rows in a window", which cannot # know whether a row is still reviewable. One unreviewable row @@ -226,20 +279,30 @@ def _review_each( ) yield self._skipped_result(playbook, exc) continue - outcome = reviewer.decide( - candidates=candidates, - request_interaction_data_models=window, - existing_playbooks=self._existing_playbooks( - consolidator, - candidates, - simulated_archived_ids, - ), - agent_context=self.request_context.configurator.get_agent_context(), - playbook_definition=( - playbook_config.extraction_definition_prompt or "" - ).strip(), - tool_context=tool_context, - ) + try: + outcome = reviewer.decide( + candidates=candidates, + request_interaction_data_models=window, + existing_playbooks=self._existing_playbooks( + consolidator, + candidates, + simulated_archived_ids, + ), + agent_context=self.request_context.configurator.get_agent_context(), + playbook_definition=( + playbook_config.extraction_definition_prompt or "" + ).strip(), + tool_context=tool_context, + ) + except PlaybookCandidateEvidenceError as exc: + unavailable = _PlaybookNotReviewableError(str(exc)) + logger.info( + "event=playbook_review_skipped user_playbook_id=%d reason=%s", + playbook.user_playbook_id, + unavailable, + ) + yield self._skipped_result(playbook, unavailable) + continue survivors = reviewer.apply_decisions( candidates=candidates, outcome=outcome, @@ -261,7 +324,6 @@ def _review_each( def _successor( reviewed: _ReviewedPlaybook, *, - run_id: str, created_at: int, ) -> UserPlaybook: if reviewed.survivor is None: @@ -269,9 +331,11 @@ def _successor( return reviewed.survivor.model_copy( update={ "user_playbook_id": 0, - # Not a real ``requests`` row: an operation-run correlation id, - # matching the convention ``apply_playbook_edit`` documents. - "request_id": run_id, + # Preserve the extraction request that owns the full historical + # window. The review run remains attributable through the revise + # lineage event below; replacing this with ``run_id`` would make + # the successor impossible to review again from full provenance. + "request_id": reviewed.original.request_id, "created_at": created_at, # CURRENT, not PENDING: retrieval reads user playbooks with # ``status_filter=[None]``, so a PENDING replacement would take @@ -287,9 +351,7 @@ def _successor( ) def _apply_edit(self, reviewed: _ReviewedPlaybook, *, run_id: str) -> None: - successor = self._successor( - reviewed, run_id=run_id, created_at=int(time.time()) - ) + successor = self._successor(reviewed, created_at=int(time.time())) # All fallible model/embedding work happens before the transaction opens # (``apply_playbook_edit`` is called with ``skip_embedding=True``), so a # later playbook never rolls this decision back. diff --git a/reflexio/server/services/storage/sqlite_storage/agent_run/_agent_run_store.py b/reflexio/server/services/storage/sqlite_storage/agent_run/_agent_run_store.py index 9065a9a50..3858b29b3 100644 --- a/reflexio/server/services/storage/sqlite_storage/agent_run/_agent_run_store.py +++ b/reflexio/server/services/storage/sqlite_storage/agent_run/_agent_run_store.py @@ -1,7 +1,7 @@ """SQLite AgentRunStore methods (the AgentRunStore bucket). -Extracted verbatim from ``_agent_run.py`` (the AgentRunStore bucket): the seven -agent-run lifecycle methods plus the three ``_..._unlocked`` run-cascade +Extracted from ``_agent_run.py`` (the AgentRunStore bucket): the agent-run +lifecycle and provenance lookup methods plus the three ``_..._unlocked`` run-cascade privates. The privates are consumed cross-bucket by the pending-tool-call methods (``cancel``/``expire``/``update_resolved``/``mark_not_applicable``) in ``SQLitePendingToolCallStoreMixin``, which reach them via MRO co-composition — @@ -213,6 +213,38 @@ def get_agent_run(self, run_id: str) -> AgentRunRecord | None: row = self._fetchone("SELECT * FROM _agent_runs WHERE id = ?", (run_id,)) return _row_to_agent_run(row) if row else None + @SQLiteStorageBase.handle_exceptions + def get_latest_finalized_agent_run_for_request( + self, + *, + org_id: str, + extractor_kind: str, + user_id: str | None, + request_id: str, + ) -> AgentRunRecord | None: + row = self._fetchone( + """ + SELECT * + FROM _agent_runs + WHERE org_id = ? + AND extractor_kind = ? + AND user_id IS ? + AND request_id = ? + AND status IN (?, ?) + ORDER BY created_at DESC, updated_at DESC, id DESC + LIMIT 1 + """, + ( + org_id, + extractor_kind, + user_id, + request_id, + AgentRunStatus.FINALIZED.value, + AgentRunStatus.FINALIZED_PENDING_TOOL.value, + ), + ) + return _row_to_agent_run(row) if row else None + @SQLiteStorageBase.handle_exceptions def update_agent_run_status( self, diff --git a/reflexio/server/services/storage/storage_base/agent_run/_agent_run_store.py b/reflexio/server/services/storage/storage_base/agent_run/_agent_run_store.py index e88a64ab6..b8bba0371 100644 --- a/reflexio/server/services/storage/storage_base/agent_run/_agent_run_store.py +++ b/reflexio/server/services/storage/storage_base/agent_run/_agent_run_store.py @@ -1,8 +1,8 @@ """Backend-neutral AgentRunStore contract (the AgentRunStore bucket). -Extracted verbatim from ``_agent_run.py`` (the AgentRunStore bucket): the seven -agent-run lifecycle methods. The residual ``AgentRunMixin`` composite stays -composed alongside this class and holds the remaining pending-tool-call / +Extracted from ``_agent_run.py`` (the AgentRunStore bucket): the agent-run +lifecycle and provenance lookup methods. The residual ``AgentRunMixin`` +composite stays composed alongside this class and holds the remaining pending-tool-call / run-tool-dependency stubs plus the shared ``build_scope_hash`` / ``human_feedback_scope`` / ``build_pending_tool_call_dedup_key`` staticmethods. """ @@ -24,6 +24,17 @@ def create_agent_run(self, record: AgentRunRecord) -> AgentRunRecord: def get_agent_run(self, run_id: str) -> AgentRunRecord | None: raise NotImplementedError(f"{type(self).__name__} does not support agent runs") + def get_latest_finalized_agent_run_for_request( + self, + *, + org_id: str, + extractor_kind: str, + user_id: str | None, + request_id: str, + ) -> AgentRunRecord | None: + """Return the newest finalized extraction run for one logical request.""" + raise NotImplementedError(f"{type(self).__name__} does not support agent runs") + def update_agent_run_status( self, run_id: str, diff --git a/tests/client/test_playbook_review_client.py b/tests/client/test_playbook_review_client.py index c35e2fa63..007cdfa55 100644 --- a/tests/client/test_playbook_review_client.py +++ b/tests/client/test_playbook_review_client.py @@ -39,4 +39,32 @@ def fake_make_request(method: str, path: str, **kwargs: Any) -> dict[str, Any]: "top_k": 25, "report_only": False, }, + "timeout": 300, } + + +def test_report_only_review_uses_extended_client_timeout(monkeypatch) -> None: + client = ReflexioClient( + api_key="test-key", + url_endpoint="http://localhost:8000", + timeout=450, + ) + captured: dict[str, Any] = {} + + def fake_make_request(method: str, path: str, **kwargs: Any) -> dict[str, Any]: + captured.update(method=method, path=path, **kwargs) + return { + "success": True, + "report_only": True, + "selected_count": 0, + "results": [], + } + + monkeypatch.setattr(client, "_make_request", fake_make_request) + + client.review_user_playbooks( + start_time=datetime(2026, 1, 1, tzinfo=UTC), + end_time=datetime(2026, 1, 2, tzinfo=UTC), + ) + + assert captured["timeout"] == 600 diff --git a/tests/server/services/playbook/test_playbook_generation_service.py b/tests/server/services/playbook/test_playbook_generation_service.py index 730c4eedf..df01ae040 100644 --- a/tests/server/services/playbook/test_playbook_generation_service.py +++ b/tests/server/services/playbook/test_playbook_generation_service.py @@ -927,6 +927,37 @@ def test_review_interaction_window_returns_honest_empty_window(): assert service._review_interaction_window(playbook_config) == [] +def test_automatic_review_reloads_the_configured_extraction_window(): + request_context = MagicMock() + service = PlaybookGenerationService( + llm_client=MagicMock(), request_context=request_context + ) + service.service_config = PlaybookGenerationServiceConfig( + request_id="request", + agent_version="v1", + user_id="user-1", + source="chat", + ) + playbook_config = PlaybookConfig( + extractor_name="test_playbook", + extraction_definition_prompt="Review grounded lessons", + request_sources_enabled=["chat"], + window_size_override=3, + ) + expected_window = [MagicMock(spec=RequestInteractionDataModel)] + extractor = MagicMock() + extractor._get_interactions.return_value = expected_window + + with patch.object( + service, "_create_extractor", return_value=extractor + ) as create_extractor: + assert service._review_interaction_window(playbook_config) == expected_window + + create_extractor.assert_called_once_with(playbook_config, service.service_config) + extractor._get_interactions.assert_called_once_with() + request_context.storage.get_interactions_by_ids.assert_not_called() + + def test_loading_a_new_generation_request_clears_the_review_window_cache(): service = PlaybookGenerationService( llm_client=MagicMock(), request_context=MagicMock() diff --git a/tests/server/services/playbook/test_playbook_review_service_integration.py b/tests/server/services/playbook/test_playbook_review_service_integration.py index b55870800..d173fc261 100644 --- a/tests/server/services/playbook/test_playbook_review_service_integration.py +++ b/tests/server/services/playbook/test_playbook_review_service_integration.py @@ -1,6 +1,7 @@ from __future__ import annotations from datetime import UTC, datetime +from hashlib import sha256 from types import SimpleNamespace from typing import Any, cast from unittest.mock import Mock, patch @@ -14,11 +15,17 @@ CandidateEvidenceUnit, CandidateReviewDecision, CandidateRevision, + PlaybookCandidateEvidenceError, PlaybookCandidateReviewOutput, PlaybookReviewOutcome, ) from reflexio.server.services.playbook.review_service import UserPlaybookReviewService from reflexio.server.services.storage.sqlite_storage import SQLiteStorage +from reflexio.server.services.storage.storage_base import ( + AgentBinding, + AgentRunRecord, + AgentRunStatus, +) pytestmark = pytest.mark.integration @@ -34,7 +41,19 @@ def _storage(tmp_path) -> SQLiteStorage: return storage -def _seed_source_interactions(storage: SQLiteStorage, count: int) -> None: +def _storage_digest(storage: SQLiteStorage) -> str: + dump = "\n".join(storage.conn.iterdump()) + return sha256(dump.encode()).hexdigest() + + +def _seed_source_interactions( + storage: SQLiteStorage, + count: int, + *, + sources: tuple[str, ...] = ("test",), + generation_request_ids: tuple[str, ...] = ("generation-request",), + generation_source_interaction_ids: tuple[int, ...] | None = None, +) -> None: for index in range(1, count + 1): request_id = f"source-request-{index}" storage.add_request( @@ -42,7 +61,7 @@ def _seed_source_interactions(storage: SQLiteStorage, count: int) -> None: request_id=request_id, user_id="user-1", created_at=100 + index, - source="test", + source=sources[(index - 1) % len(sources)], agent_version="1.0", session_id="session-1", ) @@ -64,6 +83,29 @@ def _seed_source_interactions(storage: SQLiteStorage, count: int) -> None: ], embeddings_prepared=True, ) + source_ids = list(generation_source_interaction_ids or range(1, count + 1)) + for request_id in generation_request_ids: + storage.create_agent_run( + AgentRunRecord( + id=f"run-{request_id}", + binding=AgentBinding( + org_id=storage.org_id, + extractor_kind="playbook", + user_id="user-1", + request_id=request_id, + agent_version="1.0", + source="test", + source_interaction_ids=source_ids, + window_start_interaction_id=min(source_ids), + window_end_interaction_id=max(source_ids), + ), + status=AgentRunStatus.FINALIZED, + generation_request_snapshot={ + "request_id": request_id, + "source_interaction_ids": source_ids, + }, + ) + ) def _playbook( @@ -220,6 +262,7 @@ def test_report_mode_respects_window_and_top_k_without_writes(tmp_path) -> None: _playbook(3, created_at=202), ] storage.save_user_playbooks(rows) + before_digest = _storage_digest(storage) enabled, decide, existing = _patch_review_dependencies( [_accept_output(), _reject_output()] ) @@ -236,10 +279,227 @@ def test_report_mode_respects_window_and_top_k_without_writes(tmp_path) -> None: assert response.accepted_count == 1 assert response.rejected_count == 1 assert all(result.applied is False for result in response.results) + assert _storage_digest(storage) == before_digest assert len(storage.get_user_playbooks(status_filter=[None])) == 3 assert storage.get_user_playbooks(status_filter=[Status.PENDING]) == [] +def test_manual_review_reconstructs_full_generation_window_across_sources( + tmp_path, +) -> None: + storage = _storage(tmp_path) + _seed_source_interactions(storage, 12, sources=("chat", "api")) + row = _playbook(1, created_at=200).model_copy( + update={ + "source": "generation-service", + # Candidate evidence is intentionally much smaller than the original + # extraction window recorded on the durable agent run. + "source_interaction_ids": [1], + } + ) + storage.save_user_playbooks([row]) + storage.get_last_k_interactions_grouped = Mock( # type: ignore[method-assign] + side_effect=AssertionError("manual review must not reconstruct a last-k window") + ) + reviewed_sources: set[str] = set() + reviewed_interaction_ids: list[int] = [] + + def decide(*args, **kwargs) -> PlaybookReviewOutcome: + for group in kwargs["request_interaction_data_models"]: + reviewed_sources.add(group.request.source) + reviewed_interaction_ids.extend( + interaction.interaction_id for interaction in group.interactions + ) + return _accept_output(1) + + enabled, _, existing = _patch_review_dependencies([]) + with ( + enabled, + patch( + "reflexio.server.services.playbook.review_service." + "PlaybookCandidateReviewer.decide", + side_effect=decide, + ), + existing, + ): + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert response.skipped_count == 0 + assert [result.decision for result in response.results] == ["accept"] + assert reviewed_sources == {"chat", "api"} + assert reviewed_interaction_ids == list(range(1, 13)) + + +def test_manual_review_adds_consolidated_citations_to_generation_window( + tmp_path, +) -> None: + storage = _storage(tmp_path) + _seed_source_interactions( + storage, + 3, + generation_source_interaction_ids=(1, 2), + ) + row = _playbook(3, created_at=200).model_copy( + update={"source_interaction_ids": [3]} + ) + storage.save_user_playbooks([row]) + reviewed_interaction_ids: list[int] = [] + + def decide(*args, **kwargs) -> PlaybookReviewOutcome: + reviewed_interaction_ids.extend( + interaction.interaction_id + for group in kwargs["request_interaction_data_models"] + for interaction in group.interactions + ) + return _accept_output(3) + + enabled, _, existing = _patch_review_dependencies([]) + with ( + enabled, + patch( + "reflexio.server.services.playbook.review_service." + "PlaybookCandidateReviewer.decide", + side_effect=decide, + ), + existing, + ): + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert response.skipped_count == 0 + assert reviewed_interaction_ids == [1, 2, 3] + + +def test_manual_review_skips_without_generation_window_provenance(tmp_path) -> None: + storage = _storage(tmp_path) + _seed_source_interactions(storage, 1) + storage.conn.execute("DELETE FROM _agent_runs") + storage.conn.commit() + row = _playbook(1, created_at=200) + storage.save_user_playbooks([row]) + enabled, decide, existing = _patch_review_dependencies([]) + + with enabled, decide as decide_mock, existing: + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert response.skipped_count == 1 + assert response.results[0].reason_code == "evidence_unavailable" + assert "no complete generation-window provenance" in ( + response.results[0].reason or "" + ) + decide_mock.assert_not_called() + + +def test_manual_review_skips_only_when_exact_source_interaction_is_gone( + tmp_path, +) -> None: + storage = _storage(tmp_path) + _seed_source_interactions(storage, 1) + row = _playbook(1, created_at=200) + storage.save_user_playbooks([row]) + storage.delete_request("source-request-1") + enabled, decide, existing = _patch_review_dependencies([]) + + with enabled, decide as decide_mock, existing: + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert response.skipped_count == 1 + assert response.results[0].decision == "skip" + assert response.results[0].reason_code == "evidence_unavailable" + assert "missing persisted generation-window interactions: [1]" in ( + response.results[0].reason or "" + ) + decide_mock.assert_not_called() + + +def test_manual_review_skips_when_source_request_is_gone(tmp_path) -> None: + storage = _storage(tmp_path) + _seed_source_interactions(storage, 1) + row = _playbook(1, created_at=200) + storage.save_user_playbooks([row]) + storage.get_request = Mock(return_value=None) # type: ignore[method-assign] + enabled, decide, existing = _patch_review_dependencies([]) + + with enabled, decide as decide_mock, existing: + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert response.skipped_count == 1 + assert response.results[0].decision == "skip" + assert response.results[0].reason_code == "evidence_unavailable" + assert "missing persisted source request source-request-1" in ( + response.results[0].reason or "" + ) + decide_mock.assert_not_called() + + +@pytest.mark.parametrize("mismatched_record", ["interaction", "request"]) +def test_manual_review_skips_cross_user_provenance( + tmp_path, + mismatched_record: str, +) -> None: + storage = _storage(tmp_path) + _seed_source_interactions(storage, 1) + row = _playbook(1, created_at=200) + storage.save_user_playbooks([row]) + if mismatched_record == "interaction": + interaction = storage.get_interactions_by_ids([1])[0] + storage.get_interactions_by_ids = Mock( # type: ignore[method-assign] + return_value=[interaction.model_copy(update={"user_id": "other-user"})] + ) + else: + request = storage.get_request("source-request-1") + assert request is not None + storage.get_request = Mock( # type: ignore[method-assign] + return_value=request.model_copy(update={"user_id": "other-user"}) + ) + enabled, decide, existing = _patch_review_dependencies([]) + + with enabled, decide as decide_mock, existing: + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert response.skipped_count == 1 + assert response.results[0].decision == "skip" + assert response.results[0].reason_code == "evidence_unavailable" + assert "owned by another user" in (response.results[0].reason or "") + decide_mock.assert_not_called() + + +def test_manual_review_skips_bad_candidate_evidence_and_continues(tmp_path) -> None: + storage = _storage(tmp_path) + _seed_source_interactions( + storage, + 2, + generation_request_ids=("shared-generation",), + ) + rows = [ + _playbook(1, created_at=201, request_id="shared-generation"), + _playbook(2, created_at=200, request_id="shared-generation"), + ] + storage.save_user_playbooks(rows) + enabled, decide, existing = _patch_review_dependencies( + [ + PlaybookCandidateEvidenceError( + "Reviewer cannot map C1 evidence to a local turn" + ), + _accept_output(2), + ] + ) + + with enabled, decide, existing: + response = _service(storage).run(_request(report_only=True)) + + assert response.success is True + assert [result.decision for result in response.results] == ["skip", "accept"] + assert response.skipped_count == 1 + assert response.accepted_count == 1 + assert "cannot map C1 evidence" in (response.results[0].reason or "") + + def test_apply_mode_commits_each_accept_edit_and_reject_in_order(tmp_path) -> None: storage = _storage(tmp_path) _seed_source_interactions(storage, 3) @@ -283,6 +543,7 @@ def test_apply_mode_commits_each_accept_edit_and_reject_in_order(tmp_path) -> No successor = next(item for item in current if item.user_playbook_id == successor_id) assert successor.content == "Narrow revised lesson." assert successor.source == "test" + assert successor.request_id == rows[1].request_id # The edited incumbent is SUPERSEDED and points at its replacement; only the # rejected row is ARCHIVED. @@ -310,7 +571,11 @@ def test_accepted_same_generation_playbook_is_visible_to_next_review( report_only: bool, ) -> None: storage = _storage(tmp_path) - _seed_source_interactions(storage, 2) + _seed_source_interactions( + storage, + 2, + generation_request_ids=("shared-generation",), + ) rows = [ _playbook(1, created_at=201, request_id="shared-generation"), _playbook(2, created_at=200, request_id="shared-generation"), diff --git a/tests/server/services/playbook/test_playbook_reviewer.py b/tests/server/services/playbook/test_playbook_reviewer.py index 089440176..bfabbec2f 100644 --- a/tests/server/services/playbook/test_playbook_reviewer.py +++ b/tests/server/services/playbook/test_playbook_reviewer.py @@ -17,6 +17,7 @@ CandidateEvidenceUnit, CandidateReviewDecision, CandidateRevision, + PlaybookCandidateEvidenceError, PlaybookCandidateReviewer, PlaybookCandidateReviewOutput, ) @@ -172,6 +173,73 @@ def test_reviewer_accepts_revises_rejects_with_exact_evidence_accounting(): assert "interaction_id" not in rendered +def test_reviewer_rebuilds_each_cited_span_from_exact_interaction_ids(): + interactions = [ + _interaction_model(111, "First retained historical evidence."), + _interaction_model(112, "Second retained historical evidence."), + _interaction_model(113, "Ancillary generation-window context."), + ] + candidate = _candidate( + 111, + # Historical consolidated rows can retain more cited IDs than text in + # their bounded combined span. Exact IDs, not this denormalized field, + # are the durable evidence identity. + "First retained historical evidence.", + content="Use both retained signals.", + ).model_copy(update={"source_interaction_ids": [111, 112]}) + output = PlaybookCandidateReviewOutput( + decisions=[ + CandidateReviewDecision( + id="C1", + decision="accept", + reason_code="grounded_useful", + evidence_ids=["C1-E1", "C1-E2"], + ) + ] + ) + reviewer, client = _reviewer(output) + + result = reviewer.review( + candidates=[candidate], + request_interaction_data_models=interactions, + existing_playbooks=[], + agent_context="Test agent", + playbook_definition="Reusable user guidance", + tool_context="", + ) + + assert result[0].source_interaction_ids == [111, 112] + assert result[0].source_span == "First retained historical evidence." + rendered = client.generate_chat_response.call_args.args[0][0]["content"] + assert "[C1-E1]" in rendered and "[C1-E2]" in rendered + assert "Ancillary generation-window context." in rendered + assert "[C1-E3]" not in rendered + + +def test_reviewer_raises_specific_error_when_cited_id_is_not_in_context(): + reviewer, _ = _reviewer(PlaybookCandidateReviewOutput(decisions=[])) + candidate = _candidate( + 999, + "Missing evidence.", + content="Unreviewable lesson.", + ) + + with pytest.raises( + PlaybookCandidateEvidenceError, + match="cannot map C1 evidence to a local turn", + ): + reviewer.decide( + candidates=[candidate], + request_interaction_data_models=[ + _interaction_model(111, "Different evidence.") + ], + existing_playbooks=[], + agent_context="Test agent", + playbook_definition="Reusable user guidance", + tool_context="", + ) + + def test_reviewer_fails_closed_when_candidate_accounting_is_incomplete(): interactions = [ _interaction_model(201, "First grounded correction."), diff --git a/tests/server/services/storage/sqlite_storage/test_agent_run_storage.py b/tests/server/services/storage/sqlite_storage/test_agent_run_storage.py index 497aad152..78300ee39 100644 --- a/tests/server/services/storage/sqlite_storage/test_agent_run_storage.py +++ b/tests/server/services/storage/sqlite_storage/test_agent_run_storage.py @@ -97,6 +97,103 @@ def test_sqlite_agent_run_crud_round_trip(storage): assert loaded.generation_request_snapshot == {"request_id": "request_1"} +def test_sqlite_get_latest_finalized_agent_run_for_request_filters_binding(storage): + matching = replace( + _agent_run("run_matching", AgentRunStatus.FINALIZED), + binding=replace( + _agent_run("unused", AgentRunStatus.FINALIZED).binding, + extractor_kind="playbook", + source_interaction_ids=[10, 11, 12], + ), + ) + storage.create_agent_run(matching) + storage.create_agent_run( + replace( + matching, + id="run_wrong_kind", + binding=replace(matching.binding, extractor_kind="profile"), + ) + ) + storage.create_agent_run( + replace( + matching, + id="run_not_finalized", + status=AgentRunStatus.RUNNING, + binding=replace( + matching.binding, + source_interaction_ids=[99], + ), + ) + ) + newest = replace( + matching, + id="run_newest", + status=AgentRunStatus.FINALIZED_PENDING_TOOL, + binding=replace(matching.binding, source_interaction_ids=[20, 21]), + ) + storage.create_agent_run(newest) + storage.conn.execute( + "UPDATE _agent_runs SET created_at = ?, updated_at = ? WHERE id = ?", + ("2026-01-02T00:00:00Z", "2026-01-02T00:00:00Z", newest.id), + ) + storage.conn.execute( + "UPDATE _agent_runs SET created_at = ?, updated_at = ? WHERE id = ?", + ("2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z", matching.id), + ) + storage.conn.commit() + + loaded = storage.get_latest_finalized_agent_run_for_request( + org_id="org_1", + extractor_kind="playbook", + user_id="user_1", + request_id="request_1", + ) + + assert loaded is not None + assert loaded.id == "run_newest" + assert loaded.status == AgentRunStatus.FINALIZED_PENDING_TOOL + assert loaded.binding.source_interaction_ids == [20, 21] + assert ( + storage.get_latest_finalized_agent_run_for_request( + org_id="org_1", + extractor_kind="playbook", + user_id="another-user", + request_id="request_1", + ) + is None + ) + + +def test_sqlite_latest_finalized_agent_run_breaks_timestamp_ties(storage): + binding = replace( + _agent_run("unused", AgentRunStatus.FINALIZED).binding, + extractor_kind="playbook", + ) + for run_id, source_ids in (("run_a", [10]), ("run_z", [20])): + storage.create_agent_run( + replace( + _agent_run(run_id, AgentRunStatus.FINALIZED), + binding=replace(binding, source_interaction_ids=source_ids), + ) + ) + storage.conn.execute( + "UPDATE _agent_runs SET created_at = ?, updated_at = ? WHERE id = ?", + ("2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z", run_id), + ) + storage.conn.commit() + + loaded = storage.get_latest_finalized_agent_run_for_request( + org_id="org_1", + extractor_kind="playbook", + user_id="user_1", + request_id="request_1", + ) + + assert loaded is not None + assert loaded.id == "run_z" + assert loaded.binding.source_interaction_ids == [20] + + def test_sqlite_update_agent_run_status_records_lifecycle_timestamps(storage): storage.create_agent_run(_agent_run("run_1", AgentRunStatus.RUNNING)) diff --git a/tests/server/test_api_security_middleware.py b/tests/server/test_api_security_middleware.py index 997e55458..900ba9fc7 100644 --- a/tests/server/test_api_security_middleware.py +++ b/tests/server/test_api_security_middleware.py @@ -4,7 +4,11 @@ from fastapi.testclient import TestClient from reflexio.server.api import create_app -from reflexio.server.middleware import BodySizeLimitMiddleware +from reflexio.server.middleware import ( + SYNC_REQUEST_TIMEOUT_SECONDS, + BodySizeLimitMiddleware, + TimeoutMiddleware, +) def test_cors_uses_frontend_url_allowlist(monkeypatch): @@ -138,6 +142,38 @@ async def send(message): assert response_body == b'{"detail":"Request body too large"}' +def test_playbook_review_uses_synchronous_request_timeout(monkeypatch): + observed: dict[str, float | None] = {} + + async def fake_wait_for(awaitable, *, timeout=None): + observed["timeout"] = timeout + return await awaitable + + async def call_next(_request): + from starlette.responses import Response + + return Response() + + monkeypatch.setattr(asyncio, "wait_for", fake_wait_for) + request = Request( + { + "type": "http", + "method": "POST", + "scheme": "http", + "path": "/api/review_user_playbooks", + "raw_path": b"/api/review_user_playbooks", + "query_string": b"", + "headers": [], + "client": ("testclient", 50000), + "server": ("testserver", 80), + } + ) + + asyncio.run(TimeoutMiddleware(FastAPI()).dispatch(request, call_next)) + + assert observed["timeout"] == SYNC_REQUEST_TIMEOUT_SECONDS + + def test_security_headers_are_added(monkeypatch): monkeypatch.delenv("REFLEXIO_ALLOWED_ORIGINS", raising=False)