Persist verified legacy execution correlation in Action slot

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

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.

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"

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