From ca35a4b78214daf97d4093eb7517775aa4765a5c Mon Sep 17 00:00:00 2001 From: Guangyu Date: Mon, 31 Aug 2026 03:33:34 +0000 Subject: [PATCH 1/3] feat(playbook): retire the replay optimizer vocabulary Phase 7 Task 7. Remove the replay literals from the four public Literal unions, from the publication optimizer sets, and from both SQLite optimizer allowlists. The SQLite rebuild triggers were written for the expand direction only: both returned early once every REQUIRED literal was present, so removing a literal from the target tuple would have left an existing database's permissive CHECK in place forever (OQ-3). Each trigger now also asks whether any RETIRED literal is still present, and the table copy remediates the rows the new CHECK would reject: replay jobs are relabelled 'offline_tuner_legacy', their retired stage and terminal_outcome become NULL, and replay_manifest artifact rows are deleted (the column is NOT NULL and part of a UNIQUE key). 'offline_tuner_legacy' and 'optimizer_legacy_unknown' are retained permanently as historical labels (OQ-1 option A). --- reflexio/models/api_schema/domain/entities.py | 14 +- .../server/services/playbook/publication.py | 21 +- .../services/storage/sqlite_storage/_base.py | 79 +++-- .../sqlite_storage/playbook/_optimization.py | 75 +---- .../storage/sqlite_storage/playbook/_user.py | 2 +- .../test_provisional_publication_contract.py | 4 +- .../playbook/test_publication_models.py | 81 ++++- ...ptimization_replay_contract_integration.py | 164 +++++----- .../test_sqlite_allowlist_contraction.py | 290 ++++++++++++++++++ .../test_user_playbook_publication_sqlite.py | 62 ++-- 10 files changed, 551 insertions(+), 241 deletions(-) create mode 100644 tests/server/services/storage/test_sqlite_allowlist_contraction.py diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 403a2da68..9f2840dff 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -409,9 +409,13 @@ class AgentPlaybook(BaseModel): superseded_by: int | None = None +# 'offline_tuner_legacy' and 'optimizer_legacy_unknown' were never written by +# running code. They were assigned to pre-existing rows by a one-time heuristic +# backfill (supabase/data/tenant/20260723000000:54-61) and are retained +# permanently as HISTORICAL labels: dropping them would make those rows violate +# the tenant CHECK and abort the contract migration. OptimizerKind = Literal[ "gepa", - "offline_tuner_replay", "offline_tuner_open_world", "offline_tuner_legacy", "optimizer_legacy_unknown", @@ -425,8 +429,6 @@ class AgentPlaybook(BaseModel): "evidence_frozen", "discovery_analyzed", "candidate_generated", - "replay_running", - "replay_evaluated", "held_out_analyzed", "publishing", "applied", @@ -439,16 +441,11 @@ class AgentPlaybook(BaseModel): "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", @@ -462,7 +459,6 @@ class AgentPlaybook(BaseModel): OptimizationArtifactKind = Literal[ "expected_population_manifest", "generation_selection", - "replay_manifest", "candidate", "candidate_search_projection", "open_world_evidence_bundle", diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index 244b95df2..48b3900b5 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -41,15 +41,17 @@ "confirmed_online_support", } ) -PublishableOptimizerKind = Literal[ - "gepa", "offline_tuner_replay", "offline_tuner_open_world" -] +PublishableOptimizerKind = Literal["gepa", "offline_tuner_open_world"] +# 'offline_optimizer' remains in the union because it is a value already +# PERSISTED on user_playbooks.source rows. Phase 7 removes the only optimizer +# kind that produced it; the tenant RPC's matching CASE arm goes in Task 10. PublicationSource = Literal["gepa", "offline_optimizer"] -_LEGACY_PUBLICATION_OPTIMIZERS = frozenset({"gepa", "offline_tuner_replay"}) -_DECISION_PROOF_OPTIMIZERS = frozenset( - {"gepa", "offline_tuner_replay", "offline_tuner_open_world"} -) +# Phase 7 retired 'offline_tuner_replay'. The legacy publication path is now +# GEPA-only: the open-world path publishes through its own provisional +# publisher, not through publication_source_for_optimizer. +_LEGACY_PUBLICATION_OPTIMIZERS = frozenset({"gepa"}) +_DECISION_PROOF_OPTIMIZERS = frozenset({"gepa", "offline_tuner_open_world"}) _PROJECTION_SCHEMA_VERSION = "offline-tuner-candidate-search-projection-v1" _USER_PLAYBOOK_FULL_VERSION_SCHEMA = "user-playbook-full-version-v1" _CANONICAL_DECIMAL = re.compile(r"-?(?:0|[1-9][0-9]*)(?:\.[0-9]*[1-9])?\Z") @@ -183,8 +185,11 @@ def _validate_decision_proof_optimizer(value: object) -> None: def publication_source_for_optimizer( optimizer_kind: OptimizerKind, ) -> PublicationSource: + # The 'offline_optimizer' arm was reachable only for 'offline_tuner_replay', + # which Phase 7 retired. _validate_legacy_publication_optimizer now admits + # 'gepa' alone, so the branch is gone rather than left dead. _validate_legacy_publication_optimizer(optimizer_kind) - return "offline_optimizer" if optimizer_kind != "gepa" else "gepa" + return "gepa" def incumbent_user_playbook_semantic_digest( diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index 29133d7db..f48b1b49f 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -79,6 +79,23 @@ _TRAJECTORY_FETCH_SIZE = 256 _SESSION_OUTCOME_MIGRATION_BATCH_SIZE = 256 _MINIMUM_SQLITE_VERSION = (3, 35, 0) +# Literals Phase 7 retired. The rebuild's trigger asks TWO questions, not one: +# "is every required literal present?" (the expand direction, which was all it +# ever asked) AND "is any retired literal still present?" (the contract +# direction). Without the second, an existing database whose table_sql still +# holds every SURVIVING literal satisfies the first and returns early -- the +# rebuild never runs and the permissive CHECK survives forever. +_RETIRED_OPTIMIZER_JOB_LITERALS: tuple[str, ...] = ( + "'offline_tuner_replay'", + "'replay_running'", + "'replay_evaluated'", + "'replay_unsupported'", + "'incomplete_replay_scope'", + "'insufficient_replay_cases'", + "'replay_inconclusive'", + "'replay_failed'", +) +_RETIRED_ARTIFACT_KIND_LITERALS: tuple[str, ...] = ("'replay_manifest'",) _SQLITE_INITIALIZATION_LOCK_STRIPES = 64 _sqlite_initialization_locks = tuple( threading.Lock() for _ in range(_SQLITE_INITIALIZATION_LOCK_STRIPES) @@ -2211,8 +2228,12 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: "'governance_invalidated'", "'infrastructure_failure'", ) - if all(check in table_sql for check in required_checks) and ( - "'offline_tuner_open_world'" in table_sql + if ( + all(check in table_sql for check in required_checks) + and "'offline_tuner_open_world'" in table_sql + and not any( + retired in table_sql for retired in _RETIRED_OPTIMIZER_JOB_LITERALS + ) ): return foreign_keys_enabled = bool( @@ -2235,7 +2256,6 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: optimizer_kind TEXT NOT NULL DEFAULT 'optimizer_legacy_unknown' CHECK (optimizer_kind IN ( 'gepa', - 'offline_tuner_replay', 'offline_tuner_open_world', 'offline_tuner_legacy', 'optimizer_legacy_unknown' @@ -2256,8 +2276,6 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: 'evidence_frozen', 'discovery_analyzed', 'candidate_generated', - 'replay_running', - 'replay_evaluated', 'held_out_analyzed', 'publishing', 'applied', @@ -2269,16 +2287,11 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: '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', @@ -2311,10 +2324,27 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: candidate_content_digest, search_projection_digest, publication_scope_digest, created_at, updated_at ) SELECT - job_id, optimizer_kind, target_kind, target_id, status, + job_id, + -- Phase 7 remediation. The copy would otherwise fail the new + -- CHECK on any row still carrying a retired literal. + -- 'offline_tuner_legacy' is the accurate surviving label: these + -- jobs WERE offline-tuner jobs, and OQ-1 option A keeps that + -- literal admissible. stage and terminal_outcome are nullable, + -- so their retired values become NULL rather than a wrong one. + CASE WHEN optimizer_kind = 'offline_tuner_replay' + THEN 'offline_tuner_legacy' + ELSE optimizer_kind END, + target_kind, target_id, status, best_candidate_id, successor_target_id, decision_reason, metadata_json, discovery_key, attempt_key, lease_owner, - lease_fence, lease_expires_at, stage, terminal_outcome, + lease_fence, lease_expires_at, + CASE WHEN stage IN ('replay_running', 'replay_evaluated') + THEN NULL ELSE stage END, + CASE WHEN terminal_outcome IN ( + 'replay_unsupported', 'incomplete_replay_scope', + 'insufficient_replay_cases', 'replay_inconclusive', + 'replay_failed' + ) THEN NULL ELSE terminal_outcome END, expected_population_manifest_digest, generation_selection_manifest_digest, replay_manifest_digest, candidate_content_digest, search_projection_digest, @@ -2379,7 +2409,6 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: artifact_kinds = ( "'expected_population_manifest'", "'generation_selection'", - "'replay_manifest'", "'candidate'", "'candidate_search_projection'", "'open_world_evidence_bundle'", @@ -2387,7 +2416,11 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: "'open_world_candidate'", "'open_world_attempt_decision'", ) - if all(artifact_kind in table_sql for artifact_kind in artifact_kinds): + if all( + artifact_kind in table_sql for artifact_kind in artifact_kinds + ) and not any( + retired in table_sql for retired in _RETIRED_ARTIFACT_KIND_LITERALS + ): return foreign_keys_enabled = bool( @@ -2413,7 +2446,6 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: artifact_kind TEXT NOT NULL CHECK (artifact_kind IN ( 'expected_population_manifest', 'generation_selection', - 'replay_manifest', 'candidate', 'candidate_search_projection', 'open_world_evidence_bundle', @@ -2432,6 +2464,14 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: ) """ ) + # artifact_kind is NOT NULL and part of UNIQUE (job_id, + # artifact_kind), so there is no surviving value to map + # 'replay_manifest' onto. Local development SQLite only -- no tenant + # Postgres row is touched by this path. + self.conn.execute( + "DELETE FROM playbook_optimization_artifacts " + "WHERE artifact_kind = 'replay_manifest'" + ) self.conn.execute( """ INSERT INTO playbook_optimization_artifacts_new ( @@ -3416,7 +3456,6 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: optimizer_kind TEXT NOT NULL DEFAULT 'optimizer_legacy_unknown' CHECK (optimizer_kind IN ( 'gepa', - 'offline_tuner_replay', 'offline_tuner_open_world', 'offline_tuner_legacy', 'optimizer_legacy_unknown' @@ -3437,8 +3476,6 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: 'evidence_frozen', 'discovery_analyzed', 'candidate_generated', - 'replay_running', - 'replay_evaluated', 'held_out_analyzed', 'publishing', 'applied', @@ -3450,16 +3487,11 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: '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', @@ -3524,7 +3556,6 @@ def clear_user_data(self, user_id: str) -> dict[str, int]: artifact_kind TEXT NOT NULL CHECK (artifact_kind IN ( 'expected_population_manifest', 'generation_selection', - 'replay_manifest', 'candidate', 'candidate_search_projection', 'open_world_evidence_bundle', diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py index b9607fd3d..243ddcc8c 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_optimization.py @@ -37,13 +37,6 @@ ) _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"), @@ -51,13 +44,6 @@ }, } _ACTIVE_STAGES_BY_OPTIMIZER = { - "offline_tuner_replay": ( - "evidence_frozen", - "candidate_generated", - "replay_running", - "replay_evaluated", - "publishing", - ), "offline_tuner_open_world": ( "evidence_frozen", "discovery_analyzed", @@ -66,27 +52,6 @@ ), } _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", @@ -816,15 +781,12 @@ def advance_playbook_optimization_stage( 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"): + # The 'applied' stage was reachable only for 'offline_tuner_replay', + # whose 'publishing' -> 'applied' predecessor Phase 7 removed. With + # no optimizer declaring an 'applied' predecessor the request now + # falls through to the "unknown transition" arm below and is + # refused -- exactly what the replay-only guard returned. + if stage in ("failed", "abstained"): if terminal_outcome not in _TERMINAL_OUTCOMES_BY_OPTIMIZER.get( optimizer_kind, {} ).get(stage, set()): @@ -854,31 +816,6 @@ def advance_playbook_optimization_stage( *predecessors, ), ) - elif stage == "applied": - cur = self.conn.execute( - """UPDATE playbook_optimization_jobs - SET stage = ?, - terminal_outcome = ?, - status = ?, - lease_owner = NULL, - lease_expires_at = NULL, - 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'""", - ( - stage, - terminal_outcome, - terminal_status, - advanced_at, - job_id, - fence, - advanced_at, - ), - ) else: active_stages = _ACTIVE_STAGES_BY_OPTIMIZER.get(optimizer_kind, ()) placeholders = ", ".join("?" for _ in active_stages) diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_user.py b/reflexio/server/services/storage/sqlite_storage/playbook/_user.py index fe85125e3..d16c1593f 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_user.py +++ b/reflexio/server/services/storage/sqlite_storage/playbook/_user.py @@ -312,7 +312,7 @@ def claim_user_playbook_publication( ).fetchone() if job is None: raise StorageError("publication optimizer job does not exist") - if job["optimizer_kind"] not in {"gepa", "offline_tuner_replay"}: + if job["optimizer_kind"] != "gepa": raise StorageError("publication optimizer kind is not publishable") if job["target_kind"] != "user_playbook": raise StorageError("publication target is not a user playbook") diff --git a/tests/server/services/playbook/test_provisional_publication_contract.py b/tests/server/services/playbook/test_provisional_publication_contract.py index 4a79d2d1c..f204c8cb4 100644 --- a/tests/server/services/playbook/test_provisional_publication_contract.py +++ b/tests/server/services/playbook/test_provisional_publication_contract.py @@ -238,7 +238,9 @@ def test_qualification_authority_requires_exact_digest_references(field: str) -> replace(_authority(), **{field: "invalid"}) -@pytest.mark.parametrize("optimizer_kind", ["gepa", "offline_tuner_replay"]) +@pytest.mark.parametrize( + "optimizer_kind", ["gepa", "offline_tuner_legacy", "optimizer_legacy_unknown"] +) def test_provisional_publication_rejects_non_open_world_optimizers( optimizer_kind: str, ) -> None: diff --git a/tests/server/services/playbook/test_publication_models.py b/tests/server/services/playbook/test_publication_models.py index a66e8b508..8835cb834 100644 --- a/tests/server/services/playbook/test_publication_models.py +++ b/tests/server/services/playbook/test_publication_models.py @@ -10,7 +10,15 @@ OpenWorldDeploymentLifecycleState, UserPlaybook, ) +from reflexio.models.api_schema.domain.entities import ( + OptimizationArtifactKind, + OptimizationJobStage, + OptimizationTerminalOutcome, + OptimizerKind, +) from reflexio.server.services.playbook.publication import ( + _DECISION_PROOF_OPTIMIZERS, + _LEGACY_PUBLICATION_OPTIMIZERS, LIFECYCLE_TERMINAL_REASONS, LIFECYCLE_TERMINAL_STATES, DecisionProofEnvelope, @@ -80,7 +88,6 @@ def test_lifecycle_terminal_result_is_derived_from_the_state_literal() -> None: def test_open_world_publication_literals_and_user_playbook_field_partition() -> None: assert get_args(PublishableOptimizerKind) == ( "gepa", - "offline_tuner_replay", "offline_tuner_open_world", ) assert get_args(OpenWorldDeploymentLifecycleState) == ( @@ -248,9 +255,11 @@ def test_publication_canonical_json_must_match_digest_and_bytes() -> None: def test_publication_envelopes_bind_their_declared_fields() -> None: proof = _proof() + # A publishable-but-different kind, so the envelope/payload binding is what + # raises. A retired kind would raise earlier, at the publishability check. with pytest.raises(ValueError, match="optimizer_kind"): DecisionProofEnvelope( - optimizer_kind="offline_tuner_replay", + optimizer_kind="offline_tuner_open_world", schema_version=proof.schema_version, canonical_json=proof.canonical_json, digest=proof.digest, @@ -306,9 +315,34 @@ def test_publication_request_binds_content_optimizer_and_canonical_epochs() -> N revised_content="new content", subject_epochs_json=json.dumps({"subjects": []}), ) - with pytest.raises(ValueError, match="optimizer_kind"): + # The legacy publication path admits 'gepa' alone after Phase 7. + with pytest.raises(ValueError, match="not publishable"): + PublicationRequest( + **{**common, "optimizer_kind": "offline_tuner_open_world"}, + revised_content="new content", + subject_epochs_json=_canonical({"subjects": []}), + ) + # The request/proof binding still has to hold for the kind that IS admitted. + open_world_proof_json = _canonical( + { + "decision": "apply", + "optimizer_kind": "offline_tuner_open_world", + "schema_version": "gepa-publication-proof-v1", + "source": "playbook_optimizer", + } + ) + with pytest.raises(ValueError, match="decision proof optimizer_kind"): PublicationRequest( - **{**common, "optimizer_kind": "offline_tuner_replay"}, + **{ + **common, + "decision_proof": DecisionProofEnvelope( + optimizer_kind="offline_tuner_open_world", + schema_version="gepa-publication-proof-v1", + canonical_json=open_world_proof_json, + digest=_digest(open_world_proof_json), + decision="apply", + ), + }, revised_content="new content", subject_epochs_json=_canonical({"subjects": []}), ) @@ -495,3 +529,42 @@ def test_publication_service_requires_explicit_verifier() -> None: object(), # type: ignore[arg-type] verifier=object(), # type: ignore[arg-type] ) + + +def test_the_optimizer_vocabularies_carry_no_replay_literal() -> None: + """Phase 7 removed the replay path; its literals leave the public surface. + + Asserted as EQUALITIES, so a re-added replay literal fails here as loudly as + a missing survivor. A subset assertion would pass against a surface that + quietly grew the literal back. + """ + assert set(get_args(OptimizerKind)) == { + "gepa", + "offline_tuner_open_world", + "offline_tuner_legacy", + "optimizer_legacy_unknown", + } + assert set(get_args(PublishableOptimizerKind)) == { + "gepa", + "offline_tuner_open_world", + } + assert frozenset({"gepa"}) == _LEGACY_PUBLICATION_OPTIMIZERS + assert frozenset({"gepa", "offline_tuner_open_world"}) == _DECISION_PROOF_OPTIMIZERS + + for union in ( + OptimizationJobStage, + OptimizationTerminalOutcome, + OptimizationArtifactKind, + ): + assert not [member for member in get_args(union) if "replay" in member], union + + +def test_the_two_backfilled_legacy_kinds_are_retained() -> None: + """OQ-1 option A retains the two heuristic backfill labels. + + They were assigned to pre-existing rows by a one-time backfill + (``supabase/data/tenant/20260723000000:54-61``); dropping them would make + those rows violate the tenant CHECK and abort the contract migration. + """ + assert "offline_tuner_legacy" in get_args(OptimizerKind) + assert "optimizer_legacy_unknown" in get_args(OptimizerKind) 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 0c07b801c..82df86022 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 @@ -27,13 +27,6 @@ _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", @@ -44,8 +37,6 @@ _ORDINARY_STAGES = ( "discovery_analyzed", "candidate_generated", - "replay_running", - "replay_evaluated", "held_out_analyzed", "publishing", ) @@ -67,16 +58,11 @@ "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", @@ -87,31 +73,6 @@ "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( { @@ -146,12 +107,17 @@ def storage(tmp_path: Path) -> Generator[BaseStorage]: store.conn.close() -def _replay_job( +def _durable_job( discovery_key: str, attempt_key: str, ) -> schemas.PlaybookOptimizationJob: + """A generic durable optimizer job for the identity/lease/artifact contracts. + + Phase 7 retired 'offline_tuner_replay'; 'offline_tuner_open_world' is the + surviving non-GEPA kind, and none of the contracts below are replay-specific. + """ 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, @@ -197,10 +163,10 @@ def _artifact( ) -def test_replay_job_model_exposes_typed_durable_fields() -> None: - job = _replay_job("d1", "a1") +def test_durable_job_model_exposes_typed_durable_fields() -> None: + job = _durable_job("d1", "a1") - assert job.optimizer_kind == "offline_tuner_replay" + assert job.optimizer_kind == "offline_tuner_open_world" assert job.stage == "evidence_frozen" assert job.lease_fence == 0 assert job.terminal_outcome is None @@ -208,31 +174,31 @@ def test_replay_job_model_exposes_typed_durable_fields() -> None: def test_same_discovery_key_returns_one_active_job(storage: BaseStorage) -> None: - first = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) - second = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + first = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) + second = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) assert second.job_id == first.job_id def test_same_attempt_key_returns_one_active_job(storage: BaseStorage) -> None: - first = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) - second = storage.create_or_get_playbook_optimization_job(_replay_job("d2", "a1")) + first = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) + second = storage.create_or_get_playbook_optimization_job(_durable_job("d2", "a1")) assert second.job_id == first.job_id def test_conflicting_active_identity_is_rejected(storage: BaseStorage) -> None: - storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) with pytest.raises( OptimizationJobIdentityConflictError, match="immutable optimizer job identity", ): - storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a2")) + storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a2")) 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 = _durable_job("open-world-discovery", "open-world-attempt") open_world_job.optimizer_kind = "offline_tuner_open_world" saved = storage.create_or_get_playbook_optimization_job(open_world_job) @@ -267,7 +233,7 @@ def test_gepa_publication_reclaim_contract_has_none_live_and_reclaimed_outcomes( def test_stale_lease_fence_cannot_advance_stage(storage: BaseStorage) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -286,7 +252,7 @@ def test_stale_lease_fence_cannot_advance_stage(storage: BaseStorage) -> None: storage.advance_playbook_optimization_stage( job_id=job.job_id, fence=claim.fence, - stage="candidate_generated", + stage="discovery_analyzed", now=claim.expires_at + 1, ) is False @@ -295,7 +261,7 @@ def test_stale_lease_fence_cannot_advance_stage(storage: BaseStorage) -> None: storage.advance_playbook_optimization_stage( job_id=job.job_id, fence=reclaimed.fence, - stage="candidate_generated", + stage="discovery_analyzed", now=claim.expires_at + 1, ) is True @@ -303,7 +269,7 @@ def test_stale_lease_fence_cannot_advance_stage(storage: BaseStorage) -> None: def test_current_owner_can_renew_but_stale_owner_cannot(storage: BaseStorage) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -330,7 +296,7 @@ def test_current_owner_can_renew_but_stale_owner_cannot(storage: BaseStorage) -> def test_stage_advancement_is_linear(storage: BaseStorage) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -342,7 +308,7 @@ def test_stage_advancement_is_linear(storage: BaseStorage) -> None: storage.advance_playbook_optimization_stage( job_id=job.job_id, fence=claim.fence, - stage="replay_running", + stage="held_out_analyzed", now=3_001, ) is False @@ -351,7 +317,7 @@ def test_stage_advancement_is_linear(storage: BaseStorage) -> None: storage.advance_playbook_optimization_stage( job_id=job.job_id, fence=claim.fence, - stage="candidate_generated", + stage="discovery_analyzed", now=3_001, ) is True @@ -361,10 +327,10 @@ def test_stage_advancement_is_linear(storage: BaseStorage) -> None: @pytest.mark.parametrize( ("stage", "outcome", "expected_status"), [ - ("abstained", "candidate_did_not_improve", "skipped"), - ("failed", "generation_failed", "failed"), - ("failed", "replay_failed", "failed"), - ("failed", "publication_failed", "failed"), + ("abstained", "no_grounded_hypothesis", "skipped"), + ("abstained", "heldout_evidence_failed", "skipped"), + ("failed", "analyst_unqualified", "failed"), + ("failed", "governance_invalidated", "failed"), ("failed", "infrastructure_failure", "failed"), ], ) @@ -374,7 +340,7 @@ def test_terminal_stage_records_outcome_and_releases_lease( outcome: schemas.OptimizationTerminalOutcome, expected_status: str, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -403,7 +369,7 @@ def test_terminal_stage_records_outcome_and_releases_lease( def test_stale_lease_fence_cannot_write_singleton_artifact( storage: BaseStorage, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -435,7 +401,7 @@ def test_stale_lease_fence_cannot_write_singleton_artifact( def test_artifact_upsert_canonicalizes_equivalent_json_and_requires_digest_and_content( storage: BaseStorage, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -491,7 +457,7 @@ def test_artifact_upsert_canonicalizes_equivalent_json_and_requires_digest_and_c def test_malformed_persisted_artifact_raises_typed_integrity_error( storage: BaseStorage, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -647,7 +613,7 @@ def test_previous_artifact_schema_is_upgraded_without_losing_constraints( job_id, optimizer_kind, target_kind, target_id, status, lease_owner, lease_fence, lease_expires_at, created_at, updated_at ) VALUES ( - 41, 'offline_tuner_replay', 'user_playbook', 9, 'running', + 41, 'offline_tuner_open_world', 'user_playbook', 9, 'running', 'worker-a', 3, 1000, 101, 102 )""" ) @@ -749,11 +715,11 @@ def test_optimizer_job_rebuild_preserves_deleted_id_high_water_and_repeats( conn.close() first_store = SQLiteStorage(org_id="job-sequence-first", db_path=str(db_path)) - first = first_store.create_playbook_optimization_job(_replay_job("d1", "a1")) + first = first_store.create_playbook_optimization_job(_durable_job("d1", "a1")) first_store.conn.close() second_store = SQLiteStorage(org_id="job-sequence-second", db_path=str(db_path)) second = second_store.create_playbook_optimization_job( - _replay_job("d2", "a2").model_copy(update={"target_id": 42}) + _durable_job("d2", "a2").model_copy(update={"target_id": 42}) ) second_store.conn.close() @@ -772,7 +738,7 @@ def test_empty_optimizer_job_rebuild_preserves_deleted_id_high_water( conn.close() store = SQLiteStorage(org_id="empty-job-sequence", db_path=str(db_path)) - job = store.create_playbook_optimization_job(_replay_job("d1", "a1")) + job = store.create_playbook_optimization_job(_durable_job("d1", "a1")) store.conn.close() assert job.job_id == 7 @@ -783,7 +749,7 @@ def test_artifact_rebuild_preserves_deleted_id_high_water_and_repeats( ) -> None: db_path = tmp_path / "legacy-artifact-sequence.db" store = SQLiteStorage(org_id="artifact-sequence-setup", db_path=str(db_path)) - parent = store.create_playbook_optimization_job(_replay_job("d1", "a1")) + parent = store.create_playbook_optimization_job(_durable_job("d1", "a1")) store.conn.execute("DROP INDEX idx_poa_job") store.conn.execute("DROP TABLE playbook_optimization_artifacts") store.conn.executescript( @@ -849,16 +815,16 @@ def test_artifact_rebuild_preserves_deleted_id_high_water_and_repeats( "current_stage", [ "evidence_frozen", + "discovery_analyzed", "candidate_generated", - "replay_running", - "replay_evaluated", + "held_out_analyzed", ], ) def test_applied_terminal_stage_requires_publishing( storage: BaseStorage, current_stage: schemas.OptimizationJobStage, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -889,8 +855,17 @@ def test_applied_terminal_stage_requires_publishing( assert persisted.terminal_outcome is None -def test_applied_terminal_stage_advances_from_publishing(storage: BaseStorage) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) +def test_applied_terminal_stage_is_unreachable_after_the_replay_retirement( + storage: BaseStorage, +) -> None: + """The 'publishing' -> 'applied' transition was replay-only. + + Phase 7 removed the only optimizer that declared an 'applied' predecessor, + so this transition is now refused from every stage, 'publishing' included, + and the job is left exactly as it was. Replaces the success-path test that + covered the retired path. + """ + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -904,24 +879,27 @@ def test_applied_terminal_stage_advances_from_publishing(storage: BaseStorage) - ) storage.conn.commit() - assert storage.advance_playbook_optimization_stage( - job_id=job.job_id, - fence=claim.fence, - stage="applied", - terminal_outcome="applied", - now=6_001, + assert ( + storage.advance_playbook_optimization_stage( + job_id=job.job_id, + fence=claim.fence, + stage="applied", + terminal_outcome="applied", + now=6_001, + ) + is False ) persisted = storage.get_playbook_optimization_job(job.job_id) assert persisted is not None - assert persisted.stage == "applied" - assert persisted.status == "completed" - assert persisted.terminal_outcome == "applied" + assert persisted.stage == "publishing" + assert persisted.status == "running" + assert persisted.terminal_outcome is None def test_governance_erased_terminal_outcome_round_trips( storage: BaseStorage, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) assert isinstance(storage, SQLiteStorage) storage.conn.execute( """UPDATE playbook_optimization_jobs @@ -941,7 +919,7 @@ def test_governance_erased_terminal_outcome_round_trips( def test_ordinary_stage_advance_rejects_governance_erased( storage: BaseStorage, ) -> None: - job = storage.create_or_get_playbook_optimization_job(_replay_job("d1", "a1")) + job = storage.create_or_get_playbook_optimization_job(_durable_job("d1", "a1")) claim = storage.claim_playbook_optimization_job( job_id=job.job_id, owner="worker-a", @@ -971,10 +949,10 @@ def _claimed_job_at_stage( stage: schemas.OptimizationJobStage, *, now: int, - optimizer_kind: schemas.OptimizerKind = "offline_tuner_replay", + optimizer_kind: schemas.OptimizerKind = "offline_tuner_open_world", target_id: int = 41, ) -> tuple[int, int]: - job = _replay_job(f"d-{target_id}", f"a-{target_id}") + job = _durable_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) @@ -1169,11 +1147,9 @@ def test_optimizer_kind_terminal_outcome_matrix_is_exact( 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") - ) + # Phase 7 retired the only optimizer that declared an 'applied' + # predecessor, so no (kind, stage, outcome) triple reaches it. + expected = False before = ( _optimization_job_row(storage, job_id) if not expected else None ) diff --git a/tests/server/services/storage/test_sqlite_allowlist_contraction.py b/tests/server/services/storage/test_sqlite_allowlist_contraction.py new file mode 100644 index 000000000..b5cfda540 --- /dev/null +++ b/tests/server/services/storage/test_sqlite_allowlist_contraction.py @@ -0,0 +1,290 @@ +"""The SQLite allowlist rebuild must work in the REMOVAL direction (OQ-3). + +The rebuild's trigger predicate was written for the expand direction only: it +returns early when every REQUIRED literal is already present. Removing a literal +from the target tuple leaves an old database satisfying that predicate, so the +rebuild never runs and the permissive CHECK survives forever. These tests pin +the negative predicate and the row remediation that make removal work. + +The databases here are built the way the rest of this suite builds a legacy +database: open a real ``SQLiteStorage`` so the full current schema exists, then +replace the two optimizer tables with their pre-Phase-7 permissive definitions. +Reopening the storage is what runs the rebuild. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from reflexio.server.services.storage.sqlite_storage import SQLiteStorage + +pytestmark = pytest.mark.integration + + +_LEGACY_JOBS_DDL = """ +CREATE TABLE playbook_optimization_jobs ( + job_id INTEGER PRIMARY KEY AUTOINCREMENT, + optimizer_kind TEXT NOT NULL DEFAULT 'optimizer_legacy_unknown' + CHECK (optimizer_kind IN ( + 'gepa', 'offline_tuner_replay', 'offline_tuner_open_world', + 'offline_tuner_legacy', 'optimizer_legacy_unknown' + )), + target_kind TEXT NOT NULL, + target_id INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + best_candidate_id INTEGER, + successor_target_id INTEGER, + decision_reason TEXT NOT NULL DEFAULT '', + metadata_json TEXT NOT NULL DEFAULT '{}', + discovery_key TEXT, + attempt_key TEXT, + lease_owner TEXT, + lease_fence INTEGER NOT NULL DEFAULT 0 CHECK (lease_fence >= 0), + 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', 'failed' + )), + terminal_outcome TEXT CHECK (terminal_outcome IS NULL OR terminal_outcome IN ( + 'applied', 'replay_unsupported', 'replay_failed', 'governance_erased', + 'no_grounded_hypothesis', 'analyst_unqualified', + 'heldout_evidence_failed', 'stale_incumbent', 'governance_invalidated', + 'infrastructure_failure', 'insufficient_negative_evidence', + 'insufficient_positive_evidence', 'insufficient_coverage', + 'deployment_unsupported', 'incomplete_replay_scope', + 'insufficient_replay_cases', 'replay_inconclusive', + 'candidate_regressed', 'candidate_did_not_improve', 'incumbent_changed', + 'generation_failed', 'publication_failed' + )), + expected_population_manifest_digest TEXT, + generation_selection_manifest_digest TEXT, + replay_manifest_digest TEXT, + candidate_content_digest TEXT, + search_projection_digest TEXT, + publication_scope_digest TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +) +""" + +_LEGACY_ARTIFACTS_DDL = """ +CREATE TABLE playbook_optimization_artifacts ( + artifact_id INTEGER PRIMARY KEY AUTOINCREMENT, + job_id INTEGER NOT NULL, + artifact_kind TEXT NOT NULL CHECK (artifact_kind IN ( + 'expected_population_manifest', 'generation_selection', + 'replay_manifest', 'candidate', 'candidate_search_projection', + '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, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE (job_id, artifact_kind), + FOREIGN KEY (job_id) REFERENCES playbook_optimization_jobs(job_id) + ON DELETE CASCADE +) +""" + + +def _write_legacy_optimizer_tables(db_path: Path, org_id: str) -> None: + """Leave ``db_path`` holding the pre-Phase-7 permissive optimizer tables.""" + initial = SQLiteStorage(org_id=org_id, db_path=str(db_path)) + initial.conn.close() + + conn = sqlite3.connect(db_path) + conn.execute("PRAGMA foreign_keys=OFF") + conn.execute("DROP INDEX IF EXISTS idx_poa_job") + conn.execute("DROP TABLE playbook_optimization_artifacts") + conn.execute("DROP INDEX IF EXISTS idx_poj_target") + conn.execute("DROP INDEX IF EXISTS idx_poj_status") + conn.execute("DROP INDEX IF EXISTS uq_poj_active_discovery") + conn.execute("DROP INDEX IF EXISTS uq_poj_active_attempt") + conn.execute("DROP INDEX IF EXISTS uq_poj_active_target") + conn.execute("DROP TABLE playbook_optimization_jobs") + conn.execute(_LEGACY_JOBS_DDL) + conn.execute(_LEGACY_ARTIFACTS_DDL) + conn.execute( + """INSERT INTO playbook_optimization_jobs ( + job_id, optimizer_kind, target_kind, target_id, status, + metadata_json, stage, terminal_outcome, created_at, updated_at + ) VALUES (1, 'offline_tuner_replay', 'user_playbook', 1, 'skipped', + '{"offline_tuner": {}}', 'replay_evaluated', + 'replay_inconclusive', 1, 1)""" + ) + conn.execute( + """INSERT INTO playbook_optimization_artifacts ( + artifact_id, job_id, artifact_kind, content_json, content_digest, + created_at, updated_at + ) VALUES (1, 1, 'replay_manifest', '{"a":1}', ?, 1, 1)""", + ("a" * 64,), + ) + conn.execute( + """INSERT INTO playbook_optimization_artifacts ( + artifact_id, job_id, artifact_kind, content_json, content_digest, + created_at, updated_at + ) VALUES (2, 1, 'candidate', '{"b":2}', ?, 1, 1)""", + ("b" * 64,), + ) + conn.commit() + conn.close() + + +def _table_sql(conn: sqlite3.Connection, name: str) -> str: + row = conn.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?", + (name,), + ).fetchone() + assert row is not None + return str(row["sql"]) + + +def test_the_rebuild_trigger_is_not_satisfied_by_a_legacy_schema( + tmp_path: Path, +) -> None: + """The whole OQ-3 defect in one assertion. + + A positive 'all required present' predicate returns early against this + database, because every SURVIVING literal is still in its table_sql. Only a + predicate that also asks 'is any RETIRED literal still present?' fires. + """ + from reflexio.server.services.storage.sqlite_storage import _base + + db_path = tmp_path / "legacy-trigger.db" + _write_legacy_optimizer_tables(db_path, "legacy-trigger") + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + jobs_sql = _table_sql(conn, "playbook_optimization_jobs") + artifacts_sql = _table_sql(conn, "playbook_optimization_artifacts") + finally: + conn.close() + + # The positive half of the predicate is satisfied: nothing here is missing. + assert "'offline_tuner_open_world'" in jobs_sql + assert all( + kind in artifacts_sql + for kind in ( + "'expected_population_manifest'", + "'open_world_attempt_decision'", + ) + ) + # Only the negative half can see that the schema is stale. + assert any(literal in jobs_sql for literal in _base._RETIRED_OPTIMIZER_JOB_LITERALS) + assert any( + literal in artifacts_sql for literal in _base._RETIRED_ARTIFACT_KIND_LITERALS + ) + + +def test_the_rebuild_contracts_the_allowlist_on_an_existing_database( + tmp_path: Path, +) -> None: + """After the rebuild the CHECK must refuse the retired literal.""" + db_path = tmp_path / "legacy-contract.db" + _write_legacy_optimizer_tables(db_path, "legacy-contract") + storage = SQLiteStorage(org_id="legacy-contract", db_path=str(db_path)) + try: + jobs_sql = _table_sql(storage.conn, "playbook_optimization_jobs") + artifacts_sql = _table_sql(storage.conn, "playbook_optimization_artifacts") + assert "offline_tuner_replay" not in jobs_sql + assert "replay_evaluated" not in jobs_sql + assert "replay_inconclusive" not in jobs_sql + assert "offline_tuner_legacy" in jobs_sql + # replay_manifest_digest is a COLUMN name, not a vocabulary literal. + assert "'replay_manifest'" not in artifacts_sql + + with pytest.raises(sqlite3.IntegrityError): + storage.conn.execute( + """INSERT INTO playbook_optimization_jobs ( + optimizer_kind, target_kind, target_id, created_at, + updated_at + ) VALUES ('offline_tuner_replay', 'user_playbook', 2, 1, 1)""" + ) + storage.conn.rollback() + with pytest.raises(sqlite3.IntegrityError): + storage.conn.execute( + """INSERT INTO playbook_optimization_artifacts ( + job_id, artifact_kind, content_json, content_digest, + created_at, updated_at + ) VALUES (1, 'replay_manifest', '{}', ?, 1, 1)""", + ("c" * 64,), + ) + storage.conn.rollback() + finally: + storage.conn.close() + + +def test_the_rebuild_remediates_the_rows_it_would_otherwise_reject( + tmp_path: Path, +) -> None: + """A table copy fails on a row carrying a retired value. + + The remediation is documented, not incidental: relabel the job to + 'offline_tuner_legacy' (the accurate surviving label, retained by OQ-1 + option A), null the retired stage and terminal_outcome (both nullable), and + delete the artifact rows, whose kind column is NOT NULL and part of a + UNIQUE key so there is no surviving value to map onto. + """ + db_path = tmp_path / "legacy-remediation.db" + _write_legacy_optimizer_tables(db_path, "legacy-remediation") + storage = SQLiteStorage(org_id="legacy-remediation", db_path=str(db_path)) + try: + row = storage.conn.execute( + "SELECT optimizer_kind, stage, terminal_outcome " + "FROM playbook_optimization_jobs WHERE job_id = 1" + ).fetchone() + assert row is not None + assert row["optimizer_kind"] == "offline_tuner_legacy" + assert row["stage"] is None + assert row["terminal_outcome"] is None + + kinds = [ + artifact["artifact_kind"] + for artifact in storage.conn.execute( + "SELECT artifact_kind FROM playbook_optimization_artifacts " + "WHERE job_id = 1 ORDER BY artifact_id" + ).fetchall() + ] + assert kinds == ["candidate"] + finally: + storage.conn.close() + + +def test_the_rebuild_is_idempotent(tmp_path: Path) -> None: + """A second run must return early, or every open costs a table copy.""" + db_path = tmp_path / "legacy-idempotent.db" + _write_legacy_optimizer_tables(db_path, "legacy-idempotent") + first_storage = SQLiteStorage(org_id="legacy-idempotent", db_path=str(db_path)) + try: + first_jobs = _table_sql(first_storage.conn, "playbook_optimization_jobs") + first_artifacts = _table_sql( + first_storage.conn, "playbook_optimization_artifacts" + ) + first_job_row_id = first_storage.conn.execute( + "SELECT job_id FROM playbook_optimization_jobs" + ).fetchone()["job_id"] + finally: + first_storage.conn.close() + + second_storage = SQLiteStorage(org_id="legacy-idempotent", db_path=str(db_path)) + try: + assert ( + _table_sql(second_storage.conn, "playbook_optimization_jobs") == first_jobs + ) + assert ( + _table_sql(second_storage.conn, "playbook_optimization_artifacts") + == first_artifacts + ) + assert ( + second_storage.conn.execute( + "SELECT job_id FROM playbook_optimization_jobs" + ).fetchone()["job_id"] + == first_job_row_id + ) + finally: + second_storage.conn.close() diff --git a/tests/server/services/storage/test_user_playbook_publication_sqlite.py b/tests/server/services/storage/test_user_playbook_publication_sqlite.py index 27b0e22f8..4fcdfa143 100644 --- a/tests/server/services/storage/test_user_playbook_publication_sqlite.py +++ b/tests/server/services/storage/test_user_playbook_publication_sqlite.py @@ -46,14 +46,21 @@ ) -@pytest.mark.parametrize( - ("optimizer_kind", "expected_source"), - [("gepa", "gepa"), ("offline_tuner_replay", "offline_optimizer")], -) -def test_publication_source_is_explicitly_mapped( - optimizer_kind: str, expected_source: str -) -> None: - assert publication_source_for_optimizer(optimizer_kind) == expected_source # type: ignore[arg-type] +def test_publication_source_is_explicitly_mapped() -> None: + """Phase 7 left 'gepa' as the only kind the legacy path publishes. + + The 'offline_optimizer' source was produced solely for + 'offline_tuner_replay'; with that kind retired the mapping is total and + every other kind is refused rather than silently mapped. + """ + assert publication_source_for_optimizer("gepa") == "gepa" + for retired_or_unpublishable in ( + "offline_tuner_open_world", + "offline_tuner_legacy", + "optimizer_legacy_unknown", + ): + with pytest.raises(ValueError, match="not publishable"): + publication_source_for_optimizer(retired_or_unpublishable) # type: ignore[arg-type] def _canonical(payload: dict[str, object]) -> str: @@ -314,7 +321,7 @@ def _subject_epochs_json(*, epoch: int = 0, subject_ref: str | None = None) -> s class _AcceptingVerifier: def verify(self, request: PublicationRequest) -> None: - assert request.optimizer_kind in {"gepa", "offline_tuner_replay"} + assert request.optimizer_kind == "gepa" def _service(storage: SQLiteStorage) -> UserPlaybookPublicationService: @@ -396,29 +403,22 @@ def test_publish_commits_exact_staged_projection_and_terminal_result( assert terminal == result -def test_offline_tuner_publication_persists_public_source(tmp_path: Path) -> None: +def test_non_gepa_optimizer_cannot_claim_the_legacy_publication_path( + tmp_path: Path, +) -> None: + """Replaces the retired 'offline_optimizer' source round-trip. + + 'offline_tuner_replay' was the only kind that produced + ``user_playbooks.source = 'offline_optimizer'`` through this path. Phase 7 + retired it, so what this path must now do with a non-GEPA job is refuse it + at the claim, before any successor row exists. + """ storage = _store(tmp_path) - incumbent, job = _seed(storage, optimizer_kind="offline_tuner_replay") + _, job = _seed(storage, optimizer_kind="offline_tuner_open_world") service = _service(storage) - claim = service.claim(job_id=job.job_id, owner="worker-a", worker_fence=5) - request = _request( - job_id=job.job_id, - incumbent_id=incumbent.user_playbook_id, - claim=claim, - optimizer_kind="offline_tuner_replay", - ) - result = service.publish(request) - - assert result.successor_user_playbook_id is not None - successor = storage.get_user_playbook_by_id(result.successor_user_playbook_id) - assert successor is not None - assert successor.source == "offline_optimizer" - staging = storage.conn.execute( - "SELECT optimizer_kind FROM user_playbook_publication_staging WHERE job_id = ?", - (job.job_id,), - ).fetchone() - assert staging["optimizer_kind"] == "offline_tuner_replay" + with pytest.raises(StorageError, match="not publishable"): + service.claim(job_id=job.job_id, owner="worker-a", worker_fence=5) def test_publish_lost_incumbent_cas_returns_incumbent_changed_without_orphan( @@ -876,9 +876,9 @@ def test_erasure_barrier_added_after_staging_rejects_publication( @pytest.mark.parametrize( ("job_change", "message"), [ - ({"stage": "replay_evaluated"}, "publishing"), + ({"stage": "candidate_generated"}, "publishing"), ({"attempt_key": "changed-attempt"}, "attempt"), - ({"optimizer_kind": "offline_tuner_replay"}, "optimizer"), + ({"optimizer_kind": "offline_tuner_open_world"}, "optimizer"), ({"target_id": 999}, "incumbent"), ], ) From ed2c091216e995c6e749c31b15082a3b59c87619 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 1 Sep 2026 00:18:04 +0000 Subject: [PATCH 2/3] fix(offline-tuner): close three Phase 7 review findings in the OSS package P4-2. The SQLite Phase 7 relabel turned a RUNNING 'offline_tuner_replay' job into a RUNNING 'offline_tuner_legacy' job -- a kind with no producer or consumer, holding uq_poj_active_target / uq_poj_active_discovery / uq_poj_active_attempt until the NEXT storage open let the pre-existing legacy sweep see it. The sweep in _classify_legacy_playbook_optimization_jobs runs BEFORE the relabel, so it could not. Retire the active rows in the rebuild itself, with the same status, decision_reason and lease clearing that sweep uses, so one open is enough. Observed before: REOPEN OK -> [(1, 'offline_tuner_legacy', 'running', '', 'worker-a'), (2, 'offline_tuner_legacy', 'pending', '', None), ...] SECOND OPEN -> both 'skipped', 'retired_by_replay_redesign' Observed after: both 'skipped' on the FIRST open, lease cleared. P4-3. The rebuild DELETEs every 'replay_manifest' artifact row, destroying its content_json, with no log line. Deletion stays -- artifact_kind is NOT NULL and part of UNIQUE (job_id, artifact_kind), and the tenant contract (20260830020000) drops the literal too, so retaining it on SQLite alone would leave the two backends disagreeing about the artifact vocabulary. But the inconsistency with the sibling optimizer_kind strategy, which RETAINS its retired labels precisely so historical rows survive, is now named in a comment, and the deletion emits a warning with its row count. P5-1. publication.py said the tenant RPC's 'offline_tuner_replay' -> 'offline_optimizer' CASE arm "goes in Task 10", which reads as pending. Task 10 landed on this branch: commit_user_playbook_publication now writes the staged optimizer kind straight through. Fixed the tense. P3-2 (recorded, not changed). Phase 7 contracted OptimizationTerminalOutcome by NAME, removing the four members spelled replay_*. Seven more were reachable only through the same replay arm and are equally dead -- verified independently: zero non-declaration references across 923 source files in both packages, and zero tenant-SQL writers. They are RETAINED rather than removed, because removing them is a second CHECK narrowing (a second one-way door) and would abort the validating migration on the first organization holding a historical row. Named in RETAINED_UNREACHABLE_TERMINAL_OUTCOMES with the reasoning, and pinned by tests/models/test_terminal_outcome_reachability.py, which also requires every member of the union to be classified as reachable-with-a-named-writer or retained. 'deployment_unsupported' is called out explicitly: the same spelling is also OfflineTunerUnavailableReason, a capability rejection code on a different type with ~40 live references, so grep is not evidence of reachability here. --- reflexio/models/api_schema/domain/entities.py | 32 +++++ .../server/services/playbook/publication.py | 7 +- .../services/storage/sqlite_storage/_base.py | 49 +++++++- .../test_terminal_outcome_reachability.py | 110 ++++++++++++++++ .../test_sqlite_allowlist_contraction.py | 117 ++++++++++++++++++ 5 files changed, 311 insertions(+), 4 deletions(-) create mode 100644 tests/models/test_terminal_outcome_reachability.py diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 9f2840dff..2907a75ef 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -436,6 +436,38 @@ class AgentPlaybook(BaseModel): "failed", ] +# Phase 7 removed the four members whose NAMES contain 'replay'. Seven more were +# reachable only through that same replay arm and are now equally dead: no +# writer, no reader, no tenant routine that can set them. They are RETAINED +# rather than removed, the way 'offline_tuner_legacy' is retained above -- and +# for the same two reasons. Removing them narrows the tenant CHECK a second +# time, and a CHECK narrowing is a one-way door this branch has already spent +# once; and historical rows written before the replay retirement still carry +# them, so a narrowed CHECK would abort the validating migration on the first +# organization holding one. +# +# Read a member of this set as "an outcome an earlier era could record", never +# as a state the current tuner can reach. Pinned by +# tests/models/test_terminal_outcome_reachability.py so it cannot silently grow: +# a new member added here is a new outcome somebody must show is WRITABLE. +# +# 'deployment_unsupported' is the trap. The same spelling is also +# OfflineTunerUnavailableReason -- a config-enablement rejection code in +# reflexio_ext capability_status.py, with ~40 live references. Those are a +# different vocabulary on a different type, so "grep says it is used" does not +# make this member reachable. +RETAINED_UNREACHABLE_TERMINAL_OUTCOMES: frozenset[str] = frozenset( + { + "insufficient_negative_evidence", + "insufficient_positive_evidence", + "insufficient_coverage", + "deployment_unsupported", + "candidate_regressed", + "candidate_did_not_improve", + "publication_failed", + } +) + OptimizationTerminalOutcome = Literal[ "applied", "insufficient_negative_evidence", diff --git a/reflexio/server/services/playbook/publication.py b/reflexio/server/services/playbook/publication.py index 48b3900b5..39ad4d617 100644 --- a/reflexio/server/services/playbook/publication.py +++ b/reflexio/server/services/playbook/publication.py @@ -43,8 +43,11 @@ ) PublishableOptimizerKind = Literal["gepa", "offline_tuner_open_world"] # 'offline_optimizer' remains in the union because it is a value already -# PERSISTED on user_playbooks.source rows. Phase 7 removes the only optimizer -# kind that produced it; the tenant RPC's matching CASE arm goes in Task 10. +# PERSISTED on user_playbooks.source rows. Phase 7 removed the only optimizer +# kind that produced it, and Task 10 took the tenant RPC's matching CASE arm +# with it: commit_user_playbook_publication now writes the staged optimizer kind +# straight through to user_playbooks.source (20260830020000). No path can write +# 'offline_optimizer' any more -- it is read-only history. PublicationSource = Literal["gepa", "offline_optimizer"] # Phase 7 retired 'offline_tuner_replay'. The legacy publication path is now diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index f48b1b49f..ae59af30f 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -2312,6 +2312,28 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: ) """ ) + # Phase 7 review finding P4-2. The pre-existing legacy sweep in + # _classify_legacy_playbook_optimization_jobs retires every ACTIVE + # 'offline_tuner_legacy' / 'optimizer_legacy_unknown' job, but it + # runs BEFORE this relabel. Without this statement the relabel below + # turns a running replay job into a running 'offline_tuner_legacy' + # job -- a kind with no producer or consumer -- which then holds a + # uq_poj_active_target / uq_poj_active_discovery / uq_poj_active_attempt + # slot until the next storage open lets the sweep see it. Retire it + # here with exactly the status, reason and lease clearing that sweep + # uses, so one open is enough. + self.conn.execute( + """ + UPDATE playbook_optimization_jobs + SET status = 'skipped', + decision_reason = 'retired_by_replay_redesign', + lease_owner = NULL, + lease_expires_at = NULL, + updated_at = CAST(strftime('%s', 'now') AS INTEGER) + WHERE optimizer_kind = 'offline_tuner_replay' + AND status IN ('pending', 'running') + """ + ) self.conn.execute( """ INSERT INTO playbook_optimization_jobs_new ( @@ -2468,10 +2490,33 @@ def _enforce_playbook_optimization_artifact_constraints(self) -> None: # artifact_kind), so there is no surviving value to map # 'replay_manifest' onto. Local development SQLite only -- no tenant # Postgres row is touched by this path. - self.conn.execute( + # + # Phase 7 review finding P4-3: this is the OPPOSITE strategy from the + # sibling optimizer_kind vocabulary in + # _enforce_playbook_optimization_job_constraints, where OQ-1 option A + # RETAINS 'offline_tuner_legacy' / 'optimizer_legacy_unknown' + # permanently as historical labels precisely so historical rows + # survive. Retention was available here too and was rejected: the + # tenant contract (supabase/data/tenant/20260830020000, constraint + # playbook_optimization_artifacts_artifact_kind_check) drops + # 'replay_manifest' outright, so retaining it on SQLite alone would + # leave the two backends disagreeing about the artifact + # vocabulary -- the dual-expression divergence the rest of + # Phase 7 exists to remove. The rows are destroyed rather than + # migrated, so the deletion is logged with its count instead of being + # silent. + deleted_replay_manifests = self.conn.execute( "DELETE FROM playbook_optimization_artifacts " "WHERE artifact_kind = 'replay_manifest'" - ) + ).rowcount + if deleted_replay_manifests > 0: + logger.warning( + "Phase 7 artifact contraction destroyed %d " + "'replay_manifest' artifact row(s) with their content_json: " + "the retired kind has no surviving artifact_kind to map onto " + "and the tenant contract drops it as well", + deleted_replay_manifests, + ) self.conn.execute( """ INSERT INTO playbook_optimization_artifacts_new ( diff --git a/tests/models/test_terminal_outcome_reachability.py b/tests/models/test_terminal_outcome_reachability.py new file mode 100644 index 000000000..1a15e55bd --- /dev/null +++ b/tests/models/test_terminal_outcome_reachability.py @@ -0,0 +1,110 @@ +"""``OptimizationTerminalOutcome`` retains members no path can write. + +Phase 7 removed the four replay-named members of the union. Seven more were +reachable only through that same replay arm and died with it, but removing them +would narrow the tenant CHECK a second time -- a one-way door this branch has +already spent once -- and would abort the validating migration on the first +organization still holding a historical row. So they are RETAINED, exactly the +way ``offline_tuner_legacy`` is retained in ``OptimizerKind``. + +Retention only stays honest if it is recorded. These assertions pin the retained +set so it cannot silently grow: a new member added to the union without a writer +lands in the reachable half and fails the second test, which is the prompt to +show that the new outcome can actually be written. +""" + +from __future__ import annotations + +from typing import get_args + +from reflexio.models.api_schema.domain.entities import ( + RETAINED_UNREACHABLE_TERMINAL_OUTCOMES, + OptimizationTerminalOutcome, +) +from reflexio.server.services.storage.sqlite_storage.playbook._optimization import ( + _TERMINAL_OUTCOMES_BY_OPTIMIZER, +) + +# The ten outcomes that survive Phase 7 with a path that can reach them. Six are +# written by the stage-advance allowlist below; the other four are written +# elsewhere and are named here with their writer so the split is auditable. +_REACHABLE_TERMINAL_OUTCOMES = frozenset( + { + # commit_user_playbook_publication (tenant 20260830020000:1293) + "applied", + # the same routine's CAS-lost branch (tenant 20260830020000:1147) + "incumbent_changed", + # the retention sweep's raw DML (reflexio_ext _retired_table_purge.py) + "generation_failed", + # the governance erasure path + "governance_erased", + # stage-advance: 'failed' + "infrastructure_failure", + "analyst_unqualified", + "stale_incumbent", + "governance_invalidated", + # stage-advance: 'abstained' + "no_grounded_hypothesis", + "heldout_evidence_failed", + } +) + + +def test_the_retained_unreachable_set_is_exactly_these_seven() -> None: + """Pinned by value, so growing the set is an edit somebody has to make here. + + ``deployment_unsupported`` is the one that looks alive to grep: the same + spelling is also ``OfflineTunerUnavailableReason``, a config-enablement + rejection code with dozens of live references on a different type. Those + references say nothing about this vocabulary. + """ + assert ( + frozenset( + { + "insufficient_negative_evidence", + "insufficient_positive_evidence", + "insufficient_coverage", + "deployment_unsupported", + "candidate_regressed", + "candidate_did_not_improve", + "publication_failed", + } + ) + == RETAINED_UNREACHABLE_TERMINAL_OUTCOMES + ) + assert len(RETAINED_UNREACHABLE_TERMINAL_OUTCOMES) == 7 + + +def test_the_union_is_exactly_the_reachable_set_plus_the_retained_set() -> None: + """Every member is classified: reachable, or retained-and-recorded. + + A member added to the union with neither a writer nor a retention rationale + fails here rather than accumulating quietly. + """ + members = frozenset(get_args(OptimizationTerminalOutcome)) + + assert members >= RETAINED_UNREACHABLE_TERMINAL_OUTCOMES + assert members >= _REACHABLE_TERMINAL_OUTCOMES + assert ( + members - RETAINED_UNREACHABLE_TERMINAL_OUTCOMES == _REACHABLE_TERMINAL_OUTCOMES + ) + assert len(members) == 17 + assert len(_REACHABLE_TERMINAL_OUTCOMES) == 10 + + +def test_no_retained_outcome_is_writable_through_the_stage_advance_allowlist() -> None: + """The SQLite writer allowlist is the mirror of the tenant advance RPC. + + If a retained member ever appears in it, the member is no longer + unreachable and this file's premise is wrong -- so the assertion is the + thing that would catch a silent revival, not just the growth of the set. + """ + writable = { + outcome + for stages in _TERMINAL_OUTCOMES_BY_OPTIMIZER.values() + for outcomes in stages.values() + for outcome in outcomes + } + + assert not (writable & RETAINED_UNREACHABLE_TERMINAL_OUTCOMES) + assert writable <= _REACHABLE_TERMINAL_OUTCOMES diff --git a/tests/server/services/storage/test_sqlite_allowlist_contraction.py b/tests/server/services/storage/test_sqlite_allowlist_contraction.py index b5cfda540..34ca14b96 100644 --- a/tests/server/services/storage/test_sqlite_allowlist_contraction.py +++ b/tests/server/services/storage/test_sqlite_allowlist_contraction.py @@ -14,6 +14,7 @@ from __future__ import annotations +import logging import sqlite3 from pathlib import Path @@ -255,6 +256,122 @@ def test_the_rebuild_remediates_the_rows_it_would_otherwise_reject( storage.conn.close() +@pytest.mark.parametrize("status", ["pending", "running"]) +def test_the_rebuild_leaves_no_relabelled_job_active( + tmp_path: Path, + status: str, +) -> None: + """A relabelled replay job must not survive as an ACTIVE legacy job. + + ``_classify_legacy_playbook_optimization_jobs`` retires every active + ``offline_tuner_legacy`` / ``optimizer_legacy_unknown`` job, and it runs + BEFORE this rebuild. Without the rebuild's own remediation, a running replay + job is reopened as a running ``offline_tuner_legacy`` job -- a kind with no + producer or consumer -- holding ``uq_poj_active_target`` / + ``uq_poj_active_discovery`` / ``uq_poj_active_attempt`` until the NEXT open + lets that sweep see it. One open must be enough. + """ + db_path = tmp_path / f"legacy-active-{status}.db" + _write_legacy_optimizer_tables(db_path, f"legacy-active-{status}") + conn = sqlite3.connect(db_path) + conn.execute( + """INSERT INTO playbook_optimization_jobs ( + job_id, optimizer_kind, target_kind, target_id, status, + discovery_key, attempt_key, lease_owner, lease_expires_at, + created_at, updated_at + ) VALUES (2, 'offline_tuner_replay', 'user_playbook', 2, ?, + 'discovery-2', 'attempt-2', 'worker-a', 9999, 1, 1)""", + (status,), + ) + conn.commit() + conn.close() + + storage = SQLiteStorage(org_id=f"legacy-active-{status}", db_path=str(db_path)) + try: + row = storage.conn.execute( + "SELECT optimizer_kind, status, decision_reason, lease_owner, " + "lease_expires_at FROM playbook_optimization_jobs WHERE job_id = 2" + ).fetchone() + assert row is not None + assert row["optimizer_kind"] == "offline_tuner_legacy" + assert row["status"] == "skipped" + assert row["decision_reason"] == "retired_by_replay_redesign" + assert row["lease_owner"] is None + assert row["lease_expires_at"] is None + assert ( + storage.conn.execute( + "SELECT COUNT(*) FROM playbook_optimization_jobs " + "WHERE status IN ('pending', 'running')" + ).fetchone()[0] + == 0 + ) + finally: + storage.conn.close() + + +def test_a_settled_relabelled_job_keeps_its_own_status_and_reason( + tmp_path: Path, +) -> None: + """The remediation must scope to ACTIVE rows, not rewrite settled history.""" + db_path = tmp_path / "legacy-settled.db" + _write_legacy_optimizer_tables(db_path, "legacy-settled") + conn = sqlite3.connect(db_path) + conn.execute( + """INSERT INTO playbook_optimization_jobs ( + job_id, optimizer_kind, target_kind, target_id, status, + decision_reason, created_at, updated_at + ) VALUES (3, 'offline_tuner_replay', 'user_playbook', 3, 'completed', + 'published_successor', 1, 1)""" + ) + conn.commit() + conn.close() + + storage = SQLiteStorage(org_id="legacy-settled", db_path=str(db_path)) + try: + row = storage.conn.execute( + "SELECT optimizer_kind, status, decision_reason " + "FROM playbook_optimization_jobs WHERE job_id = 3" + ).fetchone() + assert row is not None + assert row["optimizer_kind"] == "offline_tuner_legacy" + assert row["status"] == "completed" + assert row["decision_reason"] == "published_successor" + finally: + storage.conn.close() + + +def test_the_destroyed_replay_artifacts_are_logged_with_their_count( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """Deleting ``replay_manifest`` rows destroys their ``content_json``. + + The sibling ``optimizer_kind`` vocabulary retains its retired labels so + historical rows survive; this vocabulary cannot, because ``artifact_kind`` is + NOT NULL and part of ``UNIQUE (job_id, artifact_kind)`` and the tenant + contract drops the literal too. Destruction is therefore correct -- but it + must not be silent. + """ + db_path = tmp_path / "legacy-artifact-log.db" + _write_legacy_optimizer_tables(db_path, "legacy-artifact-log") + + with caplog.at_level( + logging.WARNING, + logger="reflexio.server.services.storage.sqlite_storage._base", + ): + storage = SQLiteStorage(org_id="legacy-artifact-log", db_path=str(db_path)) + storage.conn.close() + + warnings = [ + record.getMessage() + for record in caplog.records + if record.levelno >= logging.WARNING + and "replay_manifest" in record.getMessage() + ] + assert len(warnings) == 1, caplog.text + assert "destroyed 1 " in warnings[0] + + def test_the_rebuild_is_idempotent(tmp_path: Path) -> None: """A second run must return early, or every open costs a table copy.""" db_path = tmp_path / "legacy-idempotent.db" From fe85c8bff9e4cc417ebd384b5261766e18505eb9 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Tue, 1 Sep 2026 04:58:08 +0000 Subject: [PATCH 3/3] fix(storage): require every admissible optimizer kind, not a sample Addresses the CodeRabbit review on #475. The rebuild predicate's negative half catches a RETIRED literal that is still present. Its positive half could not catch the other direction: an ADMISSIBLE literal that is ABSENT. required_checks named only 'offline_tuner_open_world' among the four surviving optimizer kinds, so a schema whose CHECK omitted 'gepa' carried no retired literal, satisfied the list, and was retained forever with a constraint that rejects legitimate gepa rows. The artifact rebuild directly below already enumerates its full set of eight kinds; this makes the optimizer rebuild follow the same rule rather than remaining the outlier. Proven by contrast on a schema that is current except for the missing kind: pre-fix predicate -> 2 failed, 8 passed fixed predicate -> 10 passed --- .../services/storage/sqlite_storage/_base.py | 19 ++- .../test_sqlite_allowlist_contraction.py | 142 ++++++++++++++++++ 2 files changed, 155 insertions(+), 6 deletions(-) diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index ae59af30f..9acb53099 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -2214,10 +2214,21 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: if table_sql_row is None: return table_sql = table_sql_row["sql"] + # Every literal the FINAL constraints admit must appear, not just a + # representative few. A predicate that names some of them cannot tell + # "current" from "missing a kind": a schema whose optimizer_kind CHECK + # omits 'gepa' carries no retired literal, satisfies a partial list, + # and is retained forever with a constraint that rejects legitimate + # rows. The artifact rebuild below already enumerates its full set; + # this is that same rule. required_checks = ( "CHECK (optimizer_kind IN", "CHECK (stage IS NULL OR stage IN", "CHECK (terminal_outcome IS NULL OR terminal_outcome IN", + "'gepa'", + "'offline_tuner_open_world'", + "'offline_tuner_legacy'", + "'optimizer_legacy_unknown'", "'governance_erased'", "'discovery_analyzed'", "'held_out_analyzed'", @@ -2228,12 +2239,8 @@ def _enforce_playbook_optimization_job_constraints(self) -> None: "'governance_invalidated'", "'infrastructure_failure'", ) - if ( - all(check in table_sql for check in required_checks) - and "'offline_tuner_open_world'" in table_sql - and not any( - retired in table_sql for retired in _RETIRED_OPTIMIZER_JOB_LITERALS - ) + if all(check in table_sql for check in required_checks) and not any( + retired in table_sql for retired in _RETIRED_OPTIMIZER_JOB_LITERALS ): return foreign_keys_enabled = bool( diff --git a/tests/server/services/storage/test_sqlite_allowlist_contraction.py b/tests/server/services/storage/test_sqlite_allowlist_contraction.py index 34ca14b96..30ff5ac8f 100644 --- a/tests/server/services/storage/test_sqlite_allowlist_contraction.py +++ b/tests/server/services/storage/test_sqlite_allowlist_contraction.py @@ -405,3 +405,145 @@ def test_the_rebuild_is_idempotent(tmp_path: Path) -> None: ) finally: second_storage.conn.close() + + +# A schema that is current in every way EXCEPT that its optimizer_kind CHECK +# omits 'gepa'. It carries no retired literal, so the negative half of the +# predicate is satisfied; the question is whether the positive half notices a +# MISSING admissible kind. Before the fix it did not, because the predicate +# named only 'offline_tuner_open_world' among the four kinds. +_GEPA_LESS_JOBS_DDL = """ +CREATE TABLE playbook_optimization_jobs ( + job_id INTEGER PRIMARY KEY AUTOINCREMENT, + optimizer_kind TEXT NOT NULL DEFAULT 'optimizer_legacy_unknown' + CHECK (optimizer_kind IN ( + 'offline_tuner_open_world', + 'offline_tuner_legacy', + 'optimizer_legacy_unknown' + )), + target_kind TEXT NOT NULL, + target_id INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + best_candidate_id INTEGER, + successor_target_id INTEGER, + decision_reason TEXT NOT NULL DEFAULT '', + metadata_json TEXT NOT NULL DEFAULT '{}', + discovery_key TEXT, + attempt_key TEXT, + lease_owner TEXT, + lease_fence INTEGER NOT NULL DEFAULT 0 CHECK (lease_fence >= 0), + lease_expires_at INTEGER, + stage TEXT CHECK (stage IS NULL OR stage IN ( + 'evidence_frozen', + 'discovery_analyzed', + 'candidate_generated', + 'held_out_analyzed', + 'publishing', + 'applied', + 'abstained', + 'failed' + )), + terminal_outcome TEXT CHECK (terminal_outcome IS NULL OR terminal_outcome IN ( + 'applied', + 'insufficient_negative_evidence', + 'insufficient_positive_evidence', + 'insufficient_coverage', + 'deployment_unsupported', + 'candidate_regressed', + 'candidate_did_not_improve', + 'incumbent_changed', + 'generation_failed', + 'publication_failed', + '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, + replay_manifest_digest TEXT, + candidate_content_digest TEXT, + search_projection_digest TEXT, + publication_scope_digest TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); +""" + + +def _write_gepa_less_optimizer_table(db_path: Path, org_id: str) -> None: + """Leave ``db_path`` holding a jobs table whose CHECK omits ``'gepa'``.""" + initial = SQLiteStorage(org_id=org_id, db_path=str(db_path)) + initial.conn.close() + + conn = sqlite3.connect(db_path) + conn.execute("PRAGMA foreign_keys=OFF") + conn.execute("DROP INDEX IF EXISTS idx_poj_target") + conn.execute("DROP INDEX IF EXISTS idx_poj_status") + conn.execute("DROP INDEX IF EXISTS uq_poj_active_discovery") + conn.execute("DROP INDEX IF EXISTS uq_poj_active_attempt") + conn.execute("DROP INDEX IF EXISTS uq_poj_active_target") + conn.execute("DROP TABLE playbook_optimization_jobs") + conn.execute(_GEPA_LESS_JOBS_DDL) + conn.commit() + conn.close() + + +def test_a_schema_missing_an_admissible_optimizer_kind_is_rebuilt( + tmp_path: Path, +) -> None: + """A missing kind must trigger the rebuild, not only a retired one. + + The negative half of the predicate catches a RETIRED literal that is still + present. This is the other direction: an ADMISSIBLE literal that is absent. + Such a schema rejects legitimate ``gepa`` rows forever, and reports itself + as current, because no retired literal remains to give it away. + """ + db_path = tmp_path / "gepa_less.db" + _write_gepa_less_optimizer_table(db_path, "org_gepa_less") + + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + before = _table_sql(conn, "playbook_optimization_jobs") + conn.close() + assert "'gepa'" not in before + assert "'offline_tuner_replay'" not in before, ( + "the fixture must carry NO retired literal, so only the missing kind " + "can be what triggers the rebuild" + ) + + storage = SQLiteStorage(org_id="org_gepa_less", db_path=str(db_path)) + try: + after = _table_sql(storage.conn, "playbook_optimization_jobs") + finally: + storage.conn.close() + + assert "'gepa'" in after, ( + "the rebuild did not run: a schema whose optimizer_kind CHECK omits an " + "admissible kind was retained as current" + ) + + +def test_a_gepa_row_is_accepted_after_the_rebuild(tmp_path: Path) -> None: + """The rebuild's point: the repaired constraint admits what it must.""" + db_path = tmp_path / "gepa_row.db" + _write_gepa_less_optimizer_table(db_path, "org_gepa_row") + + storage = SQLiteStorage(org_id="org_gepa_row", db_path=str(db_path)) + try: + storage.conn.execute( + """INSERT INTO playbook_optimization_jobs ( + job_id, optimizer_kind, target_kind, target_id, status, + metadata_json, created_at, updated_at + ) VALUES (1, 'gepa', 'user_playbook', 1, 'pending', '{}', 1, 1)""" + ) + storage.conn.commit() + row = storage.conn.execute( + "SELECT optimizer_kind FROM playbook_optimization_jobs WHERE job_id = 1" + ).fetchone() + assert row["optimizer_kind"] == "gepa" + finally: + storage.conn.close()