agent: baseline 1.1 evidence materialization + builder readiness split

Separate Builder readiness (infrastructure-only) from build (recipe-evidence
gated). Add Evidence Materializer (validator/projector, never invents release
evidence). Unify Source Admission policy digest to exact policy file SHA-256
through one helper/constructor. Add HMAC-audited isolated-peer policy rebind.

Base: 2852c2a403 (exact)
Files: 10 (6 modified + 4 added), +1513/-40
Patch SHA-256: 85be2b4e369a7d39eba0f7b7f1c2641a9eabdf0276d4d48d371cfc348465ed51

active-runtime/app/aios_model_api_staging.py

@@ -265,6 +265,14 @@ class _HostTrustedCandidateBuilderAdapter:
                 raise CapabilityAuthorizationError(
                     'release source admission required'
                 ) from exc
+            if code == 'MIGRATION_PLAN_REQUIRED':
+                raise CapabilityAuthorizationError(
+                    'migration plan required'
+                ) from exc
+            if code == 'MIGRATION_PLAN_INVALID':
+                raise CapabilityAuthorizationError(
+                    'migration plan invalid'
+                ) from exc
             raise TrustedCandidateBuilderUnavailable(
                 'trusted builder request failed'
             ) from exc
@@ -331,7 +339,9 @@ def _build_silent_release_service() -> SilentReleaseActionService:
                 TRUSTED_BUILDER_CREDENTIALS_DIRECTORY / 'builder_hmac'
             ).read_bytes().strip()
             if TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT:
