diff --git a/.github/product-version-operations/l3-serial-execution-slot-migration-20261001.json b/.github/product-version-operations/l3-serial-execution-slot-migration-20261001.json new file mode 100644 index 00000000..f9414f77 --- /dev/null +++ b/.github/product-version-operations/l3-serial-execution-slot-migration-20261001.json @@ -0,0 +1,16 @@ +{ + "baseline_version": "2.7.45", + "classification_rationale": "L3-FORGE-SERIAL-EXECUTION-SLOT-MIGRATION-V1-20261001", + "component": "product", + "event_lineage": "L3-FORGE-SERIAL-EXECUTION-SLOT-MIGRATION-V1-20261001", + "expected_head": "6db6a6888a08b2ac0ca3cded937e33cf9395ef8d", + "operation_id": "l3-serial-execution-slot-migration-20261001", + "policy_revision": "forge-bootstrap-release-cadence-v2", + "product": "forge", + "projection_paths": "product-version.json", + "release_class": "PATCH", + "requested_bump": "patch", + "requested_version": null, + "schema_version": "1", + "target_version": "2.7.46" +} diff --git a/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md b/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md index bae1c3a0..c1cc0c9a 100644 --- a/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md +++ b/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md @@ -53,6 +53,24 @@ credential-bearing request fields are omitted. Missing correlation is `UNAVAILABLE`; malformed, contradictory or multi-active records are `UNSUPPORTED` with a reason and no selected Action. The view never implies a new per-Action slot, EP admission, target verification, or host effect. A -future PA-F1 migration must still prove lossless persistence and -reconciliation of actual serial execution state before claiming full -multi-slot support. +this compatibility view alone does not prove lossless persistence or +reconciliation of actual serial execution state for full multi-slot support. + +## Explicit serial execution-slot migration subset + +Schema 42 adds a separate immutable `mission_action_execution_slots` table. +`RuntimeDatabase.migrate_legacy_serial_execution_slot` copies the complete +stored legacy correlation into one Action-keyed record only when the current +Mission, Action and correlation identity are unambiguous. It pins the source +revision and canonical correlation digest; identical replay is idempotent and +changed source bytes or revision conflict. It neither changes the legacy +Mission state nor submits or replays an execution request. No installation is +automatically migrated. + +Authenticated Mission detail exposes only `execution_slots` migration status, +Action/correlation/run identity and source revision. Digests derived from the +private request are kept inside the runtime. The stored request and +private prompt remain inside the Forge runtime; malformed snapshots conflict +without exposing them. Missing or ambiguous migration stays explicit. This +subset does not qualify full PA-F1, runtime use of multiple execution slots, +target grants, admission, or parallel dispatch. diff --git a/forge/api/operations-read-openapi-v1.json b/forge/api/operations-read-openapi-v1.json index 2c279934..cd73e5e2 100644 --- a/forge/api/operations-read-openapi-v1.json +++ b/forge/api/operations-read-openapi-v1.json @@ -555,6 +555,33 @@ ], "type": "object" }, + "execution_slots": { + "additionalProperties": false, + "properties": { + "actions": { + "items": { + "additionalProperties": false, + "properties": { + "action_id": {"type": "string"}, + "correlation_id": {"type": "string"}, + "dispatch_authorized": {"const": false}, + "host_run_id": {"type": ["string", "null"]}, + "source_revision": {"type": "integer", "minimum": 1} + }, + "required": ["action_id", "correlation_id", "dispatch_authorized", "host_run_id", "source_revision"], + "type": "object" + }, + "type": "array" + }, + "contract_version": {"const": "serial-execution-slot-migration/v1"}, + "dispatch_authorized": {"const": false}, + "read_only": {"const": true}, + "reason": {"type": ["string", "null"]}, + "status": {"enum": ["MIGRATED", "UNAVAILABLE", "UNSUPPORTED"]} + }, + "required": ["actions", "contract_version", "dispatch_authorized", "read_only", "reason", "status"], + "type": "object" + }, "mission_id": { "minLength": 1, "type": "string" @@ -575,6 +602,7 @@ "action_ids", "criteria", "actions", + "execution_slots", "evidence_lineage" ], "type": "object" diff --git a/forge/operations_read_api.py b/forge/operations_read_api.py index 379a7e07..d31dbac0 100644 --- a/forge/operations_read_api.py +++ b/forge/operations_read_api.py @@ -190,6 +190,61 @@ def _planning_slots_match_current_mission(document: Mapping[str, Any], state: An return all(item["target"]["repository_id"] in scope for item in proposed) +def _project_serial_execution_slots(rows: list[sqlite3.Row], state: Any) -> dict[str, Any]: + """Expose only proven identities from the immutable legacy migration record.""" + from types import SimpleNamespace + + base = {"contract_version": "serial-execution-slot-migration/v1", "read_only": True, + "dispatch_authorized": False, "status": "UNAVAILABLE", "reason": "NOT_MIGRATED", + "actions": []} + if not rows: + serial = project_serial_correlation(state) + if serial["binding_status"] == "UNSUPPORTED": + return {**base, "status": "UNSUPPORTED", "reason": serial["reason"]} + return base + if len(rows) != 1: + raise OperationsProjectionError("MISSION_EXECUTION_SLOTS_INVALID", "Mission execution slots are inconsistent", status=409) + row = rows[0] + encoded = row[5] + try: + document = json.loads(encoded) + except (TypeError, ValueError): + document = None + expected_digest = "sha256:" + sha256(encoded.encode("utf-8")).hexdigest() + if (not isinstance(document, dict) + or set(document) != {"contract_version", "mission_id", "action_id", "source_revision", + "source_digest", "action_status", "correlation", "dispatch_authorized"} + or document["contract_version"] != base["contract_version"] + or document["mission_id"] != state.mission_id or document["mission_id"] != row[0] + or document["action_id"] != row[1] + or not isinstance(document["source_revision"], int) + or isinstance(document["source_revision"], bool) + or document["source_revision"] < 1 + or document["source_revision"] != row[2] + or not isinstance(document["action_status"], str) + or not document["action_status"] + or document["source_digest"] != row[3] + or row[4] != expected_digest or document["dispatch_authorized"] is not False + or not isinstance(document["correlation"], dict) + or document["source_digest"] != "sha256:" + sha256(json.dumps( + document["correlation"], sort_keys=True, separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8")).hexdigest()): + raise OperationsProjectionError("MISSION_EXECUTION_SLOTS_INVALID", "Mission execution slots are inconsistent", status=409) + binding = project_serial_correlation(SimpleNamespace( + mission_id=row[0], revision=row[2], + actions=({"id": row[1], "status": document["action_status"]},), + execution_correlation=document["correlation"], + )) + if binding["binding_status"] != "BOUND" or binding["action_id"] != row[1]: + raise OperationsProjectionError("MISSION_EXECUTION_SLOTS_INVALID", "Mission execution slots are inconsistent", status=409) + return {**base, "status": "MIGRATED", "reason": None, "actions": [{ + "action_id": row[1], "correlation_id": binding["correlation_id"], + "host_run_id": binding["host_run_id"], "source_revision": row[2], + "dispatch_authorized": False, + }]} + + def _project_assessment(document: object) -> dict[str, Any] | None: if not isinstance(document, Mapping): return None @@ -269,6 +324,10 @@ def mission_detail(self, mission_id: str) -> dict[str, Any]: "SELECT mission_revision,document_digest,document FROM mission_action_slot_snapshots " "WHERE mission_id=? ORDER BY mission_revision DESC LIMIT 1", (mission_id,) ).fetchone() + execution_rows = connection.execute( + "SELECT mission_id,action_id,source_revision,source_digest,document_digest,document " + "FROM mission_action_execution_slots WHERE mission_id=? ORDER BY action_id", (mission_id,) + ).fetchall() except ValueError as error: if str(error) == "unknown Mission": raise OperationsProjectionError("MISSION_MISSING", "Mission was not found", status=404) from None @@ -328,6 +387,7 @@ def mission_detail(self, mission_id: str) -> dict[str, Any]: "criteria": [_safe_text(item) for item in state.mission.get("acceptance_criteria", ())], "actions": [_project_action(item) for item in state.actions], "planning_slots": planning_slots, + "execution_slots": _project_serial_execution_slots(execution_rows, state), "serial_correlation": project_serial_correlation(state), "evidence_lineage": { "execution_evidence": _project_execution_evidence(state.execution_evidence), diff --git a/forge/runtime/database.py b/forge/runtime/database.py index 5e40381a..f4e4f81b 100644 --- a/forge/runtime/database.py +++ b/forge/runtime/database.py @@ -22,7 +22,7 @@ canonical_repository_root, repository_identity, repository_uuid) -RUNTIME_SCHEMA_VERSION = 41 +RUNTIME_SCHEMA_VERSION = 42 _REQUIRED_METADATA = frozenset(( "schema_version", "migration_version", "forge_version", "created_at", "last_migration", "integrity_status", @@ -30,7 +30,7 @@ "database_location", "last_access_at", "status", "instance_version", "initialization_version", )) _TABLES = frozenset(( - "mission_state", "mission_action_slot_snapshots", "mission_runtime_projections", "execution_context_snapshots", "architecture_reviews", "mission_recommendations", + "mission_state", "mission_action_slot_snapshots", "mission_action_execution_slots", "mission_runtime_projections", "execution_context_snapshots", "architecture_reviews", "mission_recommendations", "decision_evidence", "execution_receipts", "planning_state", "bootstrap_portfolio_state", "mission_lifecycle_events", "dispatcher_state", "runtime_metadata", "delegation_requests", "integration_evidence", "mission_id_allocations", "mission_intake_evidence", @@ -43,6 +43,15 @@ "operational_reset_state", "operational_reset_operations", "operational_reset_audit", "operational_reset_tombstones", "operational_reset_artifact_steps", )) + + +def _tables_for_schema() -> frozenset[str]: + tables = _TABLES + if RUNTIME_SCHEMA_VERSION < 41: + tables = tables - {"mission_action_slot_snapshots"} + if RUNTIME_SCHEMA_VERSION < 42: + tables = tables - {"mission_action_execution_slots"} + return tables _OPERATIONAL_RESET_TABLES = frozenset(( "operational_reset_state", "operational_reset_operations", "operational_reset_audit", "operational_reset_tombstones", "operational_reset_artifact_steps", @@ -516,6 +525,29 @@ def _require_action_slot_structure(self) -> None: if references != {("mission_state", "mission_id", "mission_id")}: raise RuntimeIntegrityError("Mission Action slot Mission reference is inconsistent") + def _require_execution_slot_structure(self) -> None: + columns = tuple((row["name"], row["type"].upper(), row["notnull"], row["pk"]) + for row in self._connection.execute("PRAGMA table_info(mission_action_execution_slots)")) + if columns != ( + ("mission_id", "TEXT", 1, 1), ("action_id", "TEXT", 1, 2), + ("source_revision", "INTEGER", 1, 0), ("source_digest", "TEXT", 1, 0), + ("document_digest", "TEXT", 1, 0), ("document", "TEXT", 1, 0), + ): + raise RuntimeIntegrityError("Mission execution slot storage shape is inconsistent") + references = {(row["table"], row["from"], row["to"]) + for row in self._connection.execute("PRAGMA foreign_key_list(mission_action_execution_slots)")} + if references != {("mission_state", "mission_id", "mission_id")}: + raise RuntimeIntegrityError("Mission execution slot Mission reference is inconsistent") + for operation in ("update", "delete"): + trigger = self._connection.execute( + "SELECT sql FROM sqlite_master WHERE type='trigger' AND name=?", + (f"mission_action_execution_slots_immutable_{operation}",), + ).fetchone() + sql = "" if trigger is None or trigger[0] is None else " ".join(trigger[0].lower().split()) + if (trigger is None or f"before {operation}" not in sql + or "raise(abort, 'mission execution slots are immutable')" not in sql): + raise RuntimeIntegrityError("Mission execution slot immutability is inconsistent") + def _require_operational_reset_structure(self) -> None: required_columns = { "operational_reset_state": { @@ -545,7 +577,7 @@ def _require_operational_reset_structure(self) -> None: ).fetchall() if len(state_rows) != 1 or int(state_rows[0]["singleton"]) != 1 or int(state_rows[0]["dataset_generation"]) < 0: raise RuntimeIntegrityError("operational reset durable state is inconsistent") - reset_tables = _TABLES if RUNTIME_SCHEMA_VERSION >= 41 else _TABLES - {"mission_action_slot_snapshots"} + reset_tables = _tables_for_schema() for table in sorted(reset_tables): prefix = "operational_reset_authorize" if table in _OPERATIONAL_RESET_TABLES else "operational_reset_block" for operation in ("insert", "update", "delete"): @@ -1944,6 +1976,42 @@ def _migrate(self, forge_version: str) -> None: except Exception: self._connection.rollback() raise + elif version == 41: + active = self._connection.execute( + "SELECT active_operation_id FROM operational_reset_state WHERE singleton=1" + ).fetchone() + if active is not None and active[0] is not None: + raise RuntimeMaintenanceActive("cannot migrate during operational reset maintenance") + try: + self._connection.execute("BEGIN IMMEDIATE") + self._connection.execute( + "CREATE TABLE IF NOT EXISTS mission_action_execution_slots (" + "mission_id TEXT NOT NULL, action_id TEXT NOT NULL," + "source_revision INTEGER NOT NULL, source_digest TEXT NOT NULL," + "document_digest TEXT NOT NULL, document TEXT NOT NULL," + "PRIMARY KEY (mission_id, action_id)," + "FOREIGN KEY (mission_id) REFERENCES mission_state(mission_id))" + ) + for operation in ("UPDATE", "DELETE"): + self._connection.execute( + f"CREATE TRIGGER IF NOT EXISTS mission_action_execution_slots_immutable_{operation.lower()} " + f"BEFORE {operation} ON mission_action_execution_slots " + "BEGIN SELECT RAISE(ABORT, 'Mission execution slots are immutable'); END" + ) + for operation in ("INSERT", "UPDATE", "DELETE"): + self._connection.execute( + f"CREATE TRIGGER IF NOT EXISTS operational_reset_block_mission_action_execution_slots_{operation.lower()} " + f"BEFORE {operation} ON mission_action_execution_slots " + "WHEN (SELECT active_operation_id FROM operational_reset_state WHERE singleton=1) IS NOT NULL " + "BEGIN SELECT RAISE(ABORT, 'Forge operational reset maintenance is active'); END" + ) + self._set_metadata({"schema_version": "42", "migration_version": "42", + "last_migration": "42", "forge_version": forge_version}) + self._connection.execute("PRAGMA user_version=42") + self._connection.commit() + except Exception: + self._connection.rollback() + raise elif version != RUNTIME_SCHEMA_VERSION: raise RuntimeIntegrityError("runtime database migration path is unavailable") @@ -2072,7 +2140,7 @@ def validate_integrity(self, *, record_status: bool = True) -> None: if check != "ok": raise RuntimeIntegrityError("SQLite integrity check failed") tables = {row["name"] for row in self._connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} - required_tables = _TABLES if RUNTIME_SCHEMA_VERSION >= 41 else _TABLES - {"mission_action_slot_snapshots"} + required_tables = _tables_for_schema() if not required_tables <= tables: raise RuntimeIntegrityError("runtime database schema is incomplete") metadata = self.metadata @@ -2089,6 +2157,8 @@ def validate_integrity(self, *, record_status: bool = True) -> None: self._require_operational_log_structure() if RUNTIME_SCHEMA_VERSION >= 41: self._require_action_slot_structure() + if RUNTIME_SCHEMA_VERSION >= 42: + self._require_execution_slot_structure() self._require_operational_reset_structure() identity = self.runtime_identity expected_identity = "forge-installation" if self._installation_scoped else repository_identity(self.repository_root) @@ -2243,6 +2313,68 @@ def record_mission_action_slots(self, graph: Mapping[str, Any]) -> dict[str, Any "document_digest": digest, "target_verification": "UNVERIFIED", "dispatch_authorized": False} + def migrate_legacy_serial_execution_slot(self, mission_id: str) -> dict[str, Any]: + """Copy one proven legacy correlation to an immutable, non-dispatchable Action slot.""" + from types import SimpleNamespace + from forge.serial_correlation_projection import project_serial_correlation + + if not isinstance(mission_id, str) or not mission_id: + raise RuntimeDatabaseError("serial slot migration requires a Mission identity") + self._reject_reset_retired_identity("mission_id", mission_id) + with self._connection: + self._connection.execute("BEGIN IMMEDIATE") + row = self._connection.execute( + "SELECT document FROM mission_state WHERE mission_id=?", (mission_id,) + ).fetchone() + if row is None: + raise RuntimeDatabaseError("serial slot migration requires an existing Mission") + state = json.loads(row["document"]) + revision = state.get("revision") + mission = state.get("mission") + if (state.get("mission_id") != mission_id or not isinstance(mission, dict) + or mission.get("id") != mission_id or not isinstance(revision, int) + or isinstance(revision, bool) or revision < 1): + raise RuntimeDatabaseError("serial slot migration Mission identity is inconsistent") + correlation = state.get("execution_correlation") + binding = project_serial_correlation(SimpleNamespace( + mission_id=mission_id, revision=revision, actions=state.get("actions"), + execution_correlation=correlation, + )) + if (binding["binding_status"] != "BOUND" + or not isinstance(binding["action_status"], str) + or not binding["action_status"]): + raise RuntimeDatabaseError("serial slot migration correlation is unavailable or ambiguous") + correlation_encoded = self._dump(correlation) + source_digest = "sha256:" + sha256(correlation_encoded.encode("utf-8")).hexdigest() + action_id = binding["action_id"] + document = { + "contract_version": "serial-execution-slot-migration/v1", + "mission_id": mission_id, "action_id": action_id, + "source_revision": revision, "source_digest": source_digest, + "action_status": binding["action_status"], + "correlation": correlation, + "dispatch_authorized": False, + } + encoded = self._dump(document) + digest = "sha256:" + sha256(encoded.encode("utf-8")).hexdigest() + prior = self._connection.execute( + "SELECT action_id,source_revision,source_digest,document_digest,document " + "FROM mission_action_execution_slots WHERE mission_id=?", (mission_id,) + ).fetchall() + if prior: + if (len(prior) != 1 or prior[0]["action_id"] != action_id + or prior[0]["source_revision"] != revision + or prior[0]["source_digest"] != source_digest + or prior[0]["document_digest"] != digest or prior[0]["document"] != encoded): + raise RuntimeDatabaseError("serial slot migration conflicts with immutable source") + else: + self._connection.execute( + "INSERT INTO mission_action_execution_slots VALUES (?,?,?,?,?,?)", + (mission_id, action_id, revision, source_digest, digest, encoded), + ) + return {"mission_id": mission_id, "action_id": action_id, + "source_revision": revision, "dispatch_authorized": False} + def create_mission_state(self, state: Any) -> dict[str, Any]: """Create one Mission state without replacing an existing record.""" document = _document(state, "mission state") diff --git a/forge/runtime/operational_reset.py b/forge/runtime/operational_reset.py index a9ebe08f..6c921cdc 100644 --- a/forge/runtime/operational_reset.py +++ b/forge/runtime/operational_reset.py @@ -65,6 +65,7 @@ class OperationalResetError(RuntimeError): # C — old planning/execution history removed from the active generation. "mission_state": "OPERATIONAL_HISTORY", "mission_action_slot_snapshots": "OPERATIONAL_HISTORY", + "mission_action_execution_slots": "OPERATIONAL_HISTORY", "execution_context_snapshots": "OPERATIONAL_HISTORY", "architecture_reviews": "OPERATIONAL_HISTORY", "mission_recommendations": "OPERATIONAL_HISTORY", @@ -129,6 +130,7 @@ class OperationalResetError(RuntimeError): "mission_intake_evidence", "mission_runtime_projections", "mission_action_slot_snapshots", + "mission_action_execution_slots", "execution_context_snapshots", "action_derivations", "scheduler_submissions", @@ -157,6 +159,7 @@ class OperationalResetError(RuntimeError): } _IMMUTABLE_DELETE_TRIGGERS: Mapping[str, tuple[str, ...]] = { + "mission_action_execution_slots": ("mission_action_execution_slots_immutable_delete",), "execution_context_snapshots": ("execution_context_snapshots_immutable_delete",), "action_derivation_canary_closures": ("action_derivation_canary_closures_immutable_delete",), "action_derivation_results": ("action_derivation_results_immutable_delete",), diff --git a/forge/serial_correlation_projection.py b/forge/serial_correlation_projection.py index e92bd922..3027dd40 100644 --- a/forge/serial_correlation_projection.py +++ b/forge/serial_correlation_projection.py @@ -30,7 +30,8 @@ def project_serial_correlation(state: Any) -> dict[str, Any]: actions = state.actions if (not isinstance(actions, (tuple, list)) or not actions or any(not isinstance(action, Mapping) or not isinstance(action.get("id"), str) - or not action["id"] for action in actions) + or not action["id"] or not isinstance(action.get("status"), str) + for action in actions) or len({action["id"] for action in actions}) != len(actions)): return {**base, "binding_status": "UNSUPPORTED", "reason": "AMBIGUOUS_ACTION_SET"} in_flight = [action for action in actions if action.get("status") in _IN_FLIGHT] diff --git a/product-version.json b/product-version.json index ef3ecf82..7f44e1c2 100644 --- a/product-version.json +++ b/product-version.json @@ -1,5 +1,5 @@ { "product": "forge", "schema_version": 1, - "version": "2.7.45" + "version": "2.7.46" } diff --git a/tests/test_operations_read_api.py b/tests/test_operations_read_api.py index 67b61dd8..8909e867 100644 --- a/tests/test_operations_read_api.py +++ b/tests/test_operations_read_api.py @@ -289,6 +289,35 @@ def test_schema40_migration_keeps_serial_mission_without_invented_slots(self) -> self.assertEqual(response.status, 200) self.assertIsNone(response.body["mission"]["planning_slots"]) + def test_schema41_migration_preserves_serial_state_without_invented_execution_slot(self) -> None: + self.create_mission() + state = self.database.get_document("mission_state", "MISSION-0042") + state["actions"][0]["status"] = "WAITING_FOR_RESULT" + state["execution_correlation"] = { + "mission_id": "MISSION-0042", "action_id": "ACTION-0042-A", + "correlation_id": "corr-0042", "runtime_prompt": "synthetic-private-prompt", + } + self.database.save_mission_state(state) + prior = self.database.get_document("mission_state", "MISSION-0042") + path = self.database.path + self.database.close() + with sqlite3.connect(path) as connection: + connection.execute("DROP TABLE mission_action_execution_slots") + connection.execute( + "UPDATE runtime_metadata SET value='41' WHERE key IN " + "('schema_version','migration_version','last_migration','database_version')" + ) + connection.execute("PRAGMA user_version=41") + self.database = RuntimeBootstrap(data_root=self.root, forge_version="test").open() + self.assertEqual(self.database.get_document("mission_state", "MISSION-0042"), prior) + self.assertEqual(self.database._connection.execute( # noqa: SLF001 + "SELECT COUNT(*) FROM mission_action_execution_slots" + ).fetchone()[0], 0) + response = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(response.status, 200) + self.assertEqual(response.body["mission"]["execution_slots"]["status"], "UNAVAILABLE") + self.assertNotIn("synthetic-private-prompt", json.dumps(response.body)) + def test_installed_mission_readback_binds_only_one_legacy_correlation(self) -> None: self.create_mission() document = self.database.get_document("mission_state", "MISSION-0042") @@ -315,6 +344,133 @@ def test_installed_mission_readback_binds_only_one_legacy_correlation(self) -> N "MULTIPLE_IN_FLIGHT_ACTIONS") self.assertNotIn("action_id", ambiguous.body["mission"]["serial_correlation"]) + def test_serial_execution_slot_migration_replay_restart_and_private_readback(self) -> None: + self.create_mission() + state = self.database.get_document("mission_state", "MISSION-0042") + state["actions"][0]["status"] = "WAITING_FOR_RESULT" + correlation = { + "request": {"mission_id": "MISSION-0042", "action_id": "ACTION-0042-A", + "correlation_id": "corr-0042", "runtime_prompt": "synthetic-private-prompt", + "credential_reference": "synthetic-private-credential"}, + "host_run_id": "run-0042", "receipt_id": "receipt-0042", + } + state["execution_correlation"] = correlation + self.database.save_mission_state(state) + receipt = self.database.migrate_legacy_serial_execution_slot("MISSION-0042") + self.assertEqual(self.database.migrate_legacy_serial_execution_slot("MISSION-0042"), receipt) + self.assertNotIn("source_digest", receipt) + self.assertNotIn("document_digest", receipt) + row = self.database._connection.execute( # noqa: SLF001 + "SELECT document FROM mission_action_execution_slots WHERE mission_id='MISSION-0042'" + ).fetchone() + self.assertEqual(json.loads(row[0])["correlation"], correlation) + self.assertEqual(self.database.get_document("mission_state", "MISSION-0042"), state) + self.database.close() + self.database = RuntimeBootstrap(data_root=self.root, forge_version="test").open() + before = self.snapshot() + response = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(response.status, 200) + slots = response.body["mission"]["execution_slots"] + self.assertEqual(slots["status"], "MIGRATED") + self.assertEqual(slots["actions"][0]["action_id"], "ACTION-0042-A") + self.assertEqual(slots["actions"][0]["correlation_id"], "corr-0042") + self.assertEqual(slots["actions"][0]["host_run_id"], "run-0042") + self.assertFalse(slots["dispatch_authorized"]) + self.assertNotIn("source_digest", json.dumps(slots)) + self.assertNotIn("document_digest", json.dumps(slots)) + self.assertNotIn("synthetic-private-prompt", json.dumps(response.body)) + self.assertNotIn("synthetic-private-credential", json.dumps(response.body)) + self.assertEqual(self.snapshot(), before) + denied = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer wrong") + self.assertEqual(denied.status, 401) + changed = json.loads(json.dumps(state)) + changed["execution_correlation"]["host_run_id"] = "different-run" + self.database.save_mission_state(changed) + with self.assertRaisesRegex(RuntimeDatabaseError, "conflicts"): + self.database.migrate_legacy_serial_execution_slot("MISSION-0042") + self.assertEqual(self.database._connection.execute( # noqa: SLF001 + "SELECT COUNT(*) FROM mission_action_execution_slots" + ).fetchone()[0], 1) + with self.assertRaises(sqlite3.IntegrityError): + self.database._connection.execute( # noqa: SLF001 + "DELETE FROM mission_action_execution_slots WHERE mission_id='MISSION-0042'" + ) + later = json.loads(json.dumps(state)) + later["revision"] = 2 + self.database.save_mission_state(later) + with self.assertRaisesRegex(RuntimeDatabaseError, "conflicts"): + self.database.migrate_legacy_serial_execution_slot("MISSION-0042") + historical = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(historical.status, 200) + self.assertEqual(historical.body["mission"]["execution_slots"]["actions"][0]["source_revision"], 1) + + def test_serial_execution_slot_migration_rejects_ambiguous_identity(self) -> None: + self.create_mission() + original = self.database.get_document("mission_state", "MISSION-0042") + for change in ("absent", "ambiguous", "wrong_mission", "wrong_action", + "missing_correlation", "malformed_status"): + with self.subTest(change=change): + state = json.loads(json.dumps(original)) + state["actions"][0]["status"] = "WAITING_FOR_RESULT" + state["execution_correlation"] = { + "mission_id": "MISSION-0042", "action_id": "ACTION-0042-A", + "correlation_id": "corr-0042", + } + if change == "absent": + state["execution_correlation"] = None + elif change == "ambiguous": + state["actions"][1]["status"] = "WAITING_FOR_RESULT" + elif change == "wrong_mission": + state["execution_correlation"]["mission_id"] = "MISSION-OTHER" + elif change == "wrong_action": + state["execution_correlation"]["action_id"] = "ACTION-0042-B" + elif change == "missing_correlation": + state["execution_correlation"].pop("correlation_id") + elif change == "malformed_status": + state["actions"].pop() + state["actions"][0]["status"] = {"state": "WAITING_FOR_RESULT"} + self.database.save_mission_state(state) + changes = self.database._connection.total_changes # noqa: SLF001 + with self.assertRaisesRegex(RuntimeDatabaseError, "unavailable or ambiguous"): + self.database.migrate_legacy_serial_execution_slot("MISSION-0042") + self.assertEqual(self.database._connection.total_changes, changes) # noqa: SLF001 + self.assertEqual(self.database._connection.execute( # noqa: SLF001 + "SELECT COUNT(*) FROM mission_action_execution_slots" + ).fetchone()[0], 0) + ambiguous = json.loads(json.dumps(original)) + ambiguous["actions"][0]["status"] = "WAITING_FOR_RESULT" + ambiguous["actions"][1]["status"] = "WAITING_FOR_RESULT" + ambiguous["execution_correlation"] = { + "mission_id": "MISSION-0042", "action_id": "ACTION-0042-A", + "correlation_id": "corr-0042", + } + self.database.save_mission_state(ambiguous) + unsupported = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(unsupported.body["mission"]["execution_slots"]["status"], "UNSUPPORTED") + + def test_serial_execution_slot_corrupt_digest_fails_closed(self) -> None: + self.create_mission() + state = self.database.get_document("mission_state", "MISSION-0042") + state["actions"][0]["status"] = "WAITING_FOR_RESULT" + state["execution_correlation"] = { + "mission_id": "MISSION-0042", "action_id": "ACTION-0042-A", + "correlation_id": "corr-0042", "runtime_prompt": "synthetic-private-prompt", + } + self.database.save_mission_state(state) + self.database.migrate_legacy_serial_execution_slot("MISSION-0042") + with self.database._connection: # noqa: SLF001 - corrupt-storage fixture + self.database._connection.execute( # noqa: SLF001 + "DROP TRIGGER mission_action_execution_slots_immutable_update" + ) + self.database._connection.execute( # noqa: SLF001 + "UPDATE mission_action_execution_slots SET document_digest=? WHERE mission_id=?", + ("sha256:" + "0" * 64, "MISSION-0042"), + ) + response = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(response.status, 409) + self.assertEqual(response.body["error"]["code"], "MISSION_EXECUTION_SLOTS_INVALID") + self.assertNotIn("synthetic-private-prompt", json.dumps(response.body)) + def test_same_revision_action_slot_graph_drift_fails_closed(self) -> None: self.create_mission() self.database.record_mission_action_slots(self._slot_graph())