feat(governed): Release Migration Plan (AIOS-BASELINE-1.1-RELEASE-MIGRATION-PLAN)

Make governed release source closure explicit so a pushed governed candidate
can enter the authoritative Git/release closure without manual runtime
overwrite or fixed-recipe bypass.

- Gitea machine transport: add _safe_branch() and exact OID/branch verification.
- release_source_admission.py (new): admit a governed source by verifying remote
  branch -> exact OID, fetching only to refs/aios/release-sources/<OID>,
  validating authoritative main ancestry + tree OID, and writing a policy-bound,
  HMAC-authenticated immutable admission record (source_admission_hmac).
- trusted candidate builder: read-only admission verifier, remains
  PrivateNetwork / read-only; missing admission surfaces SOURCE_ADMISSION_REQUIRED.
- Model API / release saga: consume admission records and return machine-actionable
  SOURCE_ADMISSION_REQUIRED (not NOT_FOUND) when a non-fixed source is unadmitted.
- systemd: add aios-release-source-admission.service (oneshot, root/gitops,
  AIOS_GITOPS_WRITES_ENABLED=false, independent source_admission_hmac credential);
  wire source_admission_hmac into trusted builder; update staging unit.
- Fixed V2R2C2/V2R2C3 recipes are not overridden by admission.
- Tests: new release_source_admission + adjusted builder/model-api/live-executor/
  readiness tests.

Verified: 13-file after_sha256 byte-exact; py_compile PASS;
Linux 326 targeted passed / 0 failed; acceptance 165; regression 1401;
systemd-analyze verify 3/3 PASS.

Base: a320ea3 (agent/baseline-1.1-principal-recovery-20260908).

active-runtime/app/aios_gitea_machine_transport.py

@@ -24,6 +24,25 @@ def _safe_oid(value: str, field: str) -> str:
     return value
 
 
