agent: baseline 1.1 evidence materialization + builder readiness split

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

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

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -157,7 +157,7 @@ def render_openapi_contract_fragment() -> str:
         "      operationId: commitFilesGoverned\n"
         "      summary: Submit one atomic governed commit containing multiple files\n"
         "      security: [{taskPrincipalAuth: []}]\n"
-        "      x-aios-task-principal: Bridge verifies a fresh aios.task-principal-assertion.v1 and exact task/job/repository/branch/generation/revision/attempt/execution-digest/capability binding; caller JSON cannot provide a principal and missing verification is denied.\n"
+        "      x-aios-task-principal: Bridge resolves a fresh server-side TaskPrincipal from the authoritative admitted+claimed Task and re-verifies its Owner epoch, then checks exact task/job/repository/branch/generation/capability binding; caller JSON cannot provide a principal and missing verification is denied. An explicitly injected legacy bearer verifier may still be used by a trusted host.\n"
         "      requestBody:\n"
         "        required: true\n"
         "        content:\n"

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -1279,6 +1279,7 @@ def serve(adapter: BridgeV2Adapter, host: str = "127.0.0.1", port: int = 18892)
 def main(
     *,
     task_admission_handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None,
+    task_principal_verifier: TaskPrincipalVerifier | None = None,
 ) -> None:
     activation_raw = os.environ.get(
         "AIOS_TASK_ADMISSION_ACTIVATION_FILE",
@@ -1333,27 +1334,33 @@ def main(
         repository_discovery_cursor_secret=discovery_cursor_secret,
     )
     runtime = GovernedRuntime(config)
-    # A TaskPrincipal verifier is deliberately supplied only by the hosting
-    # service.  Without that independently authenticated verifier, all Git
-    # mutation requests fail closed while the admission-only routes remain
-    # available for a configured Owner-approved epoch.
-    BridgeHandler.principal_verifier = None
+    # Prefer a server-side TaskPrincipal provider recovered from the same
+    # verified Task-admission runtime.  This removes the caller-side credential
+    # bootstrap loop while keeping every mutation dependent on a currently
+    # claimed Task and a freshly re-verified Owner epoch.  Hosts may still
+    # inject an independently authenticated verifier explicitly.
+    BridgeHandler.principal_verifier = task_principal_verifier
     if task_admission_handlers is None:
+        admission_runtime = None
         if os.environ.get("AIOS_TASK_ADMISSION_EPOCH_ROOT", "").strip():
             from aios_task_admission_live_adapters import (
-                build_task_admission_handlers_from_environment,
+                build_task_admission_runtime_from_environment,
             )
-            task_admission_handlers = build_task_admission_handlers_from_environment(
+            admission_runtime = build_task_admission_runtime_from_environment(
                 post_admission=runtime.project_admitted_task,
             )
         elif activation_raw and Path(activation_raw).is_file():
             from aios_task_admission_live_adapters import (
-                build_task_admission_handlers_from_activation_file,
+                build_task_admission_runtime_from_activation_file,
             )
-            task_admission_handlers = build_task_admission_handlers_from_activation_file(
+            admission_runtime = build_task_admission_runtime_from_activation_file(
                 Path(activation_raw),
                 post_admission=runtime.project_admitted_task,
             )
+        if admission_runtime is not None:
+            task_admission_handlers = admission_runtime.handlers
+            if BridgeHandler.principal_verifier is None:
+                BridgeHandler.principal_verifier = admission_runtime.principal_verifier
     serve(
         build_bridge_adapter(
             runtime,

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,33 @@ 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 (
+                    open_source_admission_store,
+                )
+
+                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 = open_source_admission_store(
+                    Path(TRUSTED_BUILDER_SOURCE_ADMISSION_ROOT),
+                    repository=builder_config.repository,
+                    policy=TRUSTED_BUILDER_POLICY,
+                    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 +413,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_task_admission_bootstrap.py

@@ -438,6 +438,71 @@ class TaskAdmissionStore:
             ).fetchone()
         return dict(row) if row else None
 
+    def get_grant_by_id(self, grant_id: str) -> TaskAdmissionGrant | None:
+        """Return one immutable admitted grant by its stable grant identity.
+
+        This is a read-only lookup used by the server-side TaskPrincipal
+        recovery path.  It never trusts caller-supplied grant content.
+        """
+        if not isinstance(grant_id, str) or not grant_id:
+            raise TaskAdmissionError("TASK_ADMISSION_GRANT_ID_INVALID", 400)
+        with self._connection() as con:
+            row = con.execute(
+                "SELECT grant_json FROM task_admission_grants WHERE grant_id=?",
+                (grant_id,),
+            ).fetchone()
+        if row is None:
+            return None
+        try:
+            raw = json.loads(row["grant_json"])
+            return TaskAdmissionGrant(
+                **{
+                    **raw,
+                    "capabilities": tuple(raw["capabilities"]),
+                    "file_allowlist": tuple(raw["file_allowlist"]),
+                }
+            )
+        except (TypeError, ValueError, KeyError, json.JSONDecodeError) as exc:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_DURABLE_GRANT_INVALID", 503
+            ) from exc
+
+    def get_grant_by_task_id(self, task_id: str) -> TaskAdmissionGrant | None:
+        """Return the immutable admitted grant bound to an authoritative Task."""
+        if not isinstance(task_id, str) or not task_id:
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_ID_INVALID", 400)
+        with self._connection() as con:
+            row = con.execute(
+                """
+                SELECT g.grant_json
+                FROM task_admission_grants AS g
+                JOIN task_admission_receipts AS r
+                  ON r.envelope_id = g.envelope_id
+                WHERE r.task_id = ?
+                """,
+                (task_id,),
+            ).fetchone()
+        if row is None:
+            return None
+        try:
+            raw = json.loads(row["grant_json"])
+            grant = TaskAdmissionGrant(
+                **{
+                    **raw,
+                    "capabilities": tuple(raw["capabilities"]),
+                    "file_allowlist": tuple(raw["file_allowlist"]),
+                }
+            )
+        except (TypeError, ValueError, KeyError, json.JSONDecodeError) as exc:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_DURABLE_GRANT_INVALID", 503
+            ) from exc
+        if grant.task_id != task_id:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_DURABLE_GRANT_TASK_MISMATCH", 503
+            )
+        return grant
+
     def save_envelope(
         self,
         *,

active-runtime/app/aios_task_admission_live_adapters.py

@@ -257,6 +257,28 @@ class AgentTaskAdmissionPort:
             raise TaskAdmissionError("TASK_ADMISSION_TASK_STATE_UNKNOWN", 503)
         return recovered
 
+    def verify_active_task(
+        self, grant: TaskAdmissionGrant
+    ) -> tuple[Mapping[str, Any], Mapping[str, Any]]:
+        """Re-read and validate the authoritative Task for execution use.
+
+        Admission may complete while the Task is still QUEUED.  A TaskPrincipal
+        may be recovered only after the Task authority reports CLAIMED/RUNNING.
+        The exact immutable metadata is revalidated on every call.
+        """
+        execution = self._execution_plan(grant)
+        task = self._read(grant, execution)
+        if task is None:
+            raise TaskAdmissionError("TASK_PRINCIPAL_TASK_NOT_FOUND", 403)
+        state = str(task.get("state") or task.get("status") or "")
+        if state not in {"CLAIMED", "RUNNING"}:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_TASK_NOT_CLAIMED",
+                409,
+                details={"state": state},
+            )
+        return task, execution
+
     def projection_result(
         self,
         grant: TaskAdmissionGrant,
@@ -286,6 +308,152 @@ class AgentTaskAdmissionPort:
             "external_job_id": external_job_id,
         }
 
+_BRIDGE_PRINCIPAL_CAPABILITIES = {
+    "acquireWriterLease": "writer.lease",
+    "commitFileGoverned": "governed.write.file",
+    "commitFilesGoverned": "governed.write.files",
+    "getGitOpsJob": "writer.lease",
+    "resumeGitOpsJob": "writer.lease",
+    "reconcileGitOpsJob": "writer.lease",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionPrincipalProvider:
+    """Recover a short-lived execution principal from durable admission state.
+
+    This provider is server-side only.  It never accepts an Owner grant,
+    TaskPrincipal, or capability from caller JSON.  Every use re-reads the
+    admitted grant, re-verifies its parent Owner epoch (including revocation),
+    re-reads the authoritative Task, and requires CLAIMED/RUNNING before a
+    mutating Bridge operation can continue.
+    """
+
+    store: Any
+    authorization: Any
+    task_port: AgentTaskAdmissionPort
+    epoch_id: str
+    clock: Any = time.time
+
+    def readiness(self) -> Mapping[str, Any]:
+        try:
+            epoch = self.authorization.verify_delegation_epoch(self.epoch_id)
+        except Exception as exc:
+            return {
+                "ready": False,
+                "mode": "server-side-task-admission",
+                "error": str(exc) or type(exc).__name__,
+            }
+        return {
+            "ready": True,
+            "mode": "server-side-task-admission",
+            "epoch_id": epoch.epoch_id,
+            "repository": epoch.repository,
+        }
+
+    @staticmethod
+    def _request_value(
+        request: Mapping[str, Any], name: str, expected: object
+    ) -> None:
+        if name in request and request[name] != expected:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_REQUEST_BINDING_MISMATCH",
+                403,
+                details={"field": name},
+            )
+
+    def __call__(
+        self,
+        transport: Mapping[str, Any],
+        request: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        operation = str(transport.get("operation") or "")
+        capability = _BRIDGE_PRINCIPAL_CAPABILITIES.get(operation)
+        if capability is None:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_BOOTSTRAP_ROUTE_REQUIRED", 403
+            )
+        task_id = request.get("task_id")
+        if not isinstance(task_id, str) or not task_id:
+            raise TaskAdmissionError("TASK_PRINCIPAL_TASK_ID_INVALID", 403)
+
+        grant = self.store.get_grant_by_task_id(task_id)
+        if grant is None:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_AUTHORIZATION_REQUIRED", 403
+            )
+        if grant.parent_epoch_id is None:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_PARENT_EPOCH_REQUIRED", 403
+            )
+
+        # Re-verify the exact parent on every use so revocation/expiry cannot be
+        # bypassed by a previously admitted Task.
+        epoch = self.authorization.verify_delegation_epoch(grant.parent_epoch_id)
+        if (
+            epoch.epoch_id != self.epoch_id
+            or epoch.repository != grant.repository
+            or epoch.trust_domain != grant.trust_domain
+            or not set(grant.capabilities).issubset(epoch.allowed_capabilities)
+            or not any(
+                grant.target_branch.startswith(prefix)
+                for prefix in epoch.allowed_branch_prefixes
+            )
+        ):
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_PARENT_EPOCH_SCOPE_MISMATCH", 403
+            )
+
+        now = int(self.clock())
+        if not grant.issued_at <= now < grant.expires_at:
+            raise TaskAdmissionError("TASK_PRINCIPAL_GRANT_EXPIRED", 403)
+        if (
+            "task.principal.issue" not in grant.capabilities
+            or capability not in grant.capabilities
+        ):
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_CAPABILITY_DENIED", 403
+            )
+
+        task, execution = self.task_port.verify_active_task(grant)
+        external_job_id = execution.get("external_job_id")
+        if not isinstance(external_job_id, str) or not external_job_id:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_EXECUTION_ID_INVALID", 503
+            )
+
+        self._request_value(request, "task_id", grant.task_id)
+        self._request_value(request, "job_id", external_job_id)
+        self._request_value(request, "repository", grant.repository)
+        self._request_value(request, "target_branch", grant.target_branch)
+        self._request_value(request, "generation", grant.generation)
+        if operation == "acquireWriterLease":
+            self._request_value(
+                request, "base_oid", grant.authority_base_oid
+            )
+
+        return {
+            "schema": "aios.task-principal-server-attestation.v1",
+            "task_id": grant.task_id,
+            "grant_id": grant.grant_id,
+            "parent_epoch_id": grant.parent_epoch_id,
+            "repository": grant.repository,
+            "target_branch": grant.target_branch,
+            "generation": grant.generation,
+            "job_id": external_job_id,
+            "capability": capability,
+            "task_state": str(task.get("state") or task.get("status") or ""),
+            "request_digest": sha256_json(dict(request)),
+            "verified_at": now,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionRuntimeBundle:
+    handlers: Mapping[str, Any]
+    principal_verifier: TaskAdmissionPrincipalProvider
+
+
 class TaskAdmissionHostError(RuntimeError):
     """Stable fail-closed host-composition error."""
 
@@ -670,7 +838,7 @@ def task_admission_policy_from_epoch(
     )
 
 
-def _build_task_admission_handlers(
+def _build_task_admission_runtime(
     *,
     caller_service: str,
     epoch_root: Path,
@@ -682,8 +850,8 @@ def _build_task_admission_handlers(
     task_control_token_file: Path,
     owner_verifier_module: Path,
     post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
-) -> Mapping[str, Any]:
-    """Compose handlers from already-trusted server-side coordinates."""
+) -> TaskAdmissionRuntimeBundle:
+    """Compose admission handlers plus automatic execution-principal recovery."""
     from aios_task_admission_bootstrap import (
         TaskAdmissionBootstrapService,
         TaskAdmissionStore,
@@ -708,34 +876,49 @@ def _build_task_admission_handlers(
         verified_epoch_provider=provider,
         signature_verifier=verifier,
     )
+    transport = LocalAgentTaskHTTPTransport(
+        base_url=task_api_url,
+        token_file=task_control_token_file,
+        require_root_owner=True,
+    )
     task_port = AgentTaskAdmissionPort(
-        LocalAgentTaskHTTPTransport(
-            base_url=task_api_url,
-            token_file=task_control_token_file,
-            require_root_owner=True,
-        ),
+        transport,
         execution_plan_provider=ExecutionPlanDirectoryProvider(
             root=execution_plan_dir,
             require_root_owner=True,
         ),
     )
+    store = TaskAdmissionStore(state_db)
     service = TaskAdmissionBootstrapService(
-        store=TaskAdmissionStore(state_db),
+        store=store,
         authorization=authorization,
         task_port=task_port,
         policy=task_admission_policy_from_epoch(epoch),
         post_admission=post_admission,
     )
-    return TaskAdmissionBridgeContract(service).handlers()
+    return TaskAdmissionRuntimeBundle(
+        handlers=TaskAdmissionBridgeContract(service).handlers(),
+        principal_verifier=TaskAdmissionPrincipalProvider(
+            store=store,
+            authorization=authorization,
+            task_port=task_port,
+            epoch_id=epoch.epoch_id,
+        ),
+    )
 
 
-def build_task_admission_handlers_from_activation_file(
+def _build_task_admission_handlers(**kwargs: Any) -> Mapping[str, Any]:
+    """Compatibility wrapper for callers that only need admission routes."""
+    return _build_task_admission_runtime(**kwargs).handlers
+
+
+def build_task_admission_runtime_from_activation_file(
     activation_file: Path,
     *,
     caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
     post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
-) -> Mapping[str, Any]:
-    """Load one root-owned, fail-closed staging activation descriptor."""
+) -> TaskAdmissionRuntimeBundle:
+    """Load admission handlers plus server-side principal recovery."""
     _host_require_regular(activation_file, root_owned=True)
     value = _host_read_json_object(activation_file)
     required = {
@@ -777,7 +960,7 @@ def build_task_admission_handlers_from_activation_file(
     if task_api_url != "http://127.0.0.1:18940":
         raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_TASK_API_INVALID")
 
-    return _build_task_admission_handlers(
+    return _build_task_admission_runtime(
         caller_service=caller_service,
         epoch_root=paths["epoch_root"],
         epoch_id=epoch_id,
@@ -791,12 +974,25 @@ def build_task_admission_handlers_from_activation_file(
     )
 
 
-def build_task_admission_handlers_from_environment(
+def build_task_admission_handlers_from_activation_file(
+    activation_file: Path,
     *,
     caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
     post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
 ) -> Mapping[str, Any]:
-    """Compose all three handlers from server-side trusted environment only."""
+    return build_task_admission_runtime_from_activation_file(
+        activation_file,
+        caller_service=caller_service,
+        post_admission=post_admission,
+    ).handlers
+
+
+def build_task_admission_runtime_from_environment(
+    *,
+    caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
+    post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+) -> TaskAdmissionRuntimeBundle:
+    """Compose admission plus automatic TaskPrincipal recovery from trusted environment."""
     required = (
         "AIOS_TASK_ADMISSION_EPOCH_ROOT",
         "AIOS_TASK_ADMISSION_EPOCH_ID",
@@ -813,7 +1009,7 @@ def build_task_admission_handlers_from_environment(
             "TASK_ADMISSION_HOST_ENV_MISSING:"
             + ",".join(sorted(missing))
         )
-    return _build_task_admission_handlers(
+    return _build_task_admission_runtime(
         caller_service=caller_service,
         epoch_root=Path(os.environ["AIOS_TASK_ADMISSION_EPOCH_ROOT"]),
         epoch_id=os.environ["AIOS_TASK_ADMISSION_EPOCH_ID"],
@@ -840,6 +1036,17 @@ def build_task_admission_handlers_from_environment(
         post_admission=post_admission,
     )
 
+def build_task_admission_handlers_from_environment(
+    *,
+    caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
+    post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+) -> Mapping[str, Any]:
+    return build_task_admission_runtime_from_environment(
+        caller_service=caller_service,
+        post_admission=post_admission,
+    ).handlers
+
+
 # ---- producer phase core v12r4 (hash-pinned sibling source) ----
 import hashlib
 import json

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]]:
@@ -297,6 +348,35 @@ class ServerOwnedEvidenceSource:
             raise BuilderError("COLLECTOR_EVIDENCE_OWNER_DENIED")
         return resolved
 
+    def assert_ready(self) -> None:
+        """Validate the server-owned evidence facility without requiring recipe evidence.
+
+        Readiness is an infrastructure/property check.  Candidate-specific evidence
+        remains mandatory for ``BuildTrustedCandidateV1`` and is validated by
+        :meth:`__call__`; an empty evidence root therefore no longer makes the
+        builder globally unavailable.
+        """
+
+        for path, code in (
+            (self._authority_git_dir, "COLLECTOR_AUTHORITY_GIT_UNAVAILABLE"),
+            (self._evidence_root, "COLLECTOR_EVIDENCE_ROOT_UNAVAILABLE"),
+        ):
+            try:
+                metadata = path.stat(follow_symlinks=False)
+            except OSError as exc:
+                raise BuilderError(code) from exc
+            if not stat.S_ISDIR(metadata.st_mode) or path.is_symlink():
+                raise BuilderError(code)
+            if os.name != "nt" and metadata.st_mode & 0o022:
+                raise BuilderError("COLLECTOR_EVIDENCE_ROOT_MODE_DENIED")
+            if (
+                path == self._evidence_root
+                and os.name != "nt"
+                and self._expected_owner_uid is not None
+                and metadata.st_uid != self._expected_owner_uid
+            ):
+                raise BuilderError("COLLECTOR_EVIDENCE_ROOT_OWNER_DENIED")
+
     def _digest(self, path: Path) -> str:
         digest = hashlib.sha256()
         with path.open("rb") as handle:
@@ -436,9 +516,17 @@ class DeterministicEvidenceCollector:
         self._config = config
         self._evidence_source = evidence_source
 
-    def collect(self, recipe_name: str, recipe: Mapping[str, str]) -> Mapping[str, Any]:
+    def assert_ready(self) -> None:
+        """Fail closed on collector configuration, not on recipe evidence presence."""
+
         if self._evidence_source is None:
             raise BuilderError("COLLECTOR_UNCONFIGURED")
+        checker = getattr(self._evidence_source, "assert_ready", None)
+        if checker is not None:
+            checker()
+
+    def collect(self, recipe_name: str, recipe: Mapping[str, str]) -> Mapping[str, Any]:
+        self.assert_ready()
         bundle = self._evidence_source(recipe_name, recipe)
         if not isinstance(bundle, Mapping) or set(bundle) != self._BUNDLE_FIELDS:
             raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
@@ -464,11 +552,32 @@ class DeterministicEvidenceCollector:
         try:
             from silent_release_controller import ReleaseCandidate
             from silent_release_hardening import ReleasePlan, TrustedCandidateBundle
+            from release_migration_plan import (
+                MigrationPlanError,
+                MigrationPlanRequired,
+            )
 
             candidate = ReleaseCandidate.from_dict(dict(bundle["candidate"]))
-            release_plan = ReleasePlan.from_dict(dict(bundle["release_plan"]))
+            release_plan_payload = dict(bundle["release_plan"])
+            migration_resolver = self._config.migration_plan_resolver
+            if migration_resolver is not None:
+                try:
+                    migration = migration_resolver(
+                        recipe_name,
+                        recipe["candidate_id"],
+                        recipe["source_commit"],
+                    )
+                except MigrationPlanRequired as exc:
+                    raise BuilderError("MIGRATION_PLAN_REQUIRED") from exc
+                except MigrationPlanError as exc:
+                    raise BuilderError("MIGRATION_PLAN_INVALID") from exc
+                if migration is not None:
+                    release_plan_payload["migration"] = migration.to_dict()
+            release_plan = ReleasePlan.from_dict(release_plan_payload)
             checks = dict(raw_checks)
             TrustedCandidateBundle(candidate, release_plan, checks).validate()
+        except BuilderError:
+            raise
         except (ImportError, KeyError, TypeError, ValueError, RuntimeError) as exc:
             raise BuilderError("COLLECTOR_EVIDENCE_INVALID") from exc
         if candidate.source_commit != recipe["source_commit"] or candidate.candidate_id != recipe["candidate_id"]:
@@ -622,14 +731,12 @@ class TrustedCandidateBuilder:
         bound = {"repository": self.config.repository, "task_id": recipe["task_id"], "candidate_id": recipe["candidate_id"], "base_oid": recipe["base_oid"], "source_commit": recipe["source_commit"], "version": recipe["version"], "purpose": recipe["purpose"]}
         if {key: request.get(key) for key in bound} != bound:
             raise BuilderError("REQUEST_SCOPE_INVALID")
-        bundle = dict(self.collector.collect(recipe_name, recipe))
-        candidate = bundle["candidate"]
-        release_plan = bundle["release_plan"]
-        checks = bundle.get("checks")
-        evidence_digests = bundle["evidence_digests"]
-        if not isinstance(checks, Mapping) or set(checks) != set(self.config.required_checks) or not all(value is True for value in checks.values()):
-            raise BuilderError("COLLECTOR_CHECKS_INCOMPLETE")
         if request["operation"] == self.config.readiness_operation:
+            self.collector.assert_ready()
+            # Accessing the identity properties validates that the signer is
+            # configured and usable without invoking the private key.
+            builder_key_id = self.signer.key_id
+            builder_key_generation = self.signer.key_generation
             result = {
                 "schema": self.config.response_schema,
                 "status": "READY",
@@ -637,8 +744,8 @@ class TrustedCandidateBuilder:
                 "recipe": recipe_name,
                 "candidate_id": recipe["candidate_id"],
                 "source_commit": recipe["source_commit"],
-                "builder_key_id": self.signer.key_id,
-                "builder_key_generation": self.signer.key_generation,
+                "builder_key_id": builder_key_id,
+                "builder_key_generation": builder_key_generation,
                 "signature_algorithm": "Ed25519",
                 "signature_purpose": "AIOS_TRUSTED_RELEASE_BUILDER",
                 "policy_sha256": self.config.policy_sha256,
@@ -653,6 +760,15 @@ class TrustedCandidateBuilder:
             )
             self._persist_nonce_ledger()
             return result
+
+        bundle = dict(self.collector.collect(recipe_name, recipe))
+        candidate = bundle["candidate"]
+        release_plan = bundle["release_plan"]
+        checks = bundle.get("checks")
+        evidence_digests = bundle["evidence_digests"]
+        if not isinstance(checks, Mapping) or set(checks) != set(self.config.required_checks) or not all(value is True for value in checks.values()):
+            raise BuilderError("COLLECTOR_CHECKS_INCOMPLETE")
+
         issued = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
         attestation = {
             "builder_attestation_id": "builder-attestation-" + _sha256(recipe_name, recipe["task_id"], candidate["artifact_sha256"])[:24],
@@ -776,6 +892,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 open_source_admission_store
+
+        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 = open_source_admission_store(
+            Path(source_admission_root),
+            repository=config.repository,
+            policy=arguments.policy,
+            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_evidence_materializer.py

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

active-runtime/app/release_migration_plan.py

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

active-runtime/app/release_source_admission.py

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

@@ -11,6 +11,7 @@ WorkingDirectory=@RUNTIME_ROOT@
 Environment=AIOS_BRIDGE_SCHEMA_VERSION=aios.chatgpt-gitops-bridge.v2
 Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
+Environment=AIOS_P0_TASK_PRINCIPAL_WRITES_ENABLED=true
 Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITEA_API_BASE=https://g.cnnetmkt.top/git/api/v1
 Environment=AIOS_GITEA_TOKEN_FILE=%d/gitea-governance-executor-token

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

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

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

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

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

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

deploy/systemd/aios-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]

docs/gitops/openapi-chatgpt-actions.yaml

@@ -410,7 +410,7 @@ paths:
       operationId: commitFilesGoverned
       summary: Submit one atomic governed commit containing multiple files
       security: [{taskPrincipalAuth: []}]
-      x-aios-task-principal: Bridge verifies a fresh aios.task-principal-assertion.v1 and exact task/job/repository/branch/generation/revision/attempt/execution-digest/capability binding; caller JSON cannot provide a principal and missing verification is denied.
+      x-aios-task-principal: Bridge resolves a fresh server-side TaskPrincipal from the authoritative admitted+claimed Task and re-verifies its Owner epoch, then checks exact task/job/repository/branch/generation/capability binding; caller JSON cannot provide a principal and missing verification is denied. An explicitly injected legacy bearer verifier may still be used by a trusted host.
       requestBody:
         required: true
         content:

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_r7_atomic_recovery_and_bridge_sandbox.py

@@ -27,6 +27,7 @@ def test_bridge_unit_keeps_strict_sandbox_and_required_projection():
     ).read_text()
     assert "ProtectSystem=strict" in text
     assert "ProtectHome=true" in text
+    assert "Environment=AIOS_P0_TASK_PRINCIPAL_WRITES_ENABLED=true" in text
     projection = re.search(
         r"^LoadCredentialEncrypted=([^:]+):/etc/credstore\.encrypted/([^\n]+)$",
         text,

tests/test_release_evidence_materializer.py

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

tests/test_release_migration_plan.py

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

tests/test_release_source_admission.py

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

tests/test_silent_release_execution_readiness.py

@@ -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_task_admission_bootstrap_recovery.py

@@ -418,8 +418,10 @@ def test_bridge_persistent_consumer_is_opt_in_and_default_fail_closed():
         ROOT / "active-runtime" / "app" / "aios_chatgpt_bridge_v2_server.py"
     ).read_text(encoding="utf-8")
     assert 'os.environ.get("AIOS_TASK_ADMISSION_EPOCH_ROOT", "").strip()' in source
-    assert "build_task_admission_handlers_from_environment" in source
+    assert "build_task_admission_runtime_from_environment" in source
     assert "task_admission_handlers is None" in source
+    assert "admission_runtime.principal_verifier" in source
+    assert "TaskPrincipal" in source
 
 
 def test_bridge_persistent_host_is_opt_in_and_activation_file_is_fail_closed():
@@ -428,7 +430,7 @@ def test_bridge_persistent_host_is_opt_in_and_activation_file_is_fail_closed():
     assert 'AIOS_TASK_ADMISSION_ACTIVATION_FILE' in text
     assert '/var/lib/aios-gitops/bridge-v2-staging/task-admission-activation.json' in text
     assert 'Path(activation_raw).is_file()' in text
-    assert 'build_task_admission_handlers_from_activation_file' in text
+    assert 'build_task_admission_runtime_from_activation_file' in text
 
 
 def test_host_default_state_database_stays_inside_existing_bridge_state_root():

tests/test_task_principal_auto_recovery.py

@@ -0,0 +1,246 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+if str(APP) not in sys.path:
+    sys.path.insert(0, str(APP))
+
+from aios_task_admission_bootstrap import (  # noqa: E402
+    OwnerDelegationEpoch,
+    TaskAdmissionError,
+    TaskAdmissionGrant,
+    TaskAdmissionStore,
+)
+from aios_task_admission_live_adapters import (  # noqa: E402
+    TaskAdmissionPrincipalProvider,
+)
+
+
+def _grant(*, capabilities=None) -> TaskAdmissionGrant:
+    return TaskAdmissionGrant(
+        grant_id="grant-1",
+        envelope_id="env-1",
+        task_id="task-1",
+        repository="aios/aios",
+        authority_base_oid="a" * 40,
+        target_branch="agent/task-1",
+        generation=1,
+        execution_digest="b" * 64,
+        capabilities=tuple(
+            capabilities
+            or (
+                "task.principal.issue",
+                "writer.lease",
+                "governed.write.file",
+                "governed.write.files",
+            )
+        ),
+        file_allowlist=("active-runtime/app/example.py",),
+        risk="medium",
+        trust_domain="aios.owner",
+        policy_digest="c" * 64,
+        authorization_kind="delegation_epoch",
+        owner_identity="owner",
+        parent_epoch_id="epoch-1",
+        issued_at=100,
+        expires_at=1000,
+        one_time_use=True,
+    )
+
+
+def _epoch(*, capabilities=None) -> OwnerDelegationEpoch:
+    return OwnerDelegationEpoch(
+        epoch_id="epoch-1",
+        owner_identity="owner",
+        trust_domain="aios.owner",
+        repository="aios/aios",
+        allowed_branch_prefixes=("agent/",),
+        allowed_capabilities=frozenset(
+            capabilities
+            or {
+                "task.principal.issue",
+                "writer.lease",
+                "governed.write.file",
+                "governed.write.files",
+            }
+        ),
+        allowed_file_prefixes=("active-runtime/app/",),
+        risk_ceiling="medium",
+        issued_at=1,
+        not_before=1,
+        expires_at=2000,
+        revocation_epoch=0,
+        policy_digest="c" * 64,
+    )
+
+
+class Store:
+    def __init__(self, grant):
+        self.grant = grant
+
+    def get_grant_by_task_id(self, task_id):
+        return self.grant if self.grant and self.grant.task_id == task_id else None
+
+
+class Authorization:
+    def __init__(self, epoch=None, error=None):
+        self.epoch = epoch or _epoch()
+        self.error = error
+        self.calls = 0
+
+    def verify_delegation_epoch(self, epoch_id):
+        self.calls += 1
+        if self.error is not None:
+            raise self.error
+        assert epoch_id == self.epoch.epoch_id
+        return self.epoch
+
+
+class Port:
+    def __init__(self, *, state="CLAIMED", job_id="job-1"):
+        self.state = state
+        self.job_id = job_id
+        self.calls = 0
+
+    def verify_active_task(self, grant):
+        self.calls += 1
+        if self.state not in {"CLAIMED", "RUNNING"}:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_TASK_NOT_CLAIMED", 409, details={"state": self.state}
+            )
+        return (
+            {"task_id": grant.task_id, "state": self.state},
+            {
+                "task_id": grant.task_id,
+                "repository": grant.repository,
+                "target_branch": grant.target_branch,
+                "generation": grant.generation,
+                "external_job_id": self.job_id,
+            },
+        )
+
+
+def _request(**updates):
+    value = {
+        "task_id": "task-1",
+        "job_id": "job-1",
+        "repository": "aios/aios",
+        "base_oid": "a" * 40,
+        "target_branch": "agent/task-1",
+        "generation": 1,
+    }
+    value.update(updates)
+    return value
+
+
+def test_provider_recovers_principal_from_claimed_task_without_caller_bearer():
+    auth = Authorization()
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=auth,
+        task_port=Port(state="CLAIMED"),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    result = provider({"operation": "acquireWriterLease", "headers": {}}, _request())
+    assert result["schema"] == "aios.task-principal-server-attestation.v1"
+    assert result["task_id"] == "task-1"
+    assert result["job_id"] == "job-1"
+    assert result["capability"] == "writer.lease"
+    assert result["task_state"] == "CLAIMED"
+    assert auth.calls == 1
+    assert provider.readiness()["ready"] is True
+
+
+def test_provider_requires_claim_before_writer_lease():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=Authorization(),
+        task_port=Port(state="QUEUED"),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_TASK_NOT_CLAIMED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("job_id", "job-other"),
+        ("repository", "other/repo"),
+        ("target_branch", "agent/other"),
+        ("generation", 2),
+        ("base_oid", "d" * 40),
+    ],
+)
+def test_provider_rejects_request_binding_drift(field, value):
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=Authorization(),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_REQUEST_BINDING_MISMATCH"):
+        provider(
+            {"operation": "acquireWriterLease"},
+            _request(**{field: value}),
+        )
+
+
+def test_provider_reverifies_revocation_on_every_use():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=Authorization(
+            error=TaskAdmissionError("TASK_ADMISSION_EPOCH_REVOKED", 403)
+        ),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_EPOCH_REVOKED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+def test_provider_rejects_missing_execution_capability():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant(capabilities=("task.principal.issue",))),
+        authorization=Authorization(),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_CAPABILITY_DENIED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+def test_provider_does_not_treat_missing_admission_as_owner_reauthorization():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(None),
+        authorization=Authorization(),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_AUTHORIZATION_REQUIRED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+def test_store_recovers_exact_grant_by_task_id(tmp_path):
+    store = TaskAdmissionStore(tmp_path / "admission.sqlite3")
+    grant = _grant()
+    store.save_grant_and_receipt(
+        envelope_id=grant.envelope_id,
+        grant=grant,
+        receipt={"schema": "fixture", "task_id": grant.task_id},
+        created_at=200,
+    )
+    recovered = store.get_grant_by_task_id(grant.task_id)
+    assert recovered == grant
+    assert store.get_grant_by_task_id("task-missing") is None

tests/test_trusted_candidate_builder.py

@@ -29,6 +29,12 @@ from aios_trusted_candidate_builder import (  # noqa: E402
     load_builder_policy,
 )
 from silent_release_controller import ReleaseCandidate, ReleaseState  # noqa: E402
+from release_migration_plan import (  # noqa: E402
+    MigrationPlan,
+    MigrationPlanRequired,
+    RuntimeComponentBinding,
+    StateMigrationContract,
+)
 from silent_release_hardening import (  # noqa: E402
     BuilderAttestation,
     BuilderAttestationKeyRecord,
@@ -313,6 +319,45 @@ def test_readiness_fails_closed_when_signer_is_unavailable() -> None:
         server.handle_request(request, connection=object())
 
 
+def test_readiness_is_independent_from_recipe_evidence_but_build_remains_fail_closed(tmp_path) -> None:
+    policy = load_builder_policy(POLICY_PATH)
+    config = TrustedCandidateBuilderConfig.from_policy(policy)
+    evidence_root = tmp_path / "evidence"
+    evidence_root.mkdir(mode=0o700)
+    authority_git = tmp_path / "repo.git"
+    authority_git.mkdir(mode=0o700)
+
+    source = ServerOwnedEvidenceSource(
+        config,
+        authority_git,
+        evidence_root,
+        expected_owner_uid=None,
+        git_runner=lambda *_args, **_kwargs: SimpleNamespace(
+            stdout="a" * 40 + "\n", returncode=0
+        ),
+    )
+    collector = DeterministicEvidenceCollector(config, evidence_source=source)
+    server, client, _ = make_seam(collector=collector)
+
+    readiness = client.create_request(
+        "V2R2C2",
+        "rc-v2r2c2-d2b5a8339a34",
+        "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6",
+        nonce="v" * 32,
+        operation=server.config.readiness_operation,
+    )
+    assert server.handle_request(readiness, connection=object())["status"] == "READY"
+
+    build = client.create_request(
+        "V2R2C2",
+        "rc-v2r2c2-d2b5a8339a34",
+        "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6",
+        nonce="w" * 32,
+    )
+    with pytest.raises(BuilderError, match="COLLECTOR_EVIDENCE_UNAVAILABLE"):
+        server.handle_request(build, connection=object())
+
+
 def test_unix_client_applies_bounded_timeout_and_fails_closed(monkeypatch) -> None:
     observed = {}
 
@@ -590,3 +635,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)