-                from release_source_admission import ReleaseSourceAdmissionStore
+                from release_source_admission import (
+                    open_source_admission_store,
+                )
 
                 try:
                     source_admission_hmac = (
@@ -342,10 +352,10 @@ def _build_silent_release_service() -> SilentReleaseActionService:
                     raise CapabilityAuthorizationError(
                         'release source admission credential is unavailable'
                     ) from exc
-                source_store = ReleaseSourceAdmissionStore(
+                source_store = open_source_admission_store(
                     Path(TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT),
                     repository=builder_config.repository,
-                    policy_sha256=builder_config.policy_sha256,
+                    policy=TRUSTED_BUILDER_POLICY,
                     hmac_key=source_admission_hmac,
                     expected_owner_uid=0,
                     writable=False,

active-runtime/app/aios_trusted_candidate_builder.py

@@ -134,6 +134,7 @@ class TrustedCandidateBuilderConfig:
     governed_recipes: Mapping[str, Mapping[str, str]]
     policy_sha256: str
     source_admission_resolver: Callable[[str, str | None, str], Mapping[str, str]] | None = None
+    migration_plan_resolver: Callable[[str, str, str], Any] | None = None
 
     @classmethod
     def from_policy(cls, policy: Mapping[str, Any]) -> "TrustedCandidateBuilderConfig":
@@ -347,6 +348,35 @@ class ServerOwnedEvidenceSource:
             raise BuilderError("COLLECTOR_EVIDENCE_OWNER_DENIED")
         return resolved
 
+    def assert_ready(self) -> None:
+        """Validate the server-owned evidence facility without requiring recipe evidence.
+
+        Readiness is an infrastructure/property check.  Candidate-specific evidence
+        remains mandatory for ``BuildTrustedCandidateV1`` and is validated by
+        :meth:`__call__`; an empty evidence root therefore no longer makes the
+        builder globally unavailable.
+        """
+
+        for path, code in (
+            (self._authority_git_dir, "COLLECTOR_AUTHORITY_GIT_UNAVAILABLE"),
+            (self._evidence_root, "COLLECTOR_EVIDENCE_ROOT_UNAVAILABLE"),
+        ):
+            try:
+                metadata = path.stat(follow_symlinks=False)
+            except OSError as exc:
+                raise BuilderError(code) from exc
+            if not stat.S_ISDIR(metadata.st_mode) or path.is_symlink():
+                raise BuilderError(code)
+            if os.name != "nt" and metadata.st_mode & 0o022:
+                raise BuilderError("COLLECTOR_EVIDENCE_ROOT_MODE_DENIED")
+            if (
+                path == self._evidence_root
+                and os.name != "nt"
+                and self._expected_owner_uid is not None
+                and metadata.st_uid != self._expected_owner_uid
+            ):
+                raise BuilderError("COLLECTOR_EVIDENCE_ROOT_OWNER_DENIED")
+
     def _digest(self, path: Path) -> str:
         digest = hashlib.sha256()
         with path.open("rb") as handle:
@@ -486,9 +516,17 @@ class DeterministicEvidenceCollector:
         self._config = config
         self._evidence_source = evidence_source
 
-    def collect(self, recipe_name: str, recipe: Mapping[str, str]) -> Mapping[str, Any]:
+    def assert_ready(self) -> None:
+        """Fail closed on collector configuration, not on recipe evidence presence."""
+
         if self._evidence_source is None:
             raise BuilderError("COLLECTOR_UNCONFIGURED")
+        checker = getattr(self._evidence_source, "assert_ready", None)
+        if checker is not None:
+            checker()
+
+    def collect(self, recipe_name: str, recipe: Mapping[str, str]) -> Mapping[str, Any]:
+        self.assert_ready()
         bundle = self._evidence_source(recipe_name, recipe)
         if not isinstance(bundle, Mapping) or set(bundle) != self._BUNDLE_FIELDS:
             raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
@@ -514,11 +552,32 @@ class DeterministicEvidenceCollector:
         try:
             from silent_release_controller import ReleaseCandidate
             from silent_release_hardening import ReleasePlan, TrustedCandidateBundle
+            from release_migration_plan import (
+                MigrationPlanError,
+                MigrationPlanRequired,
+            )
 
             candidate = ReleaseCandidate.from_dict(dict(bundle["candidate"]))
-            release_plan = ReleasePlan.from_dict(dict(bundle["release_plan"]))
+            release_plan_payload = dict(bundle["release_plan"])
+            migration_resolver = self._config.migration_plan_resolver
+            if migration_resolver is not None:
+                try:
+                    migration = migration_resolver(
+                        recipe_name,
+                        recipe["candidate_id"],
+                        recipe["source_commit"],
+                    )
+                except MigrationPlanRequired as exc:
+                    raise BuilderError("MIGRATION_PLAN_REQUIRED") from exc
+                except MigrationPlanError as exc:
+                    raise BuilderError("MIGRATION_PLAN_INVALID") from exc
+                if migration is not None:
+                    release_plan_payload["migration"] = migration.to_dict()
+            release_plan = ReleasePlan.from_dict(release_plan_payload)
             checks = dict(raw_checks)
             TrustedCandidateBundle(candidate, release_plan, checks).validate()
+        except BuilderError:
+            raise
         except (ImportError, KeyError, TypeError, ValueError, RuntimeError) as exc:
             raise BuilderError("COLLECTOR_EVIDENCE_INVALID") from exc
         if candidate.source_commit != recipe["source_commit"] or candidate.candidate_id != recipe["candidate_id"]:
@@ -672,14 +731,12 @@ class TrustedCandidateBuilder:
         bound = {"repository": self.config.repository, "task_id": recipe["task_id"], "candidate_id": recipe["candidate_id"], "base_oid": recipe["base_oid"], "source_commit": recipe["source_commit"], "version": recipe["version"], "purpose": recipe["purpose"]}
         if {key: request.get(key) for key in bound} != bound:
             raise BuilderError("REQUEST_SCOPE_INVALID")
-        bundle = dict(self.collector.collect(recipe_name, recipe))
-        candidate = bundle["candidate"]
-        release_plan = bundle["release_plan"]
-        checks = bundle.get("checks")
-        evidence_digests = bundle["evidence_digests"]
-        if not isinstance(checks, Mapping) or set(checks) != set(self.config.required_checks) or not all(value is True for value in checks.values()):
-            raise BuilderError("COLLECTOR_CHECKS_INCOMPLETE")
         if request["operation"] == self.config.readiness_operation:
+            self.collector.assert_ready()
+            # Accessing the identity properties validates that the signer is
+            # configured and usable without invoking the private key.
+            builder_key_id = self.signer.key_id
+            builder_key_generation = self.signer.key_generation
             result = {
                 "schema": self.config.response_schema,
                 "status": "READY",
@@ -687,8 +744,8 @@ class TrustedCandidateBuilder:
                 "recipe": recipe_name,
                 "candidate_id": recipe["candidate_id"],
                 "source_commit": recipe["source_commit"],
-                "builder_key_id": self.signer.key_id,
-                "builder_key_generation": self.signer.key_generation,
+                "builder_key_id": builder_key_id,
+                "builder_key_generation": builder_key_generation,
                 "signature_algorithm": "Ed25519",
                 "signature_purpose": "AIOS_TRUSTED_RELEASE_BUILDER",
                 "policy_sha256": self.config.policy_sha256,
@@ -703,6 +760,15 @@ class TrustedCandidateBuilder:
             )
             self._persist_nonce_ledger()
             return result
+
+        bundle = dict(self.collector.collect(recipe_name, recipe))
+        candidate = bundle["candidate"]
+        release_plan = bundle["release_plan"]
+        checks = bundle.get("checks")
+        evidence_digests = bundle["evidence_digests"]
+        if not isinstance(checks, Mapping) or set(checks) != set(self.config.required_checks) or not all(value is True for value in checks.values()):
+            raise BuilderError("COLLECTOR_CHECKS_INCOMPLETE")
+
         issued = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
         attestation = {
             "builder_attestation_id": "builder-attestation-" + _sha256(recipe_name, recipe["task_id"], candidate["artifact_sha256"])[:24],
@@ -830,7 +896,7 @@ def _main() -> None:
         "AIOS_TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT", ""
     ).strip()
     if source_admission_root:
-        from release_source_admission import ReleaseSourceAdmissionStore
+        from release_source_admission import open_source_admission_store
 
         try:
             source_admission_hmac = (
@@ -838,10 +904,10 @@ def _main() -> None:
             ).read_bytes().strip()
         except OSError as exc:
             raise BuilderError("SOURCE_ADMISSION_HMAC_UNAVAILABLE") from exc
-        source_store = ReleaseSourceAdmissionStore(
+        source_store = open_source_admission_store(
             Path(source_admission_root),
             repository=config.repository,
-            policy_sha256=config.policy_sha256,
+            policy=arguments.policy,
             hmac_key=source_admission_hmac,
             expected_owner_uid=0,
             writable=False,
@@ -850,6 +916,26 @@ def _main() -> None:
             config,
             source_admission_resolver=source_store.resolve_recipe,
         )
+    migration_plan_root = os.environ.get(
+        "AIOS_TRUSTED_BUILDER_MIGRATION_PLAN_ROOT", ""
+    ).strip()
+    migration_profile_root = os.environ.get(
+        "AIOS_TRUSTED_BUILDER_MIGRATION_PROFILE_ROOT", ""
+    ).strip()
+    if bool(migration_plan_root) != bool(migration_profile_root):
+        raise BuilderError("MIGRATION_PLAN_CONFIGURATION_INVALID")
+    if migration_plan_root:
+        from release_migration_plan import FileMigrationPlanResolver
+
+        migration_resolver = FileMigrationPlanResolver(
+            Path(migration_plan_root),
+            profile_root=Path(migration_profile_root),
+            expected_owner_uid=0,
+        )
+        config = replace(
+            config,
+            migration_plan_resolver=migration_resolver.resolve,
+        )
     signer = CredentialPathEd25519Signer(
         keyring.signing_private_key_path,
         key_id=os.environ["AIOS_TRUSTED_BUILDER_KEY_ID"],

active-runtime/app/release_evidence_materializer.py

@@ -0,0 +1,529 @@
+#!/usr/bin/env python3
+"""Governed, fail-closed projection of verified release evidence into Builder storage.
+
+This module never invents candidate/release evidence.  A separate governed release
+producer must first write a complete, root-owned evidence closure below
+``source_root/<version>``.  The materializer re-validates that closure with the
+same ``ServerOwnedEvidenceSource``/``DeterministicEvidenceCollector`` code used by
+Trusted Candidate Builder and atomically projects the exact bytes into the
+Builder-owned evidence root.
+
+Readiness is deliberately orthogonal: an empty Builder evidence root does not make
+Builder infrastructure globally unavailable, but every BuildTrustedCandidateV1
+request still requires a valid materialized closure.
+"""
+from __future__ import annotations
+
+import argparse
+from dataclasses import replace
+import hashlib
+import json
+import os
+from pathlib import Path
+import secrets
+import shutil
+import stat
+from typing import Any, Callable, Mapping
+import subprocess
+
+from aios_trusted_candidate_builder import (
+    BuilderError,
+    DeterministicEvidenceCollector,
+    ServerOwnedEvidenceSource,
+    TrustedCandidateBuilderConfig,
+    load_builder_policy,
+)
+from release_migration_plan import FileMigrationPlanResolver
+from release_source_admission import (
+    ReleaseSourceAdmissionStore,
+    open_source_admission_store,
+    source_admission_policy_sha256,
+)
+from release_version import validate_release_version
+
+
+MATERIALIZATION_SCHEMA = "aios.trusted-builder-evidence-materialization.v1"
+SOURCE_ADMISSION_CREDENTIAL = "source_admission_hmac"
+
+
+class EvidenceMaterializationError(RuntimeError):
+    """Stable, non-sensitive materialization failure."""
+
+
+def _canonical_json(value: Mapping[str, Any]) -> bytes:
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+    ).encode("utf-8")
+
+
+def _file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with Path(path).open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _validate_oid(value: object, name: str) -> str:
+    normalized = str(value or "").strip().lower()
+    if (
+        len(normalized) != 40
+        or any(char not in "0123456789abcdef" for char in normalized)
+    ):
+        raise EvidenceMaterializationError(f"{name.upper()}_INVALID")
+    return normalized
+
+
+def _validate_identifier(value: object, name: str) -> str:
+    normalized = str(value or "").strip()
+    if (
+        not normalized
+        or len(normalized) > 255
+        or normalized[0] in ".-"
+        or any(
+            char not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._:-"
+            for char in normalized
+        )
+    ):
+        raise EvidenceMaterializationError(f"{name.upper()}_INVALID")
+    return normalized
+
+
+def _safe_directory(path: Path, *, expected_owner_uid: int | None) -> Path:
+    candidate = Path(path)
+    try:
+        metadata = candidate.stat(follow_symlinks=False)
+        resolved = candidate.resolve(strict=True)
+    except OSError as exc:
+        raise EvidenceMaterializationError("EVIDENCE_DIRECTORY_UNAVAILABLE") from exc
+    if candidate.is_symlink() or not stat.S_ISDIR(metadata.st_mode):
+        raise EvidenceMaterializationError("EVIDENCE_DIRECTORY_INVALID")
+    if os.name != "nt" and metadata.st_mode & 0o022:
+        raise EvidenceMaterializationError("EVIDENCE_DIRECTORY_PERMISSIONS_INVALID")
+    if (
+        os.name != "nt"
+        and expected_owner_uid is not None
+        and metadata.st_uid != expected_owner_uid
+    ):
+        raise EvidenceMaterializationError("EVIDENCE_DIRECTORY_OWNER_INVALID")
+    return resolved
+
+
+def _safe_regular(
+    root: Path,
+    relative: str,
+    *,
+    expected_owner_uid: int | None,
+) -> Path:
+    root = _safe_directory(root, expected_owner_uid=expected_owner_uid)
+    relative_path = Path(relative)
+    if relative_path.is_absolute() or any(
+        part in {"", ".", ".."} for part in relative_path.parts
+    ):
+        raise EvidenceMaterializationError("EVIDENCE_PATH_INVALID")
+    current = root
+    for part in relative_path.parts:
+        current = current / part
+        if current.is_symlink():
+            raise EvidenceMaterializationError("EVIDENCE_SYMLINK_DENIED")
+    try:
+        resolved = (root / relative_path).resolve(strict=True)
+        resolved.relative_to(root)
+        metadata = resolved.stat(follow_symlinks=False)
+    except (OSError, ValueError) as exc:
+        raise EvidenceMaterializationError("EVIDENCE_INPUT_UNAVAILABLE") from exc
+    if not stat.S_ISREG(metadata.st_mode):
+        raise EvidenceMaterializationError("EVIDENCE_SPECIAL_FILE_DENIED")
+    if os.name != "nt" and metadata.st_mode & 0o022:
+        raise EvidenceMaterializationError("EVIDENCE_INPUT_PERMISSIONS_INVALID")
+    if (
+        os.name != "nt"
+        and expected_owner_uid is not None
+        and metadata.st_uid != expected_owner_uid
+    ):
+        raise EvidenceMaterializationError("EVIDENCE_INPUT_OWNER_INVALID")
+    return resolved
+
+
+def _fsync_directory(path: Path) -> None:
+    if os.name == "nt":
+        return
+    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def _copy_exact(source: Path, target: Path, *, mode: int = 0o640) -> None:
+    target.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+    descriptor = os.open(
+        target,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+        mode,
+    )
+    try:
+        with source.open("rb") as src:
+            while True:
+                chunk = src.read(1024 * 1024)
+                if not chunk:
+                    break
+                os.write(descriptor, chunk)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    os.chmod(target, mode)
+
+
+def _write_json_exclusive(path: Path, payload: Mapping[str, Any]) -> str:
+    raw = _canonical_json(payload) + b"\n"
+    descriptor = os.open(
+        path,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+        0o640,
+    )
+    try:
+        os.write(descriptor, raw)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    os.chmod(path, 0o640)
+    return hashlib.sha256(raw).hexdigest()
+
+
+class TrustedBuilderEvidenceMaterializer:
+    """Verify a governed evidence closure and atomically project exact bytes."""
+
+    def __init__(
+        self,
+        *,
+        builder_policy: Path,
+        authority_git_dir: Path,
+        source_root: Path,
+        output_root: Path,
+        source_admission_root: Path | None = None,
+        source_admission_hmac: bytes | None = None,
+        migration_plan_root: Path | None = None,
+        migration_profile_root: Path | None = None,
+        expected_owner_uid: int | None = 0,
+        git_runner: Callable[..., Any] = subprocess.run,
+        source_admission_resolver: Callable[[str, str | None, str], Mapping[str, str]] | None = None,
+        migration_plan_resolver: Callable[[str, str, str], Any] | None = None,
+    ) -> None:
+        self.builder_policy = Path(builder_policy)
+        self.authority_git_dir = Path(authority_git_dir)
+        self.source_root = Path(source_root)
+        self.output_root = Path(output_root)
+        self.expected_owner_uid = expected_owner_uid
+        self.git_runner = git_runner
+
+        try:
+            policy = load_builder_policy(self.builder_policy)
+            config = TrustedCandidateBuilderConfig.from_policy(policy)
+        except BuilderError as exc:
+            raise EvidenceMaterializationError(str(exc)) from exc
+
+        self.source_admission_store: ReleaseSourceAdmissionStore | None = None
+        if source_admission_resolver is not None:
+            config = replace(
+                config,
+                source_admission_resolver=source_admission_resolver,
+            )
+        elif source_admission_root is not None:
+            if not source_admission_hmac or len(source_admission_hmac) < 32:
+                raise EvidenceMaterializationError("SOURCE_ADMISSION_HMAC_UNAVAILABLE")
+            self.source_admission_store = open_source_admission_store(
+                Path(source_admission_root),
+                repository=config.repository,
+                policy=self.builder_policy,
+                hmac_key=source_admission_hmac,
+                expected_owner_uid=expected_owner_uid,
+                writable=False,
+            )
+            config = replace(
+                config,
+                source_admission_resolver=self.source_admission_store.resolve_recipe,
+            )
+
+        if migration_plan_resolver is not None:
+            config = replace(config, migration_plan_resolver=migration_plan_resolver)
+        elif migration_plan_root is not None or migration_profile_root is not None:
+            if migration_plan_root is None or migration_profile_root is None:
+                raise EvidenceMaterializationError(
+                    "MIGRATION_PLAN_CONFIGURATION_INVALID"
+                )
+            resolver = FileMigrationPlanResolver(
+                Path(migration_plan_root),
+                profile_root=Path(migration_profile_root),
+                expected_owner_uid=expected_owner_uid,
+            )
+            config = replace(config, migration_plan_resolver=resolver.resolve)
+
+        self.config = config
+
+    @property
+    def policy_file_sha256(self) -> str:
+        return source_admission_policy_sha256(self.builder_policy)
+
+    def _collector_for(self, evidence_root: Path) -> DeterministicEvidenceCollector:
+        source = ServerOwnedEvidenceSource(
+            self.config,
+            self.authority_git_dir,
+            evidence_root,
+            expected_owner_uid=self.expected_owner_uid,
+            git_runner=self.git_runner,
+        )
+        return DeterministicEvidenceCollector(
+            self.config,
+            evidence_source=source,
+        )
+
+    def _relative_files(self, version: str) -> tuple[str, ...]:
+        relative = [f"{version}/receipt.json"]
+        relative.extend(
+            template.format(version=version)
+            for template in self.config.evidence_layout.values()
+        )
+        return tuple(dict.fromkeys(relative))
+
+    def _validate_existing(
+        self,
+        *,
+        version: str,
+        candidate_id: str,
+        source_commit: str,
+        expected_bundle: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        try:
+            observed = self._collector_for(self.output_root).collect(
+                version,
+                self.config.resolve_recipe(version, candidate_id, source_commit),
+            )
+        except (BuilderError, RuntimeError, ValueError) as exc:
+            raise EvidenceMaterializationError("EVIDENCE_DESTINATION_CONFLICT") from exc
+        if _canonical_json(dict(observed)) != _canonical_json(dict(expected_bundle)):
+            raise EvidenceMaterializationError("EVIDENCE_DESTINATION_CONFLICT")
+        record_path = self.output_root / version / "materialization.json"
+        if not record_path.is_file():
+            raise EvidenceMaterializationError("EVIDENCE_DESTINATION_CONFLICT")
+        try:
+            record = json.loads(record_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise EvidenceMaterializationError("EVIDENCE_DESTINATION_CONFLICT") from exc
+        if not isinstance(record, Mapping):
+            raise EvidenceMaterializationError("EVIDENCE_DESTINATION_CONFLICT")
+        return dict(record)
+
+    def materialize(
+        self,
+        *,
+        version: str,
+        candidate_id: str,
+        source_commit: str,
+    ) -> dict[str, Any]:
+        try:
+            version = validate_release_version(version)
+        except ValueError as exc:
+            raise EvidenceMaterializationError("VERSION_INVALID") from exc
+        candidate_id = _validate_identifier(candidate_id, "candidate_id")
+        source_commit = _validate_oid(source_commit, "source_commit")
+
+        _safe_directory(
+            self.source_root,
+            expected_owner_uid=self.expected_owner_uid,
+        )
+        _safe_directory(
+            self.output_root,
+            expected_owner_uid=self.expected_owner_uid,
+        )
+
+        try:
+            recipe = self.config.resolve_recipe(
+                version,
+                candidate_id,
+                source_commit,
+            )
+            bundle = self._collector_for(self.source_root).collect(
+                version,
+                recipe,
+            )
+        except BuilderError as exc:
+            raise EvidenceMaterializationError(str(exc)) from exc
+        except (RuntimeError, ValueError) as exc:
+            raise EvidenceMaterializationError("EVIDENCE_SOURCE_INVALID") from exc
+
+        admission_id = ""
+        if self.source_admission_store is not None:
+            try:
+                admission_id = str(
+                    self.source_admission_store.lookup_source(source_commit)[
+                        "admission_id"
+                    ]
+                )
+            except Exception as exc:
+                raise EvidenceMaterializationError(
+                    "SOURCE_ADMISSION_INVALID"
+                ) from exc
+
+        receipt_path = _safe_regular(
+            self.source_root,
+            f"{version}/receipt.json",
+            expected_owner_uid=self.expected_owner_uid,
+        )
+        receipt_sha256 = _file_sha256(receipt_path)
+        release_plan = dict(bundle["release_plan"])
+        migration = release_plan.get("migration")
+        migration_sha256 = ""
+        if isinstance(migration, Mapping):
+            migration_sha256 = str(migration.get("sha256") or "")
+
+        materialization_payload = {
+            "schema": MATERIALIZATION_SCHEMA,
+            "repository": self.config.repository,
+            "version": version,
+            "candidate_id": candidate_id,
+            "source_commit": source_commit,
+            "tree_oid": str(bundle["candidate"]["tree_oid"]),
+            "source_admission_id": admission_id,
+            "builder_policy_file_sha256": self.policy_file_sha256,
+            "receipt_sha256": receipt_sha256,
+            "migration_sha256": migration_sha256,
+            "evidence_digests": dict(bundle["evidence_digests"]),
+        }
+        materialization_id = hashlib.sha256(
+            _canonical_json(materialization_payload)
+        ).hexdigest()
+        materialization_payload["materialization_id"] = (
+            "evidence-materialization-" + materialization_id[:24]
+        )
+
+        target = self.output_root / version
+        if target.exists() or target.is_symlink():
+            existing = self._validate_existing(
+                version=version,
+                candidate_id=candidate_id,
+                source_commit=source_commit,
+                expected_bundle=bundle,
+            )
+            if (
+                existing.get("materialization_id")
+                != materialization_payload["materialization_id"]
+            ):
+                raise EvidenceMaterializationError(
+                    "EVIDENCE_DESTINATION_CONFLICT"
+                )
+            return {
+                **materialization_payload,
+                "status": "EVIDENCE_MATERIALIZED",
+                "idempotent": True,
+            }
+
+        temporary = self.output_root / (
+            f".{version}.{os.getpid()}.{secrets.token_hex(8)}.tmp"
+        )
+        if temporary.exists() or temporary.is_symlink():
+            raise EvidenceMaterializationError("EVIDENCE_TEMP_CONFLICT")
+        temporary.mkdir(mode=0o750)
+        try:
+            for relative in self._relative_files(version):
+                source = _safe_regular(
+                    self.source_root,
+                    relative,
+                    expected_owner_uid=self.expected_owner_uid,
+                )
+                destination_relative = Path(relative).relative_to(version)
+                destination = temporary / destination_relative
+                _copy_exact(source, destination)
+                if _file_sha256(destination) != _file_sha256(source):
+                    raise EvidenceMaterializationError(
+                        "EVIDENCE_COPY_DIGEST_MISMATCH"
+                    )
+            materialization_sha256 = _write_json_exclusive(
+                temporary / "materialization.json",
+                materialization_payload,
+            )
+            _fsync_directory(temporary)
+            os.replace(temporary, target)
+            _fsync_directory(self.output_root)
+        except Exception:
+            shutil.rmtree(temporary, ignore_errors=True)
+            raise
+
+        try:
+            observed = self._collector_for(self.output_root).collect(
+                version,
+                recipe,
+            )
+        except (BuilderError, RuntimeError, ValueError) as exc:
+            raise EvidenceMaterializationError(
+                "EVIDENCE_POST_WRITE_VERIFY_FAILED"
+            ) from exc
+        if _canonical_json(dict(observed)) != _canonical_json(dict(bundle)):
+            raise EvidenceMaterializationError(
+                "EVIDENCE_POST_WRITE_VERIFY_FAILED"
+            )
+
+        return {
+            **materialization_payload,
+            "materialization_sha256": materialization_sha256,
+            "status": "EVIDENCE_MATERIALIZED",
+            "idempotent": False,
+        }
+
+
+def _credential(name: str) -> bytes:
+    root = os.environ.get("CREDENTIALS_DIRECTORY", "").strip()
+    if not root:
+        raise EvidenceMaterializationError("CREDENTIALS_DIRECTORY_REQUIRED")
+    path = Path(root) / name
+    try:
+        value = path.read_bytes().strip()
+    except OSError as exc:
+        raise EvidenceMaterializationError(
+            f"{name.upper()}_UNAVAILABLE"
+        ) from exc
+    if len(value) < 32:
+        raise EvidenceMaterializationError(f"{name.upper()}_INVALID")
+    return value
+
+
+def _main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("operation", choices=("materialize",))
+    parser.add_argument("--version", required=True)
+    parser.add_argument("--candidate-id", required=True)
+    parser.add_argument("--source-commit", required=True)
+    parser.add_argument("--builder-policy", required=True, type=Path)
+    parser.add_argument("--authority-git-dir", required=True, type=Path)
+    parser.add_argument("--source-root", required=True, type=Path)
+    parser.add_argument("--output-root", required=True, type=Path)
+    parser.add_argument("--source-admission-root", required=True, type=Path)
+    parser.add_argument("--migration-plan-root", required=True, type=Path)
+    parser.add_argument("--migration-profile-root", required=True, type=Path)
+    arguments = parser.parse_args()
+
+    try:
+        result = TrustedBuilderEvidenceMaterializer(
+            builder_policy=arguments.builder_policy,
+            authority_git_dir=arguments.authority_git_dir,
+            source_root=arguments.source_root,
+            output_root=arguments.output_root,
+            source_admission_root=arguments.source_admission_root,
+            source_admission_hmac=_credential(SOURCE_ADMISSION_CREDENTIAL),
+            migration_plan_root=arguments.migration_plan_root,
+            migration_profile_root=arguments.migration_profile_root,
+        ).materialize(
+            version=arguments.version,
+            candidate_id=arguments.candidate_id,
+            source_commit=arguments.source_commit,
+        )
+    except EvidenceMaterializationError as exc:
+        raise SystemExit(str(exc)) from exc
+    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+
+
+if __name__ == "__main__":
+    _main()

active-runtime/app/release_migration_plan.py

@@ -0,0 +1,989 @@
+#!/usr/bin/env python3
+"""Fail-closed cross-code-line release migration planning.
+
+The migration plan is data, not a shell script.  It binds an admitted source to
+the exact currently observed staging code lines, the compatible state contract,
+and the rollback/verification contract.  The plan can then be embedded in the
+existing signed ReleasePlan so Owner approval covers migration semantics without
+creating a second release authority.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import stat
+import subprocess
+import tempfile
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Callable, Mapping, Sequence
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SHA_RE = re.compile(r"^[0-9a-f]{64}$")
+IDENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:@/-]{2,255}$")
+PLAN_SCHEMA = "aios.release-migration-plan.v1"
+REQUIREMENT_SCHEMA = "aios.release-migration-requirement.v1"
+PROFILE_SCHEMA = "aios.release-migration-profile.v1"
+ALLOWED_STATE_MODES = frozenset({"REUSE_IN_PLACE_WITH_SNAPSHOT", "PRESERVE_UNUSED"})
+REQUIRED_PRECONDITIONS = (
+    "writer_disabled",
+    "task_quiescent",
+    "state_snapshot_verified",
+    "credentials_ready",
+    "rollback_ready",
+)
+REQUIRED_VERIFICATION_CONTRACTS = (
+    "runtime_identity_exact",
+    "task_principal_no_bearer_e2e",
+    "restart_recovery",
+    "credential_projection",
+    "release_consistency_no_unknown",
+)
+
+
+class MigrationPlanError(RuntimeError):
+    """Migration evidence is missing, malformed, stale, or unsafe."""
+
+
+class MigrationPlanRequired(MigrationPlanError):
+    """A cross-code-line release cannot proceed without an exact migration plan."""
+
+
+def _canonical_json(value: object) -> bytes:
+    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
+
+
+def _sha(value: object, label: str) -> str:
+    text = str(value or "").lower()
+    if not SHA_RE.fullmatch(text):
+        raise MigrationPlanError(f"{label}_invalid")
+    return text
+
+
+def _oid(value: object, label: str) -> str:
+    text = str(value or "").lower()
+    if not OID_RE.fullmatch(text):
+        raise MigrationPlanError(f"{label}_invalid")
+    return text
+
+
+def _ident(value: object, label: str) -> str:
+    text = str(value or "")
+    if not IDENT_RE.fullmatch(text):
+        raise MigrationPlanError(f"{label}_invalid")
+    return text
+
+
+def _secure_json(path: Path, *, expected_owner_uid: int | None = 0) -> dict[str, Any]:
+    try:
+        metadata = path.lstat()
+    except OSError as exc:
+        raise MigrationPlanError(f"secure_json_unavailable:{path}") from exc
+    if path.is_symlink() or not stat.S_ISREG(metadata.st_mode):
+        raise MigrationPlanError("secure_json_regular_file_required")
+    if expected_owner_uid is not None and os.name != "nt" and metadata.st_uid != expected_owner_uid:
+        raise MigrationPlanError("secure_json_owner_invalid")
+    if os.name != "nt" and metadata.st_mode & 0o022:
+        raise MigrationPlanError("secure_json_mode_invalid")
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise MigrationPlanError("secure_json_invalid") from exc
+    if not isinstance(value, dict):
+        raise MigrationPlanError("secure_json_object_required")
+    return value
+
+
+def _durable_json(path: Path, value: Mapping[str, Any], *, mode: int = 0o640) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    encoded = _canonical_json(dict(value))
+    if path.exists():
+        current = path.read_bytes()
+        if current != encoded:
+            raise MigrationPlanError("migration_record_conflict")
+        return
+    temporary = path.with_name(path.name + ".tmp")
+    descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode)
+    try:
+        with os.fdopen(descriptor, "wb") as handle:
+            handle.write(encoded)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+        if os.name != "nt":
+            os.chmod(path, mode)
+            directory_fd = os.open(path.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+    finally:
+        try:
+            temporary.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def _hash_files(root: Path, files: Sequence[str]) -> str:
+    payload: dict[str, str] = {}
+    resolved_root = root.resolve(strict=True)
+    for relative in files:
+        rel = Path(relative)
+        if rel.is_absolute() or ".." in rel.parts:
+            raise MigrationPlanError("interface_path_invalid")
+        path = (resolved_root / rel).resolve(strict=True)
+        try:
+            path.relative_to(resolved_root)
+        except ValueError as exc:
+            raise MigrationPlanError("interface_path_escape") from exc
+        if path.is_symlink() or not path.is_file():
+            raise MigrationPlanError("interface_file_invalid")
+        payload[relative] = hashlib.sha256(path.read_bytes()).hexdigest()
+    return hashlib.sha256(_canonical_json(payload)).hexdigest()
+
+
+def _git_blob(git_dir: Path, source_oid: str, relative: str) -> bytes:
+    rel = Path(relative)
+    if rel.is_absolute() or ".." in rel.parts:
+        raise MigrationPlanError("target_interface_path_invalid")
+    completed = subprocess.run(
+        ["git", f"--git-dir={git_dir}", "show", f"{source_oid}:{relative}"],
+        check=False,
+        capture_output=True,
+        timeout=30,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if completed.returncode != 0:
+        raise MigrationPlanError("target_interface_unavailable")
+    return bytes(completed.stdout)
+
+
+def _hash_git_files(git_dir: Path, source_oid: str, files: Sequence[str]) -> str:
+    payload = {
+        relative: hashlib.sha256(_git_blob(git_dir, source_oid, relative)).hexdigest()
+        for relative in files
+    }
+    return hashlib.sha256(_canonical_json(payload)).hexdigest()
+
+
+def _systemd_properties(service: str) -> dict[str, Any]:
+    _ident(service, "service")
+    completed = subprocess.run(
+        [
+            "systemctl",
+            "show",
+            service,
+            "--property=WorkingDirectory",
+            "--property=Environment",
+            "--property=FragmentPath",
+            "--property=ActiveState",
+        ],
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=20,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if completed.returncode != 0:
+        raise MigrationPlanError("systemd_observation_unavailable")
+    result: dict[str, Any] = {}
+    for line in completed.stdout.splitlines():
+        if "=" in line:
+            key, value = line.split("=", 1)
+            result[key] = value
+    if result.get("ActiveState") != "active":
+        raise MigrationPlanError("component_not_active")
+    environment: dict[str, str] = {}
+    for token in str(result.get("Environment") or "").split():
+        if "=" in token:
+            key, value = token.split("=", 1)
+            environment[key] = value.strip('"')
+    result["environment"] = environment
+    return result
+
+
+def _sqlite_schema(connection: sqlite3.Connection) -> dict[str, Any]:
+    tables = [
+        row[0]
+        for row in connection.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
+        )
+    ]
+    result: dict[str, Any] = {}
+    for table in tables:
+        if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", table):
+            raise MigrationPlanError("sqlite_table_name_invalid")
+        columns = []
+        for row in connection.execute(f"PRAGMA table_info({table})"):
+            columns.append(
+                {
+                    "cid": int(row[0]),
+                    "name": str(row[1]),
+                    "type": str(row[2]),
+                    "notnull": int(row[3]),
+                    "default": row[4],
+                    "pk": int(row[5]),
+                }
+            )
+        result[table] = {
+            "columns": columns,
+            "row_count": int(connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]),
+        }
+    return result
+
+
+def _schema_digest(schema: Mapping[str, Any]) -> str:
+    shape = {
+        table: {"columns": value["columns"]}
+        for table, value in sorted(schema.items())
+    }
+    return hashlib.sha256(_canonical_json(shape)).hexdigest()
+
+
+def _assert_additive_sqlite(before: Mapping[str, Any], after: Mapping[str, Any]) -> None:
+    for table, old in before.items():
+        if table not in after:
+            raise MigrationPlanError("sqlite_table_removed")
+        old_columns = {column["name"]: column for column in old["columns"]}
+        new_columns = {column["name"]: column for column in after[table]["columns"]}
+        for name, column in old_columns.items():
+            if new_columns.get(name) != column:
+                raise MigrationPlanError("sqlite_column_changed")
+        for name, column in new_columns.items():
+            if name in old_columns:
+                continue
+            # A new NOT NULL column is backward compatible only with a default.
+            if column["notnull"] and column["default"] is None:
+                raise MigrationPlanError("sqlite_new_column_unsafe")
+        if int(after[table]["row_count"]) != int(old["row_count"]):
+            raise MigrationPlanError("sqlite_row_count_changed_during_probe")
+
+
+@dataclass(frozen=True)
+class RuntimeComponentBinding:
+    component_id: str
+    service: str
+    code_line_id: str
+    source_oid: str
+    artifact_sha256: str | None
+    interface_sha256: str
+
+    def __post_init__(self) -> None:
+        _ident(self.component_id, "component_id")
+        _ident(self.service, "service")
+        _ident(self.code_line_id, "code_line_id")
+        _oid(self.source_oid, "source_oid")
+        if self.artifact_sha256 is not None:
+            _sha(self.artifact_sha256, "artifact_sha256")
+        _sha(self.interface_sha256, "interface_sha256")
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, value: Mapping[str, Any]) -> "RuntimeComponentBinding":
+        return cls(**dict(value))
+
+
+@dataclass(frozen=True)
+class StateMigrationContract:
+    state_id: str
+    state_ref: str
+    mode: str
+    before_schema_sha256: str
+    after_schema_sha256: str
+    snapshot_name: str
+    snapshot_sha256: str
+    snapshot_required: bool
+    rollback_compatible: bool
+
+    def __post_init__(self) -> None:
+        _ident(self.state_id, "state_id")
+        _ident(self.state_ref, "state_ref")
+        if self.mode not in ALLOWED_STATE_MODES:
+            raise MigrationPlanError("state_mode_invalid")
+        _sha(self.before_schema_sha256, "before_schema_sha256")
+        _sha(self.after_schema_sha256, "after_schema_sha256")
+        _sha(self.snapshot_sha256, "snapshot_sha256")
+        snapshot = Path(self.snapshot_name)
+        if (
+            snapshot.is_absolute()
+            or ".." in snapshot.parts
+            or not snapshot.parts
+            or str(snapshot).startswith(".")
+        ):
+            raise MigrationPlanError("snapshot_name_invalid")
+        if self.snapshot_required is not True or self.rollback_compatible is not True:
+            raise MigrationPlanError("state_recovery_contract_invalid")
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, value: Mapping[str, Any]) -> "StateMigrationContract":
+        return cls(**dict(value))
+
+
+@dataclass(frozen=True)
+class MigrationPlan:
+    migration_id: str
+    version: str
+    candidate_id: str
+    repository_identity: str
+    source_admission_id: str
+    source_commit: str
+    tree_oid: str
+    profile_id: str
+    profile_sha256: str
+    from_components: tuple[RuntimeComponentBinding, ...]
+    to_components: tuple[RuntimeComponentBinding, ...]
+    state_contracts: tuple[StateMigrationContract, ...]
+    activation_schema: str
+    activation_sha256: str
+    rollback_window_seconds: int
+    preconditions: tuple[str, ...] = REQUIRED_PRECONDITIONS
+    verification_contracts: tuple[str, ...] = REQUIRED_VERIFICATION_CONTRACTS
+    runtime_switch_capability: str = "runtime_switch"
+    rollback_contract: str = "restore_previous_runtime_and_state_snapshot.v1"
+    forward_recovery_contract: str = "forward_recovery_after_verified_rollback_block.v1"
+    writer_expected: bool = False
+    direct_db_allowed: bool = False
+    direct_runtime_overwrite_allowed: bool = False
+
+    def __post_init__(self) -> None:
+        _ident(self.migration_id, "migration_id")
+        _ident(self.version, "version")
+        _ident(self.candidate_id, "candidate_id")
+        if self.repository_identity != "aios/aios":
+            raise MigrationPlanError("repository_identity_invalid")
+        _ident(self.source_admission_id, "source_admission_id")
+        _oid(self.source_commit, "source_commit")
+        _oid(self.tree_oid, "tree_oid")
+        _ident(self.profile_id, "profile_id")
+        _sha(self.profile_sha256, "profile_sha256")
+        _ident(self.activation_schema, "activation_schema")
+        _sha(self.activation_sha256, "activation_sha256")
+        if not 60 <= int(self.rollback_window_seconds) <= 86400:
+            raise MigrationPlanError("rollback_window_invalid")
+        if tuple(self.preconditions) != REQUIRED_PRECONDITIONS:
+            raise MigrationPlanError("migration_preconditions_invalid")
+        if tuple(self.verification_contracts) != REQUIRED_VERIFICATION_CONTRACTS:
+            raise MigrationPlanError("migration_verification_contracts_invalid")
+        if self.runtime_switch_capability != "runtime_switch":
+            raise MigrationPlanError("migration_capability_invalid")
+        if self.writer_expected or self.direct_db_allowed or self.direct_runtime_overwrite_allowed:
+            raise MigrationPlanError("migration_unsafe_authority")
+        if not self.from_components or not self.to_components or not self.state_contracts:
+            raise MigrationPlanError("migration_contract_incomplete")
+        from_ids = {item.component_id for item in self.from_components}
+        to_ids = {item.component_id for item in self.to_components}
+        if from_ids != to_ids or len(from_ids) != len(self.from_components) or len(to_ids) != len(self.to_components):
+            raise MigrationPlanError("migration_component_binding_invalid")
+        if all(
+            next(item.code_line_id for item in self.from_components if item.component_id == target.component_id)
+            == target.code_line_id
+            for target in self.to_components
+        ):
+            raise MigrationPlanError("migration_not_required")
+
+    @property
+    def canonical_payload(self) -> dict[str, Any]:
+        return {
+            "schema": PLAN_SCHEMA,
+            "migration_id": self.migration_id,
+            "version": self.version,
+            "candidate_id": self.candidate_id,
+            "repository_identity": self.repository_identity,
+            "source_admission_id": self.source_admission_id,
+            "source_commit": self.source_commit,
+            "tree_oid": self.tree_oid,
+            "profile_id": self.profile_id,
+            "profile_sha256": self.profile_sha256,
+            "from_components": [item.to_dict() for item in self.from_components],
+            "to_components": [item.to_dict() for item in self.to_components],
+            "state_contracts": [item.to_dict() for item in self.state_contracts],
+            "activation_schema": self.activation_schema,
+            "activation_sha256": self.activation_sha256,
+            "rollback_window_seconds": self.rollback_window_seconds,
+            "preconditions": list(self.preconditions),
+            "verification_contracts": list(self.verification_contracts),
+            "runtime_switch_capability": self.runtime_switch_capability,
+            "rollback_contract": self.rollback_contract,
+            "forward_recovery_contract": self.forward_recovery_contract,
+            "writer_expected": self.writer_expected,
+            "direct_db_allowed": self.direct_db_allowed,
+            "direct_runtime_overwrite_allowed": self.direct_runtime_overwrite_allowed,
+        }
+
+    @property
+    def sha256(self) -> str:
+        return hashlib.sha256(_canonical_json(self.canonical_payload)).hexdigest()
+
+    def to_dict(self) -> dict[str, Any]:
+        return {**self.canonical_payload, "sha256": self.sha256}
+
+    @classmethod
+    def from_dict(cls, value: Mapping[str, Any]) -> "MigrationPlan":
+        raw = dict(value)
+        if raw.pop("schema", None) != PLAN_SCHEMA:
+            raise MigrationPlanError("migration_plan_schema_invalid")
+        claimed = raw.pop("sha256", None)
+        raw["from_components"] = tuple(RuntimeComponentBinding.from_dict(item) for item in raw["from_components"])
+        raw["to_components"] = tuple(RuntimeComponentBinding.from_dict(item) for item in raw["to_components"])
+        raw["state_contracts"] = tuple(StateMigrationContract.from_dict(item) for item in raw["state_contracts"])
+        raw["preconditions"] = tuple(raw.get("preconditions") or ())
+        raw["verification_contracts"] = tuple(raw.get("verification_contracts") or ())
+        plan = cls(**raw)
+        if claimed is not None and claimed != plan.sha256:
+            raise MigrationPlanError("migration_plan_sha256_mismatch")
+        return plan
+
+    def assert_release_binding(
+        self,
+        *,
+        version: str,
+        candidate_id: str,
+        source_commit: str,
+        tree_oid: str,
+        repository_identity: str,
+    ) -> None:
+        expected = (
+            self.version,
+            self.candidate_id,
+            self.source_commit,
+            self.tree_oid,
+            self.repository_identity,
+        )
+        actual = (version, candidate_id, source_commit, tree_oid, repository_identity)
+        if actual != expected:
+            raise MigrationPlanError("migration_release_binding_mismatch")
+
+
+@dataclass(frozen=True)
+class MigrationRequirement:
+    version: str
+    candidate_id: str
+    source_commit: str
+    repository_identity: str
+    required: bool
+    reason_code: str
+    profile_id: str
+    profile_sha256: str
+    migration_plan_sha256: str | None
+
+    @property
+    def canonical_payload(self) -> dict[str, Any]:
+        return {
+            "schema": REQUIREMENT_SCHEMA,
+            **asdict(self),
+        }
+
+    @property
+    def sha256(self) -> str:
+        return hashlib.sha256(_canonical_json(self.canonical_payload)).hexdigest()
+
+    def to_dict(self) -> dict[str, Any]:
+        return {**self.canonical_payload, "sha256": self.sha256}
+
+    @classmethod
+    def from_dict(cls, value: Mapping[str, Any]) -> "MigrationRequirement":
+        raw = dict(value)
+        if raw.pop("schema", None) != REQUIREMENT_SCHEMA:
+            raise MigrationPlanError("migration_requirement_schema_invalid")
+        claimed = raw.pop("sha256", None)
+        requirement = cls(**raw)
+        _ident(requirement.version, "version")
+        _ident(requirement.candidate_id, "candidate_id")
+        _oid(requirement.source_commit, "source_commit")
+        if requirement.repository_identity != "aios/aios":
+            raise MigrationPlanError("migration_requirement_repository_invalid")
+        _ident(requirement.reason_code, "reason_code")
+        _ident(requirement.profile_id, "profile_id")
+        _sha(requirement.profile_sha256, "profile_sha256")
+        if requirement.required:
+            _sha(requirement.migration_plan_sha256, "migration_plan_sha256")
+        elif requirement.migration_plan_sha256 is not None:
+            raise MigrationPlanError("unexpected_migration_plan")
+        if claimed is not None and claimed != requirement.sha256:
+            raise MigrationPlanError("migration_requirement_sha256_mismatch")
+        return requirement
+
+
+class FileMigrationPlanResolver:
+    """Read-only resolver used by the trusted builder.
+
+    A profile declares that a release version is cross-code-line.  Once a
+    version is declared, absence of requirement/plan state is fail-closed.
+    """
+
+    def __init__(
+        self,
+        root: Path,
+        *,
+        profile_root: Path,
+        expected_owner_uid: int | None = 0,
+    ):
+        self.root = Path(root)
+        self.profile_root = Path(profile_root)
+        self.expected_owner_uid = expected_owner_uid
+
+    def _profile_for_version(self, version: str) -> dict[str, Any] | None:
+        matches = []
+        if not self.profile_root.is_dir():
+            return None
+        for path in sorted(self.profile_root.glob("*.json")):
+            profile = _secure_json(path, expected_owner_uid=self.expected_owner_uid)
+            if profile.get("schema") != PROFILE_SCHEMA:
+                raise MigrationPlanError("migration_profile_schema_invalid")
+            if profile.get("version") == version:
+                matches.append(profile)
+        if len(matches) > 1:
+            raise MigrationPlanError("migration_profile_ambiguous")
+        return matches[0] if matches else None
+
+    def resolve(self, version: str, candidate_id: str, source_commit: str) -> MigrationPlan | None:
+        profile = self._profile_for_version(version)
+        if profile is None:
+            return None
+        base = self.root / source_commit
+        requirement_path = base / "requirement.json"
+        plan_path = base / "plan.json"
+        if not requirement_path.is_file():
+            raise MigrationPlanRequired("MIGRATION_PLAN_REQUIRED")
+        requirement = MigrationRequirement.from_dict(
+            _secure_json(requirement_path, expected_owner_uid=self.expected_owner_uid)
+        )
+        expected = (version, candidate_id, source_commit, "aios/aios")
+        actual = (
+            requirement.version,
+            requirement.candidate_id,
+            requirement.source_commit,
+            requirement.repository_identity,
+        )
+        if actual != expected:
+            raise MigrationPlanError("migration_requirement_binding_mismatch")
+        if requirement.profile_id != profile.get("profile_id"):
+            raise MigrationPlanError("migration_profile_binding_mismatch")
+        profile_sha = hashlib.sha256(_canonical_json(profile)).hexdigest()
+        if requirement.profile_sha256 != profile_sha:
+            raise MigrationPlanError("migration_profile_digest_mismatch")
+        if not requirement.required:
+            return None
+        if not plan_path.is_file():
+            raise MigrationPlanRequired("MIGRATION_PLAN_REQUIRED")
+        plan = MigrationPlan.from_dict(
+            _secure_json(plan_path, expected_owner_uid=self.expected_owner_uid)
+        )
+        plan.assert_release_binding(
+            version=version,
+            candidate_id=candidate_id,
+            source_commit=source_commit,
+            tree_oid=plan.tree_oid,
+            repository_identity="aios/aios",
+        )
+        if requirement.migration_plan_sha256 != plan.sha256:
+            raise MigrationPlanError("migration_requirement_plan_mismatch")
+        return plan
+
+
+class MigrationPreSwitchValidator:
+    """Validates a root-owned live proof against the signed migration plan."""
+
+    def __init__(self, proof_resolver: Callable[[MigrationPlan], Mapping[str, Any]]):
+        self.proof_resolver = proof_resolver
+
+    def validate(self, migration_plan: MigrationPlan | None) -> dict[str, Any]:
+        if migration_plan is None:
+            return {"required": False, "verified": True}
+        proof = dict(self.proof_resolver(migration_plan))
+        snapshot_digests = sorted(
+            contract.snapshot_sha256 for contract in migration_plan.state_contracts
+        )
+        required = {
+            "migration_plan_sha256": migration_plan.sha256,
+            "migration_id": migration_plan.migration_id,
+            "source_commit": migration_plan.source_commit,
+            "tree_oid": migration_plan.tree_oid,
+            "activation_sha256": migration_plan.activation_sha256,
+            "state_snapshot_sha256s": snapshot_digests,
+            "writer_disabled": True,
+            "task_quiescent": True,
+            "state_snapshot_verified": True,
+            "credentials_ready": True,
+            "rollback_ready": True,
+        }
+        failed = [key for key, expected in required.items() if proof.get(key) != expected]
+        if failed:
+            raise MigrationPlanError(
+                "MIGRATION_PRE_SWITCH_PROOF_INVALID:" + ",".join(sorted(failed))
+            )
+        return {"required": True, "verified": True, **proof}
+
+
+class HostMigrationCompiler:
+    def __init__(
+        self,
+        *,
+        authority_git_dir: Path,
+        profile_path: Path,
+        output_root: Path,
+        expected_owner_uid: int | None = 0,
+    ):
+        self.authority_git_dir = Path(authority_git_dir)
+        self.profile_path = Path(profile_path)
+        self.output_root = Path(output_root)
+        self.expected_owner_uid = expected_owner_uid
+
+    def _profile(self) -> dict[str, Any]:
+        profile = _secure_json(self.profile_path, expected_owner_uid=self.expected_owner_uid)
+        required = {
+            "schema",
+            "profile_id",
+            "version",
+            "repository",
+            "current_components",
+            "target_components",
+            "activation_file",
+            "state_ref",
+            "state_db",
+            "state_mode",
+            "rollback_window_seconds",
+        }
+        if set(profile) != required or profile.get("schema") != PROFILE_SCHEMA:
+            raise MigrationPlanError("migration_profile_fields_invalid")
+        if profile.get("repository") != "aios/aios":
+            raise MigrationPlanError("migration_profile_repository_invalid")
+        _ident(profile.get("profile_id"), "profile_id")
+        _ident(profile.get("version"), "version")
+        if profile.get("state_mode") not in ALLOWED_STATE_MODES:
+            raise MigrationPlanError("migration_profile_state_mode_invalid")
+        return profile
+
+    @staticmethod
+    def _component_from_live(spec: Mapping[str, Any]) -> RuntimeComponentBinding:
+        required = {
+            "component_id",
+            "service",
+            "code_line_id",
+            "expected_source_oid",
+            "source_oid_env",
+            "artifact_sha_env",
+            "interface_files",
+            "marker_files",
+        }
+        if set(spec) != required:
+            raise MigrationPlanError("current_component_profile_invalid")
+        properties = _systemd_properties(str(spec["service"]))
+        environment = properties["environment"]
+        source_oid = _oid(environment.get(str(spec["source_oid_env"])), "current_source_oid")
+        expected_oid = _oid(spec["expected_source_oid"], "expected_source_oid")
+        if source_oid != expected_oid:
+            raise MigrationPlanError("CURRENT_RUNTIME_DRIFT")
+        working = Path(str(properties.get("WorkingDirectory") or ""))
+        if not working.is_absolute() or not working.is_dir():
+            raise MigrationPlanError("current_working_directory_invalid")
+        for marker in spec["marker_files"]:
+            if not (working / str(marker)).is_file():
+                raise MigrationPlanError("current_code_line_marker_missing")
+        interface = _hash_files(working, tuple(str(item) for item in spec["interface_files"]))
+        raw_artifact = environment.get(str(spec["artifact_sha_env"]))
+        artifact = str(raw_artifact or "").lower()
+        artifact_value = artifact if SHA_RE.fullmatch(artifact) else None
+        return RuntimeComponentBinding(
+            component_id=str(spec["component_id"]),
+            service=str(spec["service"]),
+            code_line_id=str(spec["code_line_id"]),
+            source_oid=source_oid,
+            artifact_sha256=artifact_value,
+            interface_sha256=interface,
+        )
+
+    def _component_from_target(
+        self, spec: Mapping[str, Any], *, source_commit: str
+    ) -> RuntimeComponentBinding:
+        required = {"component_id", "service", "code_line_id", "interface_files"}
+        if set(spec) != required:
+            raise MigrationPlanError("target_component_profile_invalid")
+        interface_files = tuple(str(item) for item in spec["interface_files"])
+        interface = _hash_git_files(self.authority_git_dir, source_commit, interface_files)
+        return RuntimeComponentBinding(
+            component_id=str(spec["component_id"]),
+            service=str(spec["service"]),
+            code_line_id=str(spec["code_line_id"]),
+            source_oid=source_commit,
+            artifact_sha256=None,
+            interface_sha256=interface,
+        )
+
+    @staticmethod
+    def _state_compatibility(
+        state_db: Path,
+        state_ref: str,
+        *,
+        snapshot_path: Path,
+        snapshot_name: str,
+    ) -> StateMigrationContract:
+        from aios_task_admission_bootstrap import TaskAdmissionStore
+
+        if state_db.is_symlink() or not state_db.is_file():
+            raise MigrationPlanError("state_db_invalid")
+        snapshot_path.parent.mkdir(parents=True, exist_ok=True)
+        with tempfile.TemporaryDirectory(prefix="aios-migration-probe-") as temp_dir:
+            temp_root = Path(temp_dir)
+            consistent_snapshot = temp_root / "pre-migration.sqlite3"
+            probe_path = temp_root / "schema-probe.sqlite3"
+            with sqlite3.connect(
+                f"file:{state_db}?mode=ro", uri=True
+            ) as live, sqlite3.connect(consistent_snapshot) as copy:
+                live.backup(copy)
+            with sqlite3.connect(consistent_snapshot) as frozen:
+                before = _sqlite_schema(frozen)
+            # Probe target initialization only on a second private copy.  The
+            # rollback snapshot remains byte-for-byte pre-migration state.
+            with sqlite3.connect(consistent_snapshot) as source, sqlite3.connect(
+                probe_path
+            ) as probe:
+                source.backup(probe)
+            TaskAdmissionStore(probe_path)
+            with sqlite3.connect(probe_path) as migrated:
+                after = _sqlite_schema(migrated)
+            required_tables = {
+                "task_admission_envelopes",
+                "task_admission_grants",
+                "task_admission_receipts",
+            }
+            if not required_tables.issubset(after):
+                raise MigrationPlanError("INCOMPATIBLE_STATE_SCHEMA")
+            _assert_additive_sqlite(before, after)
+            snapshot_bytes = consistent_snapshot.read_bytes()
+        snapshot_sha = hashlib.sha256(snapshot_bytes).hexdigest()
+        if snapshot_path.exists():
+            if snapshot_path.is_symlink() or not snapshot_path.is_file():
+                raise MigrationPlanError("migration_snapshot_invalid")
+            if hashlib.sha256(snapshot_path.read_bytes()).hexdigest() != snapshot_sha:
+                raise MigrationPlanError("migration_snapshot_conflict")
+        else:
+            temporary = snapshot_path.with_name(snapshot_path.name + ".tmp")
+            descriptor = os.open(
+                temporary,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+                0o640,
+            )
+            try:
+                with os.fdopen(descriptor, "wb") as handle:
+                    handle.write(snapshot_bytes)
+                    handle.flush()
+                    os.fsync(handle.fileno())
+                os.replace(temporary, snapshot_path)
+                if os.name != "nt":
+                    os.chmod(snapshot_path, 0o640)
+                    directory_fd = os.open(snapshot_path.parent, os.O_RDONLY)
+                    try:
+                        os.fsync(directory_fd)
+                    finally:
+                        os.close(directory_fd)
+            finally:
+                try:
+                    temporary.unlink()
+                except FileNotFoundError:
+                    pass
+        return StateMigrationContract(
+            state_id="task-admission",
+            state_ref=state_ref,
+            mode="REUSE_IN_PLACE_WITH_SNAPSHOT",
+            before_schema_sha256=_schema_digest(before),
+            after_schema_sha256=_schema_digest(after),
+            snapshot_name=snapshot_name,
+            snapshot_sha256=snapshot_sha,
+            snapshot_required=True,
+            rollback_compatible=True,
+        )
+
+    def compile(
+        self,
+        *,
+        source_admission: Mapping[str, Any],
+        candidate_id: str,
+    ) -> tuple[MigrationRequirement, MigrationPlan]:
+        profile = self._profile()
+        version = str(source_admission.get("version") or "")
+        source_commit = _oid(source_admission.get("source_commit"), "source_commit")
+        tree_oid = _oid(source_admission.get("tree_oid"), "tree_oid")
+        admission_id = _ident(source_admission.get("admission_id"), "source_admission_id")
+        if (
+            version != profile["version"]
+            or source_admission.get("repository") != "aios/aios"
+        ):
+            raise MigrationPlanError("source_admission_profile_mismatch")
+        _ident(candidate_id, "candidate_id")
+        profile_sha = hashlib.sha256(_canonical_json(profile)).hexdigest()
+        from_components = tuple(
+            self._component_from_live(item) for item in profile["current_components"]
+        )
+        to_components = tuple(
+            self._component_from_target(item, source_commit=source_commit)
+            for item in profile["target_components"]
+        )
+        if {item.component_id for item in from_components} != {item.component_id for item in to_components}:
+            raise MigrationPlanError("migration_component_profile_mismatch")
+        required = any(
+            next(old.code_line_id for old in from_components if old.component_id == new.component_id)
+            != new.code_line_id
+            for new in to_components
+        )
+        if not required:
+            raise MigrationPlanError("migration_profile_does_not_require_transition")
+        activation_path = Path(str(profile["activation_file"]))
+        activation = _secure_json(activation_path, expected_owner_uid=self.expected_owner_uid)
+        if activation.get("schema") != "aios.task-admission-host-activation.v1":
+            raise MigrationPlanError("activation_schema_invalid")
+        state_path = Path(str(profile["state_db"]))
+        if str(activation.get("state_db") or "") != str(state_path):
+            raise MigrationPlanError("activation_state_db_mismatch")
+        root = self.output_root / source_commit
+        snapshot_name = "snapshots/task-admission.sqlite3"
+        state_contract = self._state_compatibility(
+            state_path,
+            str(profile["state_ref"]),
+            snapshot_path=root / snapshot_name,
+            snapshot_name=snapshot_name,
+        )
+        migration_id = (
+            "migration-"
+            + hashlib.sha256(
+                f"{version}\0{candidate_id}\0{source_commit}\0{profile_sha}".encode("utf-8")
+            ).hexdigest()[:24]
+        )
+        plan = MigrationPlan(
+            migration_id=migration_id,
+            version=version,
+            candidate_id=candidate_id,
+            repository_identity="aios/aios",
+            source_admission_id=admission_id,
+            source_commit=source_commit,
+            tree_oid=tree_oid,
+            profile_id=str(profile["profile_id"]),
+            profile_sha256=profile_sha,
+            from_components=from_components,
+            to_components=to_components,
+            state_contracts=(state_contract,),
+            activation_schema=str(activation["schema"]),
+            activation_sha256=hashlib.sha256(_canonical_json(activation)).hexdigest(),
+            rollback_window_seconds=int(profile["rollback_window_seconds"]),
+        )
+        requirement = MigrationRequirement(
+            version=version,
+            candidate_id=candidate_id,
+            source_commit=source_commit,
+            repository_identity="aios/aios",
+            required=True,
+            reason_code="CROSS_CODE_LINE",
+            profile_id=str(profile["profile_id"]),
+            profile_sha256=profile_sha,
+            migration_plan_sha256=plan.sha256,
+        )
+        _durable_json(root / "requirement.json", requirement.to_dict())
+        _durable_json(root / "plan.json", plan.to_dict())
+        return requirement, plan
+
+
+def _verified_source_admission(
+    *,
+    source_admission_root: Path,
+    source_commit: str,
+    builder_policy: Path,
+    credential_name: str = "source_admission_hmac",
+) -> dict[str, Any]:
+    """Resolve one source admission through its existing HMAC verifier.
+
+    The migration planner never trusts a caller-selected admission JSON path.
+    It consumes only a root-owned admission store and the independent
+    ``source_admission_hmac`` credential projected by systemd.
+    """
+
+    from release_source_admission import open_source_admission_store
+
+    credentials = os.environ.get("CREDENTIALS_DIRECTORY", "").strip()
+    if not credentials:
+        raise MigrationPlanError("source_admission_credentials_unavailable")
+    credential_path = Path(credentials) / credential_name
+    try:
+        hmac_key = credential_path.read_bytes().strip()
+    except OSError as exc:
+        raise MigrationPlanError("source_admission_hmac_unavailable") from exc
+    try:
+        store = open_source_admission_store(
+            source_admission_root,
+            repository="aios/aios",
+            policy=builder_policy,
+            hmac_key=hmac_key,
+            expected_owner_uid=0 if os.name != "nt" else None,
+            writable=False,
+        )
+        record = store.lookup_source(source_commit)
+    except Exception as exc:
+        raise MigrationPlanError("source_admission_verification_failed") from exc
+    if record is None:
+        raise MigrationPlanRequired("SOURCE_ADMISSION_REQUIRED")
+    return dict(record)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--authority-git-dir", required=True, type=Path)
+    parser.add_argument("--profile", required=True, type=Path)
+    parser.add_argument("--output-root", required=True, type=Path)
+    parser.add_argument("--source-admission-root", required=True, type=Path)
+    parser.add_argument("--builder-policy", required=True, type=Path)
+    parser.add_argument("--source-commit", required=True)
+    parser.add_argument("--candidate-id", required=True)
+    args = parser.parse_args()
+    source_commit = _oid(args.source_commit, "source_commit")
+    compiler = HostMigrationCompiler(
+        authority_git_dir=args.authority_git_dir,
+        profile_path=args.profile,
+        output_root=args.output_root,
+        expected_owner_uid=0 if os.name != "nt" else None,
+    )
+    requirement, plan = compiler.compile(
+        source_admission=_verified_source_admission(
+            source_admission_root=args.source_admission_root,
+            source_commit=source_commit,
+            builder_policy=args.builder_policy,
+        ),
+        candidate_id=args.candidate_id,
+    )
+    print(
+        json.dumps(
+            {
+                "status": "MIGRATION_PLAN_READY",
+                "migration_plan_sha256": plan.sha256,
+                "requirement_sha256": requirement.sha256,
+                "source_commit": plan.source_commit,
+                "version": plan.version,
+            },
+            sort_keys=True,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except Exception as exc:
+        print(
+            json.dumps(
+                {
+                    "status": "FAIL_CLOSED",
+                    "error": type(exc).__name__,
+                    "reason": str(exc),
+                },
+                sort_keys=True,
+            )
+        )
+        raise SystemExit(2)

active-runtime/app/release_source_admission.py

@@ -325,6 +325,195 @@ class ReleaseSourceAdmissionStore:
         )
         return dict(checked)
 
+    def rebind_policy(
+        self,
+        source_commit: str,
+        *,
+        previous_policy_sha256: str,
+        reason: str,
+        expected_admission_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Rebind one immutable source identity to an explicitly authorized policy.
+
+        Only ``policy_sha256`` and ``mac`` change in the admission record.
+        ``admission_id`` and every source/candidate/tree binding remain immutable.
+        A separate HMAC-authenticated history record makes the temporary §8.1
+        trust expansion and its restoration auditable and idempotent.
+        """
+
+        if not self.writable:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_READ_ONLY")
+        previous_policy = _digest(
+            previous_policy_sha256,
+            "previous_policy_sha256",
+        )
+        reason_value = _identifier(reason, "reason")
+        path = self._path(source_commit)
+        if not path.exists():
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_REQUIRED")
+        _secure_stat(
+            path,
+            expected_owner_uid=self.expected_owner_uid,
+            directory=False,
+        )
+        try:
+            raw = path.read_bytes()
+            decoded = json.loads(raw)
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ReleaseSourceAdmissionError(
+                "SOURCE_ADMISSION_RECORD_INVALID"
+            ) from exc
+
+        try:
+            previous = _validate_record(
+                decoded,
+                repository=self.repository,
+                policy_sha256=previous_policy,
+                hmac_key=self.hmac_key,
+            )
+        except ReleaseSourceAdmissionError:
+            # Idempotent replay after a successful rebind validates under the
+            # current policy instead of silently weakening the old-policy gate.
+            current = _validate_record(
+                decoded,
+                repository=self.repository,
+                policy_sha256=self.policy_sha256,
+                hmac_key=self.hmac_key,
+            )
+            if expected_admission_id and current["admission_id"] != expected_admission_id:
+                raise ReleaseSourceAdmissionError(
+                    "SOURCE_ADMISSION_ID_MISMATCH"
+                )
+            return dict(current)
+
+        if expected_admission_id and previous["admission_id"] != expected_admission_id:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_ID_MISMATCH")
+        rebound = {
+            key: value
+            for key, value in previous.items()
+            if key != "mac"
+        }
+        rebound["policy_sha256"] = self.policy_sha256
+        rebound["mac"] = hmac.new(
+            self.hmac_key,
+            _canonical(rebound),
+            hashlib.sha256,
+        ).hexdigest()
+        checked = _validate_record(
+            rebound,
+            repository=self.repository,
+            policy_sha256=self.policy_sha256,
+            hmac_key=self.hmac_key,
+        )
+        if checked["admission_id"] != previous["admission_id"]:
+            raise ReleaseSourceAdmissionError(
+                "SOURCE_ADMISSION_ID_CHANGED"
+            )
+        for field in (
+            "repository",
+            "version",
+            "candidate_id",
+            "base_oid",
+            "source_ref",
+            "source_commit",
+            "remote_ref_oid",
+            "tree_oid",
+            "purpose",
+            "task_id",
+        ):
+            if checked[field] != previous[field]:
+                raise ReleaseSourceAdmissionError(
+                    "SOURCE_ADMISSION_REBIND_SCOPE_CHANGED"
+                )
+
+        encoded = _canonical(checked) + b"\n"
+        history_payload = {
+            "schema": "aios.release-source-admission-policy-rebind.v1",
+            "admission_id": checked["admission_id"],
+            "source_commit": checked["source_commit"],
+            "previous_policy_sha256": previous_policy,
+            "policy_sha256": self.policy_sha256,
+            "reason": reason_value,
+            "record_sha256": hashlib.sha256(encoded).hexdigest(),
+            "rebound_at": _now(),
+        }
+        history_payload["mac"] = hmac.new(
+            self.hmac_key,
+            _canonical(history_payload),
+            hashlib.sha256,
+        ).hexdigest()
+
+        history_root = self.root / "rebind-history"
+        if not history_root.exists():
+            history_root.mkdir(mode=0o750)
+        _secure_stat(
+            history_root,
+            expected_owner_uid=self.expected_owner_uid,
+            directory=True,
+        )
+        history_name = (
+            f"{checked['source_commit']}."
+            f"{previous_policy[:12]}-{self.policy_sha256[:12]}."
+            f"{hashlib.sha256(reason_value.encode()).hexdigest()[:12]}.json"
+        )
+        history_path = history_root / history_name
+        history_bytes = _canonical(history_payload) + b"\n"
+
+        fd, raw_name = tempfile.mkstemp(
+            prefix=f".{checked['source_commit']}.rebind.",
+            suffix=".tmp",
+            dir=self.root,
+        )
+        temp_path = Path(raw_name)
+        try:
+            os.fchmod(fd, 0o640)
+            with os.fdopen(fd, "wb", closefd=True) as handle:
+                handle.write(encoded)
+                handle.flush()
+                os.fsync(handle.fileno())
+            if history_path.exists():
+                _secure_stat(
+                    history_path,
+                    expected_owner_uid=self.expected_owner_uid,
+                    directory=False,
+                )
+                if history_path.read_bytes() != history_bytes:
+                    raise ReleaseSourceAdmissionError(
+                        "SOURCE_ADMISSION_REBIND_HISTORY_CONFLICT"
+                    )
+            else:
+                history_fd = os.open(
+                    history_path,
+                    os.O_WRONLY
+                    | os.O_CREAT
+                    | os.O_EXCL
+                    | getattr(os, "O_NOFOLLOW", 0),
+                    0o640,
+                )
+                try:
+                    os.write(history_fd, history_bytes)
+                    os.fsync(history_fd)
+                finally:
+                    os.close(history_fd)
+            os.replace(temp_path, path)
+            os.chmod(path, 0o640)
+            directory_fd = os.open(
+                self.root,
+                os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
+            )
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        except Exception:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+            temp_path.unlink(missing_ok=True)
+            raise
+        return dict(checked)
+
     def lookup_source(self, source_commit: str) -> dict[str, Any] | None:
         path = self._path(source_commit)
         if not path.exists():
@@ -499,9 +688,98 @@ def _sha256_file(path: Path) -> str:
     return digest.hexdigest()
 
 
+def source_admission_policy_sha256(path: Path) -> str:
+    """Canonical Source Admission policy binding: exact policy file bytes.
+
+    Builder/controller readiness may use a canonical JSON digest internally,
+    but Source Admission records are intentionally bound to the immutable policy
+    file bytes.  Every admission producer/consumer must call this helper.
+    """
+
+    try:
+        return _sha256_file(Path(path))
+    except OSError as exc:
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_UNAVAILABLE") from exc
+
+
+def open_source_admission_store(
+    root: str | os.PathLike[str],
+    *,
+    repository: str,
+    policy: Path,
+    hmac_key: bytes,
+    expected_owner_uid: int | None,
+    writable: bool,
+) -> ReleaseSourceAdmissionStore:
+    """The only supported constructor for policy-bound Source Admission stores."""
+
+    return ReleaseSourceAdmissionStore(
+        root,
+        repository=repository,
+        policy_sha256=source_admission_policy_sha256(policy),
+        hmac_key=hmac_key,
+        expected_owner_uid=expected_owner_uid,
+        writable=writable,
+    )
+
+
+def validate_isolated_peer_policy_delta(
+    previous_policy: Path,
+    policy: Path,
+    isolated_peer_cgroup: str,
+) -> tuple[str, str, str]:
+    """Allow exactly one temporary isolated peer add/remove and no other drift."""
+
+    peer = str(isolated_peer_cgroup or "").strip()
+    if (
+        not peer.startswith("/system.slice/")
+        or not peer.endswith(".service")
+        or "*" in peer
+        or ".." in peer
+    ):
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_PEER_INVALID")
+    try:
+        from aios_trusted_candidate_builder import load_builder_policy
+
+        before = dict(load_builder_policy(Path(previous_policy)))
+        after = dict(load_builder_policy(Path(policy)))
+    except Exception as exc:
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_INVALID") from exc
+
+    before_cgroups = tuple(before.get("allowed_peer_cgroups") or ())
+    after_cgroups = tuple(after.get("allowed_peer_cgroups") or ())
+    if len(set(before_cgroups)) != len(before_cgroups) or len(set(after_cgroups)) != len(after_cgroups):
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_PEER_INVALID")
+    if any("*" in value or ".." in value for value in (*before_cgroups, *after_cgroups)):
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_PEER_INVALID")
+
+    before_without = dict(before)
+    after_without = dict(after)
+    before_without["allowed_peer_cgroups"] = []
+    after_without["allowed_peer_cgroups"] = []
+    if _canonical(before_without) != _canonical(after_without):
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_REBIND_SCOPE_CHANGED")
+
+    before_set, after_set = set(before_cgroups), set(after_cgroups)
+    added = after_set - before_set
+    removed = before_set - after_set
+    if added == {peer} and not removed:
+        direction = "ADD_ISOLATED_PEER"
+    elif removed == {peer} and not added:
+        direction = "REMOVE_ISOLATED_PEER"
+    else:
+        raise ReleaseSourceAdmissionError("SOURCE_POLICY_PEER_DELTA_INVALID")
+    return (
+        source_admission_policy_sha256(previous_policy),
+        source_admission_policy_sha256(policy),
+        direction,
+    )
+
+
 def _main() -> None:
     parser = argparse.ArgumentParser()
     subparsers = parser.add_subparsers(dest="command", required=True)
+
     admit = subparsers.add_parser("admit")
     admit.add_argument("--version", required=True)
     admit.add_argument("--task-id", required=True)
@@ -509,15 +787,77 @@ def _main() -> None:
     admit.add_argument("--source-commit", required=True)
     admit.add_argument("--policy", required=True, type=Path)
     admit.add_argument("--store-root", required=True, type=Path)
-    arguments = parser.parse_args()
 
-    if arguments.command != "admit":
-        raise SystemExit("unsupported command")
+    rebind = subparsers.add_parser("rebind-policy")
+    rebind.add_argument("--source-commit", required=True)
+    rebind.add_argument("--previous-policy", required=True, type=Path)
+    rebind.add_argument("--policy", required=True, type=Path)
+    rebind.add_argument("--isolated-peer-cgroup", required=True)
+    rebind.add_argument("--expected-admission-id", required=True)
+    rebind.add_argument("--reason", required=True)
+    rebind.add_argument("--store-root", required=True, type=Path)
+
+    arguments = parser.parse_args()
     repository = os.environ.get("AIOS_GITOPS_REPOSITORY_IDENTITY", "")
+    if not repository:
+        raise SystemExit("AIOS_GITOPS_REPOSITORY_IDENTITY is required")
+    credentials = os.environ.get("CREDENTIALS_DIRECTORY", "")
+    if not credentials:
+        raise SystemExit("CREDENTIALS_DIRECTORY is required")
+    try:
+        admission_hmac = (
+            Path(credentials) / "source_admission_hmac"
+        ).read_bytes().strip()
+    except OSError as exc:
+        raise SystemExit(
+            "source_admission_hmac credential is required"
+        ) from exc
+
+    if arguments.command == "rebind-policy":
+        try:
+            previous_sha, policy_sha, direction = (
+                validate_isolated_peer_policy_delta(
+                    arguments.previous_policy,
+                    arguments.policy,
+                    arguments.isolated_peer_cgroup,
+                )
+            )
+            store = open_source_admission_store(
+                arguments.store_root,
+                repository=repository,
+                policy=arguments.policy,
+                hmac_key=admission_hmac,
+                expected_owner_uid=(
+                    os.geteuid() if hasattr(os, "geteuid") else None
+                ),
+                writable=True,
+            )
+            result = store.rebind_policy(
+                arguments.source_commit,
+                previous_policy_sha256=previous_sha,
+                reason=f"{arguments.reason}:{direction}",
+                expected_admission_id=arguments.expected_admission_id,
+            )
+        except ReleaseSourceAdmissionError as exc:
+            raise SystemExit(str(exc)) from exc
+        print(
+            json.dumps(
+                {
+                    **result,
+                    "rebind_direction": direction,
+                    "previous_policy_sha256": previous_sha,
+                    "policy_sha256": policy_sha,
+                },
+                sort_keys=True,
+                separators=(",", ":"),
+            )
+        )
+        return
+
     api_base = os.environ.get("AIOS_GITEA_API_BASE", "")
     remote_url = os.environ.get("AIOS_GITEA_REMOTE_URL", "")
     raw_git_dir = os.environ.get("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", "")
-    if not repository or not api_base or not remote_url or not raw_git_dir:
+    if not api_base or not remote_url or not raw_git_dir:
         raise SystemExit("release source admission environment is incomplete")
     git_dir = Path(raw_git_dir)
     transport = GiteaMachineTransport(
@@ -530,17 +870,10 @@ def _main() -> None:
             "gitea-machine-token",
         ),
     )
-    credentials = os.environ.get("CREDENTIALS_DIRECTORY", "")
-    if not credentials:
-        raise SystemExit("CREDENTIALS_DIRECTORY is required")
-    try:
-        admission_hmac = (Path(credentials) / "source_admission_hmac").read_bytes().strip()
-    except OSError as exc:
-        raise SystemExit("source_admission_hmac credential is required") from exc
-    store = ReleaseSourceAdmissionStore(
+    store = open_source_admission_store(
         arguments.store_root,
         repository=repository,
-        policy_sha256=_sha256_file(arguments.policy),
+        policy=arguments.policy,
         hmac_key=admission_hmac,
         expected_owner_uid=os.geteuid() if hasattr(os, "geteuid") else None,
         writable=True,

active-runtime/app/silent_release_hardening.py

@@ -25,6 +25,12 @@ from typing import Any, Callable, Iterable, Mapping, Protocol
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 
+from release_migration_plan import (
+    MigrationPlan,
+    MigrationPlanError,
+    MigrationPreSwitchValidator,
+)
+
 from silent_release_controller import (
     CANDIDATE_ID_RE,
     OID_RE,
@@ -562,6 +568,7 @@ class ReleasePlan:
     tag: TagPlan
     finalization: FinalizationPlan
     policy_sha256: str
+    migration: MigrationPlan | None = None
 
     def __post_init__(self) -> None:
         _validate_sha("policy_sha256", self.policy_sha256)
@@ -588,16 +595,32 @@ class ReleasePlan:
             != self.finalization.expected_previous_baseline_sha256
         ):
             raise ValueError("previous baseline is inconsistent")
+        if self.migration is not None:
+            migration = self.migration
+            if (
+                migration.version != self.tag.version
+                or migration.source_commit != self.deployment.source_commit
+                or migration.tree_oid != self.deployment.tree_oid
+                or migration.repository_identity
+                != self.deployment.repository_identity
+            ):
+                raise ValueError("migration plan release binding is inconsistent")
         return self
 
     @property
     def canonical_payload(self) -> dict[str, Any]:
-        return {
+        # Legacy release-plan bytes/hashes remain unchanged when migration is
+        # not required.  A cross-code-line migration is an optional signed
+        # sub-contract and is therefore present only when required.
+        payload = {
             "deployment": self.deployment.canonical_payload,
             "finalization": self.finalization.canonical_payload,
             "policy_sha256": self.policy_sha256,
             "tag": self.tag.canonical_payload,
         }
+        if self.migration is not None:
+            payload["migration"] = self.migration.to_dict()
+        return payload
 
     @property
     def sha256(self) -> str:
@@ -612,6 +635,7 @@ class ReleasePlan:
     def from_dict(cls, payload: Mapping[str, Any]) -> "ReleasePlan":
         values = dict(payload)
         values.pop("sha256", None)
+        migration_value = values.get("migration")
         return cls(
             deployment=DeploymentPlan.from_dict(values["deployment"]),
             tag=TagPlan.from_dict(values["tag"]),
@@ -619,6 +643,11 @@ class ReleasePlan:
                 values["finalization"]
             ),
             policy_sha256=values["policy_sha256"],
+            migration=(
+                MigrationPlan.from_dict(migration_value)
+                if isinstance(migration_value, Mapping)
+                else None
+            ),
         )
 
 
@@ -1215,6 +1244,19 @@ class TrustedCandidateBundle:
             raise CapabilityAuthorizationError(
                 "trusted candidate and release plan are inconsistent"
             )
+        if plan.migration is not None:
+            try:
+                plan.migration.assert_release_binding(
+                    version=plan.tag.version,
+                    candidate_id=candidate.candidate_id,
+                    source_commit=candidate.source_commit,
+                    tree_oid=candidate.tree_oid,
+                    repository_identity=plan.deployment.repository_identity,
+                )
+            except MigrationPlanError as exc:
+                raise CapabilityAuthorizationError(
+                    "trusted candidate migration binding is invalid"
+                ) from exc
         if attestation_verifier is not None:
             if self.attestation is None:
                 raise CapabilityAuthorizationError(
@@ -3090,6 +3132,7 @@ class ReleaseOrchestrator:
         runtime_rollback: Callable[..., Mapping[str, Any]],
         tag_publish: HardenedTagController,
         finalizer: GovernedFinalizationCoordinator,
+        migration_pre_switch_validator: MigrationPreSwitchValidator | None = None,
     ):
         self.candidates = candidates
         self.plans = plans
@@ -3104,6 +3147,7 @@ class ReleaseOrchestrator:
         self.runtime_rollback = runtime_rollback
         self.tag_publish = tag_publish
         self.finalizer = finalizer
+        self.migration_pre_switch_validator = migration_pre_switch_validator
 
     def _transition(
         self,
@@ -3304,6 +3348,14 @@ class ReleaseOrchestrator:
                         plan.deployment.repository_identity
                     ),
                 )
+                if plan.migration is not None:
+                    validator = self.migration_pre_switch_validator
+                    if validator is None:
+                        raise StaleApprovalError(
+                            "MIGRATION_PLAN_REQUIRED: "
+                            "migration pre-switch validator is unavailable"
+                        )
+                    validator.validate(plan.migration)
                 saga = self._transition(
                     saga,
                     SagaState.SWITCH_PENDING,
@@ -3697,47 +3749,79 @@ class SilentReleaseActionService:
             record["source_admission_id"] = source_admission["admission_id"]
         if was_existing:
             existing = json.loads(path.read_text(encoding="utf-8"))
+            prerequisite_states = {
+                "SOURCE_ADMISSION_REQUIRED",
+                "MIGRATION_PLAN_REQUIRED",
+            }
+            existing_status = str(existing.get("status") or "")
+            existing_version = str(existing.get("version") or "")
+            existing_admission = str(existing.get("source_admission_id") or "")
+            incoming_admission = str(record.get("source_admission_id") or "")
             if (
                 existing.get("event_id") != event_id
                 or existing.get("source_commit") != source_commit
-                or str(existing.get("version") or "") != version
-                or str(existing.get("source_admission_id") or "")
-                != str(record.get("source_admission_id") or "")
+                or (existing_version and existing_version != version)
+                or (existing_admission and existing_admission != incoming_admission)
             ):
                 raise SagaConflict(
                     "release intent id is bound to different evidence"
                 )
-            if existing.get("status") == "BUILDING":
-                try:
-                    started_at = _parse_utc(
-                        str(existing.get("attempt_started_at") or "")
-                    )
-                    elapsed = (
-                        datetime.now(timezone.utc) - started_at
-                    ).total_seconds()
-                except (TypeError, ValueError):
-                    elapsed = self.builder_attempt_timeout_seconds + 1
-                if elapsed <= self.builder_attempt_timeout_seconds:
-                    return self._intent_response(existing, idempotent=True)
-                existing.update(
-                    {
-                        "failure_code": "TRUSTED_BUILDER_CRASH_RECOVERED",
-                        "last_recovery_code": "TRUSTED_BUILDER_CRASH_RECOVERED",
-                        "recovery_count": int(existing.get("recovery_count") or 0) + 1,
-                        "retryable": int(existing.get("attempt_count") or 0)
-                        < self.max_builder_attempts,
-                        "status": "BUILDER_RETRYABLE_FAILURE",
-                        "updated_at": utc_now(),
-                    }
-                )
-                _durable_json(path, existing)
+            # A source-admission prerequisite may initially know only the
+            # event/source identity.  Once the verified admission becomes
+            # available, bind the previously missing version/admission ID to
+            # the same durable intent.  Existing non-empty evidence can never
+            # be replaced.
             if (
-                existing.get("status") != "BUILDER_RETRYABLE_FAILURE"
-                or int(existing.get("attempt_count") or 0)
-                >= self.max_builder_attempts
+                (not existing_version and version)
+                or (not existing_admission and incoming_admission)
             ):
-                return self._intent_response(existing, idempotent=True)
-            record = dict(existing)
+                if existing_status != "SOURCE_ADMISSION_REQUIRED":
+                    raise SagaConflict(
+                        "release intent prerequisite binding changed"
+                    )
+                if version:
+                    existing["version"] = version
+                if incoming_admission:
+                    existing["source_admission_id"] = incoming_admission
+                existing["updated_at"] = utc_now()
+                _durable_json(path, existing)
+            if existing_status in prerequisite_states:
+                # Prerequisite blockers are not builder attempts.  Replaying
+                # the same intent after Source Admission or Migration Plan
+                # material appears is the expected autonomous recovery path.
+                record = dict(existing)
+            else:
+                if existing.get("status") == "BUILDING":
+                    try:
+                        started_at = _parse_utc(
+                            str(existing.get("attempt_started_at") or "")
+                        )
+                        elapsed = (
+                            datetime.now(timezone.utc) - started_at
+                        ).total_seconds()
+                    except (TypeError, ValueError):
+                        elapsed = self.builder_attempt_timeout_seconds + 1
+                    if elapsed <= self.builder_attempt_timeout_seconds:
+                        return self._intent_response(existing, idempotent=True)
+                    existing.update(
+                        {
+                            "failure_code": "TRUSTED_BUILDER_CRASH_RECOVERED",
+                            "last_recovery_code": "TRUSTED_BUILDER_CRASH_RECOVERED",
+                            "recovery_count": int(existing.get("recovery_count") or 0) + 1,
+                            "retryable": int(existing.get("attempt_count") or 0)
+                            < self.max_builder_attempts,
+                            "status": "BUILDER_RETRYABLE_FAILURE",
+                            "updated_at": utc_now(),
+                        }
+                    )
+                    _durable_json(path, existing)
+                if (
+                    existing.get("status") != "BUILDER_RETRYABLE_FAILURE"
+                    or int(existing.get("attempt_count") or 0)
+                    >= self.max_builder_attempts
+                ):
+                    return self._intent_response(existing, idempotent=True)
+                record = dict(existing)
         if self.candidate_builder is None:
             record.update(
                 {
@@ -3783,18 +3867,31 @@ class SilentReleaseActionService:
             _durable_json(path, record)
             return self._intent_response(record, idempotent=False)
         except CapabilityAuthorizationError as exc:
-            if str(exc) == "release source admission required":
+            prerequisite_code = {
+                "release source admission required": "SOURCE_ADMISSION_REQUIRED",
+                "migration plan required": "MIGRATION_PLAN_REQUIRED",
+            }.get(str(exc))
+            if prerequisite_code is not None:
                 record.update(
                     {
-                        "failure_code": "SOURCE_ADMISSION_REQUIRED",
-                        "retryable": False,
-                        "status": "SOURCE_ADMISSION_REQUIRED",
+                        # Prerequisite discovery is a control-plane retry, not
+                        # a business/builder execution attempt.
+                        "attempt_count": max(
+                            0, int(record.get("attempt_count") or 0) - 1
+                        ),
+                        "failure_code": prerequisite_code,
+                        "retryable": True,
+                        "status": prerequisite_code,
                     }
                 )
             else:
                 record.update(
                     {
-                        "failure_code": "TRUSTED_BUILDER_REJECTED",
+                        "failure_code": (
+                            "MIGRATION_PLAN_INVALID"
+                            if str(exc) == "migration plan invalid"
+                            else "TRUSTED_BUILDER_REJECTED"
+                        ),
                         "retryable": False,
                         "status": "BUILDER_TERMINAL_FAILURE",
                     }

config/gitops/release-migration-profiles/V2R2C2-baseline11-principal-recovery.json

@@ -0,0 +1 @@
+{"activation_file":"/var/lib/aios-gitops/bridge-v2-staging/task-admission-activation.json","current_components":[{"artifact_sha_env":"AIOS_GITOPS_ARTIFACT_SHA256","code_line_id":"legacy-governed-main-v15","component_id":"gitops-model-api-staging","expected_source_oid":"91f104ef62a3b688a9259b96e2ca247e0363b564","interface_files":["app/aios_model_api_staging.py","app/silent_release_hardening.py"],"marker_files":["app/aios_model_api_staging.py"],"service":"aios-gitops-model-api-v2r2c1-staging.service","source_oid_env":"AIOS_GITOPS_RUNTIME_OID"},{"artifact_sha_env":"AIOS_CONTROL_PLANE_ARTIFACT_SHA256","code_line_id":"legacy-p0-task-principal","component_id":"chatgpt-bridge-v2-staging","expected_source_oid":"91f104ef62a3b688a9259b96e2ca247e0363b564","interface_files":["app/aios_chatgpt_bridge_v2_server.py","app/aios_task_principal_admission.py"],"marker_files":["app/aios_task_principal_admission.py"],"service":"aios-chatgpt-bridge-v2-staging.service","source_oid_env":"AIOS_CONTROL_PLANE_TARGET_OID"}],"profile_id":"baseline11-principal-recovery-cross-code-line-v1","repository":"aios/aios","rollback_window_seconds":3600,"schema":"aios.release-migration-profile.v1","state_db":"/var/lib/aios-gitops/bridge-v2-staging/task-admission.sqlite3","state_mode":"REUSE_IN_PLACE_WITH_SNAPSHOT","state_ref":"state:task-admission","target_components":[{"code_line_id":"baseline11-release-source-admission","component_id":"gitops-model-api-staging","interface_files":["active-runtime/app/aios_model_api_staging.py","active-runtime/app/silent_release_hardening.py","active-runtime/app/aios_trusted_candidate_builder.py","active-runtime/app/release_migration_plan.py"],"service":"aios-gitops-model-api-v2r2c1-staging.service"},{"code_line_id":"baseline11-task-admission","component_id":"chatgpt-bridge-v2-staging","interface_files":["active-runtime/app/aios_chatgpt_bridge_v2_server.py","active-runtime/app/aios_task_admission_bootstrap.py","active-runtime/app/aios_task_admission_live_adapters.py"],"service":"aios-chatgpt-bridge-v2-staging.service"}],"version":"V2R2C2-baseline11-principal-recovery"}

deploy/systemd/aios-release-evidence-materializer.service

@@ -0,0 +1,41 @@
+[Unit]
+Description=AIOS governed trusted-builder evidence materializer
+After=aios-release-source-admission.service aios-release-migration-planner.service
+ConditionPathExists=/var/lib/aios-release-source-admission
+ConditionPathExists=/var/lib/aios-release-migration
+
+[Service]
+Type=oneshot
+User=root
+Group=gitops
+UMask=0027
+LoadCredentialEncrypted=source_admission_hmac:/etc/credstore.encrypted/aios-release-source-admission/source_admission_hmac
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_RELEASE_EVIDENCE_SOURCE_ROOT=/var/lib/aios-release-evidence-staging
+Environment=AIOS_TRUSTED_BUILDER_EVIDENCE_ROOT=/var/lib/aios-trusted-candidate-builder/evidence
+EnvironmentFile=-/run/aios-release-evidence-materializer/request.env
+Environment=PYTHONDONTWRITEBYTECODE=1
+ExecStart=/usr/bin/python3 -B /opt/aios/release-evidence-materializer/current/app/release_evidence_materializer.py materialize --version ${AIOS_RELEASE_VERSION} --candidate-id ${AIOS_RELEASE_CANDIDATE_ID} --source-commit ${AIOS_RELEASE_SOURCE_COMMIT} --builder-policy /opt/aios/trusted-candidate-builder/current/config/gitops/trusted-candidate-builder-policy.json --authority-git-dir /data/aios-gitops/repo.git --source-root ${AIOS_RELEASE_EVIDENCE_SOURCE_ROOT} --output-root ${AIOS_TRUSTED_BUILDER_EVIDENCE_ROOT} --source-admission-root /var/lib/aios-release-source-admission --migration-plan-root /var/lib/aios-release-migration --migration-profile-root /etc/aios-gitops/release-migration-profiles
+ExecStartPost=/usr/bin/rm -f /run/aios-release-evidence-materializer/request.env
+RuntimeDirectory=aios-release-evidence-materializer
+RuntimeDirectoryMode=0750
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateNetwork=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectControlGroups=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectKernelLogs=true
+RestrictAddressFamilies=AF_UNIX
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true
+SystemCallArchitectures=native
+CapabilityBoundingSet=
+ReadOnlyPaths=/opt/aios/release-evidence-materializer /opt/aios/trusted-candidate-builder /data/aios-gitops/repo.git /var/lib/aios-release-evidence-staging /var/lib/aios-release-source-admission /var/lib/aios-release-migration /etc/aios-gitops/release-migration-profiles /etc/credstore.encrypted
+ReadWritePaths=/var/lib/aios-trusted-candidate-builder/evidence /run/aios-release-evidence-materializer
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-release-migration-planner.service

@@ -0,0 +1,41 @@
+[Unit]
+Description=AIOS governed cross-code-line release migration planner
+After=aios-release-source-admission.service
+ConditionPathExists=/var/lib/aios-release-source-admission
+
+[Service]
+Type=oneshot
+User=root
+Group=gitops
+UMask=0027
+LoadCredentialEncrypted=source_admission_hmac:/etc/credstore.encrypted/aios-release-source-admission/source_admission_hmac
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_RELEASE_MIGRATION_PROFILE=/etc/aios-gitops/release-migration-profiles/V2R2C2-baseline11-principal-recovery.json
+EnvironmentFile=-/run/aios-release-migration-planner/request.env
+Environment=PYTHONDONTWRITEBYTECODE=1
+ExecStart=/usr/bin/python3 -B /opt/aios/release-migration/current/app/release_migration_plan.py --authority-git-dir /data/aios-gitops/repo.git --profile ${AIOS_RELEASE_MIGRATION_PROFILE} --output-root /var/lib/aios-release-migration --source-admission-root /var/lib/aios-release-source-admission --builder-policy /opt/aios/trusted-candidate-builder/current/config/gitops/trusted-candidate-builder-policy.json --source-commit ${AIOS_RELEASE_SOURCE_COMMIT} --candidate-id ${AIOS_RELEASE_CANDIDATE_ID}
+ExecStartPost=/usr/bin/rm -f /run/aios-release-migration-planner/request.env
+StateDirectory=aios-release-migration
+StateDirectoryMode=0750
+RuntimeDirectory=aios-release-migration-planner
+RuntimeDirectoryMode=0750
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateNetwork=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectControlGroups=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectKernelLogs=true
+RestrictAddressFamilies=AF_UNIX
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true
+SystemCallArchitectures=native
+CapabilityBoundingSet=
+ReadOnlyPaths=/opt/aios/release-migration /data/aios-gitops/repo.git /var/lib/aios-release-source-admission /opt/aios/trusted-candidate-builder/current/config/gitops/trusted-candidate-builder-policy.json /etc/aios-gitops/release-migration-profiles /var/lib/aios-gitops/bridge-v2-staging
+ReadWritePaths=/var/lib/aios-release-migration /run/aios-release-migration-planner
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-release-source-admission-rebind.service

@@ -0,0 +1,40 @@
+[Unit]
+Description=AIOS governed Source Admission isolated-peer policy rebind
+After=aios-release-source-admission.service
+ConditionPathExists=/var/lib/aios-release-source-admission
+
+[Service]
+Type=oneshot
+User=root
+Group=gitops
+UMask=0027
+LoadCredentialEncrypted=source_admission_hmac:/etc/credstore.encrypted/aios-release-source-admission/source_admission_hmac
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_GITOPS_REPOSITORY_IDENTITY=aios/aios
+Environment=AIOS_RELEASE_ISOLATED_PEER_CGROUP=/system.slice/aios-gitops-model-api-v2r2c2-baseline11-candidate.service
+EnvironmentFile=-/run/aios-release-source-admission-rebind/request.env
+Environment=PYTHONDONTWRITEBYTECODE=1
+ExecStart=/usr/bin/python3 -B /opt/aios/release-source-admission/current/app/release_source_admission.py rebind-policy --source-commit ${AIOS_RELEASE_SOURCE_COMMIT} --previous-policy ${AIOS_RELEASE_PREVIOUS_POLICY} --policy ${AIOS_RELEASE_POLICY} --isolated-peer-cgroup ${AIOS_RELEASE_ISOLATED_PEER_CGROUP} --expected-admission-id ${AIOS_RELEASE_EXPECTED_ADMISSION_ID} --reason ${AIOS_RELEASE_REBIND_REASON} --store-root /var/lib/aios-release-source-admission
+ExecStartPost=/usr/bin/rm -f /run/aios-release-source-admission-rebind/request.env
+RuntimeDirectory=aios-release-source-admission-rebind
+RuntimeDirectoryMode=0750
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateNetwork=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectControlGroups=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectKernelLogs=true
+RestrictAddressFamilies=AF_UNIX
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true
+SystemCallArchitectures=native
+CapabilityBoundingSet=
+ReadOnlyPaths=/opt/aios/release-source-admission /etc/credstore.encrypted
+ReadWritePaths=/var/lib/aios-release-source-admission /run/aios-release-source-admission-rebind
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-trusted-candidate-builder.service

@@ -16,6 +16,8 @@ Environment=AIOS_TRUSTED_BUILDER_GIT_DIR=/data/aios-gitops/repo.git
 Environment=AIOS_TRUSTED_BUILDER_EVIDENCE_ROOT=/var/lib/aios-trusted-candidate-builder/evidence
 Environment=AIOS_TRUSTED_BUILDER_NONCE_LEDGER=/var/lib/aios-trusted-candidate-builder/nonce-ledger.json
 Environment=AIOS_TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT=/var/lib/aios-release-source-admission
+Environment=AIOS_TRUSTED_BUILDER_MIGRATION_PLAN_ROOT=/var/lib/aios-release-migration
+Environment=AIOS_TRUSTED_BUILDER_MIGRATION_PROFILE_ROOT=/etc/aios-gitops/release-migration-profiles
 Environment=AIOS_TRUSTED_BUILDER_KEY_ID=builder-attestation-ed25519-g1
 Environment=AIOS_TRUSTED_BUILDER_KEY_GENERATION=1
 Environment=PYTHONDONTWRITEBYTECODE=1
@@ -39,7 +41,7 @@ MemoryDenyWriteExecute=true
 RestrictSUIDSGID=true
 SystemCallArchitectures=native
 CapabilityBoundingSet=
-ReadOnlyPaths=/opt/aios/trusted-candidate-builder /data/aios-gitops/repo.git /var/lib/aios-trusted-candidate-builder/evidence -/var/lib/aios-release-source-admission
+ReadOnlyPaths=/opt/aios/trusted-candidate-builder /data/aios-gitops/repo.git /var/lib/aios-trusted-candidate-builder/evidence -/var/lib/aios-release-source-admission -/var/lib/aios-release-migration -/etc/aios-gitops/release-migration-profiles
 ReadWritePaths=/var/lib/aios-trusted-candidate-builder
 
 [Install]

tests/test_model_api_trusted_candidate_builder.py

@@ -504,3 +504,24 @@ def test_host_adapter_accepts_intent_only_without_caller_candidate_id():
     assert calls[0][1] == ''
     assert calls[0][2] == COMMIT
     assert len(calls[0][3]) >= 32
+
+
+def test_host_adapter_maps_migration_plan_prerequisite_to_typed_blocker():
+    from aios_trusted_candidate_builder import BuilderError
+
+    api = importlib.import_module("aios_model_api_staging")
+
+    class MigrationRequiredClient:
+        def build(self, version, candidate_id, source_commit, *, nonce):
+            raise BuilderError("MIGRATION_PLAN_REQUIRED")
+
+    with pytest.raises(
+        CapabilityAuthorizationError, match="migration plan required"
+    ):
+        api._HostTrustedCandidateBuilderAdapter(MigrationRequiredClient()).build(
+            {
+                "event_id": "event-migration-required",
+                "source_commit": COMMIT,
+                "version": "V2R2C2-baseline11-principal-recovery",
+            }
+        )

tests/test_release_evidence_materializer.py

@@ -0,0 +1,318 @@
+from __future__ import annotations
+
+import dataclasses
+import hashlib
+import json
+import os
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
+
+from aios_trusted_candidate_builder import (  # noqa: E402
+    BuilderError,
+    TrustedCandidateBuilderConfig,
+    load_builder_policy,
+)
+from release_evidence_materializer import (  # noqa: E402
+    EvidenceMaterializationError,
+    TrustedBuilderEvidenceMaterializer,
+)
+from silent_release_controller import ReleaseCandidate, ReleaseState  # noqa: E402
+from silent_release_hardening import (  # noqa: E402
+    DeploymentPlan,
+    FinalizationPlan,
+    ReleasePlan,
+    TagPlan,
+    TrustedCandidateBundle,
+)
+
+
+POLICY_PATH = ROOT / "config" / "gitops" / "trusted-candidate-builder-policy.json"
+
+
+def canonical(value: dict) -> bytes:
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode()
+
+
+def digest(label: str) -> str:
+    return hashlib.sha256(label.encode()).hexdigest()
+
+
+def write_bytes(path: Path, value: bytes) -> str:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    path.write_bytes(value)
+    path.chmod(0o600)
+    return hashlib.sha256(value).hexdigest()
+
+
+def write_json(path: Path, value: dict) -> str:
+    return write_bytes(path, canonical(value))
+
+
+def build_valid_source(root: Path) -> tuple[dict, dict, dict]:
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
+    recipe = config.fixed_recipe("V2R2C2")
+    version = recipe["version"]
+    version_root = root / version
+    version_root.mkdir(parents=True, mode=0o700)
+
+    tree_oid = "a" * 40
+    artifact_sha = write_bytes(version_root / "artifact.tar.gz", b"artifact-bytes")
+    manifest_sha = write_bytes(version_root / "manifest.json", b'{"manifest":"v1"}\n')
+    closure_sha = write_bytes(version_root / "checksums.sha256", b"checksums-v1\n")
+    policy_sha = write_bytes(
+        version_root / "trusted-release-policy.json",
+        b'{"trusted":"release-policy"}\n',
+    )
+
+    tests = {"passed": True, "receipt": "real-tests"}
+    canary = {"passed": True, "receipt": "real-canary"}
+    gitea = {
+        "consistent": True,
+        "remote_ref_oid": recipe["source_commit"],
+        "repository": config.repository,
+        "source_commit": recipe["source_commit"],
+        "tree_oid": tree_oid,
+    }
+    rollback_contract_sha = digest("rollback-contract")
+    rollback = {
+        "ready": True,
+        "rollback_sha256": rollback_contract_sha,
+        "source_commit": recipe["source_commit"],
+    }
+    secret_scan = {"artifact_sha256": artifact_sha, "passed": True}
+    write_json(version_root / "tests.json", tests)
+    write_json(version_root / "canary.json", canary)
+    write_json(version_root / "gitea.json", gitea)
+    write_json(version_root / "rollback.json", rollback)
+    write_json(version_root / "secret-scan.json", secret_scan)
+
+    candidate = ReleaseCandidate(
+        candidate_id=recipe["candidate_id"],
+        source_commit=recipe["source_commit"],
+        tree_oid=tree_oid,
+        artifact_sha256=artifact_sha,
+        manifest_sha256=manifest_sha,
+        tests=tests,
+        canary=canary,
+        rollback_ready=True,
+        git_consistency={
+            "consistent": True,
+            "repository": config.repository,
+            "source_commit": recipe["source_commit"],
+            "tree_oid": tree_oid,
+        },
+        gitea_consistency=gitea,
+        state=ReleaseState.READY_FOR_APPROVAL,
+    )
+    plan = ReleasePlan(
+        deployment=DeploymentPlan(
+            service="aios-silent-release-controller",
+            release_path="/srv/release/r1",
+            release_root="/srv/release",
+            cutover_link="/srv/release/current",
+            dropin_path="/srv/release/r1/deploy/systemd/controller.service",
+            dropin_sha256=digest("dropin"),
+            systemd_unit_sha256=digest("unit"),
+            rollback_path="/srv/release/rollback-r1",
+            rollback_sha256=rollback_contract_sha,
+            source_commit=recipe["source_commit"],
+            tree_oid=tree_oid,
+            artifact_sha256=artifact_sha,
+            manifest_sha256=manifest_sha,
+            package_closure_sha256=closure_sha,
+            authoritative_git_dir="/data/aios-gitops/repo.git",
+            repository_identity=config.repository,
+            baseline_before_sha256=digest("baseline"),
+            registry_before_sha256=digest("registry"),
+            production_tag_refset_before_sha256=digest("tagset"),
+            health_contract_version="trusted-builder/v1",
+            timeout_policy={"seconds": 10},
+            retry_policy={"attempts": 1},
+            policy_sha256=policy_sha,
+            health_contract={"checked": True},
+        ),
+        tag=TagPlan(
+            version="trusted-v2r2c2-001",
+            tag_type="annotated",
+            expected_tag_oid="b" * 40,
+            expected_target_oid=recipe["source_commit"],
+            evidence_closure_oid="c" * 40,
+            approval_plan_id="owner-plan-trusted-v2r2c2",
+            repository_identity=config.repository,
+            gitea_owner="aios",
+            gitea_repository="aios",
+            current_remote_tag_oid=None,
+            remote_absent_evidence_sha256=digest("tag-absence"),
+            verification_contract="trusted-builder/v1",
+        ),
+        finalization=FinalizationPlan(
+            version="trusted-v2r2c2-001",
+            expected_tag_oid="b" * 40,
+            expected_target_oid=recipe["source_commit"],
+            expected_previous_baseline_sha256=digest("baseline"),
+            registry_expected_record={"base": recipe["source_commit"]},
+            baseline_expected_record={"base": recipe["source_commit"]},
+            final_consistency_contract="trusted-builder/v1",
+        ),
+        policy_sha256=policy_sha,
+    )
+    checks = {name: True for name in config.required_checks}
+    TrustedCandidateBundle(candidate, plan, checks).validate()
+    receipt = {
+        "schema": "aios.trusted-candidate-builder.evidence.v1",
+        "version": version,
+        "candidate": candidate.to_dict(),
+        "release_plan": plan.to_dict(),
+    }
+    write_json(version_root / "receipt.json", receipt)
+    return recipe, candidate.to_dict(), plan.to_dict()
+
+
+def fake_git(tree_oid: str):
+    def runner(argv, **_kwargs):
+        if "rev-parse" in argv:
+            return SimpleNamespace(stdout=tree_oid + "\n", returncode=0)
+        return SimpleNamespace(stdout="", returncode=0)
+    return runner
+
+
+def make_materializer(tmp_path: Path, source_root: Path, output_root: Path, tree_oid: str):
+    authority = tmp_path / "repo.git"
+    authority.mkdir(mode=0o700)
+    output_root.mkdir(mode=0o700)
+    return TrustedBuilderEvidenceMaterializer(
+        builder_policy=POLICY_PATH,
+        authority_git_dir=authority,
+        source_root=source_root,
+        output_root=output_root,
+        expected_owner_uid=None,
+        git_runner=fake_git(tree_oid),
+    )
+
+
+def test_materializer_projects_only_a_fully_validated_closure_and_is_idempotent(tmp_path: Path) -> None:
+    source_root = tmp_path / "source"
+    source_root.mkdir(mode=0o700)
+    recipe, candidate, _plan = build_valid_source(source_root)
+    output_root = tmp_path / "builder-evidence"
+    materializer = make_materializer(
+        tmp_path, source_root, output_root, candidate["tree_oid"]
+    )
+
+    result = materializer.materialize(
+        version=recipe["version"],
+        candidate_id=recipe["candidate_id"],
+        source_commit=recipe["source_commit"],
+    )
+    assert result["status"] == "EVIDENCE_MATERIALIZED"
+    assert result["idempotent"] is False
+    target = output_root / recipe["version"]
+    assert (target / "receipt.json").is_file()
+    assert (target / "materialization.json").is_file()
+    assert (target / "artifact.tar.gz").read_bytes() == b"artifact-bytes"
+
+    replay = materializer.materialize(
+        version=recipe["version"],
+        candidate_id=recipe["candidate_id"],
+        source_commit=recipe["source_commit"],
+    )
+    assert replay["status"] == "EVIDENCE_MATERIALIZED"
+    assert replay["idempotent"] is True
+    assert replay["materialization_id"] == result["materialization_id"]
+
+
+def test_materializer_refuses_tampered_or_incomplete_source_without_writing_output(tmp_path: Path) -> None:
+    source_root = tmp_path / "source"
+    source_root.mkdir(mode=0o700)
+    recipe, candidate, _plan = build_valid_source(source_root)
+    (source_root / recipe["version"] / "manifest.json").write_bytes(b"tampered")
+    (source_root / recipe["version"] / "manifest.json").chmod(0o600)
+    output_root = tmp_path / "builder-evidence"
+    materializer = make_materializer(
+        tmp_path, source_root, output_root, candidate["tree_oid"]
+    )
+
+    with pytest.raises(EvidenceMaterializationError, match="COLLECTOR_DIGEST_MISMATCH"):
+        materializer.materialize(
+            version=recipe["version"],
+            candidate_id=recipe["candidate_id"],
+            source_commit=recipe["source_commit"],
+        )
+    assert not (output_root / recipe["version"]).exists()
+
+
+def test_materializer_denies_symlinked_evidence(tmp_path: Path) -> None:
+    source_root = tmp_path / "source"
+    source_root.mkdir(mode=0o700)
+    recipe, candidate, _plan = build_valid_source(source_root)
+    manifest = source_root / recipe["version"] / "manifest.json"
+    real = source_root / recipe["version"] / "manifest.real"
+    manifest.rename(real)
+    manifest.symlink_to(real.name)
+    output_root = tmp_path / "builder-evidence"
+    materializer = make_materializer(
+        tmp_path, source_root, output_root, candidate["tree_oid"]
+    )
+
+    with pytest.raises(EvidenceMaterializationError):
+        materializer.materialize(
+            version=recipe["version"],
+            candidate_id=recipe["candidate_id"],
+            source_commit=recipe["source_commit"],
+        )
+    assert not (output_root / recipe["version"]).exists()
+
+
+def test_materializer_rejects_conflicting_existing_destination(tmp_path: Path) -> None:
+    source_root = tmp_path / "source"
+    source_root.mkdir(mode=0o700)
+    recipe, candidate, _plan = build_valid_source(source_root)
+    output_root = tmp_path / "builder-evidence"
+    materializer = make_materializer(
+        tmp_path, source_root, output_root, candidate["tree_oid"]
+    )
+    materializer.materialize(
+        version=recipe["version"],
+        candidate_id=recipe["candidate_id"],
+        source_commit=recipe["source_commit"],
+    )
+    target = output_root / recipe["version"] / "artifact.tar.gz"
+    target.chmod(0o600)
+    target.write_bytes(b"conflict")
+    target.chmod(0o600)
+
+    with pytest.raises(
+        EvidenceMaterializationError, match="EVIDENCE_DESTINATION_CONFLICT"
+    ):
+        materializer.materialize(
+            version=recipe["version"],
+            candidate_id=recipe["candidate_id"],
+            source_commit=recipe["source_commit"],
+        )
+
+
+def test_materializer_does_not_create_receipt_when_source_receipt_is_missing(tmp_path: Path) -> None:
+    source_root = tmp_path / "source"
+    source_root.mkdir(mode=0o700)
+    recipe, candidate, _plan = build_valid_source(source_root)
+    (source_root / recipe["version"] / "receipt.json").unlink()
+    output_root = tmp_path / "builder-evidence"
+    materializer = make_materializer(
+        tmp_path, source_root, output_root, candidate["tree_oid"]
+    )
+
+    with pytest.raises(EvidenceMaterializationError):
+        materializer.materialize(
+            version=recipe["version"],
+            candidate_id=recipe["candidate_id"],
+            source_commit=recipe["source_commit"],
+        )
+    assert not (output_root / recipe["version"] / "receipt.json").exists()

tests/test_release_migration_plan.py

@@ -0,0 +1,501 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import sqlite3
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+for import_root in (ROOT / "active-runtime" / "app", ROOT):
+    if str(import_root) not in sys.path:
+        sys.path.insert(0, str(import_root))
+
+import release_migration_plan as migration_runtime  # noqa: E402
+from release_migration_plan import (  # noqa: E402
+    FileMigrationPlanResolver,
+    HostMigrationCompiler,
+    MigrationPlan,
+    MigrationPlanError,
+    MigrationPlanRequired,
+    MigrationPreSwitchValidator,
+    MigrationRequirement,
+    RuntimeComponentBinding,
+    StateMigrationContract,
+)
+from silent_release_hardening import (  # noqa: E402
+    DeploymentPlan,
+    FinalizationPlan,
+    ReleasePlan,
+    TagPlan,
+)
+
+COMMIT = "1" * 40
+TREE = "2" * 40
+SHA = "3" * 64
+POLICY = "4" * 64
+VERSION = "V2R2C2-baseline11-principal-recovery"
+CANDIDATE = "rc-v2r2c2-baseline11-principal-recovery-111111111111"
+
+
+def migration_plan() -> MigrationPlan:
+    return MigrationPlan(
+        migration_id="migration-test-0000000000000001",
+        version=VERSION,
+        candidate_id=CANDIDATE,
+        repository_identity="aios/aios",
+        source_admission_id="source-admission-test-0000000001",
+        source_commit=COMMIT,
+        tree_oid=TREE,
+        profile_id="baseline11-principal-recovery-cross-code-line-v1",
+        profile_sha256="5" * 64,
+        from_components=(
+            RuntimeComponentBinding(
+                component_id="gitops-model-api-staging",
+                service="aios-gitops-model-api-v2r2c1-staging.service",
+                code_line_id="legacy-governed-main-v15",
+                source_oid="6" * 40,
+                artifact_sha256="7" * 64,
+                interface_sha256="8" * 64,
+            ),
+            RuntimeComponentBinding(
+                component_id="chatgpt-bridge-v2-staging",
+                service="aios-chatgpt-bridge-v2-staging.service",
+                code_line_id="legacy-p0-task-principal",
+                source_oid="6" * 40,
+                artifact_sha256="9" * 64,
+                interface_sha256="a" * 64,
+            ),
+        ),
+        to_components=(
+            RuntimeComponentBinding(
+                component_id="gitops-model-api-staging",
+                service="aios-gitops-model-api-v2r2c1-staging.service",
+                code_line_id="baseline11-release-source-admission",
+                source_oid=COMMIT,
+                artifact_sha256=None,
+                interface_sha256="b" * 64,
+            ),
+            RuntimeComponentBinding(
+                component_id="chatgpt-bridge-v2-staging",
+                service="aios-chatgpt-bridge-v2-staging.service",
+                code_line_id="baseline11-task-admission",
+                source_oid=COMMIT,
+                artifact_sha256=None,
+                interface_sha256="c" * 64,
+            ),
+        ),
+        state_contracts=(
+            StateMigrationContract(
+                state_id="task-admission",
+                state_ref="state:task-admission",
+                mode="REUSE_IN_PLACE_WITH_SNAPSHOT",
+                before_schema_sha256="d" * 64,
+                after_schema_sha256="e" * 64,
+                snapshot_name="snapshots/task-admission.sqlite3",
+                snapshot_sha256="f" * 64,
+                snapshot_required=True,
+                rollback_compatible=True,
+            ),
+        ),
+        activation_schema="aios.task-admission-host-activation.v1",
+        activation_sha256="0" * 64,
+        rollback_window_seconds=3600,
+    )
+
+
+def release_plan(*, migration: MigrationPlan | None = None) -> ReleasePlan:
+    deployment = DeploymentPlan(
+        service="aios-gitops-model-api-v2r2c1-staging.service",
+        release_path="/opt/aios/releases/" + COMMIT,
+        release_root="/opt/aios/releases",
+        cutover_link="/opt/aios/current",
+        dropin_path="/opt/aios/releases/" + COMMIT + "/deployment/production.conf",
+        dropin_sha256="a" * 64,
+        systemd_unit_sha256="b" * 64,
+        rollback_path="/opt/aios/releases/rollback",
+        rollback_sha256="c" * 64,
+        source_commit=COMMIT,
+        tree_oid=TREE,
+        artifact_sha256=SHA,
+        manifest_sha256="6" * 64,
+        package_closure_sha256="7" * 64,
+        authoritative_git_dir="/data/aios-gitops/repo.git",
+        repository_identity="aios/aios",
+        baseline_before_sha256="8" * 64,
+        registry_before_sha256="9" * 64,
+        production_tag_refset_before_sha256="a" * 64,
+        health_contract_version="runtime/v1",
+        timeout_policy={"seconds": 10},
+        retry_policy={"attempts": 1},
+        policy_sha256=POLICY,
+        health_contract={"checked": True},
+    )
+    tag = TagPlan(
+        version=VERSION,
+        tag_type="annotated",
+        expected_tag_oid="b" * 40,
+        expected_target_oid=COMMIT,
+        evidence_closure_oid="c" * 40,
+        approval_plan_id="owner-plan-migration-test",
+        repository_identity="aios/aios",
+        gitea_owner="aios",
+        gitea_repository="aios",
+        current_remote_tag_oid=None,
+        remote_absent_evidence_sha256="d" * 64,
+        verification_contract="migration/v1",
+    )
+    finalization = FinalizationPlan(
+        version=VERSION,
+        expected_tag_oid="b" * 40,
+        expected_target_oid=COMMIT,
+        expected_previous_baseline_sha256="8" * 64,
+        registry_expected_record={"version": VERSION},
+        baseline_expected_record={"version": VERSION},
+        final_consistency_contract="migration/v1",
+    )
+    return ReleasePlan(
+        deployment=deployment,
+        tag=tag,
+        finalization=finalization,
+        policy_sha256=POLICY,
+        migration=migration,
+    )
+
+
+def test_release_plan_legacy_bytes_are_unchanged_without_migration() -> None:
+    plan = release_plan()
+    legacy_payload = {
+        "deployment": plan.deployment.canonical_payload,
+        "finalization": plan.finalization.canonical_payload,
+        "policy_sha256": plan.policy_sha256,
+        "tag": plan.tag.canonical_payload,
+    }
+    expected = hashlib.sha256(
+        json.dumps(legacy_payload, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest()
+    assert "migration" not in plan.canonical_payload
+    assert plan.sha256 == expected
+    assert ReleasePlan.from_dict(plan.to_dict()).sha256 == expected
+
+
+def test_migration_is_part_of_release_plan_hash_and_candidate_binding() -> None:
+    migration = migration_plan()
+    plan = release_plan(migration=migration)
+    assert plan.validate().migration == migration
+    assert ReleasePlan.from_dict(plan.to_dict()).sha256 == plan.sha256
+    assert release_plan().sha256 != plan.sha256
+    migration.assert_release_binding(
+        version=VERSION,
+        candidate_id=CANDIDATE,
+        source_commit=COMMIT,
+        tree_oid=TREE,
+        repository_identity="aios/aios",
+    )
+    with pytest.raises(MigrationPlanError, match="binding"):
+        migration.assert_release_binding(
+            version=VERSION,
+            candidate_id="rc-different",
+            source_commit=COMMIT,
+            tree_oid=TREE,
+            repository_identity="aios/aios",
+        )
+
+
+def test_file_resolver_requires_plan_for_profile_and_rejects_tamper(tmp_path: Path) -> None:
+    profile_root = tmp_path / "profiles"
+    plan_root = tmp_path / "plans"
+    profile_root.mkdir()
+    profile = {
+        "schema": "aios.release-migration-profile.v1",
+        "profile_id": "baseline11-principal-recovery-cross-code-line-v1",
+        "version": VERSION,
+    }
+    profile_path = profile_root / "profile.json"
+    profile_path.write_text(
+        json.dumps(profile, sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+    resolver = FileMigrationPlanResolver(
+        plan_root, profile_root=profile_root, expected_owner_uid=None
+    )
+    with pytest.raises(MigrationPlanRequired, match="MIGRATION_PLAN_REQUIRED"):
+        resolver.resolve(VERSION, CANDIDATE, COMMIT)
+
+    migration = migration_plan()
+    requirement = MigrationRequirement(
+        version=VERSION,
+        candidate_id=CANDIDATE,
+        source_commit=COMMIT,
+        repository_identity="aios/aios",
+        required=True,
+        reason_code="CROSS_CODE_LINE",
+        profile_id=profile["profile_id"],
+        profile_sha256=hashlib.sha256(
+            (json.dumps(profile, sort_keys=True, separators=(",", ":")) + "\n").encode()
+        ).hexdigest(),
+        migration_plan_sha256=migration.sha256,
+    )
+    source_root = plan_root / COMMIT
+    source_root.mkdir(parents=True)
+    (source_root / "requirement.json").write_text(
+        json.dumps(requirement.to_dict(), sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+    (source_root / "plan.json").write_text(
+        json.dumps(migration.to_dict(), sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+    assert resolver.resolve(VERSION, CANDIDATE, COMMIT).sha256 == migration.sha256
+
+    raw = json.loads((source_root / "plan.json").read_text())
+    raw["rollback_window_seconds"] = 12
+    (source_root / "plan.json").write_text(json.dumps(raw), encoding="utf-8")
+    with pytest.raises(MigrationPlanError):
+        resolver.resolve(VERSION, CANDIDATE, COMMIT)
+
+
+def test_state_probe_is_additive_and_persists_pre_migration_snapshot(tmp_path: Path) -> None:
+    live = tmp_path / "legacy.sqlite3"
+    with sqlite3.connect(live) as connection:
+        connection.execute(
+            "CREATE TABLE legacy_task_claims (task_id TEXT PRIMARY KEY, status TEXT NOT NULL)"
+        )
+        connection.execute(
+            "INSERT INTO legacy_task_claims(task_id,status) VALUES('task-1','DONE')"
+        )
+        connection.commit()
+
+    snapshot = tmp_path / "out" / "snapshots" / "task-admission.sqlite3"
+    contract = HostMigrationCompiler._state_compatibility(
+        live,
+        "state:task-admission",
+        snapshot_path=snapshot,
+        snapshot_name="snapshots/task-admission.sqlite3",
+    )
+
+    assert contract.mode == "REUSE_IN_PLACE_WITH_SNAPSHOT"
+    assert contract.snapshot_sha256 == hashlib.sha256(snapshot.read_bytes()).hexdigest()
+    with sqlite3.connect(snapshot) as connection:
+        assert connection.execute(
+            "SELECT status FROM legacy_task_claims WHERE task_id='task-1'"
+        ).fetchone()[0] == "DONE"
+        # The rollback snapshot is pre-migration; target tables are created
+        # only on the private compatibility probe.
+        tables = {
+            row[0]
+            for row in connection.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'"
+            )
+        }
+        assert "task_admission_grants" not in tables
+    assert contract.before_schema_sha256 != contract.after_schema_sha256
+
+
+def test_migration_pre_switch_proof_is_exact_and_fail_closed() -> None:
+    migration = migration_plan()
+    proof = {
+        "migration_plan_sha256": migration.sha256,
+        "migration_id": migration.migration_id,
+        "source_commit": migration.source_commit,
+        "tree_oid": migration.tree_oid,
+        "activation_sha256": migration.activation_sha256,
+        "state_snapshot_sha256s": ["f" * 64],
+        "writer_disabled": True,
+        "task_quiescent": True,
+        "state_snapshot_verified": True,
+        "credentials_ready": True,
+        "rollback_ready": True,
+    }
+    validator = MigrationPreSwitchValidator(lambda _plan: dict(proof))
+    assert validator.validate(migration)["verified"] is True
+    proof["task_quiescent"] = False
+    with pytest.raises(MigrationPlanError, match="PRE_SWITCH"):
+        validator.validate(migration)
+
+
+def test_host_compiler_generates_bound_plan_from_live_read_only_probe(
+    tmp_path: Path, monkeypatch
+) -> None:
+    runtime_a = tmp_path / "runtime-model"
+    runtime_b = tmp_path / "runtime-bridge"
+    for runtime in (runtime_a, runtime_b):
+        (runtime / "app").mkdir(parents=True)
+    (runtime_a / "app" / "model.py").write_text("MODEL=1\n")
+    (runtime_a / "app" / "marker.py").write_text("MARKER=1\n")
+    (runtime_b / "app" / "bridge.py").write_text("BRIDGE=1\n")
+    (runtime_b / "app" / "legacy.py").write_text("LEGACY=1\n")
+
+    state_db = tmp_path / "task-admission.sqlite3"
+    with sqlite3.connect(state_db) as connection:
+        connection.execute(
+            "CREATE TABLE legacy_task_claims (task_id TEXT PRIMARY KEY, status TEXT NOT NULL)"
+        )
+        connection.execute(
+            "INSERT INTO legacy_task_claims VALUES('task-live','DONE')"
+        )
+        connection.commit()
+    activation = tmp_path / "activation.json"
+    activation.write_text(
+        json.dumps(
+            {
+                "schema": "aios.task-admission-host-activation.v1",
+                "state_db": str(state_db),
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    profile = {
+        "schema": "aios.release-migration-profile.v1",
+        "profile_id": "test-cross-code-line-profile-v1",
+        "version": VERSION,
+        "repository": "aios/aios",
+        "current_components": [
+            {
+                "component_id": "model",
+                "service": "model.service",
+                "code_line_id": "legacy-model",
+                "expected_source_oid": "6" * 40,
+                "source_oid_env": "SOURCE",
+                "artifact_sha_env": "ARTIFACT",
+                "interface_files": ["app/model.py"],
+                "marker_files": ["app/marker.py"],
+            },
+            {
+                "component_id": "bridge",
+                "service": "bridge.service",
+                "code_line_id": "legacy-bridge",
+                "expected_source_oid": "6" * 40,
+                "source_oid_env": "SOURCE",
+                "artifact_sha_env": "ARTIFACT",
+                "interface_files": ["app/bridge.py", "app/legacy.py"],
+                "marker_files": ["app/legacy.py"],
+            },
+        ],
+        "target_components": [
+            {
+                "component_id": "model",
+                "service": "model.service",
+                "code_line_id": "baseline-model",
+                "interface_files": ["active-runtime/app/model.py"],
+            },
+            {
+                "component_id": "bridge",
+                "service": "bridge.service",
+                "code_line_id": "baseline-bridge",
+                "interface_files": ["active-runtime/app/bridge.py"],
+            },
+        ],
+        "activation_file": str(activation),
+        "state_ref": "state:task-admission",
+        "state_db": str(state_db),
+        "state_mode": "REUSE_IN_PLACE_WITH_SNAPSHOT",
+        "rollback_window_seconds": 3600,
+    }
+    profile_path = tmp_path / "profile.json"
+    profile_path.write_text(
+        json.dumps(profile, sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+
+    def fake_systemd(service: str):
+        root = runtime_a if service == "model.service" else runtime_b
+        return {
+            "ActiveState": "active",
+            "WorkingDirectory": str(root),
+            "environment": {
+                "SOURCE": "6" * 40,
+                "ARTIFACT": "7" * 64,
+            },
+        }
+
+    monkeypatch.setattr(migration_runtime, "_systemd_properties", fake_systemd)
+    monkeypatch.setattr(
+        migration_runtime,
+        "_hash_git_files",
+        lambda _git, _source, files: hashlib.sha256(
+            "\0".join(files).encode()
+        ).hexdigest(),
+    )
+
+    compiler = HostMigrationCompiler(
+        authority_git_dir=tmp_path / "authority.git",
+        profile_path=profile_path,
+        output_root=tmp_path / "migration-output",
+        expected_owner_uid=None,
+    )
+    requirement, plan = compiler.compile(
+        source_admission={
+            "schema": "aios.release-source-admission.v1",
+            "admission_id": "source-admission-compiler-test-001",
+            "repository": "aios/aios",
+            "source_commit": COMMIT,
+            "tree_oid": TREE,
+            "version": VERSION,
+            "candidate_id": CANDIDATE,
+        },
+        candidate_id=CANDIDATE,
+    )
+
+    assert requirement.required is True
+    assert requirement.migration_plan_sha256 == plan.sha256
+    snapshot = (
+        tmp_path
+        / "migration-output"
+        / COMMIT
+        / plan.state_contracts[0].snapshot_name
+    )
+    assert snapshot.is_file()
+    assert hashlib.sha256(snapshot.read_bytes()).hexdigest() == plan.state_contracts[0].snapshot_sha256
+    # The planner probes a private copy only; live legacy state is untouched.
+    with sqlite3.connect(state_db) as connection:
+        assert connection.execute(
+            "SELECT status FROM legacy_task_claims WHERE task_id='task-live'"
+        ).fetchone()[0] == "DONE"
+        assert connection.execute(
+            "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='task_admission_grants'"
+        ).fetchone()[0] == 0
+
+
+def test_migration_systemd_units_keep_write_gate_and_builder_network_closed() -> None:
+    planner = (
+        ROOT / "deploy" / "systemd" / "aios-release-migration-planner.service"
+    ).read_text(encoding="utf-8")
+    builder = (
+        ROOT / "deploy" / "systemd" / "aios-trusted-candidate-builder.service"
+    ).read_text(encoding="utf-8")
+    assert "AIOS_GITOPS_WRITES_ENABLED=false" in planner
+    assert "LoadCredentialEncrypted=source_admission_hmac:" in planner
+    assert "builder_hmac" not in planner
+    assert "PrivateNetwork=true" in planner
+    assert "ReadOnlyPaths=" in planner and "/data/aios-gitops/repo.git" in planner
+    assert "AIOS_TRUSTED_BUILDER_MIGRATION_PLAN_ROOT=/var/lib/aios-release-migration" in builder
+    assert "AIOS_TRUSTED_BUILDER_MIGRATION_PROFILE_ROOT=/etc/aios-gitops/release-migration-profiles" in builder
+    assert "PrivateNetwork=true" in builder
+
+
+def test_baseline11_migration_profile_binds_observed_old_code_lines() -> None:
+    profile = json.loads(
+        (
+            ROOT
+            / "config"
+            / "gitops"
+            / "release-migration-profiles"
+            / "V2R2C2-baseline11-principal-recovery.json"
+        ).read_text(encoding="utf-8")
+    )
+    assert profile["schema"] == "aios.release-migration-profile.v1"
+    assert profile["version"] == VERSION
+    assert profile["state_mode"] == "REUSE_IN_PLACE_WITH_SNAPSHOT"
+    current = {item["component_id"]: item for item in profile["current_components"]}
+    assert current["gitops-model-api-staging"]["expected_source_oid"] == "91f104ef62a3b688a9259b96e2ca247e0363b564"
+    assert current["chatgpt-bridge-v2-staging"]["expected_source_oid"] == "91f104ef62a3b688a9259b96e2ca247e0363b564"
+    assert "app/aios_task_principal_admission.py" in current["chatgpt-bridge-v2-staging"]["marker_files"]
+    target = {item["component_id"]: item for item in profile["target_components"]}
+    assert "active-runtime/app/aios_task_admission_live_adapters.py" in target["chatgpt-bridge-v2-staging"]["interface_files"]

tests/test_release_source_admission.py

@@ -14,6 +14,9 @@ from release_source_admission import (  # noqa: E402
     ReleaseSourceAdmissionError,
     ReleaseSourceAdmissionStore,
     ReleaseSourceAdmitter,
+    open_source_admission_store,
+    source_admission_policy_sha256,
+    validate_isolated_peer_policy_delta,
 )
 
 
@@ -250,3 +253,135 @@ def test_store_rejects_group_writable_record(tmp_path: Path) -> None:
         match="SOURCE_ADMISSION_PERMISSIONS_INVALID",
     ):
         store.lookup_source(SOURCE)
+
+
+def test_source_admission_policy_binding_is_exact_file_bytes(tmp_path: Path) -> None:
+    policy = tmp_path / "policy.json"
+    policy.write_bytes(b'{\n  "schema": "example"\n}\n')
+    assert source_admission_policy_sha256(policy) == __import__("hashlib").sha256(
+        policy.read_bytes()
+    ).hexdigest()
+
+
+def test_policy_rebind_preserves_admission_identity_and_source_bindings(tmp_path: Path) -> None:
+    policy_before = json.loads(
+        (ROOT / "config" / "gitops" / "trusted-candidate-builder-policy.json").read_text(
+            encoding="utf-8"
+        )
+    )
+    peer = "/system.slice/aios-gitops-model-api-v2r2c2-baseline11-candidate.service"
+    policy_after = json.loads(json.dumps(policy_before))
+    assert peer not in policy_after["allowed_peer_cgroups"]
+    policy_after["allowed_peer_cgroups"].append(peer)
+
+    previous_path = tmp_path / "policy-before.json"
+    policy_path = tmp_path / "policy-after.json"
+    previous_path.write_text(
+        json.dumps(policy_before, sort_keys=True, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    policy_path.write_text(
+        json.dumps(policy_after, sort_keys=True, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    previous_sha, new_sha, direction = validate_isolated_peer_policy_delta(
+        previous_path, policy_path, peer
+    )
+    assert direction == "ADD_ISOLATED_PEER"
+
+    owner = os.geteuid() if hasattr(os, "geteuid") else None
+    old_store = ReleaseSourceAdmissionStore(
+        tmp_path / "admissions",
+        repository="aios/aios",
+        policy_sha256=previous_sha,
+        hmac_key=b"h" * 32,
+        expected_owner_uid=owner,
+        writable=True,
+    )
+    original = ReleaseSourceAdmitter(old_store, FakeBackend()).admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+    new_store = ReleaseSourceAdmissionStore(
+        tmp_path / "admissions",
+        repository="aios/aios",
+        policy_sha256=new_sha,
+        hmac_key=b"h" * 32,
+        expected_owner_uid=owner,
+        writable=True,
+    )
+    rebound = new_store.rebind_policy(
+        SOURCE,
+        previous_policy_sha256=previous_sha,
+        reason="isolated-candidate-bootstrap:ADD_ISOLATED_PEER",
+        expected_admission_id=original["admission_id"],
+    )
+    assert rebound["admission_id"] == original["admission_id"]
+    assert rebound["candidate_id"] == original["candidate_id"]
+    assert rebound["source_commit"] == original["source_commit"]
+    assert rebound["source_ref"] == original["source_ref"]
+    assert rebound["tree_oid"] == original["tree_oid"]
+    assert rebound["policy_sha256"] == new_sha
+    assert new_store.lookup_source(SOURCE)["policy_sha256"] == new_sha
+
+    restore_sha, restored_sha, restore_direction = validate_isolated_peer_policy_delta(
+        policy_path, previous_path, peer
+    )
+    assert restore_direction == "REMOVE_ISOLATED_PEER"
+    assert restore_sha == new_sha
+    assert restored_sha == previous_sha
+    restored_store = ReleaseSourceAdmissionStore(
+        tmp_path / "admissions",
+        repository="aios/aios",
+        policy_sha256=restored_sha,
+        hmac_key=b"h" * 32,
+        expected_owner_uid=owner,
+        writable=True,
+    )
+    restored = restored_store.rebind_policy(
+        SOURCE,
+        previous_policy_sha256=restore_sha,
+        reason="isolated-candidate-bootstrap:REMOVE_ISOLATED_PEER",
+        expected_admission_id=original["admission_id"],
+    )
+    assert restored["admission_id"] == original["admission_id"]
+    assert restored["policy_sha256"] == previous_sha
+    assert len(list((tmp_path / "admissions" / "rebind-history").glob("*.json"))) == 2
+
+
+def test_policy_rebind_rejects_any_non_peer_policy_drift(tmp_path: Path) -> None:
+    before = json.loads(
+        (ROOT / "config" / "gitops" / "trusted-candidate-builder-policy.json").read_text(
+            encoding="utf-8"
+        )
+    )
+    peer = "/system.slice/aios-gitops-model-api-v2r2c2-baseline11-candidate.service"
+    after = json.loads(json.dumps(before))
+    after["allowed_peer_cgroups"].append(peer)
+    after["request_timeout_seconds"] = before["request_timeout_seconds"] + 1
+    before_path = tmp_path / "before.json"
+    after_path = tmp_path / "after.json"
+    before_path.write_text(json.dumps(before) + "\n", encoding="utf-8")
+    after_path.write_text(json.dumps(after) + "\n", encoding="utf-8")
+    with pytest.raises(
+        ReleaseSourceAdmissionError,
+        match="SOURCE_POLICY_REBIND_SCOPE_CHANGED",
+    ):
+        validate_isolated_peer_policy_delta(before_path, after_path, peer)
+
+
+def test_open_source_admission_store_uses_exact_policy_file_digest(tmp_path: Path) -> None:
+    policy = tmp_path / "policy.json"
+    policy.write_text('{"a":1}\n', encoding="utf-8")
+    owner = os.geteuid() if hasattr(os, "geteuid") else None
+    store = open_source_admission_store(
+        tmp_path / "admissions",
+        repository="aios/aios",
+        policy=policy,
+        hmac_key=b"h" * 32,
+        expected_owner_uid=owner,
+        writable=True,
+    )
+    assert store.policy_sha256 == source_admission_policy_sha256(policy)

tests/test_silent_release_execution_readiness.py

@@ -790,5 +790,65 @@ def test_release_intent_surfaces_source_admission_required_as_typed_blocker(tmp_
 
     assert result["status"] == "SOURCE_ADMISSION_REQUIRED"
     assert result["failure_code"] == "SOURCE_ADMISSION_REQUIRED"
-    assert result["retryable"] is False
+    assert result["retryable"] is True
+    assert result["attempt_count"] == 0
     assert result["side_effects_executed"] is False
+
+
+def test_source_admission_prerequisite_replays_same_intent_after_verified_binding(tmp_path):
+    service, _ = trusted_service(tmp_path)
+
+    class AdmissionRequiredBuilder:
+        def build(self, _payload):
+            raise CapabilityAuthorizationError("release source admission required")
+
+    service.candidate_builder = AdmissionRequiredBuilder()
+    first = service.create_release_intent(
+        {"event_id": "webhook-source-admission-replay-001", "source_commit": COMMIT}
+    )
+    assert first["status"] == "SOURCE_ADMISSION_REQUIRED"
+    assert first["attempt_count"] == 0
+
+    service.source_admission_resolver = lambda source_commit: {
+        "schema": "aios.release-source-admission.v1",
+        "admission_id": "source-admission-replay-001",
+        "source_commit": source_commit,
+        "version": "V2R2C4",
+    }
+    second = service.create_release_intent(
+        {"event_id": "webhook-source-admission-replay-001", "source_commit": COMMIT}
+    )
+    assert second["intent_id"] == first["intent_id"]
+    assert second["source_admission_id"] == "source-admission-replay-001"
+    assert second["version"] == "V2R2C4"
+    assert second["status"] == "SOURCE_ADMISSION_REQUIRED"
+    assert second["attempt_count"] == 0
+
+
+def test_migration_plan_required_is_machine_repairable_without_attempt_budget(tmp_path):
+    service, _ = trusted_service(tmp_path)
+
+    class MigrationRequiredBuilder:
+        def build(self, _payload):
+            raise CapabilityAuthorizationError("migration plan required")
+
+    service.candidate_builder = MigrationRequiredBuilder()
+    result = service.create_release_intent(
+        {
+            "event_id": "webhook-migration-plan-required-001",
+            "source_commit": COMMIT,
+        }
+    )
+    assert result["status"] == "MIGRATION_PLAN_REQUIRED"
+    assert result["failure_code"] == "MIGRATION_PLAN_REQUIRED"
+    assert result["retryable"] is True
+    assert result["attempt_count"] == 0
+    replay = service.create_release_intent(
+        {
+            "event_id": "webhook-migration-plan-required-001",
+            "source_commit": COMMIT,
+        }
+    )
+    assert replay["intent_id"] == result["intent_id"]
+    assert replay["status"] == "MIGRATION_PLAN_REQUIRED"
+    assert replay["attempt_count"] == 0

tests/test_trusted_candidate_builder.py

@@ -29,6 +29,12 @@ from aios_trusted_candidate_builder import (  # noqa: E402
     load_builder_policy,
 )
 from silent_release_controller import ReleaseCandidate, ReleaseState  # noqa: E402
+from release_migration_plan import (  # noqa: E402
+    MigrationPlan,
+    MigrationPlanRequired,
+    RuntimeComponentBinding,
+    StateMigrationContract,
+)
 from silent_release_hardening import (  # noqa: E402
     BuilderAttestation,
     BuilderAttestationKeyRecord,
@@ -313,6 +319,45 @@ def test_readiness_fails_closed_when_signer_is_unavailable() -> None:
         server.handle_request(request, connection=object())
 
 
+def test_readiness_is_independent_from_recipe_evidence_but_build_remains_fail_closed(tmp_path) -> None:
+    policy = load_builder_policy(POLICY_PATH)
+    config = TrustedCandidateBuilderConfig.from_policy(policy)
+    evidence_root = tmp_path / "evidence"
+    evidence_root.mkdir(mode=0o700)
+    authority_git = tmp_path / "repo.git"
+    authority_git.mkdir(mode=0o700)
+
+    source = ServerOwnedEvidenceSource(
+        config,
+        authority_git,
+        evidence_root,
+        expected_owner_uid=None,
+        git_runner=lambda *_args, **_kwargs: SimpleNamespace(
+            stdout="a" * 40 + "\n", returncode=0
+        ),
+    )
+    collector = DeterministicEvidenceCollector(config, evidence_source=source)
+    server, client, _ = make_seam(collector=collector)
+
+    readiness = client.create_request(
+        "V2R2C2",
+        "rc-v2r2c2-d2b5a8339a34",
+        "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6",
+        nonce="v" * 32,
+        operation=server.config.readiness_operation,
+    )
+    assert server.handle_request(readiness, connection=object())["status"] == "READY"
+
+    build = client.create_request(
+        "V2R2C2",
+        "rc-v2r2c2-d2b5a8339a34",
+        "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6",
+        nonce="w" * 32,
+    )
+    with pytest.raises(BuilderError, match="COLLECTOR_EVIDENCE_UNAVAILABLE"):
+        server.handle_request(build, connection=object())
+
+
 def test_unix_client_applies_bounded_timeout_and_fails_closed(monkeypatch) -> None:
     observed = {}
 
@@ -668,3 +713,99 @@ def test_governed_source_admission_rejects_caller_candidate_rebinding() -> None:
     )
     with pytest.raises(BuilderError, match="REQUEST_RELEASE_BINDING_INVALID"):
         config.resolve_recipe("V2R2C4", "rc-attacker", "2" * 40)
+
+
+def _migration_for_fixed_recipe(config: TrustedCandidateBuilderConfig) -> MigrationPlan:
+    recipe = config.fixed_recipe("V2R2C2")
+    digest = lambda label: hashlib.sha256(
+        f"{label}\0{recipe['base_oid']}".encode()
+    ).hexdigest()
+    return MigrationPlan(
+        migration_id="migration-builder-test-000000001",
+        version="V2R2C2",
+        candidate_id=recipe["candidate_id"],
+        repository_identity="aios/aios",
+        source_admission_id="source-admission-builder-test-001",
+        source_commit=recipe["source_commit"],
+        tree_oid=digest("tree")[:40],
+        profile_id="migration-profile-builder-test-v1",
+        profile_sha256="1" * 64,
+        from_components=(
+            RuntimeComponentBinding(
+                component_id="gitops-model-api-staging",
+                service="aios-gitops-model-api-v2r2c1-staging.service",
+                code_line_id="legacy-v1",
+                source_oid="2" * 40,
+                artifact_sha256="3" * 64,
+                interface_sha256="4" * 64,
+            ),
+        ),
+        to_components=(
+            RuntimeComponentBinding(
+                component_id="gitops-model-api-staging",
+                service="aios-gitops-model-api-v2r2c1-staging.service",
+                code_line_id="baseline11-v1",
+                source_oid=recipe["source_commit"],
+                artifact_sha256=None,
+                interface_sha256="5" * 64,
+            ),
+        ),
+        state_contracts=(
+            StateMigrationContract(
+                state_id="task-admission",
+                state_ref="state:task-admission",
+                mode="REUSE_IN_PLACE_WITH_SNAPSHOT",
+                before_schema_sha256="6" * 64,
+                after_schema_sha256="7" * 64,
+                snapshot_name="snapshots/task-admission.sqlite3",
+                snapshot_sha256="8" * 64,
+                snapshot_required=True,
+                rollback_compatible=True,
+            ),
+        ),
+        activation_schema="aios.task-admission-host-activation.v1",
+        activation_sha256="9" * 64,
+        rollback_window_seconds=3600,
+    )
+
+
+def test_collector_injects_root_owned_migration_plan_into_release_plan():
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
+    migration = _migration_for_fixed_recipe(config)
+    config = dataclasses.replace(
+        config,
+        migration_plan_resolver=lambda version, candidate_id, source_commit: migration,
+    )
+    recipe = config.fixed_recipe("V2R2C2")
+    def migration_compatible_bundle(recipe_name, recipe_value):
+        bundle = dict(passing_bundle(recipe_name, recipe_value))
+        release = dict(bundle["release_plan"])
+        tag = dict(release["tag"])
+        finalization = dict(release["finalization"])
+        tag["version"] = "V2R2C2"
+        finalization["version"] = "V2R2C2"
+        release["tag"] = tag
+        release["finalization"] = finalization
+        bundle["release_plan"] = release
+        return bundle
+
+    result = DeterministicEvidenceCollector(
+        config, evidence_source=migration_compatible_bundle
+    ).collect("V2R2C2", recipe)
+    assert result["release_plan"]["migration"]["sha256"] == migration.sha256
+    assert ReleasePlan.from_dict(result["release_plan"]).migration.sha256 == migration.sha256
+
+
+def test_collector_surfaces_missing_migration_plan_as_typed_builder_blocker():
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
+    config = dataclasses.replace(
+        config,
+        migration_plan_resolver=lambda *_args: (_ for _ in ()).throw(
+            MigrationPlanRequired("MIGRATION_PLAN_REQUIRED")
+        ),
+    )
+    recipe = config.fixed_recipe("V2R2C2")
+    with pytest.raises(BuilderError, match="MIGRATION_PLAN_REQUIRED"):
+        DeterministicEvidenceCollector(
+            config, evidence_source=passing_bundle
+        ).collect("V2R2C2", recipe)