diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 65463f4a..fdb4cf53 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", @@ -412,9 +418,11 @@ class AgentPlaybook(BaseModel): OptimizationJobStage = Literal[ "evidence_frozen", + "discovery_analyzed", "candidate_generated", "replay_running", "replay_evaluated", + "held_out_analyzed", "publishing", "applied", "abstained", @@ -438,6 +446,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 +461,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 @@ -603,6 +620,149 @@ 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, strict=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, strict=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 + + @model_validator(mode="after") + 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=true requires every class to pass required" + ) + return self + + 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/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/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 f7e87c08..29133d7d 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -2202,9 +2202,17 @@ 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 + "'offline_tuner_open_world'" in table_sql ): return foreign_keys_enabled = bool( @@ -2228,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' )), @@ -2245,9 +2254,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 +2280,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 +2383,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 +2416,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, @@ -3394,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' )), @@ -3411,9 +3435,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 +3461,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 +3527,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, @@ -3508,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 41b938d9..b9607fd3 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,32 +29,79 @@ canonical_json_bytes, ) from reflexio.server.services.storage.error import ( + OpenWorldQualificationConflictError, OptimizationArtifactIntegrityError, OptimizationJobIdentityConflictError, OptimizationJobLeaseLiveError, StorageError, ) -_FAILURE_OUTCOMES = {"generation_failed", "replay_failed", "publication_failed"} -_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", +_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"), + }, } -_STAGE_PREDECESSORS: dict[str, str] = { - "candidate_generated": "evidence_frozen", - "replay_running": "candidate_generated", - "replay_evaluated": "replay_running", - "publishing": "replay_evaluated", - "applied": "publishing", +_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": { + "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", + "analyst_unqualified", + "stale_incumbent", + "governance_invalidated", + }, + "abstained": { + "no_grounded_hypothesis", + "analyst_unqualified", + "heldout_evidence_failed", + "stale_incumbent", + "governance_invalidated", + }, + }, } @@ -212,6 +261,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, @@ -714,34 +804,55 @@ def advance_playbook_optimization_stage( now: int | None = None, ) -> bool: advanced_at = self._lease_now(now) - predecessor = _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 predecessor 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 = ? WHERE job_id = ? AND status IN ('pending', 'running') + AND optimizer_kind = ? 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, + optimizer_kind, + fence, + advanced_at, + *predecessors, + ), ) elif stage == "applied": cur = self.conn.execute( @@ -754,6 +865,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'""", @@ -768,8 +880,10 @@ def advance_playbook_optimization_stage( ), ) else: + active_stages = _ACTIVE_STAGES_BY_OPTIMIZER.get(optimizer_kind, ()) + placeholders = ", ".join("?" for _ in active_stages) cur = self.conn.execute( - """UPDATE playbook_optimization_jobs + f"""UPDATE playbook_optimization_jobs SET stage = ?, terminal_outcome = ?, status = ?, @@ -778,23 +892,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', - 'candidate_generated', - 'replay_running', - 'replay_evaluated', - '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(): @@ -879,6 +990,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..d7bc7f69 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,53 @@ 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. + + Raises: + StorageError: If a stored row fails to validate against + ``OpenWorldQualificationRecord`` when reconstructed. + """ + 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_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/models/test_open_world_qualification_record.py b/tests/models/test_open_world_qualification_record.py new file mode 100644 index 00000000..d55f3c48 --- /dev/null +++ b/tests/models/test_open_world_qualification_record.py @@ -0,0 +1,251 @@ +"""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_passed_true_is_rejected_when_one_class_failed() -> None: + 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_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: + 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] + + +# --------------------------------------------------------------------------- +# 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/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. 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..f09cc8e3 --- /dev/null +++ b/tests/server/services/storage/sqlite_storage/test_open_world_analysis_identity.py @@ -0,0 +1,264 @@ +"""Generic SQLite identity coverage for the open-world analysis vocabulary.""" + +from __future__ import annotations + +import sqlite3 +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_open_world", + 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( + ("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 + + +@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: + 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", + [ + "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/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..d436242a --- /dev/null +++ b/tests/server/services/storage/sqlite_storage/test_open_world_qualification_cache.py @@ -0,0 +1,513 @@ +"""SQLite contract for the immutable open-world qualification cache.""" + +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +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 + +pytestmark = pytest.mark.integration + +_COMPONENT_DIGEST = "a" * 64 +_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: + 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_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, +) -> None: + class_counts = [count.model_dump() for count in _class_counts()] + 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, + passed, class_counts_json, observation_digests_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + _COMPONENT_DIGEST, + _SUITE_DIGEST, + schemas.OPEN_WORLD_QUALIFICATION_RECORD_SCHEMA_VERSION, + _RESULT_DIGEST, + 1, + json.dumps(class_counts), + json.dumps([]), + 1_700_000_000, + ), + ) + storage.conn.commit() + + 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, + ) + + +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}, + { + "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] + + ( + 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() + + +# --------------------------------------------------------------------------- +# 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", + "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) 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..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 @@ -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,114 @@ 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", +) +_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", + "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", + "analyst_unqualified", + "stale_incumbent", + "governance_invalidated", + } + ), + "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( @@ -122,12 +231,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( @@ -253,6 +363,9 @@ def test_stage_advancement_is_linear(storage: BaseStorage) -> None: [ ("abstained", "candidate_did_not_improve", "skipped"), ("failed", "generation_failed", "failed"), + ("failed", "replay_failed", "failed"), + ("failed", "publication_failed", "failed"), + ("failed", "infrastructure_failure", "failed"), ], ) def test_terminal_stage_records_outcome_and_releases_lease( @@ -587,20 +700,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") @@ -846,6 +966,278 @@ 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, + optimizer_kind: schemas.OptimizerKind = "offline_tuner_replay", + target_id: int = 41, +) -> tuple[int, int]: + 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, + 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 + + +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 + 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 + ) + before = ( + _optimization_job_row(storage, job_id) if not expected else None + ) + + assert ( + storage.advance_playbook_optimization_stage( + job_id=job_id, + fence=fence, + stage=cast(schemas.OptimizationJobStage, target_stage), + now=7_001, + ) + 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) + assert persisted.status == "running" + + +def test_optimizer_kind_terminal_outcome_matrix_is_exact( + storage: BaseStorage, +) -> 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 + ] + ) + before = ( + _optimization_job_row(storage, job_id) if not expected else None + ) + + 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 + ) + 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 == ( + 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") + ) + before = ( + _optimization_job_row(storage, job_id) if not expected else None + ) + + 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 + ) + 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) + 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( + storage: BaseStorage, +) -> None: + 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: + 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(