Persist non-dispatchable Mission Action planning slots

.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"
+}

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.

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],

forge/runtime/database.py

@@ -22,15 +22,15 @@
                         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",
     "runtime_id", "installation_id", "repository_identity", "repository_root", "database_version",
     "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")

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",

product-version.json

@@ -1,5 +1,5 @@
 {
   "product": "forge",
   "schema_version": 1,
-  "version": "2.7.41"
+  "version": "2.7.42"
 }