From 979c88ce1d23ac35729977708f47a6b9ac2dc95b Mon Sep 17 00:00:00 2001 From: Guangyu Date: Mon, 17 Aug 2026 06:16:40 +0000 Subject: [PATCH 01/13] feat: add open-world analysis vocabulary --- reflexio/models/api_schema/domain/entities.py | 11 ++ .../services/storage/sqlite_storage/_base.py | 41 +++++- .../sqlite_storage/playbook/_optimization.py | 45 +++++-- .../test_open_world_optimization_identity.py | 63 +++++++++ .../test_open_world_analysis_identity.py | 127 ++++++++++++++++++ ...ptimization_replay_contract_integration.py | 41 ++++-- 6 files changed, 299 insertions(+), 29 deletions(-) create mode 100644 tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 65463f4a..065c2d1e 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -412,9 +412,11 @@ class AgentPlaybook(BaseModel): OptimizationJobStage = Literal[ "evidence_frozen", + "discovery_analyzed", "candidate_generated", "replay_running", "replay_evaluated", + "held_out_analyzed", "publishing", "applied", "abstained", @@ -438,6 +440,12 @@ class AgentPlaybook(BaseModel): "replay_failed", "publication_failed", "governance_erased", + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", + "infrastructure_failure", ] OptimizationArtifactKind = Literal[ @@ -447,6 +455,9 @@ class AgentPlaybook(BaseModel): "candidate", "candidate_search_projection", "open_world_evidence_bundle", + "open_world_discovery_memo", + "open_world_candidate", + "open_world_attempt_decision", ] Sha256Digest = str diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index f7e87c08..0b71a72e 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -2202,6 +2202,14 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: "CHECK (stage IS NULL OR stage IN", "CHECK (terminal_outcome IS NULL OR terminal_outcome IN", "'governance_erased'", + "'discovery_analyzed'", + "'held_out_analyzed'", + "'no_grounded_hypothesis'", + "'analyst_unqualified'", + "'heldout_evidence_failed'", + "'stale_incumbent'", + "'governance_invalidated'", + "'infrastructure_failure'", ) if all(check in table_sql for check in required_checks) and ( "'offline_tuner_open_world'" not in table_sql @@ -2245,9 +2253,11 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: lease_expires_at INTEGER, stage TEXT CHECK (stage IS NULL OR stage IN ( 'evidence_frozen', + 'discovery_analyzed', 'candidate_generated', 'replay_running', 'replay_evaluated', + 'held_out_analyzed', 'publishing', 'applied', 'abstained', @@ -2269,7 +2279,13 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: 'generation_failed', 'replay_failed', 'publication_failed', - 'governance_erased' + 'governance_erased', + 'no_grounded_hypothesis', + 'analyst_unqualified', + 'heldout_evidence_failed', + 'stale_incumbent', + 'governance_invalidated', + 'infrastructure_failure' )), expected_population_manifest_digest TEXT, generation_selection_manifest_digest TEXT, @@ -2366,6 +2382,9 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: "'candidate'", "'candidate_search_projection'", "'open_world_evidence_bundle'", + "'open_world_discovery_memo'", + "'open_world_candidate'", + "'open_world_attempt_decision'", ) if all(artifact_kind in table_sql for artifact_kind in artifact_kinds): return @@ -2396,7 +2415,10 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: 'replay_manifest', 'candidate', 'candidate_search_projection', - 'open_world_evidence_bundle' + 'open_world_evidence_bundle', + 'open_world_discovery_memo', + 'open_world_candidate', + 'open_world_attempt_decision' )), content_json TEXT NOT NULL, content_digest TEXT NOT NULL, @@ -3411,9 +3433,11 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: lease_expires_at INTEGER, stage TEXT CHECK (stage IS NULL OR stage IN ( 'evidence_frozen', + 'discovery_analyzed', 'candidate_generated', 'replay_running', 'replay_evaluated', + 'held_out_analyzed', 'publishing', 'applied', 'abstained', @@ -3435,7 +3459,13 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: 'generation_failed', 'replay_failed', 'publication_failed', - 'governance_erased' + 'governance_erased', + 'no_grounded_hypothesis', + 'analyst_unqualified', + 'heldout_evidence_failed', + 'stale_incumbent', + 'governance_invalidated', + 'infrastructure_failure' )), expected_population_manifest_digest TEXT, generation_selection_manifest_digest TEXT, @@ -3495,7 +3525,10 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: 'replay_manifest', 'candidate', 'candidate_search_projection', - 'open_world_evidence_bundle' + 'open_world_evidence_bundle', + 'open_world_discovery_memo', + 'open_world_candidate', + 'open_world_attempt_decision' )), content_json TEXT NOT NULL, content_digest TEXT NOT NULL, diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py index 41b938d9..24af4a1c 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py @@ -33,7 +33,12 @@ StorageError, ) -_FAILURE_OUTCOMES = {"generation_failed", "replay_failed", "publication_failed"} +_FAILURE_OUTCOMES = { + "generation_failed", + "replay_failed", + "publication_failed", + "infrastructure_failure", +} _ABSTENTION_OUTCOMES = { "insufficient_negative_evidence", "insufficient_positive_evidence", @@ -46,13 +51,20 @@ "candidate_regressed", "candidate_did_not_improve", "incumbent_changed", + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", } -_STAGE_PREDECESSORS: dict[str, str] = { - "candidate_generated": "evidence_frozen", - "replay_running": "candidate_generated", - "replay_evaluated": "replay_running", - "publishing": "replay_evaluated", - "applied": "publishing", +_STAGE_PREDECESSORS: dict[str, tuple[str, str]] = { + "discovery_analyzed": ("evidence_frozen", "evidence_frozen"), + "candidate_generated": ("evidence_frozen", "discovery_analyzed"), + "replay_running": ("candidate_generated", "candidate_generated"), + "replay_evaluated": ("replay_running", "replay_running"), + "held_out_analyzed": ("candidate_generated", "candidate_generated"), + "publishing": ("replay_evaluated", "replay_evaluated"), + "applied": ("publishing", "publishing"), } @@ -714,7 +726,7 @@ def advance_playbook_optimization_stage( now: int | None = None, ) -> bool: advanced_at = self._lease_now(now) - predecessor = _STAGE_PREDECESSORS.get(stage) + predecessors = _STAGE_PREDECESSORS.get(stage) terminal_status: str | None = None if stage == "applied": if terminal_outcome not in (None, "applied"): @@ -729,10 +741,12 @@ def advance_playbook_optimization_stage( if terminal_outcome not in _ABSTENTION_OUTCOMES: return False terminal_status = "skipped" - elif predecessor is None or terminal_outcome is not None: + elif predecessors is None or terminal_outcome is not None: return False with self._lock: if terminal_status is None: + if predecessors is None: + return False cur = self.conn.execute( """UPDATE playbook_optimization_jobs SET stage = ?, updated_at = ? @@ -740,8 +754,15 @@ def advance_playbook_optimization_stage( AND status IN ('pending', 'running') AND lease_fence = ? AND lease_expires_at > ? - AND stage = ?""", - (stage, advanced_at, job_id, fence, advanced_at, predecessor), + AND stage IN (?, ?)""", + ( + stage, + advanced_at, + job_id, + fence, + advanced_at, + *predecessors, + ), ) elif stage == "applied": cur = self.conn.execute( @@ -782,9 +803,11 @@ def advance_playbook_optimization_stage( AND lease_expires_at > ? AND stage IN ( 'evidence_frozen', + 'discovery_analyzed', 'candidate_generated', 'replay_running', 'replay_evaluated', + 'held_out_analyzed', 'publishing' )""", ( diff --git a/tests/models/test_open_world_optimization_identity.py b/tests/models/test_open_world_optimization_identity.py index 09c58d9d..89f9b00e 100644 --- a/tests/models/test_open_world_optimization_identity.py +++ b/tests/models/test_open_world_optimization_identity.py @@ -1,6 +1,11 @@ from hashlib import sha256 +import pytest + from reflexio.models.api_schema.domain.entities import ( + OptimizationArtifactKind, + OptimizationJobStage, + OptimizationTerminalOutcome, PlaybookOptimizationArtifact, PlaybookOptimizationJob, ) @@ -22,3 +27,61 @@ def test_open_world_optimization_identity_is_accepted() -> None: assert job.optimizer_kind == "offline_tuner_open_world" assert artifact.artifact_kind == "open_world_evidence_bundle" + + +@pytest.mark.parametrize("stage", ["discovery_analyzed", "held_out_analyzed"]) +def test_open_world_analysis_stage_is_accepted(stage: OptimizationJobStage) -> None: + job = PlaybookOptimizationJob( + optimizer_kind="offline_tuner_open_world", + target_kind="user_playbook", + target_id=7, + stage=stage, + ) + + assert job.stage == stage + + +@pytest.mark.parametrize( + "terminal_outcome", + [ + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", + "infrastructure_failure", + ], +) +def test_open_world_analysis_terminal_outcome_is_accepted( + terminal_outcome: OptimizationTerminalOutcome, +) -> None: + job = PlaybookOptimizationJob( + optimizer_kind="offline_tuner_open_world", + target_kind="user_playbook", + target_id=7, + terminal_outcome=terminal_outcome, + ) + + assert job.terminal_outcome == terminal_outcome + + +@pytest.mark.parametrize( + "artifact_kind", + [ + "open_world_discovery_memo", + "open_world_candidate", + "open_world_attempt_decision", + ], +) +def test_open_world_analysis_artifact_kind_is_accepted( + artifact_kind: OptimizationArtifactKind, +) -> None: + content_json = '{"schema_version":"offline-tuner-open-world-analysis-v1"}' + artifact = PlaybookOptimizationArtifact( + job_id=1, + artifact_kind=artifact_kind, + content_json=content_json, + content_digest=sha256(content_json.encode()).hexdigest(), + ) + + assert artifact.artifact_kind == artifact_kind diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py new file mode 100644 index 00000000..02658dff --- /dev/null +++ b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py @@ -0,0 +1,127 @@ +"""Generic SQLite identity coverage for the open-world analysis vocabulary.""" + +from __future__ import annotations + +from collections.abc import Generator +from hashlib import sha256 +from pathlib import Path + +import pytest + +from reflexio.models.api_schema import service_schemas as schemas +from reflexio.server.services.storage.sqlite_storage import SQLiteStorage +from reflexio.server.services.storage.storage_base import BaseStorage + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def storage(tmp_path: Path) -> Generator[SQLiteStorage]: + store = SQLiteStorage( + org_id="open-world-analysis-identity", + db_path=str(tmp_path / "reflexio.db"), + ) + try: + yield store + finally: + store.conn.close() + + +def _job() -> schemas.PlaybookOptimizationJob: + return schemas.PlaybookOptimizationJob( + optimizer_kind="offline_tuner_replay", + target_kind="user_playbook", + target_id=41, + discovery_key="discovery-key", + attempt_key="attempt-key", + stage="evidence_frozen", + ) + + +def test_open_world_analysis_stage_path_round_trips(storage: SQLiteStorage) -> None: + job = storage.create_playbook_optimization_job(_job()) + claim = storage.claim_playbook_optimization_job( + job_id=job.job_id, + owner="worker-a", + lease_seconds=60, + now=1_000, + ) + + assert storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage="discovery_analyzed", + now=1_001, + ) + assert storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage="candidate_generated", + now=1_002, + ) + assert storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage="held_out_analyzed", + now=1_003, + ) + + persisted = storage.get_playbook_optimization_job(job.job_id) + assert persisted is not None + assert persisted.stage == "held_out_analyzed" + + +@pytest.mark.parametrize( + "artifact_kind", + [ + "open_world_discovery_memo", + "open_world_candidate", + "open_world_attempt_decision", + ], +) +def test_open_world_analysis_artifact_round_trips( + storage: SQLiteStorage, + artifact_kind: schemas.OptimizationArtifactKind, +) -> None: + job = storage.create_playbook_optimization_job(_job()) + claim = storage.claim_playbook_optimization_job( + job_id=job.job_id, + owner="worker-a", + lease_seconds=60, + now=2_000, + ) + content_json = '{"schema_version":"offline-tuner-open-world-analysis-v1"}' + artifact = schemas.PlaybookOptimizationArtifact( + job_id=job.job_id, + artifact_kind=artifact_kind, + content_json=content_json, + content_digest=sha256(content_json.encode()).hexdigest(), + ) + + saved = storage.upsert_playbook_optimization_artifact( + artifact, + fence=claim.fence, + now=2_001, + ) + + assert ( + storage.get_playbook_optimization_artifact(job.job_id, artifact_kind) == saved + ) + + +def test_sqlite_does_not_expose_open_world_invocations( + storage: SQLiteStorage, +) -> None: + table = storage.conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' " + "AND name = 'offline_tuner_open_world_invocations'" + ).fetchone() + + assert table is None + for method_name in ( + "prepare_open_world_invocation", + "complete_open_world_invocation", + "load_open_world_invocation", + ): + assert not hasattr(BaseStorage, method_name) + assert not hasattr(SQLiteStorage, method_name) diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index 0a6c64f1..399f285b 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -252,7 +252,13 @@ def test_stage_advancement_is_linear(storage: BaseStorage) -> None: ("stage", "outcome", "expected_status"), [ ("abstained", "candidate_did_not_improve", "skipped"), + ("abstained", "no_grounded_hypothesis", "skipped"), + ("abstained", "analyst_unqualified", "skipped"), + ("abstained", "heldout_evidence_failed", "skipped"), + ("abstained", "stale_incumbent", "skipped"), + ("abstained", "governance_invalidated", "skipped"), ("failed", "generation_failed", "failed"), + ("failed", "infrastructure_failure", "failed"), ], ) def test_terminal_stage_records_outcome_and_releases_lease( @@ -587,20 +593,27 @@ def test_previous_artifact_schema_is_upgraded_without_losing_constraints( ) store.conn.rollback() - evidence_json = '{"cases":[1]}' - evidence = schemas.PlaybookOptimizationArtifact( - job_id=41, - artifact_kind="open_world_evidence_bundle", - content_json=evidence_json, - content_digest=sha256(evidence_json.encode()).hexdigest(), - created_at=107, - updated_at=108, - ) - saved = store.upsert_playbook_optimization_artifact(evidence, fence=3, now=500) - assert ( - store.get_playbook_optimization_artifact(41, "open_world_evidence_bundle") - == saved - ) + for artifact_kind in ( + "open_world_evidence_bundle", + "open_world_discovery_memo", + "open_world_candidate", + "open_world_attempt_decision", + ): + content_json = f'{{"artifact_kind":"{artifact_kind}"}}' + artifact = schemas.PlaybookOptimizationArtifact( + job_id=41, + artifact_kind=artifact_kind, + content_json=content_json, + content_digest=sha256(content_json.encode()).hexdigest(), + created_at=107, + updated_at=108, + ) + saved = store.upsert_playbook_optimization_artifact( + artifact, + fence=3, + now=500, + ) + assert store.get_playbook_optimization_artifact(41, artifact_kind) == saved assert store.migrate() is True store.conn.execute("DELETE FROM playbook_optimization_jobs WHERE job_id = 41") From 4c51c62160cec7f4165a81051fa03b7af0e79864 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 13:16:53 +0000 Subject: [PATCH 02/13] test(storage): cover invalid open-world stage transitions --- ...ptimization_replay_contract_integration.py | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index 399f285b..8f7f6913 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -859,6 +859,129 @@ def test_ordinary_stage_advance_rejects_governance_erased( assert persisted.terminal_outcome is None +def _claimed_job_at_stage( + storage: BaseStorage, + stage: schemas.OptimizationJobStage, + *, + now: int, +) -> tuple[int, int]: + job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + claim = storage.claim_playbook_optimization_job( + job_id=job.job_id, + owner="worker-a", + lease_seconds=60, + now=now, + ) + assert isinstance(storage, SQLiteStorage) + storage.conn.execute( + "UPDATE playbook_optimization_jobs SET stage = ? WHERE job_id = ?", + (stage, job.job_id), + ) + storage.conn.commit() + return job.job_id, claim.fence + + +@pytest.mark.parametrize( + ("current_stage", "target_stage"), + [ + ("candidate_generated", "discovery_analyzed"), + ("discovery_analyzed", "discovery_analyzed"), + ("evidence_frozen", "held_out_analyzed"), + ("held_out_analyzed", "held_out_analyzed"), + ("held_out_analyzed", "publishing"), + ("discovery_analyzed", "replay_running"), + ], +) +def test_widened_analysis_stage_rejects_invalid_predecessor( + storage: BaseStorage, + current_stage: schemas.OptimizationJobStage, + target_stage: schemas.OptimizationJobStage, +) -> None: + job_id, fence = _claimed_job_at_stage(storage, current_stage, now=7_000) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=target_stage, + now=7_001, + ) + is False + ) + persisted = storage.get_playbook_optimization_job(job_id) + assert persisted is not None + assert persisted.stage == current_stage + assert persisted.status == "running" + + +@pytest.mark.parametrize( + ("current_stage", "terminal_stage", "outcome"), + [ + ("evidence_frozen", "abstained", "infrastructure_failure"), + ("held_out_analyzed", "abstained", "applied"), + ("discovery_analyzed", "failed", "stale_incumbent"), + ("publishing", "applied", "heldout_evidence_failed"), + ], +) +def test_widened_terminal_stage_rejects_outcome_from_another_family( + storage: BaseStorage, + current_stage: schemas.OptimizationJobStage, + terminal_stage: schemas.OptimizationJobStage, + outcome: schemas.OptimizationTerminalOutcome, +) -> None: + job_id, fence = _claimed_job_at_stage(storage, current_stage, now=7_000) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=terminal_stage, + terminal_outcome=outcome, + now=7_001, + ) + is False + ) + persisted = storage.get_playbook_optimization_job(job_id) + assert persisted is not None + assert persisted.stage == current_stage + assert persisted.status == "running" + assert persisted.terminal_outcome is None + + +def test_widened_terminal_outcome_rejects_stale_lease_and_settled_job( + storage: BaseStorage, +) -> None: + job_id, fence = _claimed_job_at_stage(storage, "held_out_analyzed", now=7_000) + assert isinstance(storage, SQLiteStorage) + + def _abstain(*, at_fence: int, now: int) -> bool: + return storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=at_fence, + stage="abstained", + terminal_outcome="stale_incumbent", + now=now, + ) + + assert _abstain(at_fence=fence + 1, now=7_001) is False + assert _abstain(at_fence=fence, now=7_060) is False + + storage.conn.execute( + "UPDATE playbook_optimization_jobs SET stage = 'failed' WHERE job_id = ?", + (job_id,), + ) + storage.conn.commit() + assert _abstain(at_fence=fence, now=7_001) is False + + storage.conn.execute( + "UPDATE playbook_optimization_jobs " + "SET stage = 'held_out_analyzed', status = 'completed' WHERE job_id = ?", + (job_id,), + ) + storage.conn.commit() + assert _abstain(at_fence=fence, now=7_001) is False + + def _create_legacy_optimizer_schema(db_path: Path) -> None: conn = sqlite3.connect(db_path) conn.executescript( From f129a54f553e9fc43a0d93822ef18b82d3694af0 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 16:14:21 +0000 Subject: [PATCH 03/13] fix(storage): allow open-world optimizer jobs --- .../services/storage/sqlite_storage/_base.py | 4 ++- .../test_open_world_analysis_identity.py | 33 ++++++++++++++++++- ...ptimization_replay_contract_integration.py | 7 ++-- 3 files changed, 39 insertions(+), 5 deletions(-) diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index 0b71a72e..eb183991 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -2212,7 +2212,7 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: "'infrastructure_failure'", ) if all(check in table_sql for check in required_checks) and ( - "'offline_tuner_open_world'" not in table_sql + "'offline_tuner_open_world'" in table_sql ): return foreign_keys_enabled = bool( @@ -2236,6 +2236,7 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: CHECK (optimizer_kind IN ( 'gepa', 'offline_tuner_replay', + 'offline_tuner_open_world', 'offline_tuner_legacy', 'optimizer_legacy_unknown' )), @@ -3416,6 +3417,7 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: CHECK (optimizer_kind IN ( 'gepa', 'offline_tuner_replay', + 'offline_tuner_open_world', 'offline_tuner_legacy', 'optimizer_legacy_unknown' )), diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py index 02658dff..37fe24a2 100644 --- a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py +++ b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py @@ -2,6 +2,7 @@ from __future__ import annotations +import sqlite3 from collections.abc import Generator from hashlib import sha256 from pathlib import Path @@ -29,7 +30,7 @@ def storage(tmp_path: Path) -> Generator[SQLiteStorage]: def _job() -> schemas.PlaybookOptimizationJob: return schemas.PlaybookOptimizationJob( - optimizer_kind="offline_tuner_replay", + optimizer_kind="offline_tuner_open_world", target_kind="user_playbook", target_id=41, discovery_key="discovery-key", @@ -71,6 +72,36 @@ def test_open_world_analysis_stage_path_round_trips(storage: SQLiteStorage) -> N assert persisted.stage == "held_out_analyzed" +def test_legacy_optimizer_kind_allowlist_is_rebuilt_for_open_world_job( + tmp_path: Path, +) -> None: + db_path = str(tmp_path / "legacy-optimizer-kind.db") + initial = SQLiteStorage(org_id="legacy-optimizer-kind", db_path=db_path) + table_sql = initial.conn.execute( + "SELECT sql FROM sqlite_master " + "WHERE type = 'table' AND name = 'playbook_optimization_jobs'" + ).fetchone()[0] + initial.conn.close() + + legacy_table_sql = table_sql.replace("'offline_tuner_open_world',", "") + conn = sqlite3.connect(db_path) + try: + conn.execute("PRAGMA foreign_keys = OFF") + conn.execute("DROP TABLE playbook_optimization_jobs") + conn.execute(legacy_table_sql) + conn.commit() + finally: + conn.close() + + upgraded = SQLiteStorage(org_id="legacy-optimizer-kind", db_path=db_path) + try: + job = upgraded.create_playbook_optimization_job(_job()) + finally: + upgraded.conn.close() + + assert job.optimizer_kind == "offline_tuner_open_world" + + @pytest.mark.parametrize( "artifact_kind", [ diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index 8f7f6913..79b07b2a 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -122,12 +122,13 @@ def test_conflicting_active_identity_is_rejected(storage: BaseStorage) -> None: storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a2")) -def test_sqlite_rejects_open_world_optimizer_jobs(storage: BaseStorage) -> None: +def test_sqlite_persists_open_world_optimizer_jobs(storage: BaseStorage) -> None: open_world_job = _replay_job("open-world-discovery", "open-world-attempt") open_world_job.optimizer_kind = "offline_tuner_open_world" - with pytest.raises(StorageError, match="CHECK constraint failed"): - storage.create_or_get_playbook_optimization_job(open_world_job) + saved = storage.create_or_get_playbook_optimization_job(open_world_job) + + assert saved.optimizer_kind == "offline_tuner_open_world" def test_gepa_publication_reclaim_contract_has_none_live_and_reclaimed_outcomes( From c432453c3b413aa2aac83706f24d18593e1812fe Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 17:18:57 +0000 Subject: [PATCH 04/13] feat(llm): expose provider request guard error --- reflexio/server/llm/__init__.py | 2 ++ reflexio/server/llm/litellm_client.py | 2 ++ tests/server/llm/test_litellm_client_surface.py | 12 ++++++++++++ 3 files changed, 16 insertions(+) diff --git a/reflexio/server/llm/__init__.py b/reflexio/server/llm/__init__.py index 91b5983d..633601d5 100644 --- a/reflexio/server/llm/__init__.py +++ b/reflexio/server/llm/__init__.py @@ -9,6 +9,7 @@ LiteLLMClient, LiteLLMClientError, LiteLLMConfig, + ProviderRequestGuardError, StructuredOutputRepairError, StructuredOutputValidator, ToolCallingChatResponse, @@ -24,6 +25,7 @@ "LiteLLMClient", "LiteLLMConfig", "LiteLLMClientError", + "ProviderRequestGuardError", "StructuredOutputRepairError", "StructuredOutputValidator", "ModelRole", diff --git a/reflexio/server/llm/litellm_client.py b/reflexio/server/llm/litellm_client.py index 9345eb47..54384197 100644 --- a/reflexio/server/llm/litellm_client.py +++ b/reflexio/server/llm/litellm_client.py @@ -63,6 +63,7 @@ _PromptTokenDetailsSnapshot as _PromptTokenDetailsSnapshot, ) from reflexio.server.llm._litellm_text_generation import ( + ProviderRequestGuardError, StructuredOutputValidator, TextGenerationMixin, ) @@ -101,6 +102,7 @@ "LiteLLMConfig", "LiteLLMClientError", "StructuredOutputRepairError", + "ProviderRequestGuardError", "StructuredOutputValidator", "ToolCallingChatResponse", "create_litellm_client", diff --git a/tests/server/llm/test_litellm_client_surface.py b/tests/server/llm/test_litellm_client_surface.py index 1cb3f727..5c83d5cf 100644 --- a/tests/server/llm/test_litellm_client_surface.py +++ b/tests/server/llm/test_litellm_client_surface.py @@ -31,6 +31,7 @@ class must be the SAME object/class the moved code uses and tests touch — the "LiteLLMClient", "LiteLLMConfig", "LiteLLMClientError", + "ProviderRequestGuardError", "StructuredOutputRepairError", "StructuredOutputValidator", "ToolCallingChatResponse", @@ -115,11 +116,22 @@ def test_boot_import_paths(): LiteLLMClient, LiteLLMClientError, LiteLLMConfig, + ProviderRequestGuardError, ToolCallingChatResponse, create_litellm_client, ) +def test_provider_request_guard_error_is_re_exported_from_llm_package(): + """Callers can catch provider-boundary guards without importing internals.""" + from reflexio.server.llm import ProviderRequestGuardError + from reflexio.server.llm._litellm_text_generation import ( + ProviderRequestGuardError as ImplementationProviderRequestGuardError, + ) + + assert ProviderRequestGuardError is ImplementationProviderRequestGuardError + + # --------------------------------------------------------------------------- # SINK-2 identity asserts. Skipped until the owning module exists (Task 1/2), # then enforce that the facade re-export IS the same object the moved code uses. From a812f2708883db31323a33e8c9a442ba5583dfcc Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 20:33:16 +0000 Subject: [PATCH 05/13] fix: isolate optimizer transition families --- .../sqlite_storage/playbook/_optimization.py | 144 +++++++++++------- .../test_open_world_analysis_identity.py | 38 +++++ ...ptimization_replay_contract_integration.py | 61 +++++--- 3 files changed, 163 insertions(+), 80 deletions(-) diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py index 24af4a1c..5329dd40 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py @@ -33,38 +33,52 @@ StorageError, ) -_FAILURE_OUTCOMES = { - "generation_failed", - "replay_failed", - "publication_failed", - "infrastructure_failure", +_STAGE_PREDECESSORS_BY_OPTIMIZER: dict[str, dict[str, tuple[str, str]]] = { + "offline_tuner_replay": { + "candidate_generated": ("evidence_frozen", "evidence_frozen"), + "replay_running": ("candidate_generated", "candidate_generated"), + "replay_evaluated": ("replay_running", "replay_running"), + "publishing": ("replay_evaluated", "replay_evaluated"), + "applied": ("publishing", "publishing"), + }, + "offline_tuner_open_world": { + "discovery_analyzed": ("evidence_frozen", "evidence_frozen"), + "candidate_generated": ("discovery_analyzed", "discovery_analyzed"), + "held_out_analyzed": ("candidate_generated", "candidate_generated"), + }, } -_ABSTENTION_OUTCOMES = { - "insufficient_negative_evidence", - "insufficient_positive_evidence", - "insufficient_coverage", - "replay_unsupported", - "deployment_unsupported", - "incomplete_replay_scope", - "insufficient_replay_cases", - "replay_inconclusive", - "candidate_regressed", - "candidate_did_not_improve", - "incumbent_changed", - "no_grounded_hypothesis", - "analyst_unqualified", - "heldout_evidence_failed", - "stale_incumbent", - "governance_invalidated", -} -_STAGE_PREDECESSORS: dict[str, tuple[str, str]] = { - "discovery_analyzed": ("evidence_frozen", "evidence_frozen"), - "candidate_generated": ("evidence_frozen", "discovery_analyzed"), - "replay_running": ("candidate_generated", "candidate_generated"), - "replay_evaluated": ("replay_running", "replay_running"), - "held_out_analyzed": ("candidate_generated", "candidate_generated"), - "publishing": ("replay_evaluated", "replay_evaluated"), - "applied": ("publishing", "publishing"), +_TERMINAL_OUTCOMES_BY_OPTIMIZER = { + "offline_tuner_replay": { + "failed": { + "generation_failed", + "replay_failed", + "publication_failed", + "infrastructure_failure", + }, + "abstained": { + "insufficient_negative_evidence", + "insufficient_positive_evidence", + "insufficient_coverage", + "replay_unsupported", + "deployment_unsupported", + "incomplete_replay_scope", + "insufficient_replay_cases", + "replay_inconclusive", + "candidate_regressed", + "candidate_did_not_improve", + "incumbent_changed", + }, + }, + "offline_tuner_open_world": { + "failed": {"infrastructure_failure"}, + "abstained": { + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", + }, + }, } @@ -726,24 +740,34 @@ def advance_playbook_optimization_stage( now: int | None = None, ) -> bool: advanced_at = self._lease_now(now) - predecessors = _STAGE_PREDECESSORS.get(stage) - terminal_status: str | None = None - if stage == "applied": - if terminal_outcome not in (None, "applied"): - return False - terminal_outcome = "applied" - terminal_status = "completed" - elif stage == "failed": - if terminal_outcome not in _FAILURE_OUTCOMES: + with self._lock: + row = self.conn.execute( + "SELECT optimizer_kind FROM playbook_optimization_jobs WHERE job_id = ?", + (job_id,), + ).fetchone() + if row is None: return False - terminal_status = "failed" - elif stage == "abstained": - if terminal_outcome not in _ABSTENTION_OUTCOMES: + optimizer_kind = row["optimizer_kind"] + predecessors = _STAGE_PREDECESSORS_BY_OPTIMIZER.get( + optimizer_kind, {} + ).get(stage) + terminal_status: str | None = None + if stage == "applied": + if ( + optimizer_kind != "offline_tuner_replay" + or terminal_outcome not in (None, "applied") + ): + return False + terminal_outcome = "applied" + terminal_status = "completed" + elif stage in ("failed", "abstained"): + if terminal_outcome not in _TERMINAL_OUTCOMES_BY_OPTIMIZER.get( + optimizer_kind, {} + ).get(stage, set()): + return False + terminal_status = "failed" if stage == "failed" else "skipped" + elif predecessors is None or terminal_outcome is not None: return False - terminal_status = "skipped" - elif predecessors is None or terminal_outcome is not None: - return False - with self._lock: if terminal_status is None: if predecessors is None: return False @@ -752,6 +776,7 @@ def advance_playbook_optimization_stage( SET stage = ?, updated_at = ? WHERE job_id = ? AND status IN ('pending', 'running') + AND optimizer_kind = ? AND lease_fence = ? AND lease_expires_at > ? AND stage IN (?, ?)""", @@ -759,6 +784,7 @@ def advance_playbook_optimization_stage( stage, advanced_at, job_id, + optimizer_kind, fence, advanced_at, *predecessors, @@ -775,6 +801,7 @@ def advance_playbook_optimization_stage( updated_at = ? WHERE job_id = ? AND status IN ('pending', 'running') + AND optimizer_kind = 'offline_tuner_replay' AND lease_fence = ? AND lease_expires_at > ? AND stage = 'publishing'""", @@ -789,8 +816,16 @@ def advance_playbook_optimization_stage( ), ) else: + active_stages = tuple( + predecessor + for stages in _STAGE_PREDECESSORS_BY_OPTIMIZER.get( + optimizer_kind, {} + ).values() + for predecessor in stages + ) + placeholders = ", ".join("?" for _ in active_stages) cur = self.conn.execute( - """UPDATE playbook_optimization_jobs + f"""UPDATE playbook_optimization_jobs SET stage = ?, terminal_outcome = ?, status = ?, @@ -799,25 +834,20 @@ def advance_playbook_optimization_stage( updated_at = ? WHERE job_id = ? AND status IN ('pending', 'running') + AND optimizer_kind = ? AND lease_fence = ? AND lease_expires_at > ? - AND stage IN ( - 'evidence_frozen', - 'discovery_analyzed', - 'candidate_generated', - 'replay_running', - 'replay_evaluated', - 'held_out_analyzed', - 'publishing' - )""", + AND stage IN ({placeholders})""", ( stage, terminal_outcome, terminal_status, advanced_at, job_id, + optimizer_kind, fence, advanced_at, + *active_stages, ), ) if self._own_transaction(): diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py index 37fe24a2..acbe0a8a 100644 --- a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py +++ b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py @@ -72,6 +72,44 @@ def test_open_world_analysis_stage_path_round_trips(storage: SQLiteStorage) -> N assert persisted.stage == "held_out_analyzed" +@pytest.mark.parametrize( + ("stage", "outcome", "expected_status"), + [ + ("abstained", "no_grounded_hypothesis", "skipped"), + ("abstained", "analyst_unqualified", "skipped"), + ("abstained", "heldout_evidence_failed", "skipped"), + ("abstained", "stale_incumbent", "skipped"), + ("abstained", "governance_invalidated", "skipped"), + ("failed", "infrastructure_failure", "failed"), + ], +) +def test_open_world_terminal_outcomes_are_durable( + storage: SQLiteStorage, + stage: schemas.OptimizationJobStage, + outcome: schemas.OptimizationTerminalOutcome, + expected_status: str, +) -> None: + job = storage.create_playbook_optimization_job(_job()) + claim = storage.claim_playbook_optimization_job( + job_id=job.job_id, + owner="worker-a", + lease_seconds=60, + now=2_000, + ) + + assert storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage=stage, + terminal_outcome=outcome, + now=2_001, + ) + persisted = storage.get_playbook_optimization_job(job.job_id) + assert persisted is not None + assert persisted.status == expected_status + assert persisted.terminal_outcome == outcome + + def test_legacy_optimizer_kind_allowlist_is_rebuilt_for_open_world_job( tmp_path: Path, ) -> None: diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index 79b07b2a..e005ed62 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -253,12 +253,9 @@ def test_stage_advancement_is_linear(storage: BaseStorage) -> None: ("stage", "outcome", "expected_status"), [ ("abstained", "candidate_did_not_improve", "skipped"), - ("abstained", "no_grounded_hypothesis", "skipped"), - ("abstained", "analyst_unqualified", "skipped"), - ("abstained", "heldout_evidence_failed", "skipped"), - ("abstained", "stale_incumbent", "skipped"), - ("abstained", "governance_invalidated", "skipped"), ("failed", "generation_failed", "failed"), + ("failed", "replay_failed", "failed"), + ("failed", "publication_failed", "failed"), ("failed", "infrastructure_failure", "failed"), ], ) @@ -865,8 +862,11 @@ def _claimed_job_at_stage( stage: schemas.OptimizationJobStage, *, now: int, + optimizer_kind: schemas.OptimizerKind = "offline_tuner_replay", ) -> tuple[int, int]: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = _replay_job("d1", "a1") + job.optimizer_kind = optimizer_kind + job = storage.create_or_get_playbook_optimization_job(job) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -883,22 +883,21 @@ def _claimed_job_at_stage( @pytest.mark.parametrize( - ("current_stage", "target_stage"), + ("optimizer_kind", "current_stage", "target_stage"), [ - ("candidate_generated", "discovery_analyzed"), - ("discovery_analyzed", "discovery_analyzed"), - ("evidence_frozen", "held_out_analyzed"), - ("held_out_analyzed", "held_out_analyzed"), - ("held_out_analyzed", "publishing"), - ("discovery_analyzed", "replay_running"), + ("offline_tuner_open_world", "candidate_generated", "replay_running"), + ("offline_tuner_replay", "candidate_generated", "held_out_analyzed"), ], ) -def test_widened_analysis_stage_rejects_invalid_predecessor( +def test_optimizer_kind_rejects_cross_family_stage( storage: BaseStorage, + optimizer_kind: schemas.OptimizerKind, current_stage: schemas.OptimizationJobStage, target_stage: schemas.OptimizationJobStage, ) -> None: - job_id, fence = _claimed_job_at_stage(storage, current_stage, now=7_000) + job_id, fence = _claimed_job_at_stage( + storage, current_stage, now=7_000, optimizer_kind=optimizer_kind + ) assert ( storage.advance_playbook_optimization_stage( @@ -916,21 +915,32 @@ def test_widened_analysis_stage_rejects_invalid_predecessor( @pytest.mark.parametrize( - ("current_stage", "terminal_stage", "outcome"), + ("optimizer_kind", "current_stage", "terminal_stage", "outcome"), [ - ("evidence_frozen", "abstained", "infrastructure_failure"), - ("held_out_analyzed", "abstained", "applied"), - ("discovery_analyzed", "failed", "stale_incumbent"), - ("publishing", "applied", "heldout_evidence_failed"), + ( + "offline_tuner_open_world", + "candidate_generated", + "abstained", + "candidate_did_not_improve", + ), + ( + "offline_tuner_replay", + "candidate_generated", + "abstained", + "no_grounded_hypothesis", + ), ], ) -def test_widened_terminal_stage_rejects_outcome_from_another_family( +def test_optimizer_kind_rejects_cross_family_terminal_outcome( storage: BaseStorage, + optimizer_kind: schemas.OptimizerKind, current_stage: schemas.OptimizationJobStage, terminal_stage: schemas.OptimizationJobStage, outcome: schemas.OptimizationTerminalOutcome, ) -> None: - job_id, fence = _claimed_job_at_stage(storage, current_stage, now=7_000) + job_id, fence = _claimed_job_at_stage( + storage, current_stage, now=7_000, optimizer_kind=optimizer_kind + ) assert ( storage.advance_playbook_optimization_stage( @@ -952,7 +962,12 @@ def test_widened_terminal_stage_rejects_outcome_from_another_family( def test_widened_terminal_outcome_rejects_stale_lease_and_settled_job( storage: BaseStorage, ) -> None: - job_id, fence = _claimed_job_at_stage(storage, "held_out_analyzed", now=7_000) + job_id, fence = _claimed_job_at_stage( + storage, + "held_out_analyzed", + now=7_000, + optimizer_kind="offline_tuner_open_world", + ) assert isinstance(storage, SQLiteStorage) def _abstain(*, at_fence: int, now: int) -> bool: From 7eb59c018692e0fbbe9e76493c9c9a3b275b27c4 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 21:53:44 +0000 Subject: [PATCH 06/13] fix: allow open-world terminal transitions Keep SQLite terminal stage membership aligned with the optimizer-family matrix and cover every stage/outcome combination. --- .../sqlite_storage/playbook/_optimization.py | 35 ++- ...ptimization_replay_contract_integration.py | 279 +++++++++++++----- 2 files changed, 227 insertions(+), 87 deletions(-) diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py index 5329dd40..c4c42331 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py @@ -47,6 +47,21 @@ "held_out_analyzed": ("candidate_generated", "candidate_generated"), }, } +_ACTIVE_STAGES_BY_OPTIMIZER = { + "offline_tuner_replay": ( + "evidence_frozen", + "candidate_generated", + "replay_running", + "replay_evaluated", + "publishing", + ), + "offline_tuner_open_world": ( + "evidence_frozen", + "discovery_analyzed", + "candidate_generated", + "held_out_analyzed", + ), +} _TERMINAL_OUTCOMES_BY_OPTIMIZER = { "offline_tuner_replay": { "failed": { @@ -748,14 +763,14 @@ def advance_playbook_optimization_stage( if row is None: return False optimizer_kind = row["optimizer_kind"] - predecessors = _STAGE_PREDECESSORS_BY_OPTIMIZER.get( - optimizer_kind, {} - ).get(stage) + predecessors = _STAGE_PREDECESSORS_BY_OPTIMIZER.get(optimizer_kind, {}).get( + stage + ) terminal_status: str | None = None if stage == "applied": - if ( - optimizer_kind != "offline_tuner_replay" - or terminal_outcome not in (None, "applied") + if optimizer_kind != "offline_tuner_replay" or terminal_outcome not in ( + None, + "applied", ): return False terminal_outcome = "applied" @@ -816,13 +831,7 @@ def advance_playbook_optimization_stage( ), ) else: - active_stages = tuple( - predecessor - for stages in _STAGE_PREDECESSORS_BY_OPTIMIZER.get( - optimizer_kind, {} - ).values() - for predecessor in stages - ) + active_stages = _ACTIVE_STAGES_BY_OPTIMIZER.get(optimizer_kind, ()) placeholders = ", ".join("?" for _ in active_stages) cur = self.conn.execute( f"""UPDATE playbook_optimization_jobs diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index e005ed62..3ad3be2c 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -7,6 +7,7 @@ from collections.abc import Generator from hashlib import sha256 from pathlib import Path +from typing import cast import pytest from pydantic import ValidationError @@ -25,6 +26,94 @@ pytestmark = pytest.mark.integration +_STAGE_PATHS_BY_OPTIMIZER: dict[str, tuple[str, ...]] = { + "offline_tuner_replay": ( + "evidence_frozen", + "candidate_generated", + "replay_running", + "replay_evaluated", + "publishing", + ), + "offline_tuner_open_world": ( + "evidence_frozen", + "discovery_analyzed", + "candidate_generated", + "held_out_analyzed", + ), +} +_ORDINARY_STAGES = ( + "discovery_analyzed", + "candidate_generated", + "replay_running", + "replay_evaluated", + "held_out_analyzed", + "publishing", +) +_ALL_TERMINAL_OUTCOMES = ( + "applied", + "insufficient_negative_evidence", + "insufficient_positive_evidence", + "insufficient_coverage", + "replay_unsupported", + "deployment_unsupported", + "incomplete_replay_scope", + "insufficient_replay_cases", + "replay_inconclusive", + "candidate_regressed", + "candidate_did_not_improve", + "incumbent_changed", + "generation_failed", + "replay_failed", + "publication_failed", + "governance_erased", + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", + "infrastructure_failure", +) +_TERMINAL_OUTCOMES_BY_OPTIMIZER: dict[str, dict[str, frozenset[str]]] = { + "offline_tuner_replay": { + "failed": frozenset( + { + "generation_failed", + "replay_failed", + "publication_failed", + "infrastructure_failure", + } + ), + "abstained": frozenset( + { + "insufficient_negative_evidence", + "insufficient_positive_evidence", + "insufficient_coverage", + "replay_unsupported", + "deployment_unsupported", + "incomplete_replay_scope", + "insufficient_replay_cases", + "replay_inconclusive", + "candidate_regressed", + "candidate_did_not_improve", + "incumbent_changed", + } + ), + }, + "offline_tuner_open_world": { + "failed": frozenset({"infrastructure_failure"}), + "abstained": frozenset( + { + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", + } + ), + }, +} + + @pytest.fixture def storage(tmp_path: Path) -> Generator[BaseStorage]: store = SQLiteStorage( @@ -863,9 +952,11 @@ def _claimed_job_at_stage( *, now: int, optimizer_kind: schemas.OptimizerKind = "offline_tuner_replay", + target_id: int = 41, ) -> tuple[int, int]: - job = _replay_job("d1", "a1") + job = _replay_job(f"d-{target_id}", f"a-{target_id}") job.optimizer_kind = optimizer_kind + job.target_id = target_id job = storage.create_or_get_playbook_optimization_job(job) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, @@ -882,81 +973,121 @@ def _claimed_job_at_stage( return job.job_id, claim.fence -@pytest.mark.parametrize( - ("optimizer_kind", "current_stage", "target_stage"), - [ - ("offline_tuner_open_world", "candidate_generated", "replay_running"), - ("offline_tuner_replay", "candidate_generated", "held_out_analyzed"), - ], -) -def test_optimizer_kind_rejects_cross_family_stage( - storage: BaseStorage, - optimizer_kind: schemas.OptimizerKind, - current_stage: schemas.OptimizationJobStage, - target_stage: schemas.OptimizationJobStage, -) -> None: - job_id, fence = _claimed_job_at_stage( - storage, current_stage, now=7_000, optimizer_kind=optimizer_kind - ) - - assert ( - storage.advance_playbook_optimization_stage( - job_id=job_id, - fence=fence, - stage=target_stage, - now=7_001, - ) - is False - ) - persisted = storage.get_playbook_optimization_job(job_id) - assert persisted is not None - assert persisted.stage == current_stage - assert persisted.status == "running" - - -@pytest.mark.parametrize( - ("optimizer_kind", "current_stage", "terminal_stage", "outcome"), - [ - ( - "offline_tuner_open_world", - "candidate_generated", - "abstained", - "candidate_did_not_improve", - ), - ( - "offline_tuner_replay", - "candidate_generated", - "abstained", - "no_grounded_hypothesis", - ), - ], -) -def test_optimizer_kind_rejects_cross_family_terminal_outcome( +def test_optimizer_kind_stage_matrix_is_exact(storage: BaseStorage) -> None: + """Reject every non-edge, including every cross-family stage.""" + case = 0 + for optimizer_kind, stages in _STAGE_PATHS_BY_OPTIMIZER.items(): + for current_stage in stages: + for target_stage in _ORDINARY_STAGES: + case += 1 + job_id, fence = _claimed_job_at_stage( + storage, + cast(schemas.OptimizationJobStage, current_stage), + now=7_000, + optimizer_kind=cast(schemas.OptimizerKind, optimizer_kind), + target_id=10_000 + case, + ) + expected = ( + stages.index(target_stage) == stages.index(current_stage) + 1 + if target_stage in stages + else False + ) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=cast(schemas.OptimizationJobStage, target_stage), + now=7_001, + ) + is expected + ) + persisted = storage.get_playbook_optimization_job(job_id) + assert persisted is not None + assert persisted.stage == (target_stage if expected else current_stage) + assert persisted.status == "running" + + +def test_optimizer_kind_terminal_outcome_matrix_is_exact( storage: BaseStorage, - optimizer_kind: schemas.OptimizerKind, - current_stage: schemas.OptimizationJobStage, - terminal_stage: schemas.OptimizationJobStage, - outcome: schemas.OptimizationTerminalOutcome, ) -> None: - job_id, fence = _claimed_job_at_stage( - storage, current_stage, now=7_000, optimizer_kind=optimizer_kind - ) - - assert ( - storage.advance_playbook_optimization_stage( - job_id=job_id, - fence=fence, - stage=terminal_stage, - terminal_outcome=outcome, - now=7_001, - ) - is False - ) - persisted = storage.get_playbook_optimization_job(job_id) - assert persisted is not None - assert persisted.stage == current_stage - assert persisted.status == "running" - assert persisted.terminal_outcome is None + """Reject every terminal outcome assigned to the other family.""" + case = 0 + for optimizer_kind, stages in _STAGE_PATHS_BY_OPTIMIZER.items(): + for current_stage in stages: + for terminal_stage in ("failed", "abstained"): + for outcome in _ALL_TERMINAL_OUTCOMES: + case += 1 + job_id, fence = _claimed_job_at_stage( + storage, + cast(schemas.OptimizationJobStage, current_stage), + now=7_000, + optimizer_kind=cast(schemas.OptimizerKind, optimizer_kind), + target_id=20_000 + case, + ) + expected = ( + outcome + in _TERMINAL_OUTCOMES_BY_OPTIMIZER[optimizer_kind][ + terminal_stage + ] + ) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=cast(schemas.OptimizationJobStage, terminal_stage), + terminal_outcome=cast( + schemas.OptimizationTerminalOutcome, outcome + ), + now=7_001, + ) + is expected + ) + persisted = storage.get_playbook_optimization_job(job_id) + assert persisted is not None + assert persisted.stage == ( + terminal_stage if expected else current_stage + ) + assert persisted.terminal_outcome == (outcome if expected else None) + assert persisted.status == ( + ("failed" if terminal_stage == "failed" else "skipped") + if expected + else "running" + ) + + for outcome in (None, *_ALL_TERMINAL_OUTCOMES): + case += 1 + job_id, fence = _claimed_job_at_stage( + storage, + cast(schemas.OptimizationJobStage, current_stage), + now=7_000, + optimizer_kind=cast(schemas.OptimizerKind, optimizer_kind), + target_id=30_000 + case, + ) + expected = ( + optimizer_kind == "offline_tuner_replay" + and current_stage == "publishing" + and outcome in (None, "applied") + ) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage="applied", + terminal_outcome=cast( + schemas.OptimizationTerminalOutcome | None, outcome + ), + now=7_001, + ) + is expected + ) + persisted = storage.get_playbook_optimization_job(job_id) + assert persisted is not None + assert persisted.stage == ("applied" if expected else current_stage) + assert persisted.terminal_outcome == ("applied" if expected else None) + assert persisted.status == ("completed" if expected else "running") def test_widened_terminal_outcome_rejects_stale_lease_and_settled_job( From 33279b9d111f0d685a9956780da8e7b7f0ddb682 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 22:14:29 +0000 Subject: [PATCH 07/13] test: cover invalid optimizer transitions Exercise malformed stages, outcomes, and inactive optimizer kinds while proving rejected writes leave SQLite jobs unchanged. --- ...ptimization_replay_contract_integration.py | 87 +++++++++++++++++++ 1 file changed, 87 insertions(+) diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index 3ad3be2c..2136af77 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -49,6 +49,19 @@ "held_out_analyzed", "publishing", ) +_INVALID_STAGE_REQUESTS = ( + ("evidence_frozen", None), + ("failed", None), + ("abstained", None), + ("unknown_stage", None), + ("unknown_stage", "unknown_terminal_outcome"), + ("failed", "unknown_terminal_outcome"), + ("abstained", "unknown_terminal_outcome"), + *( + (stage, "infrastructure_failure") + for stage in ("evidence_frozen", *_ORDINARY_STAGES) + ), +) _ALL_TERMINAL_OUTCOMES = ( "applied", "insufficient_negative_evidence", @@ -973,6 +986,80 @@ def _claimed_job_at_stage( return job.job_id, claim.fence +def _optimization_job_row(storage: BaseStorage, job_id: int) -> dict[str, object]: + assert isinstance(storage, SQLiteStorage) + row = storage.conn.execute( + "SELECT * FROM playbook_optimization_jobs WHERE job_id = ?", (job_id,) + ).fetchone() + assert row is not None + return dict(row) + + +def test_invalid_stage_inputs_leave_sqlite_job_unchanged( + storage: BaseStorage, +) -> None: + case = 0 + for optimizer_kind in _STAGE_PATHS_BY_OPTIMIZER: + for stage, terminal_outcome in _INVALID_STAGE_REQUESTS: + case += 1 + job_id, fence = _claimed_job_at_stage( + storage, + "evidence_frozen", + now=7_000, + optimizer_kind=cast(schemas.OptimizerKind, optimizer_kind), + target_id=50_000 + case, + ) + before = _optimization_job_row(storage, job_id) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=cast(schemas.OptimizationJobStage, stage), + terminal_outcome=cast( + schemas.OptimizationTerminalOutcome | None, terminal_outcome + ), + now=7_001, + ) + is False + ) + assert _optimization_job_row(storage, job_id) == before + + for optimizer_kind in ( + "gepa", + "offline_tuner_legacy", + "optimizer_legacy_unknown", + ): + for stage, terminal_outcome in ( + ("candidate_generated", None), + ("failed", "infrastructure_failure"), + ("abstained", "candidate_did_not_improve"), + ): + case += 1 + job_id, fence = _claimed_job_at_stage( + storage, + "evidence_frozen", + now=7_000, + optimizer_kind=cast(schemas.OptimizerKind, optimizer_kind), + target_id=60_000 + case, + ) + before = _optimization_job_row(storage, job_id) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=cast(schemas.OptimizationJobStage, stage), + terminal_outcome=cast( + schemas.OptimizationTerminalOutcome | None, terminal_outcome + ), + now=7_001, + ) + is False + ) + assert _optimization_job_row(storage, job_id) == before + + def test_optimizer_kind_stage_matrix_is_exact(storage: BaseStorage) -> None: """Reject every non-edge, including every cross-family stage.""" case = 0 From 7961955e90b7c58ccdc0ada399569ba1e3b5af88 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 22:30:07 +0000 Subject: [PATCH 08/13] test: preserve rejected optimizer job rows --- ...ok_optimization_replay_contract_integration.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index 2136af77..a1b2fb26 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -1079,6 +1079,9 @@ def test_optimizer_kind_stage_matrix_is_exact(storage: BaseStorage) -> None: if target_stage in stages else False ) + before = ( + _optimization_job_row(storage, job_id) if not expected else None + ) assert ( storage.advance_playbook_optimization_stage( @@ -1089,6 +1092,8 @@ def test_optimizer_kind_stage_matrix_is_exact(storage: BaseStorage) -> None: ) is expected ) + if before is not None: + assert _optimization_job_row(storage, job_id) == before persisted = storage.get_playbook_optimization_job(job_id) assert persisted is not None assert persisted.stage == (target_stage if expected else current_stage) @@ -1118,6 +1123,9 @@ def test_optimizer_kind_terminal_outcome_matrix_is_exact( terminal_stage ] ) + before = ( + _optimization_job_row(storage, job_id) if not expected else None + ) assert ( storage.advance_playbook_optimization_stage( @@ -1131,6 +1139,8 @@ def test_optimizer_kind_terminal_outcome_matrix_is_exact( ) is expected ) + if before is not None: + assert _optimization_job_row(storage, job_id) == before persisted = storage.get_playbook_optimization_job(job_id) assert persisted is not None assert persisted.stage == ( @@ -1157,6 +1167,9 @@ def test_optimizer_kind_terminal_outcome_matrix_is_exact( and current_stage == "publishing" and outcome in (None, "applied") ) + before = ( + _optimization_job_row(storage, job_id) if not expected else None + ) assert ( storage.advance_playbook_optimization_stage( @@ -1170,6 +1183,8 @@ def test_optimizer_kind_terminal_outcome_matrix_is_exact( ) is expected ) + if before is not None: + assert _optimization_job_row(storage, job_id) == before persisted = storage.get_playbook_optimization_job(job_id) assert persisted is not None assert persisted.stage == ("applied" if expected else current_stage) From b5a7dfb7c6e9656f0c6c0e63bc04bc85cae5041d Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 18 Aug 2026 23:56:48 +0000 Subject: [PATCH 09/13] feat(storage): cache open-world analyst qualification Add the shared strict qualification-result record and its immutable tenant-local SQLite cache, keyed by (component_identity_digest, suite_digest). The record pins the exact v1 schema version, all seven safety-critical class counts in canonical order (nonnegative, passed_required never above required), SHA-256 component/suite/result digests, and sorted, unique observation digests. It carries no customer data or model output. Persistence is idempotent for a semantically exact replay and rejects a semantic conflict with OpenWorldQualificationConflictError; the first insert controls created_at, which is excluded from conflict equality. Database triggers reject raw UPDATE and DELETE so a cached qualification cannot be edited out from under an attempt. --- reflexio/models/api_schema/domain/entities.py | 139 +++++++++ reflexio/server/services/storage/error.py | 4 + .../services/storage/sqlite_storage/_base.py | 22 ++ .../sqlite_storage/playbook/_optimization.py | 91 ++++++ .../storage_base/playbook/_optimization.py | 44 +++ .../test_open_world_qualification_record.py | 151 ++++++++++ .../test_open_world_qualification_cache.py | 272 ++++++++++++++++++ 7 files changed, 723 insertions(+) create mode 100644 tests/models/test_open_world_qualification_record.py create mode 100644 tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 065c2d1e..c91d843f 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -7,6 +7,7 @@ from pydantic import ( BaseModel, + ConfigDict, Field, PrivateAttr, field_validator, @@ -122,6 +123,11 @@ "PlaybookOptimizationCandidate", "PlaybookOptimizationEvaluation", "PlaybookOptimizationEvent", + "OpenWorldQualificationClass", + "OPEN_WORLD_QUALIFICATION_CLASSES", + "OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION", + "OpenWorldQualificationClassCount", + "OpenWorldQualificationRecord", "AgentPlaybookSourceWindow", "agent_playbook_to_snapshot", "RunPlaybookAggregationRequest", @@ -614,6 +620,139 @@ class PlaybookOptimizationEvent(BaseModel): created_at: int = Field(default_factory=lambda: int(datetime.now(UTC).timestamp())) +OpenWorldQualificationClass = Literal[ + "citation_fidelity", + "abstention", + "support", + "refutation", + "insufficiency", + "unsupported_causal_claim_rejection", + "prompt_injection_resistance", +] + +OPEN_WORLD_QUALIFICATION_CLASSES: Final[tuple[OpenWorldQualificationClass, ...]] = ( + "citation_fidelity", + "abstention", + "support", + "refutation", + "insufficiency", + "unsupported_causal_claim_rejection", + "prompt_injection_resistance", +) + +OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION: Final[str] = ( + "offline-tuner-open-world-qualification-result-v1" +) + + +def _validate_lowercase_sha256(label: str, value: str) -> str: + """Return ``value`` when it is a lowercase SHA-256 hex digest. + + Args: + label (str): Field name used in the raised error message. + value (str): Candidate digest. + + Returns: + str: The validated digest. + + Raises: + ValueError: If ``value`` is not 64 lowercase hex characters. + """ + if len(value) != 64 or any(char not in "0123456789abcdef" for char in value): + raise ValueError(f"{label} must be lowercase SHA-256 hex") + return value + + +class OpenWorldQualificationClassCount(BaseModel): + """Diagnostic required/passed counts for one safety-critical class. + + Counts are never combined into a score: pass-all qualification is decided + by the reducer, and these values exist only to explain one result. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + qualification_class: OpenWorldQualificationClass + required: int = Field(ge=0) + passed_required: int = Field(ge=0) + + @model_validator(mode="after") + def validate_passed_within_required(self) -> Self: + if self.passed_required > self.required: + raise ValueError("qualification passed_required may not exceed required") + return self + + +class OpenWorldQualificationRecord(BaseModel): + """One immutable pass-all qualification result for an analyst identity. + + The record carries no customer data or model output: only the pinned + component identity, the suite it was measured against, the canonical + result digest, per-class diagnostic counts for every one of the seven + safety-critical classes in canonical order, and the sorted, unique + digests of the observations that produced it. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["offline-tuner-open-world-qualification-result-v1"] = ( + "offline-tuner-open-world-qualification-result-v1" + ) + component_identity_digest: Sha256Digest + suite_digest: Sha256Digest + result_digest: Sha256Digest + class_counts: tuple[OpenWorldQualificationClassCount, ...] + passed: bool + observation_digests: tuple[Sha256Digest, ...] = () + created_at: int = Field( + default_factory=lambda: int(datetime.now(UTC).timestamp()), ge=0 + ) + + @field_validator("component_identity_digest", "suite_digest", "result_digest") + @classmethod + def validate_identity_digests(cls, value: str) -> str: + return _validate_lowercase_sha256("qualification digest", value) + + @field_validator("class_counts") + @classmethod + def validate_class_counts( + cls, value: tuple[OpenWorldQualificationClassCount, ...] + ) -> tuple[OpenWorldQualificationClassCount, ...]: + observed = tuple(count.qualification_class for count in value) + if observed != OPEN_WORLD_QUALIFICATION_CLASSES: + raise ValueError( + "qualification class_counts must list every safety-critical " + "class exactly once in canonical order" + ) + return value + + @field_validator("observation_digests") + @classmethod + def validate_observation_digests(cls, value: tuple[str, ...]) -> tuple[str, ...]: + for digest in value: + _validate_lowercase_sha256("qualification observation digest", digest) + if list(value) != sorted(set(value)): + raise ValueError( + "qualification observation digests must be sorted and unique" + ) + return value + + def semantic_key(self) -> tuple[Any, ...]: + """Return the conflict-detection identity, excluding ``created_at``.""" + return ( + self.schema_version, + self.component_identity_digest, + self.suite_digest, + self.result_digest, + tuple( + (count.qualification_class, count.required, count.passed_required) + for count in self.class_counts + ), + self.passed, + self.observation_digests, + ) + + class AgentPlaybookSourceWindow(BaseModel): """Replayable source window snapshotted when an agent playbook is generated.""" diff --git a/reflexio/server/services/storage/error.py b/reflexio/server/services/storage/error.py index fe14734b..3a04c301 100644 --- a/reflexio/server/services/storage/error.py +++ b/reflexio/server/services/storage/error.py @@ -30,6 +30,10 @@ class OptimizationArtifactIntegrityError(StorageError): """Raised when a durable optimizer artifact is malformed or conflicts.""" +class OpenWorldQualificationConflictError(StorageError): + """Raised when a cached qualification key resolves to a conflicting result.""" + + def require_non_empty_session_id(value: Any) -> str: """Return a stripped, non-empty request ``session_id`` or raise ``StorageError``. diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index eb183991..29133d7d 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -3543,6 +3543,28 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: CREATE INDEX IF NOT EXISTS idx_poa_job ON playbook_optimization_artifacts(job_id); +CREATE TABLE IF NOT EXISTS offline_tuner_open_world_qualifications ( + component_identity_digest TEXT NOT NULL, + suite_digest TEXT NOT NULL, + schema_version TEXT NOT NULL, + result_digest TEXT NOT NULL, + passed INTEGER NOT NULL CHECK (passed IN (0, 1)), + class_counts_json TEXT NOT NULL, + observation_digests_json TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (component_identity_digest, suite_digest) +); +CREATE TRIGGER IF NOT EXISTS offline_tuner_open_world_qualifications_no_update +BEFORE UPDATE ON offline_tuner_open_world_qualifications +BEGIN + SELECT RAISE(ABORT, 'open-world qualification records are immutable'); +END; +CREATE TRIGGER IF NOT EXISTS offline_tuner_open_world_qualifications_no_delete +BEFORE DELETE ON offline_tuner_open_world_qualifications +BEGIN + SELECT RAISE(ABORT, 'open-world qualification records are immutable'); +END; + CREATE TABLE IF NOT EXISTS playbook_optimization_candidates ( candidate_id INTEGER PRIMARY KEY AUTOINCREMENT, job_id INTEGER NOT NULL, diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py index c4c42331..8b16cbe2 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py @@ -10,6 +10,8 @@ from reflexio.models.api_schema.domain.entities import canonicalize_artifact_json from reflexio.models.api_schema.service_schemas import ( + OpenWorldQualificationClassCount, + OpenWorldQualificationRecord, OptimizationArtifactKind, OptimizationJobClaim, OptimizationJobStage, @@ -27,6 +29,7 @@ canonical_json_bytes, ) from reflexio.server.services.storage.error import ( + OpenWorldQualificationConflictError, OptimizationArtifactIntegrityError, OptimizationJobIdentityConflictError, OptimizationJobLeaseLiveError, @@ -253,6 +256,47 @@ def _validate_gepa_publication_prepare_metadata( _json_loads, ) +_QUALIFICATION_SELECT_SQL = """SELECT * FROM offline_tuner_open_world_qualifications + WHERE component_identity_digest = ? AND suite_digest = ?""" + +_QUALIFICATION_INSERT_SQL = """INSERT INTO offline_tuner_open_world_qualifications + (component_identity_digest, suite_digest, schema_version, result_digest, + passed, class_counts_json, observation_digests_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""" + + +def _qualification_insert_values( + record: OpenWorldQualificationRecord, +) -> tuple[Any, ...]: + return ( + record.component_identity_digest, + record.suite_digest, + record.schema_version, + record.result_digest, + int(record.passed), + json.dumps([count.model_dump() for count in record.class_counts]), + json.dumps(list(record.observation_digests)), + record.created_at, + ) + + +def _row_to_open_world_qualification_record( + row: sqlite3.Row, +) -> OpenWorldQualificationRecord: + return OpenWorldQualificationRecord( + schema_version=row["schema_version"], + component_identity_digest=row["component_identity_digest"], + suite_digest=row["suite_digest"], + result_digest=row["result_digest"], + class_counts=tuple( + OpenWorldQualificationClassCount(**count) + for count in json.loads(row["class_counts_json"]) + ), + passed=bool(row["passed"]), + observation_digests=tuple(json.loads(row["observation_digests_json"])), + created_at=row["created_at"], + ) + def _row_to_playbook_optimization_candidate( row: sqlite3.Row, @@ -941,6 +985,53 @@ def upsert_playbook_optimization_artifact( self.conn.rollback() raise + @SQLiteStorageBase.handle_exceptions + def persist_open_world_qualification_record( + self, record: OpenWorldQualificationRecord + ) -> OpenWorldQualificationRecord: + with self._lock: + owns_transaction = self._own_transaction() + if owns_transaction: + self.conn.execute("BEGIN IMMEDIATE") + try: + existing = self.conn.execute( + _QUALIFICATION_SELECT_SQL, + (record.component_identity_digest, record.suite_digest), + ).fetchone() + if existing is not None: + stored = _row_to_open_world_qualification_record(existing) + if stored.semantic_key() != record.semantic_key(): + raise OpenWorldQualificationConflictError( + "open-world qualification record conflicts with the " + "record already cached for this identity and suite" + ) + else: + self.conn.execute( + _QUALIFICATION_INSERT_SQL, + _qualification_insert_values(record), + ) + stored = record + if owns_transaction: + self.conn.commit() + return stored + except Exception: + if owns_transaction: + self.conn.rollback() + raise + + @SQLiteStorageBase.handle_exceptions + def load_open_world_qualification_record( + self, + *, + component_identity_digest: str, + suite_digest: str, + ) -> OpenWorldQualificationRecord | None: + row = self._fetchone( + _QUALIFICATION_SELECT_SQL, + (component_identity_digest, suite_digest), + ) + return None if row is None else _row_to_open_world_qualification_record(row) + @SQLiteStorageBase.handle_exceptions def get_playbook_optimization_artifact( self, diff --git a/reflexio/server/services/storage/storage_base/playbook/_optimization.py b/reflexio/server/services/storage/storage_base/playbook/_optimization.py index d3ae42e3..0e6fcbb1 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_optimization.py +++ b/reflexio/server/services/storage/storage_base/playbook/_optimization.py @@ -3,6 +3,7 @@ from abc import abstractmethod from reflexio.models.api_schema.domain import ( + OpenWorldQualificationRecord, OptimizationArtifactKind, OptimizationJobClaim, OptimizationJobStage, @@ -169,6 +170,49 @@ def get_playbook_optimization_artifact( "Storage backend does not support replay-gated playbook optimization" ) + def persist_open_world_qualification_record( + self, record: OpenWorldQualificationRecord + ) -> OpenWorldQualificationRecord: + """Persist one immutable qualification result and return the stored row. + + The cache key is ``(component_identity_digest, suite_digest)``. The + first insert controls ``created_at``; replaying a semantically + identical record is idempotent and returns the stored row unchanged. + + Args: + record (OpenWorldQualificationRecord): The result to persist. + + Returns: + OpenWorldQualificationRecord: The durable record for this key. + + Raises: + OpenWorldQualificationConflictError: If a record already exists for + the key and differs in any field other than ``created_at``. + """ + raise NotImplementedError( + "Storage backend does not support open-world analyst qualification" + ) + + def load_open_world_qualification_record( + self, + *, + component_identity_digest: str, + suite_digest: str, + ) -> OpenWorldQualificationRecord | None: + """Load the cached qualification result for one exact identity/suite pair. + + Args: + component_identity_digest (str): Pinned analyst component identity. + suite_digest (str): Canonical qualification-suite digest. + + Returns: + OpenWorldQualificationRecord | None: The stored record, or ``None`` + when this key has never been qualified. + """ + raise NotImplementedError( + "Storage backend does not support open-world analyst qualification" + ) + @abstractmethod def update_playbook_optimization_job( self, diff --git a/tests/models/test_open_world_qualification_record.py b/tests/models/test_open_world_qualification_record.py new file mode 100644 index 00000000..6fc6307e --- /dev/null +++ b/tests/models/test_open_world_qualification_record.py @@ -0,0 +1,151 @@ +"""Strictness coverage for the shared open-world qualification record.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from reflexio.models.api_schema.domain.entities import ( + OPEN_WORLD_QUALIFICATION_CLASSES, + OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION, + OpenWorldQualificationClassCount, + OpenWorldQualificationRecord, +) + +_COMPONENT_DIGEST = "a" * 64 +_SUITE_DIGEST = "b" * 64 +_RESULT_DIGEST = "c" * 64 + + +def _class_counts( + **overrides: tuple[int, int], +) -> tuple[OpenWorldQualificationClassCount, ...]: + return tuple( + OpenWorldQualificationClassCount( + qualification_class=qualification_class, + required=overrides.get(qualification_class, (2, 2))[0], + passed_required=overrides.get(qualification_class, (2, 2))[1], + ) + for qualification_class in OPEN_WORLD_QUALIFICATION_CLASSES + ) + + +def _record(**overrides: object) -> OpenWorldQualificationRecord: + payload: dict[str, object] = { + "component_identity_digest": _COMPONENT_DIGEST, + "suite_digest": _SUITE_DIGEST, + "result_digest": _RESULT_DIGEST, + "class_counts": _class_counts(), + "passed": True, + "observation_digests": ("0" * 64, "1" * 64), + "created_at": 1_700_000_000, + } + payload.update(overrides) + return OpenWorldQualificationRecord(**payload) # type: ignore[arg-type] + + +def test_seven_ordered_classes_round_trip() -> None: + record = _record() + + assert record.schema_version == OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION + assert record.schema_version == "offline-tuner-open-world-qualification-result-v1" + assert len(OPEN_WORLD_QUALIFICATION_CLASSES) == 7 + assert OPEN_WORLD_QUALIFICATION_CLASSES == ( + "citation_fidelity", + "abstention", + "support", + "refutation", + "insufficiency", + "unsupported_causal_claim_rejection", + "prompt_injection_resistance", + ) + assert tuple(count.qualification_class for count in record.class_counts) == ( + OPEN_WORLD_QUALIFICATION_CLASSES + ) + + +def test_schema_version_must_be_the_exact_v1_value() -> None: + with pytest.raises(ValidationError): + _record(schema_version="offline-tuner-open-world-qualification-result-v2") + + +def test_unknown_field_is_rejected() -> None: + with pytest.raises(ValidationError): + _record(rationale="free-form model prose") + + +def test_missing_class_is_rejected() -> None: + with pytest.raises(ValidationError): + _record(class_counts=_class_counts()[:-1]) + + +def test_reordered_classes_are_rejected() -> None: + reordered = _class_counts() + with pytest.raises(ValidationError): + _record(class_counts=(reordered[1], reordered[0], *reordered[2:])) + + +def test_duplicated_class_is_rejected() -> None: + counts = _class_counts() + with pytest.raises(ValidationError): + _record(class_counts=(counts[0], *counts)) + + +def test_negative_counts_are_rejected() -> None: + with pytest.raises(ValidationError): + _record(class_counts=_class_counts(abstention=(-1, 0))) + with pytest.raises(ValidationError): + _record(class_counts=_class_counts(abstention=(2, -1))) + + +def test_passed_required_may_not_exceed_required() -> None: + with pytest.raises(ValidationError): + _record(class_counts=_class_counts(support=(1, 2))) + + +def test_zero_required_class_is_accepted() -> None: + record = _record(class_counts=_class_counts(refutation=(0, 0))) + + assert record.class_counts[3].required == 0 + + +@pytest.mark.parametrize("field", ["component_identity_digest", "suite_digest"]) +def test_identity_digests_must_be_sha256(field: str) -> None: + with pytest.raises(ValidationError): + _record(**{field: "not-a-digest"}) + + +def test_result_digest_must_be_sha256() -> None: + with pytest.raises(ValidationError): + _record(result_digest="C" * 64) + + +def test_observation_digests_must_be_sha256() -> None: + with pytest.raises(ValidationError): + _record(observation_digests=("0" * 63,)) + + +def test_observation_digests_must_be_sorted() -> None: + with pytest.raises(ValidationError): + _record(observation_digests=("1" * 64, "0" * 64)) + + +def test_observation_digests_must_be_unique() -> None: + with pytest.raises(ValidationError): + _record(observation_digests=("0" * 64, "0" * 64)) + + +def test_empty_observation_digests_are_accepted() -> None: + assert _record(observation_digests=()).observation_digests == () + + +def test_negative_created_at_is_rejected() -> None: + with pytest.raises(ValidationError): + _record(created_at=-1) + + +def test_record_is_immutable() -> None: + record = _record() + + with pytest.raises(ValidationError): + record.passed = False # type: ignore[misc] diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py new file mode 100644 index 00000000..c78cbc9d --- /dev/null +++ b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py @@ -0,0 +1,272 @@ +"""SQLite contract for the immutable open-world qualification cache.""" + +from __future__ import annotations + +import sqlite3 +from collections.abc import Generator +from pathlib import Path + +import pytest + +from reflexio.models.api_schema import service_schemas as schemas +from reflexio.server.services.storage.error import ( + OpenWorldQualificationConflictError, +) +from reflexio.server.services.storage.sqlite_storage import SQLiteStorage +from reflexio.server.services.storage.storage_base import BaseStorage + +pytestmark = pytest.mark.integration + +_COMPONENT_DIGEST = "a" * 64 +_SUITE_DIGEST = "b" * 64 +_RESULT_DIGEST = "c" * 64 + + +@pytest.fixture +def db_path(tmp_path: Path) -> str: + return str(tmp_path / "reflexio.db") + + +@pytest.fixture +def storage(db_path: str) -> Generator[SQLiteStorage]: + store = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + yield store + finally: + store.conn.close() + + +def _class_counts() -> tuple[schemas.OpenWorldQualificationClassCount, ...]: + return tuple( + schemas.OpenWorldQualificationClassCount( + qualification_class=qualification_class, + required=3, + passed_required=3, + ) + for qualification_class in schemas.OPEN_WORLD_QUALIFICATION_CLASSES + ) + + +def _record(**overrides: object) -> schemas.OpenWorldQualificationRecord: + payload: dict[str, object] = { + "component_identity_digest": _COMPONENT_DIGEST, + "suite_digest": _SUITE_DIGEST, + "result_digest": _RESULT_DIGEST, + "class_counts": _class_counts(), + "passed": True, + "observation_digests": ("0" * 64, "1" * 64), + "created_at": 1_700_000_000, + } + payload.update(overrides) + return schemas.OpenWorldQualificationRecord(**payload) # type: ignore[arg-type] + + +def test_persist_then_load_round_trips(storage: SQLiteStorage) -> None: + persisted = storage.persist_open_world_qualification_record(_record()) + + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + + assert persisted == _record() + assert loaded == persisted + + +def test_load_returns_none_for_unknown_key(storage: SQLiteStorage) -> None: + storage.persist_open_world_qualification_record(_record()) + + assert ( + storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest="d" * 64, + ) + is None + ) + assert ( + storage.load_open_world_qualification_record( + component_identity_digest="d" * 64, + suite_digest=_SUITE_DIGEST, + ) + is None + ) + + +def test_failed_result_is_persisted_and_loadable(storage: SQLiteStorage) -> None: + failing = _class_counts()[:-1] + ( + schemas.OpenWorldQualificationClassCount( + qualification_class="prompt_injection_resistance", + required=3, + passed_required=2, + ), + ) + record = _record(class_counts=failing, passed=False) + + storage.persist_open_world_qualification_record(record) + + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + assert loaded is not None + assert loaded.passed is False + assert loaded.class_counts[-1].passed_required == 2 + + +def test_exact_replay_is_idempotent(storage: SQLiteStorage) -> None: + first = storage.persist_open_world_qualification_record(_record()) + second = storage.persist_open_world_qualification_record(_record()) + + assert second == first + rows = storage.conn.execute( + "SELECT COUNT(*) FROM offline_tuner_open_world_qualifications" + ).fetchone()[0] + assert rows == 1 + + +def test_first_insert_controls_created_at(storage: SQLiteStorage) -> None: + first = storage.persist_open_world_qualification_record( + _record(created_at=1_700_000_000) + ) + replay = storage.persist_open_world_qualification_record( + _record(created_at=1_900_000_000) + ) + + assert first.created_at == 1_700_000_000 + assert replay.created_at == 1_700_000_000 + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + assert loaded is not None + assert loaded.created_at == 1_700_000_000 + + +@pytest.mark.parametrize( + "conflicting", + [ + {"result_digest": "d" * 64}, + {"passed": False}, + {"observation_digests": ("0" * 64,)}, + { + "class_counts": _class_counts()[:-1] + + ( + schemas.OpenWorldQualificationClassCount( + qualification_class="prompt_injection_resistance", + required=4, + passed_required=4, + ), + ) + }, + ], +) +def test_semantic_conflict_is_rejected( + storage: SQLiteStorage, + conflicting: dict[str, object], +) -> None: + storage.persist_open_world_qualification_record(_record()) + + with pytest.raises(OpenWorldQualificationConflictError): + storage.persist_open_world_qualification_record(_record(**conflicting)) + + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + assert loaded == _record() + + +def test_distinct_keys_are_independent(storage: SQLiteStorage) -> None: + storage.persist_open_world_qualification_record(_record()) + other = _record(suite_digest="d" * 64, result_digest="e" * 64) + + storage.persist_open_world_qualification_record(other) + + assert ( + storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest="d" * 64, + ) + == other + ) + + +def test_record_survives_close_and_reopen(db_path: str) -> None: + store = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + store.persist_open_world_qualification_record(_record()) + finally: + store.conn.close() + + reopened = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + loaded = reopened.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + finally: + reopened.conn.close() + + assert loaded == _record() + + +def test_raw_sql_update_is_rejected(storage: SQLiteStorage) -> None: + storage.persist_open_world_qualification_record(_record()) + + with pytest.raises(sqlite3.IntegrityError): + storage.conn.execute( + "UPDATE offline_tuner_open_world_qualifications SET passed = 0" + ) + storage.conn.rollback() + + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + assert loaded == _record() + + +def test_raw_sql_delete_is_rejected(storage: SQLiteStorage) -> None: + storage.persist_open_world_qualification_record(_record()) + + with pytest.raises(sqlite3.IntegrityError): + storage.conn.execute("DELETE FROM offline_tuner_open_world_qualifications") + storage.conn.rollback() + + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + assert loaded == _record() + + +def test_immutability_triggers_survive_reopen(db_path: str) -> None: + store = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + store.persist_open_world_qualification_record(_record()) + finally: + store.conn.close() + + reopened = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + with pytest.raises(sqlite3.IntegrityError): + reopened.conn.execute("DELETE FROM offline_tuner_open_world_qualifications") + reopened.conn.rollback() + finally: + reopened.conn.close() + + +def test_qualification_cache_surface_is_narrow() -> None: + for method_name in ( + "persist_open_world_qualification_record", + "load_open_world_qualification_record", + ): + assert hasattr(BaseStorage, method_name) + assert hasattr(SQLiteStorage, method_name) + for method_name in ( + "delete_open_world_qualification_record", + "update_open_world_qualification_record", + "list_open_world_qualification_records", + ): + assert not hasattr(BaseStorage, method_name) + assert not hasattr(SQLiteStorage, method_name) From 961d691867b2496332c9413878ae58376f940656 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 19 Aug 2026 00:20:12 +0000 Subject: [PATCH 10/13] test(storage): strict open-world qualification models, concurrent writers, upgrade Add strict=True to OpenWorldQualificationClassCount/Record so bool-as-int, int-as-bool, numeric strings, and floats are rejected instead of silently coerced. Add independent-connection concurrent-writer tests (identical writes converge, conflicting writes yield one winner + N-1 typed conflicts + one intact row) and an upgrade test proving normal SQLiteStorage init reinstalls the qualification table and both immutability triggers on a pre-Task-9 database, leaving unrelated data intact. --- reflexio/models/api_schema/domain/entities.py | 4 +- .../test_open_world_qualification_record.py | 88 +++++++++ .../test_open_world_qualification_cache.py | 183 ++++++++++++++++++ 3 files changed, 273 insertions(+), 2 deletions(-) diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index c91d843f..dbe5e746 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -670,7 +670,7 @@ class OpenWorldQualificationClassCount(BaseModel): by the reducer, and these values exist only to explain one result. """ - model_config = ConfigDict(extra="forbid", frozen=True) + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) qualification_class: OpenWorldQualificationClass required: int = Field(ge=0) @@ -693,7 +693,7 @@ class OpenWorldQualificationRecord(BaseModel): digests of the observations that produced it. """ - model_config = ConfigDict(extra="forbid", frozen=True) + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) schema_version: Literal["offline-tuner-open-world-qualification-result-v1"] = ( "offline-tuner-open-world-qualification-result-v1" diff --git a/tests/models/test_open_world_qualification_record.py b/tests/models/test_open_world_qualification_record.py index 6fc6307e..60a230b4 100644 --- a/tests/models/test_open_world_qualification_record.py +++ b/tests/models/test_open_world_qualification_record.py @@ -149,3 +149,91 @@ def test_record_is_immutable() -> None: with pytest.raises(ValidationError): record.passed = False # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# Strictness: coercible-but-wrong-typed values must be rejected outright, +# not silently converted. Pydantic's default (lax) mode would happily accept +# every value below; these guard the ``strict=True`` model config. +# --------------------------------------------------------------------------- + + +def test_bool_is_rejected_for_class_count_required() -> None: + with pytest.raises(ValidationError): + OpenWorldQualificationClassCount( + qualification_class="abstention", required=True, passed_required=0 + ) + + +def test_bool_is_rejected_for_class_count_passed_required() -> None: + with pytest.raises(ValidationError): + OpenWorldQualificationClassCount( + qualification_class="abstention", required=2, passed_required=False + ) + + +def test_numeric_string_is_rejected_for_class_count_required() -> None: + with pytest.raises(ValidationError): + OpenWorldQualificationClassCount( + qualification_class="abstention", + required="2", # type: ignore[arg-type] + passed_required=0, + ) + + +def test_numeric_string_is_rejected_for_class_count_passed_required() -> None: + with pytest.raises(ValidationError): + OpenWorldQualificationClassCount( + qualification_class="abstention", + required=2, + passed_required="2", # type: ignore[arg-type] + ) + + +def test_float_is_rejected_for_class_count_required() -> None: + with pytest.raises(ValidationError): + OpenWorldQualificationClassCount( + qualification_class="abstention", + required=2.0, # type: ignore[arg-type] + passed_required=0, + ) + + +def test_int_is_rejected_for_record_passed() -> None: + with pytest.raises(ValidationError): + _record(passed=1) + + +def test_numeric_string_is_rejected_for_record_passed() -> None: + with pytest.raises(ValidationError): + _record(passed="true") + + +def test_numeric_string_is_rejected_for_created_at() -> None: + with pytest.raises(ValidationError): + _record(created_at="1700000000") + + +def test_bool_is_rejected_for_created_at() -> None: + with pytest.raises(ValidationError): + _record(created_at=True) + + +def test_float_is_rejected_for_created_at() -> None: + with pytest.raises(ValidationError): + _record(created_at=1_700_000_000.0) + + +def test_int_is_rejected_for_digest_fields() -> None: + with pytest.raises(ValidationError): + _record(component_identity_digest=int(_COMPONENT_DIGEST, 16)) # type: ignore[arg-type] + + +def test_list_is_rejected_for_class_counts() -> None: + with pytest.raises(ValidationError): + _record(class_counts=list(_class_counts())) # type: ignore[arg-type] + + +def test_list_is_rejected_for_observation_digests() -> None: + with pytest.raises(ValidationError): + _record(observation_digests=["0" * 64, "1" * 64]) # type: ignore[arg-type] diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py index c78cbc9d..e5f47ed3 100644 --- a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py +++ b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py @@ -4,6 +4,7 @@ import sqlite3 from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor from pathlib import Path import pytest @@ -21,6 +22,10 @@ _SUITE_DIGEST = "b" * 64 _RESULT_DIGEST = "c" * 64 +# Bounded concurrency for the independent-connection writer tests below: enough +# threads to exercise real SQLite lock contention without saturating the host. +_CONCURRENT_WRITERS = 6 + @pytest.fixture def db_path(tmp_path: Path) -> str: @@ -256,6 +261,184 @@ def test_immutability_triggers_survive_reopen(db_path: str) -> None: reopened.conn.close() +# --------------------------------------------------------------------------- +# Concurrent writers on INDEPENDENT SQLite connections (separate SQLiteStorage +# instances over the same db file, not one instance shared across threads) -- +# real cross-connection lock contention, not just the in-process RLock. +# Synchronization is via ThreadPoolExecutor.submit()/.result(); no sleeps. +# --------------------------------------------------------------------------- + + +def test_concurrent_identical_writes_converge_on_independent_connections( + db_path: str, +) -> None: + stores = [ + SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + for _ in range(_CONCURRENT_WRITERS) + ] + try: + with ThreadPoolExecutor(max_workers=_CONCURRENT_WRITERS) as pool: + futures = [ + pool.submit(store.persist_open_world_qualification_record, _record()) + for store in stores + ] + results = [future.result() for future in futures] + finally: + for store in stores: + store.conn.close() + + assert all(result == _record() for result in results) + + verify_store = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + rows = verify_store.conn.execute( + "SELECT COUNT(*) FROM offline_tuner_open_world_qualifications" + ).fetchone()[0] + loaded = verify_store.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + finally: + verify_store.conn.close() + assert rows == 1 + assert loaded == _record() + + +def test_concurrent_conflicting_writes_yield_one_winner_and_one_intact_row( + db_path: str, +) -> None: + variants = [ + _record(result_digest=f"{index:064x}") for index in range(_CONCURRENT_WRITERS) + ] + stores = [ + SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + for _ in variants + ] + try: + with ThreadPoolExecutor(max_workers=_CONCURRENT_WRITERS) as pool: + futures = [ + pool.submit(store.persist_open_world_qualification_record, variant) + for store, variant in zip(stores, variants, strict=True) + ] + outcomes: list[tuple[str, object]] = [] + for future in futures: + try: + outcomes.append(("winner", future.result())) + except OpenWorldQualificationConflictError as exc: + outcomes.append(("conflict", exc)) + finally: + for store in stores: + store.conn.close() + + winners = [record for kind, record in outcomes if kind == "winner"] + conflicts = [exc for kind, exc in outcomes if kind == "conflict"] + + assert len(winners) == 1 + assert len(conflicts) == len(variants) - 1 + assert all( + isinstance(exc, OpenWorldQualificationConflictError) for exc in conflicts + ) + assert winners[0] in variants + + verify_store = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + rows = verify_store.conn.execute( + "SELECT COUNT(*) FROM offline_tuner_open_world_qualifications" + ).fetchone()[0] + loaded = verify_store.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + finally: + verify_store.conn.close() + assert rows == 1 + assert loaded == winners[0] + + +# --------------------------------------------------------------------------- +# Upgrade from a pre-Task-9 schema: a DB that predates this table must +# self-upgrade on normal storage initialization, in place, with existing data +# intact. Dropping the table (SQLite auto-drops its triggers with it) on a +# fully-initialized modern schema reproduces exactly what an old DB on disk +# looks like, without hand-reconstructing legacy DDL. +# --------------------------------------------------------------------------- + + +def _qualification_schema_objects( + conn: sqlite3.Connection, +) -> tuple[set[str], set[str]]: + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master" + " WHERE type = 'table' AND name = 'offline_tuner_open_world_qualifications'" + ) + } + triggers = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master" + " WHERE type = 'trigger'" + " AND name LIKE 'offline_tuner_open_world_qualifications_%'" + ) + } + return tables, triggers + + +def test_upgrade_from_pre_task_9_schema_installs_table_and_triggers( + db_path: str, +) -> None: + baseline = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + # optimizer_kind="gepa" (not a legacy kind) so re-running migrate() on + # the second, upgraded instance does not itself retire this sentinel + # job as part of unrelated legacy-optimizer cleanup -- that would + # confound this test's "unrelated data survives" assertion below. + sentinel_job = baseline.create_playbook_optimization_job( + schemas.PlaybookOptimizationJob( + optimizer_kind="gepa", target_kind="agent_playbook", target_id=1 + ) + ) + finally: + baseline.conn.close() + + raw = sqlite3.connect(db_path) + try: + raw.execute("DROP TABLE offline_tuner_open_world_qualifications") + raw.commit() + assert _qualification_schema_objects(raw) == (set(), set()) + finally: + raw.close() + + upgraded = SQLiteStorage(org_id="open-world-qualification", db_path=db_path) + try: + tables, triggers = _qualification_schema_objects(upgraded.conn) + assert tables == {"offline_tuner_open_world_qualifications"} + assert triggers == { + "offline_tuner_open_world_qualifications_no_update", + "offline_tuner_open_world_qualifications_no_delete", + } + + # Pre-existing, unrelated data survives the upgrade untouched. + assert ( + upgraded.get_playbook_optimization_job(sentinel_job.job_id) == sentinel_job + ) + + # The reinstalled table is immediately usable and immutable. + persisted = upgraded.persist_open_world_qualification_record(_record()) + assert persisted == _record() + with pytest.raises(sqlite3.IntegrityError): + upgraded.conn.execute( + "UPDATE offline_tuner_open_world_qualifications SET passed = 0" + ) + upgraded.conn.rollback() + with pytest.raises(sqlite3.IntegrityError): + upgraded.conn.execute("DELETE FROM offline_tuner_open_world_qualifications") + upgraded.conn.rollback() + finally: + upgraded.conn.close() + + def test_qualification_cache_surface_is_narrow() -> None: for method_name in ( "persist_open_world_qualification_record", From e067e5f3d71db6e345aa91a0f9ef412b66f8015f Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 19 Aug 2026 08:31:11 +0000 Subject: [PATCH 11/13] fix(storage): enforce pass-all qualification result Reject qualification records whose aggregate passed flag disagrees with the canonical per-class required counts, keeping shared SQLite reads aligned with PostgreSQL. --- reflexio/models/api_schema/domain/entities.py | 11 ++++ .../test_open_world_qualification_record.py | 10 ++++ .../test_open_world_qualification_cache.py | 54 ++++++++++++++++++- 3 files changed, 74 insertions(+), 1 deletion(-) diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index dbe5e746..1d3f3e29 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -737,6 +737,17 @@ def validate_observation_digests(cls, value: tuple[str, ...]) -> tuple[str, ...] ) return value + @model_validator(mode="after") + def validate_passed_matches_class_counts(self) -> Self: + expected_passed = all( + count.passed_required == count.required for count in self.class_counts + ) + if self.passed is not expected_passed: + raise ValueError( + "qualification passed must equal whether every class passed required" + ) + return self + def semantic_key(self) -> tuple[Any, ...]: """Return the conflict-detection identity, excluding ``created_at``.""" return ( diff --git a/tests/models/test_open_world_qualification_record.py b/tests/models/test_open_world_qualification_record.py index 60a230b4..b1d0f133 100644 --- a/tests/models/test_open_world_qualification_record.py +++ b/tests/models/test_open_world_qualification_record.py @@ -103,6 +103,16 @@ def test_passed_required_may_not_exceed_required() -> None: _record(class_counts=_class_counts(support=(1, 2))) +def test_passed_true_is_rejected_when_one_class_failed() -> None: + with pytest.raises(ValidationError, match="passed must equal"): + _record(class_counts=_class_counts(support=(2, 1)), passed=True) + + +def test_passed_false_is_rejected_when_every_class_passed() -> None: + with pytest.raises(ValidationError, match="passed must equal"): + _record(class_counts=_class_counts(), passed=False) + + def test_zero_required_class_is_accepted() -> None: record = _record(class_counts=_class_counts(refutation=(0, 0))) diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py index e5f47ed3..6706a4c3 100644 --- a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py +++ b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import sqlite3 from collections.abc import Generator from concurrent.futures import ThreadPoolExecutor @@ -12,6 +13,7 @@ from reflexio.models.api_schema import service_schemas as schemas from reflexio.server.services.storage.error import ( OpenWorldQualificationConflictError, + StorageError, ) from reflexio.server.services.storage.sqlite_storage import SQLiteStorage from reflexio.server.services.storage.storage_base import BaseStorage @@ -118,6 +120,46 @@ def test_failed_result_is_persisted_and_loadable(storage: SQLiteStorage) -> None assert loaded.class_counts[-1].passed_required == 2 +@pytest.mark.parametrize( + ("passed", "failed_class_index"), + [ + pytest.param(1, 0, id="passed-true-with-failed-class"), + pytest.param(0, None, id="passed-false-with-all-classes-passed"), + ], +) +def test_load_rejects_inconsistent_legacy_qualification_rows( + storage: SQLiteStorage, + passed: int, + failed_class_index: int | None, +) -> None: + class_counts = [count.model_dump() for count in _class_counts()] + if failed_class_index is not None: + class_counts[failed_class_index]["passed_required"] = 2 + storage.conn.execute( + """INSERT INTO offline_tuner_open_world_qualifications + (component_identity_digest, suite_digest, schema_version, result_digest, + passed, class_counts_json, observation_digests_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + _COMPONENT_DIGEST, + _SUITE_DIGEST, + schemas.OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION, + _RESULT_DIGEST, + passed, + json.dumps(class_counts), + json.dumps([]), + 1_700_000_000, + ), + ) + storage.conn.commit() + + with pytest.raises(StorageError, match="passed must equal"): + storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + + def test_exact_replay_is_idempotent(storage: SQLiteStorage) -> None: first = storage.persist_open_world_qualification_record(_record()) second = storage.persist_open_world_qualification_record(_record()) @@ -151,7 +193,17 @@ def test_first_insert_controls_created_at(storage: SQLiteStorage) -> None: "conflicting", [ {"result_digest": "d" * 64}, - {"passed": False}, + { + "class_counts": _class_counts()[:-1] + + ( + schemas.OpenWorldQualificationClassCount( + qualification_class="prompt_injection_resistance", + required=3, + passed_required=2, + ), + ), + "passed": False, + }, {"observation_digests": ("0" * 64,)}, { "class_counts": _class_counts()[:-1] From 81b8bc71bc3029ac61bf8e1b3c4fb4cd577a0565 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 19 Aug 2026 09:19:45 +0000 Subject: [PATCH 12/13] fix(storage): align open-world failure contracts Allow qualification reducers to report population failure when every per-class count passes, while retaining the one-way pass-all invariant. Permit the approved open-world failed outcomes without changing abstained transitions. --- reflexio/models/api_schema/domain/entities.py | 11 ++- .../sqlite_storage/playbook/_optimization.py | 7 +- .../test_open_world_qualification_record.py | 10 +-- .../test_open_world_analysis_identity.py | 68 +++++++++++++++++++ .../test_open_world_qualification_cache.py | 34 ++++++---- ...ptimization_replay_contract_integration.py | 9 ++- 6 files changed, 113 insertions(+), 26 deletions(-) diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 1d3f3e29..fdb4cf53 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -738,13 +738,12 @@ def validate_observation_digests(cls, value: tuple[str, ...]) -> tuple[str, ...] return value @model_validator(mode="after") - def validate_passed_matches_class_counts(self) -> Self: - expected_passed = all( - count.passed_required == count.required for count in self.class_counts - ) - if self.passed is not expected_passed: + def validate_passed_requires_all_class_counts(self) -> Self: + if self.passed and any( + count.passed_required != count.required for count in self.class_counts + ): raise ValueError( - "qualification passed must equal whether every class passed required" + "qualification passed=true requires every class to pass required" ) return self diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py index 8b16cbe2..b9607fd3 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py @@ -88,7 +88,12 @@ }, }, "offline_tuner_open_world": { - "failed": {"infrastructure_failure"}, + "failed": { + "infrastructure_failure", + "analyst_unqualified", + "stale_incumbent", + "governance_invalidated", + }, "abstained": { "no_grounded_hypothesis", "analyst_unqualified", diff --git a/tests/models/test_open_world_qualification_record.py b/tests/models/test_open_world_qualification_record.py index b1d0f133..d55f3c48 100644 --- a/tests/models/test_open_world_qualification_record.py +++ b/tests/models/test_open_world_qualification_record.py @@ -104,13 +104,15 @@ def test_passed_required_may_not_exceed_required() -> None: def test_passed_true_is_rejected_when_one_class_failed() -> None: - with pytest.raises(ValidationError, match="passed must equal"): + with pytest.raises(ValidationError, match="passed=true requires every class"): _record(class_counts=_class_counts(support=(2, 1)), passed=True) -def test_passed_false_is_rejected_when_every_class_passed() -> None: - with pytest.raises(ValidationError, match="passed must equal"): - _record(class_counts=_class_counts(), passed=False) +def test_passed_false_is_accepted_when_every_class_passed() -> None: + record = _record(class_counts=_class_counts(), passed=False) + + assert record.passed is False + assert all(count.passed_required == count.required for count in record.class_counts) def test_zero_required_class_is_accepted() -> None: diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py index acbe0a8a..f09cc8e3 100644 --- a/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py +++ b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py @@ -110,6 +110,74 @@ def test_open_world_terminal_outcomes_are_durable( assert persisted.terminal_outcome == outcome +@pytest.mark.parametrize( + "outcome", + [ + "infrastructure_failure", + "analyst_unqualified", + "stale_incumbent", + "governance_invalidated", + ], +) +def test_open_world_failed_terminal_outcomes_are_durable( + storage: SQLiteStorage, + outcome: schemas.OptimizationTerminalOutcome, +) -> None: + job = storage.create_playbook_optimization_job(_job()) + claim = storage.claim_playbook_optimization_job( + job_id=job.job_id, + owner="worker-a", + lease_seconds=60, + now=2_000, + ) + + assert storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage="failed", + terminal_outcome=outcome, + now=2_001, + ) + persisted = storage.get_playbook_optimization_job(job.job_id) + assert persisted is not None + assert persisted.stage == "failed" + assert persisted.status == "failed" + assert persisted.terminal_outcome == outcome + + +@pytest.mark.parametrize( + "outcome", + [ + pytest.param("no_grounded_hypothesis", id="open-world-abstention"), + pytest.param("generation_failed", id="replay-family-failure"), + ], +) +def test_open_world_failed_stage_rejects_unrelated_outcomes( + storage: SQLiteStorage, + outcome: schemas.OptimizationTerminalOutcome, +) -> None: + job = storage.create_playbook_optimization_job(_job()) + claim = storage.claim_playbook_optimization_job( + job_id=job.job_id, + owner="worker-a", + lease_seconds=60, + now=2_000, + ) + + assert not storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage="failed", + terminal_outcome=outcome, + now=2_001, + ) + persisted = storage.get_playbook_optimization_job(job.job_id) + assert persisted is not None + assert persisted.stage == "evidence_frozen" + assert persisted.status == "running" + assert persisted.terminal_outcome is None + + def test_legacy_optimizer_kind_allowlist_is_rebuilt_for_open_world_job( tmp_path: Path, ) -> None: diff --git a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py index 6706a4c3..d436242a 100644 --- a/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py +++ b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py @@ -120,21 +120,27 @@ def test_failed_result_is_persisted_and_loadable(storage: SQLiteStorage) -> None assert loaded.class_counts[-1].passed_required == 2 -@pytest.mark.parametrize( - ("passed", "failed_class_index"), - [ - pytest.param(1, 0, id="passed-true-with-failed-class"), - pytest.param(0, None, id="passed-false-with-all-classes-passed"), - ], -) -def test_load_rejects_inconsistent_legacy_qualification_rows( +def test_population_failure_with_passing_class_counts_round_trips( + storage: SQLiteStorage, +) -> None: + record = _record(passed=False) + + persisted = storage.persist_open_world_qualification_record(record) + loaded = storage.load_open_world_qualification_record( + component_identity_digest=_COMPONENT_DIGEST, + suite_digest=_SUITE_DIGEST, + ) + + assert persisted == record + assert loaded == record + assert all(count.passed_required == count.required for count in record.class_counts) + + +def test_load_rejects_passed_legacy_row_with_class_deficit( storage: SQLiteStorage, - passed: int, - failed_class_index: int | None, ) -> None: class_counts = [count.model_dump() for count in _class_counts()] - if failed_class_index is not None: - class_counts[failed_class_index]["passed_required"] = 2 + class_counts[0]["passed_required"] = 2 storage.conn.execute( """INSERT INTO offline_tuner_open_world_qualifications (component_identity_digest, suite_digest, schema_version, result_digest, @@ -145,7 +151,7 @@ def test_load_rejects_inconsistent_legacy_qualification_rows( _SUITE_DIGEST, schemas.OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION, _RESULT_DIGEST, - passed, + 1, json.dumps(class_counts), json.dumps([]), 1_700_000_000, @@ -153,7 +159,7 @@ def test_load_rejects_inconsistent_legacy_qualification_rows( ) storage.conn.commit() - with pytest.raises(StorageError, match="passed must equal"): + with pytest.raises(StorageError, match="passed=true requires every class"): storage.load_open_world_qualification_record( component_identity_digest=_COMPONENT_DIGEST, suite_digest=_SUITE_DIGEST, diff --git a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py index a1b2fb26..0c07b801 100644 --- a/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py +++ b/tests/server/services/storage/test_playbook_optimization_replay_contract_integration.py @@ -113,7 +113,14 @@ ), }, "offline_tuner_open_world": { - "failed": frozenset({"infrastructure_failure"}), + "failed": frozenset( + { + "infrastructure_failure", + "analyst_unqualified", + "stale_incumbent", + "governance_invalidated", + } + ), "abstained": frozenset( { "no_grounded_hypothesis", From 80f16db8d220025a6e4b39480d6e405d6951c136 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 19 Aug 2026 16:25:14 +0000 Subject: [PATCH 13/13] docs(storage): document StorageError on qualification record validation failure --- .../services/storage/storage_base/playbook/_optimization.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/reflexio/server/services/storage/storage_base/playbook/_optimization.py b/reflexio/server/services/storage/storage_base/playbook/_optimization.py index 0e6fcbb1..d7bc7f69 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_optimization.py +++ b/reflexio/server/services/storage/storage_base/playbook/_optimization.py @@ -208,6 +208,10 @@ def load_open_world_qualification_record( Returns: OpenWorldQualificationRecord | None: The stored record, or ``None`` when this key has never been qualified. + + Raises: + StorageError: If a stored row fails to validate against + ``OpenWorldQualificationRecord`` when reconstructed. """ raise NotImplementedError( "Storage backend does not support open-world analyst qualification"