+def _safe_branch(value: str) -> str:
+    branch = str(value or "")
+    if (
+        not branch
+        or len(branch) > 240
+        or branch != branch.strip()
+        or branch.startswith(("/", "-", "."))
+        or branch.endswith(("/", ".", ".lock"))
+        or ".." in branch
+        or "//" in branch
+        or "@{" in branch
+        or "\\" in branch
+        or any(part in {"", ".", ".."} for part in branch.split("/"))
+        or any(ch not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._/-" for ch in branch)
+    ):
+        raise GiteaTransportError("BRANCH_INVALID")
+    return branch
+
+
 @dataclass(slots=True)
 class GiteaMachineTransport:
     api_base: str
@@ -91,6 +110,7 @@ class GiteaMachineTransport:
         return self.repository_info()
 
     def read_branch(self, branch: str) -> str | None:
+        branch = _safe_branch(branch)
         owner, name = self.repository_id.split("/", 1)
         try:
             value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")[1]
@@ -136,6 +156,99 @@ class GiteaMachineTransport:
         finally:
             askpass.unlink(missing_ok=True)
 
+    def fetch_branch_exact(self, branch: str, expected_oid: str) -> str:
+        """Fetch one exact remote branch tip into a quarantine ref.
+
+        This never updates ``main`` or any caller-visible branch.  The remote
+        ref is read before and after the fetch so a concurrent branch move
+        cannot be silently admitted.
+        """
+
+        branch = _safe_branch(branch)
+        expected_oid = _safe_oid(expected_oid, "expected_oid")
+        remote_before = self.read_branch(branch)
+        if remote_before != expected_oid:
+            raise GiteaTransportError("GITEA_SOURCE_REF_MISMATCH")
+        target_ref = f"refs/aios/release-sources/{expected_oid}"
+
+        current = subprocess.run(
+            [
+                "git",
+                f"--git-dir={self.authority_git_dir}",
+                "rev-parse",
+                "--verify",
+                f"{target_ref}^{{commit}}",
+            ],
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=self.timeout,
+        )
+        if current.returncode == 0 and current.stdout.strip().lower() == expected_oid:
+            remote_after = self.read_branch(branch)
+            if remote_after != expected_oid:
+                raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
+            return expected_oid
+
+        askpass, additions = self._askpass()
+        environment = os.environ.copy()
+        environment.update(additions)
+        try:
+            result = subprocess.run(
+                [
+                    "git",
+                    f"--git-dir={self.authority_git_dir}",
+                    "fetch",
+                    "--no-tags",
+                    "--no-write-fetch-head",
+                    self.remote_url,
+                    f"refs/heads/{branch}:{target_ref}",
+                ],
+                text=True,
+                capture_output=True,
+                check=False,
+                env=environment,
+                timeout=self.timeout,
+            )
+            if result.returncode != 0:
+                raise GiteaTransportError("GITEA_SOURCE_FETCH_FAILED")
+        finally:
+            askpass.unlink(missing_ok=True)
+
+        local = subprocess.run(
+            [
+                "git",
+                f"--git-dir={self.authority_git_dir}",
+                "rev-parse",
+                "--verify",
+                f"{target_ref}^{{commit}}",
+            ],
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=self.timeout,
+        )
+        local_oid = local.stdout.strip().lower()
+        if local.returncode != 0 or local_oid != expected_oid:
+            subprocess.run(
+                [
+                    "git",
+                    f"--git-dir={self.authority_git_dir}",
+                    "update-ref",
+                    "-d",
+                    target_ref,
+                ],
+                text=True,
+                capture_output=True,
+                check=False,
+                timeout=self.timeout,
+            )
+            raise GiteaTransportError("GITEA_SOURCE_FETCH_MISMATCH")
+        remote_after = self.read_branch(branch)
+        if remote_after != expected_oid:
+            raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
+        return expected_oid
+
     def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str:
         _safe_oid(expected_oid, "expected_oid")
         _safe_oid(new_oid, "new_oid")

active-runtime/app/aios_model_api_staging.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, urllib.request, uuid
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from dataclasses import replace
 from pathlib import Path
 from baseline_authority import BaselineAuthority, BaselineAuthorityError
 from silent_release_controller import ReleaseCandidate
@@ -114,6 +115,9 @@ TRUSTED_BUILDER_SOCKET = Path(os.getenv(
 TRUSTED_BUILDER_PUBLIC_KEYRING = Path(os.getenv(
     'AIOS_TRUSTED_CANDIDATE_BUILDER_PUBLIC_KEYRING', ''
 ))
+TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT = str(os.getenv(
+    'AIOS_TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT', ''
+)).strip()
 TRUSTED_RELEASE_POLICY = Path(os.getenv(
     'AIOS_SILENT_RELEASE_TRUSTED_POLICY',
     str(WS / 'config' / 'gitops' / 'silent-release-policy.json'),
@@ -253,6 +257,22 @@ class _HostTrustedCandidateBuilderAdapter:
                 nonce=secrets.token_hex(32),
             )
         except BuilderError as exc:
+            code = str(exc)
+            if code in {'SOURCE_ADMISSION_REQUIRED', 'RECIPE_UNKNOWN'} or (
+                code == 'REQUEST_RELEASE_BINDING_INVALID'
+                and not intent.get('version')
+            ):
+                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
@@ -290,6 +310,7 @@ def _build_silent_release_service() -> SilentReleaseActionService:
     builder_attestation_verifier = None
     trusted_policy = None
     builder_readiness_probe = None
+    source_admission_resolver = None
     if TRUSTED_BUILDER_ENABLED:
         # Every path is opt-in.  In particular, this API receives its own
         # client HMAC credential, never the builder's signing private key.
@@ -317,6 +338,31 @@ def _build_silent_release_service() -> SilentReleaseActionService:
             client_hmac = (
                 TRUSTED_BUILDER_CREDENTIALS_DIRECTORY / 'builder_hmac'
             ).read_bytes().strip()
+            if TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT:
+                from release_source_admission import ReleaseSourceAdmissionStore
+
+                try:
+                    source_admission_hmac = (
+                        TRUSTED_BUILDER_CREDENTIALS_DIRECTORY
+                        / 'source_admission_hmac'
+                    ).read_bytes().strip()
+                except OSError as exc:
+                    raise CapabilityAuthorizationError(
+                        'release source admission credential is unavailable'
+                    ) from exc
+                source_store = ReleaseSourceAdmissionStore(
+                    Path(TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT),
+                    repository=builder_config.repository,
+                    policy_sha256=builder_config.policy_sha256,
+                    hmac_key=source_admission_hmac,
+                    expected_owner_uid=0,
+                    writable=False,
+                )
+                builder_config = replace(
+                    builder_config,
+                    source_admission_resolver=source_store.resolve_recipe,
+                )
+                source_admission_resolver = source_store.lookup_source
             client_keyring = BuilderKeyring(builder_hmac=client_hmac)
             client = UnixTrustedCandidateBuilderClient(
                 builder_config,
@@ -365,6 +411,7 @@ def _build_silent_release_service() -> SilentReleaseActionService:
         builder_attestation_verifier=builder_attestation_verifier,
         trusted_policy=trusted_policy,
         builder_readiness_probe=builder_readiness_probe,
+        source_admission_resolver=source_admission_resolver,
         allow_legacy_candidate_input=False,
         expected_repository=REPOSITORY_IDENTITY,
         policy_sha256=policy_sha,

active-runtime/app/aios_trusted_candidate_builder.py

@@ -15,7 +15,7 @@ import struct
 import subprocess
 import time
 from datetime import datetime, timezone
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any, Callable, Mapping, Protocol
 
@@ -133,6 +133,8 @@ class TrustedCandidateBuilderConfig:
     recipes: Mapping[str, Mapping[str, str]]
     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":
@@ -227,14 +229,63 @@ class TrustedCandidateBuilderConfig:
         return recipe
 
     def resolve_recipe(self, version: str, candidate_id: str | None, source_commit: str) -> Mapping[str, str]:
-        recipe = self.fixed_recipe(version)
-        # Intent-only callers do not choose a candidate identity.  The
-        # policy-owned recipe supplies it; explicit worker requests are
-        # still checked against the same immutable recipe.
-        effective_candidate_id = str(candidate_id or recipe.get("candidate_id") or "")
-        if {"version": recipe.get("version"), "candidate_id": recipe.get("candidate_id"), "source_commit": recipe.get("source_commit")} != {"version": version, "candidate_id": effective_candidate_id, "source_commit": source_commit}:
+        recipe = self.recipes.get(version)
+        if recipe is not None:
+            resolved = self.fixed_recipe(version)
+            # Fixed/governed-derived policy recipes remain immutable.  A
+            # source admission can never override one of these identities.
+            effective_candidate_id = str(candidate_id or resolved.get("candidate_id") or "")
+            if {
+                "version": resolved.get("version"),
+                "candidate_id": resolved.get("candidate_id"),
+                "source_commit": resolved.get("source_commit"),
+            } != {
+                "version": version,
+                "candidate_id": effective_candidate_id,
+                "source_commit": source_commit,
+            }:
+                raise BuilderError("REQUEST_RELEASE_BINDING_INVALID")
+            return resolved
+
+        resolver = self.source_admission_resolver
+        if resolver is None:
+            raise BuilderError("RECIPE_UNKNOWN")
+        try:
+            admitted = dict(resolver(version, candidate_id, source_commit))
+        except Exception as exc:
+            raise BuilderError("SOURCE_ADMISSION_REQUIRED") from exc
+        required = {
+            "mode",
+            "task_id",
+            "candidate_id",
+            "base_oid",
+            "source_commit",
+            "version",
+            "purpose",
+        }
+        if (
+            set(admitted) != required
+            or admitted.get("mode") != "governed-source-admission"
+            or admitted.get("version") != version
+            or admitted.get("source_commit") != source_commit
+            or not isinstance(admitted.get("candidate_id"), str)
+            or not admitted.get("candidate_id")
+        ):
+            raise BuilderError("SOURCE_ADMISSION_INVALID")
+        effective_candidate_id = str(candidate_id or admitted["candidate_id"])
+        if effective_candidate_id != admitted["candidate_id"]:
             raise BuilderError("REQUEST_RELEASE_BINDING_INVALID")
-        return recipe
+        for name in ("base_oid", "source_commit"):
+            value = admitted.get(name)
+            if (
+                not isinstance(value, str)
+                or len(value) != 40
+                or any(char not in "0123456789abcdef" for char in value.lower())
+            ):
+                raise BuilderError("SOURCE_ADMISSION_INVALID")
+        if admitted.get("purpose") != "AIOS_TRUSTED_RELEASE_BUILDER":
+            raise BuilderError("SOURCE_ADMISSION_INVALID")
+        return admitted
 
 
 def _invalid_governed_recipes() -> Mapping[str, Mapping[str, str]]:
@@ -464,11 +515,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"]:
@@ -776,6 +848,50 @@ def _main() -> None:
     if not credentials:
         raise SystemExit("CREDENTIALS_DIRECTORY is required")
     keyring = BuilderKeyring.from_credentials_directory(Path(credentials))
+    source_admission_root = os.environ.get(
+        "AIOS_TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT", ""
+    ).strip()
+    if source_admission_root:
+        from release_source_admission import ReleaseSourceAdmissionStore
+
+        try:
+            source_admission_hmac = (
+                Path(credentials) / "source_admission_hmac"
+            ).read_bytes().strip()
+        except OSError as exc:
+            raise BuilderError("SOURCE_ADMISSION_HMAC_UNAVAILABLE") from exc
+        source_store = ReleaseSourceAdmissionStore(
+            Path(source_admission_root),
+            repository=config.repository,
+            policy_sha256=config.policy_sha256,
+            hmac_key=source_admission_hmac,
+            expected_owner_uid=0,
+            writable=False,
+        )
+        config = replace(
+            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_migration_plan.py

@@ -0,0 +1,997 @@
+#!/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 _policy_sha256(path: Path) -> str:
+    try:
+        raw = path.read_bytes()
+    except OSError as exc:
+        raise MigrationPlanError("builder_policy_unavailable") from exc
+    return hashlib.sha256(raw).hexdigest()
+
+
+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 ReleaseSourceAdmissionStore
+
+    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 = ReleaseSourceAdmissionStore(
+            source_admission_root,
+            repository="aios/aios",
+            policy_sha256=_policy_sha256(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

@@ -0,0 +1,561 @@
+#!/usr/bin/env python3
+"""Governed admission of an exact remote Git commit into the release source closure.
+
+The admitter is intentionally separate from the trusted candidate builder:
+it may use a narrowly-scoped Gitea read/fetch credential, while the builder
+remains PrivateNetwork=true and reads only the authoritative Git object store
+plus immutable admission records.
+
+An admission never moves ``main`` and never treats an arbitrary caller-supplied
+OID as trusted.  It binds:
+  remote branch ref -> exact full commit OID -> authoritative Git object/tree
+  -> current authoritative main ancestry -> immutable release recipe metadata.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+import re
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping, Protocol
+
+from authoritative_git import (
+    AuthoritativeGitError,
+    authoritative_head_oid,
+    authoritative_lineage,
+    run_authoritative_git,
+)
+from aios_gitea_machine_transport import GiteaMachineTransport, GiteaTransportError
+from release_version import validate_release_version
+
+
+SCHEMA = "aios.release-source-admission.v1"
+RECIPE_MODE = "governed-source-admission"
+_OID_RE = re.compile(r"^[0-9a-f]{40}$")
+_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
+_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,199}$")
+_BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,239}$")
+_REF_PREFIX = "refs/heads/"
+
+
+class ReleaseSourceAdmissionError(RuntimeError):
+    """A stable fail-closed source-admission error code."""
+
+
+def _canonical(value: Mapping[str, Any]) -> bytes:
+    return json.dumps(
+        dict(value),
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+    ).encode("utf-8")
+
+
+def _oid(value: object, field: str) -> str:
+    text = str(value or "").strip().lower()
+    if not _OID_RE.fullmatch(text):
+        raise ReleaseSourceAdmissionError(f"{field.upper()}_INVALID")
+    return text
+
+
+def _digest(value: object, field: str) -> str:
+    text = str(value or "").strip().lower()
+    if not _DIGEST_RE.fullmatch(text):
+        raise ReleaseSourceAdmissionError(f"{field.upper()}_INVALID")
+    return text
+
+
+def _identifier(value: object, field: str) -> str:
+    text = str(value or "").strip()
+    if not _IDENTIFIER_RE.fullmatch(text) or ".." in text or "\x00" in text:
+        raise ReleaseSourceAdmissionError(f"{field.upper()}_INVALID")
+    return text
+
+
+def _source_ref(value: object) -> str:
+    text = str(value or "").strip()
+    if not text.startswith(_REF_PREFIX):
+        raise ReleaseSourceAdmissionError("SOURCE_REF_INVALID")
+    branch = text[len(_REF_PREFIX):]
+    if (
+        not _BRANCH_RE.fullmatch(branch)
+        or ".." in branch
+        or "//" in branch
+        or "@{" in branch
+        or branch.endswith(("/", ".", ".lock"))
+        or any(part in {"", ".", ".."} for part in branch.split("/"))
+    ):
+        raise ReleaseSourceAdmissionError("SOURCE_REF_INVALID")
+    return text
+
+
+def _version(value: object) -> str:
+    try:
+        return validate_release_version(value)
+    except (TypeError, ValueError) as exc:
+        raise ReleaseSourceAdmissionError("VERSION_INVALID") from exc
+
+
+def _version_slug(version: str) -> str:
+    value = re.sub(r"[^A-Za-z0-9]+", "-", version.lower()).strip("-")
+    if not value:
+        raise ReleaseSourceAdmissionError("VERSION_INVALID")
+    return value[:80]
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _secure_stat(path: Path, *, expected_owner_uid: int | None, directory: bool) -> None:
+    try:
+        info = path.lstat()
+    except OSError as exc:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_UNAVAILABLE") from exc
+    if path.is_symlink():
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_UNSAFE")
+    if directory and not path.is_dir():
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_UNSAFE")
+    if not directory and not path.is_file():
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_UNSAFE")
+    if expected_owner_uid is not None and info.st_uid != expected_owner_uid:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_OWNER_MISMATCH")
+    # Group/other write would allow an untrusted process to forge release input.
+    if info.st_mode & 0o022:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_PERMISSIONS_INVALID")
+
+
+def _validate_record(
+    value: Mapping[str, Any],
+    *,
+    repository: str,
+    policy_sha256: str,
+    hmac_key: bytes,
+) -> dict[str, Any]:
+    if not isinstance(value, Mapping):
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_RECORD_INVALID")
+    required = {
+        "schema",
+        "admission_id",
+        "repository",
+        "task_id",
+        "version",
+        "candidate_id",
+        "base_oid",
+        "source_ref",
+        "source_commit",
+        "remote_ref_oid",
+        "tree_oid",
+        "purpose",
+        "policy_sha256",
+        "admitted_at",
+        "mac",
+    }
+    if set(value) != required or value.get("schema") != SCHEMA:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_RECORD_INVALID")
+    record = dict(value)
+    if _identifier(record["repository"], "repository") != repository:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_REPOSITORY_MISMATCH")
+    _identifier(record["task_id"], "task_id")
+    record["version"] = _version(record["version"])
+    _identifier(record["candidate_id"], "candidate_id")
+    _identifier(record["purpose"], "purpose")
+    _identifier(record["admission_id"], "admission_id")
+    record["base_oid"] = _oid(record["base_oid"], "base_oid")
+    record["source_commit"] = _oid(record["source_commit"], "source_commit")
+    record["remote_ref_oid"] = _oid(record["remote_ref_oid"], "remote_ref_oid")
+    record["tree_oid"] = _oid(record["tree_oid"], "tree_oid")
+    record["source_ref"] = _source_ref(record["source_ref"])
+    record["policy_sha256"] = _digest(record["policy_sha256"], "policy_sha256")
+    if record["policy_sha256"] != policy_sha256:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_POLICY_MISMATCH")
+    if record["remote_ref_oid"] != record["source_commit"]:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_REMOTE_MISMATCH")
+    admitted_at = record.get("admitted_at")
+    if not isinstance(admitted_at, str) or not admitted_at.endswith("Z") or len(admitted_at) < 20:
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_TIME_INVALID")
+    supplied_mac = record.get("mac")
+    if (
+        not isinstance(supplied_mac, str)
+        or not re.fullmatch(r"[0-9a-f]{64}", supplied_mac)
+    ):
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_MAC_INVALID")
+    unsigned = {key: value for key, value in record.items() if key != "mac"}
+    expected_mac = hmac.new(hmac_key, _canonical(unsigned), hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(supplied_mac, expected_mac):
+        raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_MAC_INVALID")
+    return record
+
+
+class ReleaseSourceBackend(Protocol):
+    def authoritative_main_oid(self) -> str: ...
+    def remote_ref_oid(self, source_ref: str) -> str: ...
+    def fetch_exact_ref(self, source_ref: str, source_commit: str) -> str: ...
+    def tree_oid(self, source_commit: str) -> str: ...
+    def require_lineage(self, base_oid: str, source_commit: str) -> None: ...
+
+
+class ReleaseSourceAdmissionStore:
+    """Durable immutable admission records keyed by exact source commit."""
+
+    def __init__(
+        self,
+        root: str | os.PathLike[str],
+        *,
+        repository: str,
+        policy_sha256: str,
+        hmac_key: bytes,
+        expected_owner_uid: int | None = None,
+        writable: bool = False,
+    ) -> None:
+        self.root = Path(root)
+        self.repository = _identifier(repository, "repository")
+        self.policy_sha256 = _digest(policy_sha256, "policy_sha256")
+        if not isinstance(hmac_key, (bytes, bytearray)) or len(hmac_key) < 32:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_HMAC_INVALID")
+        self.hmac_key = bytes(hmac_key)
+        self.expected_owner_uid = expected_owner_uid
+        self.writable = bool(writable)
+        if self.writable and not self.root.exists():
+            self.root.mkdir(parents=True, mode=0o750)
+        if self.root.exists():
+            _secure_stat(
+                self.root,
+                expected_owner_uid=self.expected_owner_uid,
+                directory=True,
+            )
+
+    def _path(self, source_commit: str) -> Path:
+        return self.root / f"{_oid(source_commit, 'source_commit')}.json"
+
+    def register(self, record: Mapping[str, Any]) -> dict[str, Any]:
+        if not self.writable:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_READ_ONLY")
+        supplied = dict(record)
+        if "mac" not in supplied:
+            supplied["mac"] = hmac.new(
+                self.hmac_key,
+                _canonical(supplied),
+                hashlib.sha256,
+            ).hexdigest()
+        checked = _validate_record(
+            supplied,
+            repository=self.repository,
+            policy_sha256=self.policy_sha256,
+            hmac_key=self.hmac_key,
+        )
+        path = self._path(checked["source_commit"])
+        encoded = _canonical(checked) + b"\n"
+        if path.exists():
+            _secure_stat(
+                path,
+                expected_owner_uid=self.expected_owner_uid,
+                directory=False,
+            )
+            try:
+                existing = path.read_bytes()
+            except OSError as exc:
+                raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_UNAVAILABLE") from exc
+            if existing == encoded:
+                return dict(checked)
+            try:
+                previous = _validate_record(
+                    json.loads(existing),
+                    repository=self.repository,
+                    policy_sha256=self.policy_sha256,
+                    hmac_key=self.hmac_key,
+                )
+            except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+                raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_RECORD_INVALID") from exc
+            comparable_previous = {
+                key: value
+                for key, value in previous.items()
+                if key not in {"admitted_at", "mac"}
+            }
+            comparable_checked = {
+                key: value
+                for key, value in checked.items()
+                if key not in {"admitted_at", "mac"}
+            }
+            if comparable_previous == comparable_checked:
+                return dict(previous)
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_CONFLICT")
+        self.root.mkdir(parents=True, exist_ok=True, mode=0o750)
+        _secure_stat(
+            self.root,
+            expected_owner_uid=self.expected_owner_uid,
+            directory=True,
+        )
+        fd, raw_name = tempfile.mkstemp(
+            prefix=f".{checked['source_commit']}.",
+            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())
+            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
+        _secure_stat(
+            path,
+            expected_owner_uid=self.expected_owner_uid,
+            directory=False,
+        )
+        return dict(checked)
+
+    def lookup_source(self, source_commit: str) -> dict[str, Any] | None:
+        path = self._path(source_commit)
+        if not path.exists():
+            return None
+        _secure_stat(
+            path,
+            expected_owner_uid=self.expected_owner_uid,
+            directory=False,
+        )
+        try:
+            raw = path.read_bytes()
+            if len(raw) > 64 * 1024:
+                raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_RECORD_INVALID")
+            decoded = json.loads(raw)
+        except ReleaseSourceAdmissionError:
+            raise
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_RECORD_INVALID") from exc
+        return _validate_record(
+            decoded,
+            repository=self.repository,
+            policy_sha256=self.policy_sha256,
+            hmac_key=self.hmac_key,
+        )
+
+    def resolve_recipe(
+        self,
+        version: str,
+        candidate_id: str | None,
+        source_commit: str,
+    ) -> dict[str, str]:
+        record = self.lookup_source(source_commit)
+        if record is None:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_NOT_FOUND")
+        if record["version"] != version:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_VERSION_MISMATCH")
+        supplied_candidate = str(candidate_id or "")
+        if supplied_candidate and supplied_candidate != record["candidate_id"]:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_CANDIDATE_MISMATCH")
+        return {
+            "mode": RECIPE_MODE,
+            "task_id": record["task_id"],
+            "candidate_id": record["candidate_id"],
+            "base_oid": record["base_oid"],
+            "source_commit": record["source_commit"],
+            "version": record["version"],
+            "purpose": record["purpose"],
+        }
+
+
+@dataclass(slots=True)
+class GiteaReleaseSourceBackend:
+    transport: GiteaMachineTransport
+    git_dir: Path
+
+    def authoritative_main_oid(self) -> str:
+        return authoritative_head_oid(git_dir=self.git_dir)
+
+    def remote_ref_oid(self, source_ref: str) -> str:
+        ref = _source_ref(source_ref)
+        branch = ref[len(_REF_PREFIX):]
+        value = self.transport.read_branch(branch)
+        if value is None:
+            raise ReleaseSourceAdmissionError("SOURCE_REF_NOT_FOUND")
+        return _oid(value, "remote_ref_oid")
+
+    def fetch_exact_ref(self, source_ref: str, source_commit: str) -> str:
+        ref = _source_ref(source_ref)
+        branch = ref[len(_REF_PREFIX):]
+        try:
+            value = self.transport.fetch_branch_exact(
+                branch,
+                _oid(source_commit, "source_commit"),
+            )
+        except GiteaTransportError as exc:
+            raise ReleaseSourceAdmissionError("SOURCE_FETCH_FAILED") from exc
+        return _oid(value, "fetched_source_commit")
+
+    def tree_oid(self, source_commit: str) -> str:
+        source = _oid(source_commit, "source_commit")
+        try:
+            result = run_authoritative_git(
+                ["rev-parse", "--verify", f"{source}^{{tree}}"],
+                git_dir=self.git_dir,
+            )
+        except AuthoritativeGitError as exc:
+            raise ReleaseSourceAdmissionError("SOURCE_COMMIT_NOT_FOUND") from exc
+        return _oid(result.stdout.strip().lower(), "tree_oid")
+
+    def require_lineage(self, base_oid: str, source_commit: str) -> None:
+        try:
+            authoritative_lineage(
+                _oid(base_oid, "base_oid"),
+                _oid(source_commit, "source_commit"),
+                git_dir=self.git_dir,
+            )
+        except AuthoritativeGitError as exc:
+            raise ReleaseSourceAdmissionError("SOURCE_LINEAGE_MISMATCH") from exc
+
+
+class ReleaseSourceAdmitter:
+    def __init__(
+        self,
+        store: ReleaseSourceAdmissionStore,
+        backend: ReleaseSourceBackend,
+        *,
+        purpose: str = "AIOS_TRUSTED_RELEASE_BUILDER",
+    ) -> None:
+        if not store.writable:
+            raise ReleaseSourceAdmissionError("SOURCE_ADMISSION_STORE_READ_ONLY")
+        self.store = store
+        self.backend = backend
+        self.purpose = _identifier(purpose, "purpose")
+
+    def admit(
+        self,
+        *,
+        version: str,
+        task_id: str,
+        source_ref: str,
+        source_commit: str,
+    ) -> dict[str, Any]:
+        version_value = _version(version)
+        task_value = _identifier(task_id, "task_id")
+        ref_value = _source_ref(source_ref)
+        source_value = _oid(source_commit, "source_commit")
+
+        remote_before = self.backend.remote_ref_oid(ref_value)
+        if remote_before != source_value:
+            raise ReleaseSourceAdmissionError("SOURCE_REF_MOVED")
+        base_oid = _oid(self.backend.authoritative_main_oid(), "base_oid")
+        fetched = _oid(
+            self.backend.fetch_exact_ref(ref_value, source_value),
+            "fetched_source_commit",
+        )
+        if fetched != source_value:
+            raise ReleaseSourceAdmissionError("SOURCE_FETCH_MISMATCH")
+        remote_after = self.backend.remote_ref_oid(ref_value)
+        if remote_after != source_value:
+            raise ReleaseSourceAdmissionError("SOURCE_REF_MOVED")
+        self.backend.require_lineage(base_oid, source_value)
+        tree_oid = _oid(self.backend.tree_oid(source_value), "tree_oid")
+
+        candidate_id = f"rc-{_version_slug(version_value)}-{source_value[:12]}"
+        admission_id = "source-admission-" + hashlib.sha256(
+            f"{self.store.repository}\0{version_value}\0{source_value}\0{ref_value}".encode("utf-8")
+        ).hexdigest()[:24]
+        record = {
+            "schema": SCHEMA,
+            "admission_id": admission_id,
+            "repository": self.store.repository,
+            "task_id": task_value,
+            "version": version_value,
+            "candidate_id": candidate_id,
+            "base_oid": base_oid,
+            "source_ref": ref_value,
+            "source_commit": source_value,
+            "remote_ref_oid": remote_after,
+            "tree_oid": tree_oid,
+            "purpose": self.purpose,
+            "policy_sha256": self.store.policy_sha256,
+            "admitted_at": _now(),
+        }
+        return self.store.register(record)
+
+
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+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)
+    admit.add_argument("--source-ref", required=True)
+    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")
+    repository = os.environ.get("AIOS_GITOPS_REPOSITORY_IDENTITY", "")
+    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:
+        raise SystemExit("release source admission environment is incomplete")
+    git_dir = Path(raw_git_dir)
+    transport = GiteaMachineTransport(
+        api_base=api_base,
+        repository_id=repository,
+        remote_url=remote_url,
+        authority_git_dir=git_dir,
+        credential_name=os.environ.get(
+            "AIOS_GITEA_MACHINE_CREDENTIAL_NAME",
+            "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(
+        arguments.store_root,
+        repository=repository,
+        policy_sha256=_sha256_file(arguments.policy),
+        hmac_key=admission_hmac,
+        expected_owner_uid=os.geteuid() if hasattr(os, "geteuid") else None,
+        writable=True,
+    )
+    result = ReleaseSourceAdmitter(
+        store,
+        GiteaReleaseSourceBackend(transport=transport, git_dir=git_dir),
+    ).admit(
+        version=arguments.version,
+        task_id=arguments.task_id,
+        source_ref=arguments.source_ref,
+        source_commit=arguments.source_commit,
+    )
+    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+
+
+if __name__ == "__main__":
+    _main()

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,
@@ -3531,6 +3583,7 @@ class SilentReleaseActionService:
         trusted_policy: TrustedReleasePolicy | None = None,
         orchestrator: ReleaseOrchestrator | None = None,
         builder_readiness_probe: Callable[[], Mapping[str, Any]] | None = None,
+        source_admission_resolver: Callable[[str], Mapping[str, Any] | None] | None = None,
         allow_legacy_candidate_input: bool = False,
         expected_repository: str = "",
         policy_sha256: str = "",
@@ -3547,6 +3600,7 @@ class SilentReleaseActionService:
         self.trusted_policy = trusted_policy
         self.orchestrator = orchestrator
         self.builder_readiness_probe = builder_readiness_probe
+        self.source_admission_resolver = source_admission_resolver
         self.allow_legacy_candidate_input = bool(
             allow_legacy_candidate_input
         )
@@ -3638,6 +3692,40 @@ class SilentReleaseActionService:
         version = str(payload.get("version") or "")
         if version:
             _validate_identifier("version", version)
+
+        source_admission: dict[str, Any] | None = None
+        resolver = self.source_admission_resolver
+        if resolver is not None:
+            try:
+                raw_admission = resolver(source_commit)
+            except Exception as exc:
+                raise CapabilityAuthorizationError(
+                    "release source admission is invalid"
+                ) from exc
+            if raw_admission is not None:
+                source_admission = dict(raw_admission)
+                admitted_source = str(
+                    source_admission.get("source_commit") or ""
+                )
+                admitted_version = str(source_admission.get("version") or "")
+                admission_id = str(
+                    source_admission.get("admission_id") or ""
+                )
+                if (
+                    admitted_source != source_commit
+                    or not admitted_version
+                    or not admission_id
+                ):
+                    raise CapabilityAuthorizationError(
+                        "release source admission is invalid"
+                    )
+                _validate_identifier("version", admitted_version)
+                _validate_identifier("admission_id", admission_id)
+                if version and version != admitted_version:
+                    raise CapabilityAuthorizationError(
+                        "release source admission version mismatch"
+                    )
+                version = admitted_version
         intent_id = "intent-" + hashlib.sha256(
             f"{event_id}\0{source_commit}".encode("utf-8")
         ).hexdigest()[:24]
@@ -3657,47 +3745,83 @@ class SilentReleaseActionService:
         }
         if version:
             record["version"] = version
+        if source_admission is not None:
+            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 (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(
                 {
@@ -3742,14 +3866,36 @@ class SilentReleaseActionService:
             )
             _durable_json(path, record)
             return self._intent_response(record, idempotent=False)
-        except CapabilityAuthorizationError:
-            record.update(
-                {
-                    "failure_code": "TRUSTED_BUILDER_REJECTED",
-                    "retryable": False,
-                    "status": "BUILDER_TERMINAL_FAILURE",
-                }
-            )
+        except CapabilityAuthorizationError as exc:
+            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(
+                    {
+                        # 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": (
+                            "MIGRATION_PLAN_INVALID"
+                            if str(exc) == "migration plan invalid"
+                            else "TRUSTED_BUILDER_REJECTED"
+                        ),
+                        "retryable": False,
+                        "status": "BUILDER_TERMINAL_FAILURE",
+                    }
+                )
             _durable_json(path, record)
             return self._intent_response(record, idempotent=False)
         except (ConnectionError, OSError, TimeoutError):

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-gitops-model-api-trusted-builder.conf.example

@@ -7,8 +7,10 @@ Environment=AIOS_TRUSTED_CANDIDATE_BUILDER_SOCKET=/run/aios-trusted-candidate-bu
 Environment=AIOS_TRUSTED_CANDIDATE_BUILDER_POLICY=/opt/aios/gitops-model-api/<VERIFIED_PACKAGE_OID>/runtime/config/gitops/trusted-candidate-builder-policy.json
 Environment=AIOS_TRUSTED_CANDIDATE_BUILDER_PUBLIC_KEYRING=/etc/aios-gitops/trusted-builder/public-keyring.json
 Environment=AIOS_TRUSTED_CANDIDATE_BUILDER_CREDENTIALS_DIRECTORY=%d
+Environment=AIOS_TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT=/var/lib/aios-release-source-admission
 Environment=AIOS_SILENT_RELEASE_TRUSTED_POLICY=/opt/aios/gitops-model-api/<VERIFIED_PACKAGE_OID>/runtime/config/gitops/trusted-release-policy.json
 LoadCredential=builder_hmac:/etc/aios-gitops/credential-references/trusted-candidate-builder-hmac
+LoadCredential=source_admission_hmac:/etc/aios-gitops/credential-references/release-source-admission-hmac
 # The credential source above is a reference only; no secret value belongs here.
 # Keep execution shadow-disabled and do not add Owner/deploy/tag/baseline switches.
 Environment=AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false

deploy/systemd/aios-gitops-model-api-v2r2c1-staging.service

@@ -27,6 +27,7 @@ Environment=AIOS_GITOPS_DEPLOYMENT_STATE=STAGING
 Environment=AIOS_GITOPS_MODEL_API_SERVICE=aios-gitops-model-api-v2r2c1-staging.service
 Environment=AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false
 Environment=AIOS_SILENT_RELEASE_STATE_ROOT=/var/lib/aios-gitops/model-api-staging/silent-release
+Environment=AIOS_TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT=/var/lib/aios-release-source-admission
 Environment=AIOS_BRIDGE_V2_URL=http://127.0.0.1:18892
 Environment=AIOS_GITOPS_HOST=127.0.0.1
 Environment=AIOS_GITOPS_PORT=18893
@@ -44,7 +45,7 @@ ProtectControlGroups=true
 LockPersonality=true
 RestrictSUIDSGID=true
 RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
-ReadOnlyPaths=@RUNTIME_ROOT@ /data/aios-gitops/repo.git /var/lib/aios-gitops-runtime-anchor/baseline
+ReadOnlyPaths=@RUNTIME_ROOT@ /data/aios-gitops/repo.git /var/lib/aios-gitops-runtime-anchor/baseline -/var/lib/aios-release-source-admission
 ReadWritePaths=/var/lib/aios-gitops/model-api-staging
 
 [Install]

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.service

@@ -0,0 +1,41 @@
+[Unit]
+Description=AIOS governed release source admission
+After=network-online.target aios-gitea.service
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+User=root
+Group=gitops
+UMask=0027
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_GITOPS_REPOSITORY_IDENTITY=aios/aios
+Environment=AIOS_GITOPS_AUTHORITATIVE_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_GITEA_API_BASE=https://g.cnnetmkt.top/git/api/v1
+Environment=AIOS_GITEA_REMOTE_URL=https://g.cnnetmkt.top/git/aios/aios.git
+Environment=AIOS_GITEA_MACHINE_CREDENTIAL_NAME=gitea-machine-token
+EnvironmentFile=/run/aios-release-source-admission/request.env
+LoadCredentialEncrypted=gitea-machine-token:/etc/aios/credentials/gitea-machine.token.cred
+LoadCredentialEncrypted=source_admission_hmac:/etc/credstore.encrypted/aios-release-source-admission/source_admission_hmac
+ExecStart=/usr/bin/python3 -B /opt/aios/release-source-admission/current/app/release_source_admission.py admit --version ${AIOS_RELEASE_VERSION} --task-id ${AIOS_RELEASE_TASK_ID} --source-ref ${AIOS_RELEASE_SOURCE_REF} --source-commit ${AIOS_RELEASE_SOURCE_COMMIT} --policy /opt/aios/trusted-candidate-builder/current/config/gitops/trusted-candidate-builder-policy.json --store-root /var/lib/aios-release-source-admission
+ExecStartPost=/usr/bin/rm -f /run/aios-release-source-admission/request.env
+RuntimeDirectory=aios-release-source-admission
+RuntimeDirectoryMode=0750
+StateDirectory=aios-release-source-admission
+StateDirectoryMode=0750
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+CapabilityBoundingSet=
+ReadOnlyPaths=/opt/aios/release-source-admission /opt/aios/trusted-candidate-builder /etc/aios/credentials /etc/credstore.encrypted
+ReadWritePaths=/data/aios-gitops/repo.git /var/lib/aios-release-source-admission /run/aios-release-source-admission

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

@@ -10,10 +10,14 @@ Group=root
 UMask=0077
 LoadCredentialEncrypted=signing_private_key:/etc/credstore.encrypted/aios-trusted-candidate-builder/signing_private_key
 LoadCredentialEncrypted=builder_hmac:/etc/credstore.encrypted/aios-trusted-candidate-builder/builder_hmac
+LoadCredentialEncrypted=source_admission_hmac:/etc/credstore.encrypted/aios-release-source-admission/source_admission_hmac
 Environment=AIOS_TRUSTED_BUILDER_POLICY=/opt/aios/trusted-candidate-builder/current/config/gitops/trusted-candidate-builder-policy.json
 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
@@ -37,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
+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_live_executor_contract.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import tempfile
+from types import SimpleNamespace
 from pathlib import Path
 
 import pytest
 
 import aios_gitops_job_executor as executor_module
 from aios_authority_git_executor import AuthorityGitError, AuthorityGitExecutor
+import aios_gitea_machine_transport as gitea_transport_module
 from aios_gitea_machine_transport import GiteaMachineTransport
 from aios_gitea_machine_broker import CampaignCapabilityReceipt, GiteaMachineBroker, MachineCredentialProjection
 from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore, JobState
@@ -162,3 +164,77 @@ def test_machine_transport_branch_cas_is_not_force_push(monkeypatch):
     monkeypatch.setattr(GiteaMachineTransport, "_push_branch_fast_forward", lambda _self, branch, new: calls.append((branch, new)))
     assert transport.compare_and_swap_branch(executor_module.BRANCH, OLD, NEW) == NEW
     assert calls == [(executor_module.BRANCH, NEW)]
+
+
+def test_machine_transport_fetches_exact_branch_tip_into_quarantine_ref(monkeypatch, tmp_path):
+    transport = GiteaMachineTransport(
+        "http://127.0.0.1:3000/api/v1",
+        "aios/aios",
+        "https://g.cnnetmkt.top/git/aios/aios.git",
+        tmp_path / "repo.git",
+    )
+    reads = iter([NEW, NEW])
+    monkeypatch.setattr(
+        GiteaMachineTransport,
+        "read_branch",
+        lambda _self, _branch: next(reads),
+    )
+    askpass = tmp_path / "askpass"
+    askpass.write_text("#!/bin/sh\n", encoding="utf-8")
+    monkeypatch.setattr(
+        GiteaMachineTransport,
+        "_askpass",
+        lambda _self: (askpass, {}),
+    )
+    calls = []
+
+    def run(argv, **kwargs):
+        calls.append(list(argv))
+        if "fetch" in argv:
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        if "rev-parse" in argv:
+            # Before fetch the quarantine ref is absent; after fetch it resolves.
+            seen_fetch = any("fetch" in call for call in calls[:-1])
+            return SimpleNamespace(
+                returncode=0 if seen_fetch else 1,
+                stdout=(NEW + "\n") if seen_fetch else "",
+                stderr="",
+            )
+        if "update-ref" in argv:
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        raise AssertionError(argv)
+
+    monkeypatch.setattr(gitea_transport_module.subprocess, "run", run)
+
+    assert transport.fetch_branch_exact(executor_module.BRANCH, NEW) == NEW
+    fetch = next(call for call in calls if "fetch" in call)
+    assert "--no-tags" in fetch
+    assert "--no-write-fetch-head" in fetch
+    assert f"refs/heads/{executor_module.BRANCH}:refs/aios/release-sources/{NEW}" in fetch
+    assert not any("--force" in item or item == "-f" for item in fetch)
+
+
+def test_machine_transport_does_not_fetch_when_remote_tip_moved(monkeypatch, tmp_path):
+    transport = GiteaMachineTransport(
+        "http://127.0.0.1:3000/api/v1",
+        "aios/aios",
+        "https://g.cnnetmkt.top/git/aios/aios.git",
+        tmp_path / "repo.git",
+    )
+    monkeypatch.setattr(
+        GiteaMachineTransport,
+        "read_branch",
+        lambda _self, _branch: OLD,
+    )
+    monkeypatch.setattr(
+        gitea_transport_module.subprocess,
+        "run",
+        lambda *args, **kwargs: (_ for _ in ()).throw(
+            AssertionError("git must not run after remote mismatch")
+        ),
+    )
+    with pytest.raises(
+        gitea_transport_module.GiteaTransportError,
+        match="GITEA_SOURCE_REF_MISMATCH",
+    ):
+        transport.fetch_branch_exact(executor_module.BRANCH, NEW)

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_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

@@ -0,0 +1,252 @@
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
+
+from release_source_admission import (  # noqa: E402
+    ReleaseSourceAdmissionError,
+    ReleaseSourceAdmissionStore,
+    ReleaseSourceAdmitter,
+)
+
+
+MAIN = "1" * 40
+SOURCE = "2" * 40
+TREE = "3" * 40
+POLICY = "4" * 64
+REF = "refs/heads/agent/baseline-1.1-principal-recovery-20260908"
+VERSION = "V2R2C4"
+TASK = "AIOS-BASELINE-1.1-PRODUCTION-CLOSURE"
+
+
+class FakeBackend:
+    def __init__(
+        self,
+        *,
+        remote: str = SOURCE,
+        fetched: str = SOURCE,
+        tree: str = TREE,
+        main: str = MAIN,
+        lineage: bool = True,
+    ) -> None:
+        self.remote = remote
+        self.fetched = fetched
+        self.tree = tree
+        self.main = main
+        self.lineage = lineage
+        self.fetch_calls: list[tuple[str, str]] = []
+        self.lineage_calls: list[tuple[str, str]] = []
+
+    def authoritative_main_oid(self) -> str:
+        return self.main
+
+    def remote_ref_oid(self, source_ref: str) -> str:
+        assert source_ref == REF
+        return self.remote
+
+    def fetch_exact_ref(self, source_ref: str, source_commit: str) -> str:
+        self.fetch_calls.append((source_ref, source_commit))
+        return self.fetched
+
+    def tree_oid(self, source_commit: str) -> str:
+        assert source_commit == SOURCE
+        return self.tree
+
+    def require_lineage(self, base_oid: str, source_commit: str) -> None:
+        self.lineage_calls.append((base_oid, source_commit))
+        if not self.lineage:
+            raise ReleaseSourceAdmissionError("SOURCE_LINEAGE_MISMATCH")
+
+
+def make_store(tmp_path: Path, *, writable: bool = True) -> ReleaseSourceAdmissionStore:
+    owner = os.geteuid() if hasattr(os, "geteuid") else None
+    return ReleaseSourceAdmissionStore(
+        tmp_path / "admissions",
+        repository="aios/aios",
+        policy_sha256=POLICY,
+        hmac_key=b"h" * 32,
+        expected_owner_uid=owner,
+        writable=writable,
+    )
+
+
+def test_admit_binds_remote_ref_fetch_tree_main_lineage_and_recipe(tmp_path: Path) -> None:
+    store = make_store(tmp_path)
+    backend = FakeBackend()
+
+    record = ReleaseSourceAdmitter(store, backend).admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+
+    assert record["schema"] == "aios.release-source-admission.v1"
+    assert record["repository"] == "aios/aios"
+    assert record["source_commit"] == SOURCE
+    assert record["remote_ref_oid"] == SOURCE
+    assert record["source_ref"] == REF
+    assert record["base_oid"] == MAIN
+    assert record["tree_oid"] == TREE
+    assert record["version"] == VERSION
+    assert record["task_id"] == TASK
+    assert record["candidate_id"] == "rc-v2r2c4-" + SOURCE[:12]
+    assert backend.fetch_calls == [(REF, SOURCE)]
+    assert backend.lineage_calls == [(MAIN, SOURCE)]
+
+    recipe = store.resolve_recipe(VERSION, None, SOURCE)
+    assert recipe == {
+        "mode": "governed-source-admission",
+        "task_id": TASK,
+        "candidate_id": record["candidate_id"],
+        "base_oid": MAIN,
+        "source_commit": SOURCE,
+        "version": VERSION,
+        "purpose": "AIOS_TRUSTED_RELEASE_BUILDER",
+    }
+
+
+def test_repeated_exact_admission_is_idempotent(tmp_path: Path) -> None:
+    store = make_store(tmp_path)
+    admitter = ReleaseSourceAdmitter(store, FakeBackend())
+
+    first = admitter.admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+    second = admitter.admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+
+    assert second == first
+    assert len(list((tmp_path / "admissions").glob("*.json"))) == 1
+
+
+@pytest.mark.parametrize(
+    ("backend", "error"),
+    (
+        (FakeBackend(remote="5" * 40), "SOURCE_REF_MOVED"),
+        (FakeBackend(fetched="5" * 40), "SOURCE_FETCH_MISMATCH"),
+        (FakeBackend(lineage=False), "SOURCE_LINEAGE_MISMATCH"),
+    ),
+)
+def test_admission_rejects_moved_fetch_mismatch_or_non_descendant(
+    tmp_path: Path,
+    backend: FakeBackend,
+    error: str,
+) -> None:
+    store = make_store(tmp_path)
+    with pytest.raises(ReleaseSourceAdmissionError, match=error):
+        ReleaseSourceAdmitter(store, backend).admit(
+            version=VERSION,
+            task_id=TASK,
+            source_ref=REF,
+            source_commit=SOURCE,
+        )
+    assert not list((tmp_path / "admissions").glob("*.json"))
+
+
+@pytest.mark.parametrize(
+    "source_ref",
+    (
+        "agent/bad",
+        "refs/tags/v1",
+        "refs/heads/../main",
+        "refs/heads/agent//bad",
+        "refs/heads/agent/bad.lock",
+    ),
+)
+def test_admission_rejects_unsafe_or_non_head_refs(
+    tmp_path: Path, source_ref: str
+) -> None:
+    store = make_store(tmp_path)
+    with pytest.raises(ReleaseSourceAdmissionError, match="SOURCE_REF_INVALID"):
+        ReleaseSourceAdmitter(store, FakeBackend()).admit(
+            version=VERSION,
+            task_id=TASK,
+            source_ref=source_ref,
+            source_commit=SOURCE,
+        )
+
+
+def test_store_rejects_caller_candidate_rebinding(tmp_path: Path) -> None:
+    store = make_store(tmp_path)
+    record = ReleaseSourceAdmitter(store, FakeBackend()).admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+    with pytest.raises(
+        ReleaseSourceAdmissionError,
+        match="SOURCE_ADMISSION_CANDIDATE_MISMATCH",
+    ):
+        store.resolve_recipe(VERSION, "rc-attacker", SOURCE)
+    assert record["candidate_id"] != "rc-attacker"
+
+
+def test_store_rejects_policy_change_and_record_tamper(tmp_path: Path) -> None:
+    store = make_store(tmp_path)
+    record = ReleaseSourceAdmitter(store, FakeBackend()).admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+    path = tmp_path / "admissions" / f"{SOURCE}.json"
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    payload["task_id"] = "other-task"
+    path.write_text(
+        json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(
+        ReleaseSourceAdmissionError,
+        match="SOURCE_ADMISSION_MAC_INVALID",
+    ):
+        store.lookup_source(SOURCE)
+
+    owner = os.geteuid() if hasattr(os, "geteuid") else None
+    other_policy = ReleaseSourceAdmissionStore(
+        tmp_path / "admissions",
+        repository="aios/aios",
+        policy_sha256="6" * 64,
+        hmac_key=b"h" * 32,
+        expected_owner_uid=owner,
+        writable=False,
+    )
+    with pytest.raises(
+        ReleaseSourceAdmissionError,
+        match="SOURCE_ADMISSION_POLICY_MISMATCH",
+    ):
+        other_policy.lookup_source(SOURCE)
+
+
+def test_store_rejects_group_writable_record(tmp_path: Path) -> None:
+    store = make_store(tmp_path)
+    ReleaseSourceAdmitter(store, FakeBackend()).admit(
+        version=VERSION,
+        task_id=TASK,
+        source_ref=REF,
+        source_commit=SOURCE,
+    )
+    path = tmp_path / "admissions" / f"{SOURCE}.json"
+    os.chmod(path, 0o660)
+    with pytest.raises(
+        ReleaseSourceAdmissionError,
+        match="SOURCE_ADMISSION_PERMISSIONS_INVALID",
+    ):
+        store.lookup_source(SOURCE)

tests/test_silent_release_execution_readiness.py

@@ -726,3 +726,129 @@ def test_systemd_shadow_unit_has_both_gates_and_no_secret_environment():
     assert "PRIVATE_KEY=" not in unit
     assert "TOKEN=" not in unit
     assert "PASSWORD=" not in unit
+
+
+def test_release_intent_derives_version_from_governed_source_admission(tmp_path):
+    service, _ = trusted_service(tmp_path)
+    service.source_admission_resolver = lambda source_commit: {
+        "schema": "aios.release-source-admission.v1",
+        "admission_id": "source-admission-test-001",
+        "source_commit": source_commit,
+        "version": "V2R2C4",
+    }
+
+    result = service.create_release_intent(
+        {"event_id": "webhook-admitted-source-001", "source_commit": COMMIT}
+    )
+
+    assert result["version"] == "V2R2C4"
+    assert result["source_admission_id"] == "source-admission-test-001"
+    intent_path = service.intents_root / f"{result['intent_id']}.json"
+    persisted = json.loads(intent_path.read_text(encoding="utf-8"))
+    assert persisted["version"] == "V2R2C4"
+    assert persisted["source_admission_id"] == "source-admission-test-001"
+
+
+def test_release_intent_rejects_version_rebinding_against_source_admission(tmp_path):
+    service, _ = trusted_service(tmp_path)
+    service.source_admission_resolver = lambda source_commit: {
+        "schema": "aios.release-source-admission.v1",
+        "admission_id": "source-admission-test-002",
+        "source_commit": source_commit,
+        "version": "V2R2C4",
+    }
+
+    with pytest.raises(
+        CapabilityAuthorizationError,
+        match="release source admission version mismatch",
+    ):
+        service.create_release_intent(
+            {
+                "event_id": "webhook-admitted-source-002",
+                "source_commit": COMMIT,
+                "version": "V2R2C5",
+            }
+        )
+
+
+def test_release_intent_surfaces_source_admission_required_as_typed_blocker(tmp_path):
+    service, _ = trusted_service(tmp_path)
+
+    class AdmissionRequiredBuilder:
+        def build(self, _payload):
+            raise CapabilityAuthorizationError(
+                "release source admission required"
+            )
+
+    service.candidate_builder = AdmissionRequiredBuilder()
+    result = service.create_release_intent(
+        {
+            "event_id": "webhook-source-admission-required-001",
+            "source_commit": COMMIT,
+        }
+    )
+
+    assert result["status"] == "SOURCE_ADMISSION_REQUIRED"
+    assert result["failure_code"] == "SOURCE_ADMISSION_REQUIRED"
+    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,
@@ -590,3 +596,177 @@ def test_systemd_socket_and_service_are_af_unix_hardened_and_credential_bound()
     assert policy["allowed_peer_cgroups"] == [
         "/system.slice/aios-gitops-model-api-v2r2c1-staging.service"
     ]
+
+
+def test_governed_source_admission_resolves_unknown_version_without_overriding_fixed_recipe() -> None:
+    policy = load_builder_policy(POLICY_PATH)
+    calls: list[tuple[str, str | None, str]] = []
+
+    def resolver(version: str, candidate_id: str | None, source_commit: str):
+        calls.append((version, candidate_id, source_commit))
+        return {
+            "mode": "governed-source-admission",
+            "task_id": "AIOS-BASELINE-1.1-PRODUCTION-CLOSURE",
+            "candidate_id": "rc-v2r2c4-" + source_commit[:12],
+            "base_oid": "1" * 40,
+            "source_commit": source_commit,
+            "version": version,
+            "purpose": "AIOS_TRUSTED_RELEASE_BUILDER",
+        }
+
+    config = dataclasses.replace(
+        TrustedCandidateBuilderConfig.from_policy(policy),
+        source_admission_resolver=resolver,
+    )
+    client = UnixTrustedCandidateBuilderClient(
+        config,
+        BuilderKeyring(builder_hmac=b"h" * 32),
+        clock=lambda: NOW,
+    )
+    source = "2" * 40
+
+    request = client.create_request(
+        "V2R2C4",
+        None,
+        source,
+        nonce="z" * 32,
+    )
+
+    assert request["candidate_id"] == "rc-v2r2c4-" + source[:12]
+    assert request["base_oid"] == "1" * 40
+    assert request["source_commit"] == source
+    assert request["task_id"] == "AIOS-BASELINE-1.1-PRODUCTION-CLOSURE"
+    assert calls == [("V2R2C4", None, source)]
+
+    with pytest.raises(BuilderError, match="REQUEST_RELEASE_BINDING_INVALID"):
+        config.resolve_recipe(
+            "V2R2C2",
+            None,
+            source,
+        )
+    # Fixed policy versions never fall through to the admission resolver.
+    assert calls == [("V2R2C4", None, source)]
+
+
+def test_unknown_release_source_requires_explicit_admission() -> None:
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
+    with pytest.raises(BuilderError, match="RECIPE_UNKNOWN"):
+        config.resolve_recipe("V2R2C4", None, "2" * 40)
+
+
+def test_governed_source_admission_rejects_caller_candidate_rebinding() -> None:
+    policy = load_builder_policy(POLICY_PATH)
+
+    def resolver(version: str, candidate_id: str | None, source_commit: str):
+        return {
+            "mode": "governed-source-admission",
+            "task_id": "task-admitted-source",
+            "candidate_id": "rc-v2r2c4-" + source_commit[:12],
+            "base_oid": "1" * 40,
+            "source_commit": source_commit,
+            "version": version,
+            "purpose": "AIOS_TRUSTED_RELEASE_BUILDER",
+        }
+
+    config = dataclasses.replace(
+        TrustedCandidateBuilderConfig.from_policy(policy),
+        source_admission_resolver=resolver,
+    )
+    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)