From 15850024678e13664a4edce0830144365abbe990 Mon Sep 17 00:00:00 2001 From: Peter van Tol Date: Thu, 1 Oct 2026 19:22:02 +0200 Subject: [PATCH] Persist revision-bound non-dispatchable Action slots --- .../l3-durable-action-slots-20261001.json | 16 +++ .../PARALLEL_ACTION_DURABLE_SLOTS_V1.md | 28 +++++ forge/operations_read_api.py | 37 ++++++ forge/runtime/database.py | 118 +++++++++++++++++- forge/runtime/operational_reset.py | 2 + product-version.json | 2 +- tests/test_installed_forge_update.py | 26 ++-- tests/test_operations_read_api.py | 108 ++++++++++++++++ 8 files changed, 321 insertions(+), 16 deletions(-) create mode 100644 .github/product-version-operations/l3-durable-action-slots-20261001.json create mode 100644 docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md diff --git a/.github/product-version-operations/l3-durable-action-slots-20261001.json b/.github/product-version-operations/l3-durable-action-slots-20261001.json new file mode 100644 index 00000000..f452eec2 --- /dev/null +++ b/.github/product-version-operations/l3-durable-action-slots-20261001.json @@ -0,0 +1,16 @@ +{ + "baseline_version": "2.7.41", + "classification_rationale": "L3-FORGE-DURABLE-ACTION-SLOTS-V1-20261001", + "component": "product", + "event_lineage": "L3-FORGE-DURABLE-ACTION-SLOTS-V1-20261001", + "expected_head": "8330037ebc99b35b9f423d46efe27397e632cdab", + "operation_id": "l3-durable-action-slots-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.42" +} diff --git a/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md b/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md new file mode 100644 index 00000000..7b0b9897 --- /dev/null +++ b/docs/architecture/PARALLEL_ACTION_DURABLE_SLOTS_V1.md @@ -0,0 +1,28 @@ +# Durable parallel Action planning slots V1 + +This is a bounded `PA-F1` storage and readback subset. It preserves the +existing serial Mission execution path and does not qualify complete PA-F1, +fan-out, reservations, EP admission, host execution, or parallel dispatch. + +`RuntimeDatabase.record_mission_action_slots` accepts the strict +`parallel-action-graph/v1` document. It requires an existing Architecture- +approved Mission at the exact current state revision, the exact stored Action +ID set, and every target repository inside the approved Mission scope. The +validator checks target and baseline shape, predecessor identity, repository- +bound evidence digests, and a cycle-free graph. Scope membership is only a +planning containment check; it is not an EP binding or execution grant. + +One immutable normalized document is stored per Mission revision in the +canonical SQLite runtime. Identical replay returns the same digest; changed +bytes at that revision fail closed. A later Mission revision can record a new +document, while the earlier document remains available for audit. Schema 41 +adds this table without changing legacy serial Mission records. Operational +reset maintenance fences its writes. + +Authenticated `GET /v1/missions/{mission_id}` includes `planning_slots` from +the same read-only snapshot, or `null` when no document exists. It verifies +the stored digest and contract, reports the revision as `CURRENT` or `STALE`, +and always reports `target_verification: UNVERIFIED` and +`dispatch_authorized: false`. A slot never supplies correlation, admission, +or execution evidence. Installed artifact qualification and full PA-F1/F2/F3 +remain separate roadmap work. diff --git a/forge/operations_read_api.py b/forge/operations_read_api.py index 71e5eb04..a5d072c3 100644 --- a/forge/operations_read_api.py +++ b/forge/operations_read_api.py @@ -9,6 +9,7 @@ from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime, timedelta +from hashlib import sha256 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path @@ -25,6 +26,7 @@ from .installed_health import InstalledHealthError, InstalledHealthSnapshotService from .mission_cli import _status_projection as mission_status_projection from .models.producer import redact_action_summary +from .parallel_action_contract import ParallelActionContractError, validate_peer_graph from .runtime.data_root import DataRootResolver @@ -230,6 +232,10 @@ def mission_detail(self, mission_id: str) -> dict[str, Any]: try: with self._runtime_snapshot() as (connection, _metadata): projection, state = mission_status_projection(connection, mission_id) + slot_row = connection.execute( + "SELECT mission_revision,document_digest,document FROM mission_action_slot_snapshots " + "WHERE mission_id=? ORDER BY mission_revision DESC LIMIT 1", (mission_id,) + ).fetchone() except ValueError as error: if str(error) == "unknown Mission": raise OperationsProjectionError("MISSION_MISSING", "Mission was not found", status=404) from None @@ -241,6 +247,36 @@ def mission_detail(self, mission_id: str) -> dict[str, Any]: raise OperationsProjectionError( "MISSION_AMBIGUOUS", "Mission identity is inconsistent", status=409, ) + planning_slots: dict[str, Any] | None = None + if slot_row is not None: + encoded = slot_row[2] + expected_digest = "sha256:" + sha256(encoded.encode("utf-8")).hexdigest() + try: + document = json.loads(encoded) + except (TypeError, ValueError): + document = None + try: + validated = validate_peer_graph({key: value for key, value in document.items() + if key != "dispatch_authorized"}) if isinstance(document, dict) else None + except ParallelActionContractError: + validated = None + if (slot_row[1] != expected_digest or not isinstance(document, dict) + or document != validated + or document.get("mission_id") != mission_id + or document.get("mission_revision") != slot_row[0] + or document.get("dispatch_authorized") is not False): + raise OperationsProjectionError( + "MISSION_ACTION_SLOTS_INVALID", "Mission Action slots are inconsistent", status=409, + ) + planning_slots = { + "contract_version": document.get("contract_version"), + "mission_revision": slot_row[0], + "document_digest": expected_digest, + "freshness": "CURRENT" if slot_row[0] == state.revision else "STALE", + "target_verification": "UNVERIFIED", + "dispatch_authorized": False, + "actions": document.get("actions"), + } times = [item.get("occurred_at") for item in state.state_history if isinstance(item, Mapping)] valid_times = [item for item in times if _parse_time(item) is not None] observed_at = max( @@ -249,6 +285,7 @@ def mission_detail(self, mission_id: str) -> dict[str, Any]: projection.update({ "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, "evidence_lineage": { "execution_evidence": _project_execution_evidence(state.execution_evidence), "execution_attempts": [_project_execution_evidence(item) for item in state.execution_history], diff --git a/forge/runtime/database.py b/forge/runtime/database.py index 92eaea04..9d6e6a4b 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 = 40 +RUNTIME_SCHEMA_VERSION = 41 _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_runtime_projections", "execution_context_snapshots", "architecture_reviews", "mission_recommendations", + "mission_state", "mission_action_slot_snapshots", "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", @@ -503,6 +503,19 @@ def _require_action_derivation_result_structure(self) -> None: or f"before {operation}" not in sql or fragment not in sql): raise RuntimeIntegrityError(f"durable action-derivation result migration found incompatible {name} trigger") + def _require_action_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_slot_snapshots)")) + if columns != ( + ("mission_id", "TEXT", 1, 1), ("mission_revision", "INTEGER", 1, 2), + ("document_digest", "TEXT", 1, 0), ("document", "TEXT", 1, 0), + ): + raise RuntimeIntegrityError("Mission Action slot storage shape is inconsistent") + references = {(row["table"], row["from"], row["to"]) + for row in self._connection.execute("PRAGMA foreign_key_list(mission_action_slot_snapshots)")} + if references != {("mission_state", "mission_id", "mission_id")}: + raise RuntimeIntegrityError("Mission Action slot Mission reference is inconsistent") + def _require_operational_reset_structure(self) -> None: required_columns = { "operational_reset_state": { @@ -532,7 +545,8 @@ 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") - for table in sorted(_TABLES): + reset_tables = _TABLES if RUNTIME_SCHEMA_VERSION >= 41 else _TABLES - {"mission_action_slot_snapshots"} + 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"): name = f"{prefix}_{table}_{operation}" @@ -1901,6 +1915,35 @@ def _migrate(self, forge_version: str) -> None: except Exception: self._connection.rollback() raise + elif version == 40: + 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_slot_snapshots (" + "mission_id TEXT NOT NULL, mission_revision INTEGER NOT NULL," + "document_digest TEXT NOT NULL, document TEXT NOT NULL," + "PRIMARY KEY (mission_id, mission_revision)," + "FOREIGN KEY (mission_id) REFERENCES mission_state(mission_id))" + ) + for operation in ("INSERT", "UPDATE", "DELETE"): + self._connection.execute( + f"CREATE TRIGGER IF NOT EXISTS operational_reset_block_mission_action_slot_snapshots_{operation.lower()} " + f"BEFORE {operation} ON mission_action_slot_snapshots " + "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": "41", "migration_version": "41", + "last_migration": "41", "forge_version": forge_version}) + self._connection.execute("PRAGMA user_version=41") + self._connection.commit() + except Exception: + self._connection.rollback() + raise elif version != RUNTIME_SCHEMA_VERSION: raise RuntimeIntegrityError("runtime database migration path is unavailable") @@ -2029,7 +2072,8 @@ 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'")} - if not _TABLES <= tables: + required_tables = _TABLES if RUNTIME_SCHEMA_VERSION >= 41 else _TABLES - {"mission_action_slot_snapshots"} + if not required_tables <= tables: raise RuntimeIntegrityError("runtime database schema is incomplete") metadata = self.metadata if not _REQUIRED_METADATA <= metadata.keys(): @@ -2043,6 +2087,8 @@ def validate_integrity(self, *, record_status: bool = True) -> None: self._require_peer_configuration_structure() self._require_execution_host_exchange_audit_structure() self._require_operational_log_structure() + if RUNTIME_SCHEMA_VERSION >= 41: + self._require_action_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) @@ -2123,6 +2169,70 @@ def save_mission_state(self, state: Any) -> dict[str, Any]: self._write_mission_state_in_transaction(document, mission_id, lifecycle, context) return document + def record_mission_action_slots(self, graph: Mapping[str, Any]) -> dict[str, Any]: + """Pin non-dispatchable planning slots to one current, approved Mission revision. + + This records no admission, EP binding, execution grant, or lifecycle change. + A second proposal for the same revision must be byte-equivalent after + canonical validation; a stale revision cannot be written. + """ + from forge.parallel_action_contract import validate_peer_graph + + normalized = validate_peer_graph(graph) + mission_id = normalized["mission_id"] + revision = normalized["mission_revision"] + encoded = self._dump(normalized) + digest = "sha256:" + sha256(encoded.encode("utf-8")).hexdigest() + 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("Action slots require an existing Mission") + state = json.loads(row["document"]) + mission = state.get("mission") + if (state.get("mission_id") != mission_id or state.get("revision") != revision + or not isinstance(mission, dict) or mission.get("id") != mission_id + or mission.get("status") != "approved_for_engineering" + or state.get("status") in {"COMPLETED", "ARCHIVED", "FAILED"}): + raise RuntimeDatabaseError("Action slots require the current approved Mission revision") + stored_actions = state.get("actions") + if (not isinstance(stored_actions, list) + or len(stored_actions) != len(normalized["actions"]) + or {action.get("id") for action in stored_actions if isinstance(action, dict)} + != {action["action_id"] for action in normalized["actions"]}): + raise RuntimeDatabaseError("Action slots must match the current Mission Action set") + proposed = {action["action_id"]: action for action in normalized["actions"]} + for action in stored_actions: + predecessors = action.get("dependencies") + if (not isinstance(predecessors, list) + or set(predecessors) != { + edge["predecessor_action_id"] for edge in proposed[action["id"]]["dependencies"] + }): + raise RuntimeDatabaseError("Action slot predecessors must match the stored Mission graph") + scope = mission.get("scope") + if (not isinstance(scope, list) or not scope + or any(action["target"]["repository_id"] not in scope + for action in normalized["actions"])): + raise RuntimeDatabaseError("Action slot target is outside the approved Mission scope") + previous = self._connection.execute( + "SELECT document_digest,document FROM mission_action_slot_snapshots " + "WHERE mission_id=? AND mission_revision=?", (mission_id, revision) + ).fetchone() + if previous is not None: + if previous["document_digest"] != digest or previous["document"] != encoded: + raise RuntimeDatabaseError("Action slots conflict with the pinned Mission revision") + else: + self._connection.execute( + "INSERT INTO mission_action_slot_snapshots VALUES (?,?,?,?)", + (mission_id, revision, digest, encoded), + ) + return {"mission_id": mission_id, "mission_revision": revision, + "document_digest": digest, "target_verification": "UNVERIFIED", + "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 733860a9..a9ebe08f 100644 --- a/forge/runtime/operational_reset.py +++ b/forge/runtime/operational_reset.py @@ -64,6 +64,7 @@ class OperationalResetError(RuntimeError): "mission_id_allocations": "SECURITY_AND_AUTHORITY_LEDGER", # C — old planning/execution history removed from the active generation. "mission_state": "OPERATIONAL_HISTORY", + "mission_action_slot_snapshots": "OPERATIONAL_HISTORY", "execution_context_snapshots": "OPERATIONAL_HISTORY", "architecture_reviews": "OPERATIONAL_HISTORY", "mission_recommendations": "OPERATIONAL_HISTORY", @@ -127,6 +128,7 @@ class OperationalResetError(RuntimeError): "integration_evidence", "mission_intake_evidence", "mission_runtime_projections", + "mission_action_slot_snapshots", "execution_context_snapshots", "action_derivations", "scheduler_submissions", diff --git a/product-version.json b/product-version.json index 6ac6c40e..be37a67a 100644 --- a/product-version.json +++ b/product-version.json @@ -1,5 +1,5 @@ { "product": "forge", "schema_version": 1, - "version": "2.7.41" + "version": "2.7.42" } diff --git a/tests/test_installed_forge_update.py b/tests/test_installed_forge_update.py index 6d3cdc64..1f39b5a6 100644 --- a/tests/test_installed_forge_update.py +++ b/tests/test_installed_forge_update.py @@ -225,7 +225,8 @@ def test_2738_to_2739_preserves_unpaired_instance_and_adds_peer_lifecycle_tables "peer_configuration_digest": before["peer_state_digest"], }) self.assertEqual(update.transition_schemas(request), (39, 40)) - database = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.39").open() + with patch.object(runtime_database, "RUNTIME_SCHEMA_VERSION", 40): + database = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.39").open() database.close() after = update.database_snapshot(self.data_root / "forge.db") proof = update.verify_preservation(before, after, request) @@ -233,18 +234,20 @@ def test_2738_to_2739_preserves_unpaired_instance_and_adds_peer_lifecycle_tables self.assertEqual(after["peer_state_digest"], before["peer_state_digest"]) def test_2738_to_2739_preserves_exact_paired_binding_and_history(self) -> None: - database = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.38").open() + with patch.object(runtime_database, "RUNTIME_SCHEMA_VERSION", 40): + database = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.38").open() runtime_id = database.runtime_identity.runtime_id installation_id = database.metadata["installation_id"] database.close() - configured = EngineeringPlatformPeerConfigurationService(self.data_root).configure( - binding_id="ep-selected", endpoint="https://ep.test", - expected_ep_instance_id="ep-instance", ep_consumer_id="consumer-old", - execution_host_id="ep-host", ep_project_id="project", - ep_repository_id="repository", repository_identity="forge-repository", - credential_reference=SecretReference.parse("keychain://forge.ep/consumer"), - operator_id="operator", - ) + with patch.object(runtime_database, "RUNTIME_SCHEMA_VERSION", 40): + configured = EngineeringPlatformPeerConfigurationService(self.data_root).configure( + binding_id="ep-selected", endpoint="https://ep.test", + expected_ep_instance_id="ep-instance", ep_consumer_id="consumer-old", + execution_host_id="ep-host", ep_project_id="project", + ep_repository_id="repository", repository_identity="forge-repository", + credential_reference=SecretReference.parse("keychain://forge.ep/consumer"), + operator_id="operator", + ) with sqlite3.connect(self.data_root / "forge.db") as connection: connection.execute("DROP TABLE execution_host_peer_detach_operations") connection.execute("DROP TABLE execution_host_peer_generation") @@ -259,7 +262,8 @@ def test_2738_to_2739_preserves_exact_paired_binding_and_history(self) -> None: "runtime_id": runtime_id, "installation_id": installation_id, "peer_configuration_digest": configured.configuration_digest, }) - migrated = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.39").open() + with patch.object(runtime_database, "RUNTIME_SCHEMA_VERSION", 40): + migrated = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.39").open() migrated.close() after = update.database_snapshot(self.data_root / "forge.db") self.assertEqual(update.verify_preservation(before, after, request)["status"], "PASS") diff --git a/tests/test_operations_read_api.py b/tests/test_operations_read_api.py index 75fc93ae..1ab604e7 100644 --- a/tests/test_operations_read_api.py +++ b/tests/test_operations_read_api.py @@ -5,6 +5,7 @@ from http.client import HTTPConnection import json from pathlib import Path +import sqlite3 from tempfile import TemporaryDirectory from threading import Thread from types import SimpleNamespace @@ -20,6 +21,7 @@ from forge.execution_host_configuration import EngineeringPlatformPeerConfigurationService from forge.operations_read_api import InstalledOperationsReadService, OperationsReadAPI, make_server from forge.runtime import RuntimeBootstrap +from forge.runtime.database import RuntimeDatabaseError from forge.secure_store import SecretReference from forge.server_runtime import ForgeServerAPI, make_server as make_forge_server from forge.state import MissionStateStore @@ -157,6 +159,112 @@ def create_mission(self, number: str = "0042") -> None: mission, (intent,), (first, second), occurred_at="2026-09-21T05:00:00Z", ) + def test_durable_action_slots_replay_restart_readback_and_stale_revision(self) -> None: + self.create_mission() + graph = self._slot_graph() + receipt = self.database.record_mission_action_slots(graph) + self.assertEqual(self.database.record_mission_action_slots(graph), receipt) + self.database.close() + self.database = RuntimeBootstrap(data_root=self.root, forge_version="test").open() + response = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(response.status, 200) + slots = response.body["mission"]["planning_slots"] + self.assertEqual(slots["freshness"], "CURRENT") + self.assertEqual(slots["document_digest"], receipt["document_digest"]) + self.assertFalse(slots["dispatch_authorized"]) + self.assertEqual([item["action_id"] for item in slots["actions"]], + ["ACTION-0042-A", "ACTION-0042-B"]) + with self.database._connection: # noqa: SLF001 - simulated later Mission revision + row = self.database._connection.execute( # noqa: SLF001 + "SELECT document FROM mission_state WHERE mission_id='MISSION-0042'" + ).fetchone() + state = json.loads(row[0]) + state["revision"] = 2 + self.database._connection.execute( # noqa: SLF001 + "UPDATE mission_state SET document=? WHERE mission_id='MISSION-0042'", + (json.dumps(state),), + ) + stale = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(stale.body["mission"]["planning_slots"]["freshness"], "STALE") + with self.assertRaisesRegex(RuntimeDatabaseError, "current approved Mission revision"): + self.database.record_mission_action_slots(graph) + + def test_action_slots_reject_conflict_scope_and_unstored_action(self) -> None: + self.create_mission() + graph = self._slot_graph() + self.database.record_mission_action_slots(graph) + changed = json.loads(json.dumps(graph)) + changed["actions"][0]["target"]["baseline_revision"] = "different" + with self.assertRaisesRegex(RuntimeDatabaseError, "conflict"): + self.database.record_mission_action_slots(changed) + outside = json.loads(json.dumps(graph)) + outside["actions"][0]["target"]["repository_id"] = "other-repository" + outside["actions"][1]["dependencies"][0]["required_evidence"]["repository_id"] = "other-repository" + with self.assertRaisesRegex(RuntimeDatabaseError, "scope"): + self.database.record_mission_action_slots(outside) + missing = json.loads(json.dumps(graph)) + missing["actions"][1]["action_id"] = "ACTION-UNKNOWN" + with self.assertRaisesRegex(RuntimeDatabaseError, "Action set"): + self.database.record_mission_action_slots(missing) + altered_edges = json.loads(json.dumps(graph)) + altered_edges["actions"][1]["dependencies"] = [] + with self.assertRaisesRegex(RuntimeDatabaseError, "predecessors"): + self.database.record_mission_action_slots(altered_edges) + count = self.database._connection.execute( # noqa: SLF001 + "SELECT COUNT(*) FROM mission_action_slot_snapshots" + ).fetchone()[0] + self.assertEqual(count, 1) + + def test_action_slots_reject_missing_mission_and_corrupt_readback(self) -> None: + graph = self._slot_graph() + with self.assertRaisesRegex(RuntimeDatabaseError, "existing Mission"): + self.database.record_mission_action_slots(graph) + self.create_mission() + self.database.record_mission_action_slots(graph) + with self.database._connection: # noqa: SLF001 - controlled corrupt-storage fixture + self.database._connection.execute( # noqa: SLF001 + "UPDATE mission_action_slot_snapshots 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_ACTION_SLOTS_INVALID") + + def test_schema40_migration_keeps_serial_mission_without_invented_slots(self) -> None: + self.create_mission() + prior = self.database.get_document("mission_state", "MISSION-0042") + self.database.close() + with sqlite3.connect(self.root / "forge.db") as connection: + connection.execute("DROP TABLE mission_action_slot_snapshots") + connection.execute( + "UPDATE runtime_metadata SET value='40' WHERE key IN " + "('schema_version','migration_version','last_migration','database_version')" + ) + connection.execute("PRAGMA user_version=40") + self.database = RuntimeBootstrap(data_root=self.root, forge_version="test").open() + self.assertEqual(self.database.get_document("mission_state", "MISSION-0042"), prior) + response = self.api.handle("GET", "/v1/missions/MISSION-0042", "Bearer " + CREDENTIAL) + self.assertEqual(response.status, 200) + self.assertIsNone(response.body["mission"]["planning_slots"]) + + @staticmethod + def _slot_graph() -> dict: + return { + "contract_version": "parallel-action-graph/v1", "mission_id": "MISSION-0042", + "mission_revision": 1, "actions": [ + {"action_id": action, "target": { + "ep_instance_id": "ep-instance-1", "project_id": "forge-project", + "repository_id": "forge-repository", "baseline_revision": "baseline-1", + }, "dependencies": [] if action.endswith("-A") else [{ + "predecessor_action_id": "ACTION-0042-A", + "required_evidence": {"kind": "REPOSITORY_REVISION", + "repository_id": "forge-repository", + "content_digest": "sha256:" + "a" * 64}, + }]} + for action in ("ACTION-0042-A", "ACTION-0042-B") + ], + } + def test_unconfigured_project_is_empty_and_read_only(self) -> None: before = self.snapshot() denied = self.api.handle("GET", "/v1/projects", None)