feat(release): add host-local trusted candidate builder boundary

active-runtime/app/aios_model_api_staging.py

@@ -4,6 +4,7 @@ import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, urllib.
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from baseline_authority import BaselineAuthority, BaselineAuthorityError
+from silent_release_controller import ReleaseCandidate
 from typing import Any
 
 try:
@@ -28,11 +29,18 @@ from orchestrator.action_monitor import ActionMonitor
 from silent_release_hardening import (
     ApprovalKeyRing,
     ApprovalRevocationStore,
+    BuilderAttestation,
+    BuilderAttestationKeyRing,
+    BuilderAttestationVerifier,
     CapabilityAuthorizationError,
+    ReleasePlan,
     SagaConflict,
     SignedApprovalVerifier,
     SignedReleasePlanApproval,
     SilentReleaseActionService,
+    TrustedCandidateBuilderUnavailable,
+    TrustedCandidateBundle,
+    TrustedReleasePolicy,
 )
 try:
     from aios_gitops_job_controller import GitOpsJobController, JobError, ServiceUnavailable, JobStore
@@ -82,13 +90,33 @@ SILENT_RELEASE_STATE_ROOT = Path(os.getenv(
     'AIOS_SILENT_RELEASE_STATE_ROOT',
     '/var/lib/aios-gitops/silent-release',
 ))
-SILENT_RELEASE_EXECUTION_ENABLED = str(os.getenv(
-    'AIOS_SILENT_RELEASE_EXECUTION_ENABLED', 'false'
-)).lower() in {'1', 'true', 'yes', 'on'}
+# This immutable staging API can prepare only a shadow candidate.  An
+# environment value must not turn its HTTP process into a release executor.
+SILENT_RELEASE_EXECUTION_ENABLED = False
 SILENT_RELEASE_PUBLIC_KEYS = Path(os.getenv(
     'AIOS_SILENT_RELEASE_PUBLIC_KEYS',
     '/etc/aios-gitops/silent-release-owner-public-keys.json',
 ))
+TRUSTED_BUILDER_ENABLED = str(os.getenv(
+    'AIOS_TRUSTED_CANDIDATE_BUILDER_ENABLED', 'false'
+)).lower() in {'1', 'true', 'yes', 'on'}
+TRUSTED_BUILDER_POLICY = Path(os.getenv(
+    'AIOS_TRUSTED_CANDIDATE_BUILDER_POLICY', ''
+))
+TRUSTED_BUILDER_CREDENTIALS_DIRECTORY = Path(os.getenv(
+    'AIOS_TRUSTED_CANDIDATE_BUILDER_CREDENTIALS_DIRECTORY',
+    os.getenv('CREDENTIALS_DIRECTORY', ''),
+))
+TRUSTED_BUILDER_SOCKET = Path(os.getenv(
+    'AIOS_TRUSTED_CANDIDATE_BUILDER_SOCKET',
+    '/run/aios-trusted-candidate-builder.sock',
+))
+TRUSTED_BUILDER_PUBLIC_KEYRING = Path(os.getenv(
+    'AIOS_TRUSTED_CANDIDATE_BUILDER_PUBLIC_KEYRING', ''
+))
+TRUSTED_RELEASE_POLICY = Path(os.getenv(
+    'AIOS_SILENT_RELEASE_TRUSTED_POLICY', ''
+))
 monitor = ActionMonitor()
 
 BASE_MODELS = [
@@ -166,6 +194,79 @@ def _file_sha256(path: Path) -> str:
             digest.update(chunk)
     return digest.hexdigest()
 
+
+def _trusted_release_policy_from_file(path: Path) -> TrustedReleasePolicy:
+    payload = json.loads(path.read_text(encoding='utf-8'))
+    required = {
+        'services', 'release_roots', 'cutover_links', 'dropin_paths',
+        'authoritative_git_dirs', 'repository_identities',
+    }
+    if not isinstance(payload, dict) or set(payload) != required:
+        raise ValueError('trusted release policy schema is invalid')
+    values = {}
+    for name in required:
+        raw = payload[name]
+        if not isinstance(raw, list) or not raw or not all(
+            isinstance(item, str) and item for item in raw
+        ):
+            raise ValueError('trusted release policy values are invalid')
+        values[name] = frozenset(raw)
+    return TrustedReleasePolicy(**values)
+
+
+class _HostTrustedCandidateBuilderAdapter:
+    """Maps only a complete host-owned silent-release bundle response.
+
+    The API never manufactures candidate evidence or a builder signature.  A
+    host builder protocol that returns a different response is therefore
+    unavailable, rather than silently downgraded to caller-owned evidence.
+    """
+
+    def __init__(self, client):
+        self._client = client
+
+    def build(self, intent):
+        import secrets
+        from aios_trusted_candidate_builder import BuilderError
+
+        try:
+            version = str(intent['version'])
+            candidate_id = str(intent['candidate_id'])
+            source_commit = str(intent['source_commit'])
+        except (KeyError, TypeError) as exc:
+            raise CapabilityAuthorizationError(
+                'trusted builder requires a release worker request'
+            ) from exc
+        try:
+            response = self._client.build(
+                version,
+                candidate_id,
+                source_commit,
+                nonce=secrets.token_hex(32),
+            )
+        except BuilderError as exc:
+            raise TrustedCandidateBuilderUnavailable(
+                'trusted builder request failed'
+            ) from exc
+        raw_bundle = response.get('silent_release_bundle')
+        if not isinstance(raw_bundle, dict):
+            raise CapabilityAuthorizationError(
+                'trusted builder response is not a silent release bundle'
+            )
+        try:
+            return TrustedCandidateBundle(
+                candidate=ReleaseCandidate.from_dict(raw_bundle['candidate']),
+                release_plan=ReleasePlan.from_dict(raw_bundle['release_plan']),
+                checks=raw_bundle['checks'],
+                attestation=BuilderAttestation.from_dict(
+                    raw_bundle['attestation'], require_fingerprint=True
+                ),
+            )
+        except (KeyError, TypeError, ValueError) as exc:
+            raise CapabilityAuthorizationError(
+                'trusted builder response is invalid'
+            ) from exc
+
 def _build_silent_release_service() -> SilentReleaseActionService:
     verifier=None
     if SILENT_RELEASE_PUBLIC_KEYS.is_file():
@@ -177,10 +278,86 @@ def _build_silent_release_service() -> SilentReleaseActionService:
         )
     policy_path=WS / 'config' / 'gitops' / 'silent-release-policy.json'
     policy_sha=_file_sha256(policy_path) if policy_path.is_file() else ''
+    candidate_builder = None
+    builder_attestation_verifier = None
+    trusted_policy = None
+    builder_readiness_probe = 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.
+        if not (
+            TRUSTED_BUILDER_POLICY.is_file()
+            and TRUSTED_BUILDER_CREDENTIALS_DIRECTORY.is_dir()
+            and TRUSTED_BUILDER_PUBLIC_KEYRING.is_file()
+            and TRUSTED_RELEASE_POLICY.is_file()
+        ):
+            raise CapabilityAuthorizationError(
+                'trusted builder configuration is incomplete'
+            )
+        try:
+            from aios_trusted_candidate_builder import (
+                BuilderKeyring,
+                TrustedCandidateBuilderConfig,
+                UnixTrustedCandidateBuilderClient,
+                load_builder_policy,
+                readiness_probe,
+            )
+            builder_policy = load_builder_policy(TRUSTED_BUILDER_POLICY)
+            builder_config = TrustedCandidateBuilderConfig.from_policy(
+                builder_policy
+            )
+            client_hmac = (
+                TRUSTED_BUILDER_CREDENTIALS_DIRECTORY / 'builder_hmac'
+            ).read_bytes().strip()
+            client_keyring = BuilderKeyring(builder_hmac=client_hmac)
+            client = UnixTrustedCandidateBuilderClient(
+                builder_config,
+                client_keyring,
+                socket_path=TRUSTED_BUILDER_SOCKET,
+            )
+            candidate_builder = _HostTrustedCandidateBuilderAdapter(client)
+            builder_attestation_verifier = BuilderAttestationVerifier(
+                BuilderAttestationKeyRing.from_file(
+                    TRUSTED_BUILDER_PUBLIC_KEYRING
+                )
+            )
+            trusted_policy = _trusted_release_policy_from_file(
+                TRUSTED_RELEASE_POLICY
+            )
+
+            def builder_readiness_probe():
+                if not TRUSTED_BUILDER_SOCKET.exists():
+                    return {'status': 'UNAVAILABLE'}
+                remote = dict(readiness_probe(client))
+                if remote.get('policy_sha256') != builder_config.policy_sha256:
+                    return {'status': 'UNAVAILABLE'}
+                builder_attestation_verifier.keyring.require_active_identity(
+                    key_id=str(remote.get('builder_key_id') or ''),
+                    generation=remote.get('builder_key_generation'),
+                    algorithm=str(remote.get('signature_algorithm') or ''),
+                    purpose=str(remote.get('signature_purpose') or ''),
+                )
+                return {
+                    **remote,
+                    'hmac_credential_loaded': True,
+                    'peer_verification_configured': True,
+                    'public_key_trusted': True,
+                    'policy_loaded': True,
+                }
+
+        except Exception as exc:
+            raise CapabilityAuthorizationError(
+                'trusted builder configuration is invalid'
+            ) from exc
     return SilentReleaseActionService(
         SILENT_RELEASE_STATE_ROOT,
         execution_enabled=SILENT_RELEASE_EXECUTION_ENABLED,
         approval_verifier=verifier,
+        candidate_builder=candidate_builder,
+        builder_attestation_verifier=builder_attestation_verifier,
+        trusted_policy=trusted_policy,
+        builder_readiness_probe=builder_readiness_probe,
+        allow_legacy_candidate_input=False,
         expected_repository=REPOSITORY_IDENTITY,
         policy_sha256=policy_sha,
     )
@@ -466,8 +643,11 @@ class H(BaseHTTPRequestHandler):
         q=urllib.parse.parse_qs(u.query)
         try:
             if path in ('/health','/api/health'):
+                builder_readiness = silent_release_service().get_builder_readiness()
                 return self.sendj({
                     'status': 'ok',
+                    'process_liveness': 'PASS',
+                    'builder_readiness': builder_readiness,
                     'service': os.getenv('AIOS_GITOPS_MODEL_API_SERVICE', 'aios-gitops-model-api-v2r2c1-staging.service'),
                     'workspace': str(WS),
                     'version': 'V2R2C1',
@@ -706,8 +886,15 @@ class H(BaseHTTPRequestHandler):
             if path == '/createReleaseCandidate':
                 a=self.need('release:candidate')
                 if not a: return
+                service = silent_release_service()
+                readiness = service.get_builder_readiness()
+                if readiness.get('status') != 'READY':
+                    return self.sendj({
+                        'error': 'TRUSTED_BUILDER_UNAVAILABLE',
+                        'builder_readiness': readiness,
+                    }, 503)
                 return self.sendj(
-                    silent_release_service().create_release_candidate(data)
+                    service.create_release_candidate(data)
                 )
             if path == '/createReleaseIntent':
                 a=self.need('release:candidate')
@@ -747,6 +934,13 @@ class H(BaseHTTPRequestHandler):
             error=redact(e)
             monitor.record_failure('do_POST', error, {'path':path})
             return self.sendj({'error':error},403)
+        except TrustedCandidateBuilderUnavailable:
+            monitor.record_failure(
+                'do_POST', 'TRUSTED_BUILDER_UNAVAILABLE', {'path': path}
+            )
+            return self.sendj(
+                {'error': 'TRUSTED_BUILDER_UNAVAILABLE'}, 503
+            )
         except SagaConflict as e:
             error=redact(e)
             monitor.record_failure('do_POST', error, {'path':path})

active-runtime/app/aios_trusted_candidate_builder.py

@@ -0,0 +1,794 @@
+#!/usr/bin/env python3
+"""Fail-closed AF_UNIX builder for server-owned trusted candidate evidence."""
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import socket
+import stat
+import struct
+import subprocess
+import time
+from datetime import datetime, timezone
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Mapping, Protocol
+
+
+class BuilderError(RuntimeError):
+    """A stable, non-sensitive error code returned by the builder boundary."""
+
+
+def _canonical_json(value: Mapping[str, Any]) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+
+
+def _sha256(*parts: str) -> str:
+    return hashlib.sha256("\0".join(parts).encode("utf-8")).hexdigest()
+
+
+@dataclass(frozen=True)
+class PeerIdentity:
+    pid: int
+    uid: int
+    gid: int
+
+
+class DomainPurposeSigner(Protocol):
+    """A signer deliberately limited to the builder's one domain purpose."""
+
+    @property
+    def key_id(self) -> str:
+        ...
+
+    @property
+    def key_generation(self) -> int:
+        ...
+
+    def sign_for_purpose(self, purpose: str, payload: bytes) -> bytes:
+        ...
+
+
+class CredentialPathEd25519Signer:
+    """Reads a systemd-mounted credential path; never exports or logs key bytes."""
+
+    def __init__(self, credential_path: Path, *, key_id: str, key_generation: int, allowed_purpose: str):
+        self._credential_path = Path(credential_path)
+        self._key_id = str(key_id)
+        self._key_generation = key_generation
+        self._allowed_purpose = str(allowed_purpose)
+        if not self._key_id or not isinstance(key_generation, int) or key_generation < 1:
+            raise BuilderError("SIGNING_IDENTITY_INVALID")
+        if not self._credential_path.is_file():
+            raise BuilderError("SIGNING_CREDENTIAL_UNAVAILABLE")
+        try:
+            from cryptography.hazmat.primitives.serialization import load_pem_private_key
+            from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+            self._private_key = load_pem_private_key(self._credential_path.read_bytes(), password=None)
+            if not isinstance(self._private_key, Ed25519PrivateKey):
+                raise TypeError("signing key is not Ed25519")
+        except Exception as exc:  # do not expose parse/key material details
+            raise BuilderError("SIGNING_CREDENTIAL_INVALID") from exc
+
+    @property
+    def key_id(self) -> str:
+        return self._key_id
+
+    @property
+    def key_generation(self) -> int:
+        return self._key_generation
+
+    def sign_for_purpose(self, purpose: str, payload: bytes) -> bytes:
+        if purpose != self._allowed_purpose:
+            raise BuilderError("SIGNING_PURPOSE_DENIED")
+        return self._private_key.sign(payload)
+
+
+@dataclass(frozen=True)
+class BuilderKeyring:
+    builder_hmac: bytes
+    signing_private_key_path: Path | None = None
+
+    def __post_init__(self) -> None:
+        if len(self.builder_hmac) < 32:
+            raise BuilderError("BUILDER_HMAC_INVALID")
+
+    @classmethod
+    def from_credentials_directory(cls, credentials_directory: Path) -> "BuilderKeyring":
+        root = Path(credentials_directory)
+        hmac_path = root / "builder_hmac"
+        signing_path = root / "signing_private_key"
+        try:
+            secret = hmac_path.read_bytes().strip()
+        except OSError as exc:
+            raise BuilderError("BUILDER_HMAC_UNAVAILABLE") from exc
+        if not signing_path.is_file():
+            raise BuilderError("SIGNING_CREDENTIAL_UNAVAILABLE")
+        return cls(builder_hmac=secret, signing_private_key_path=signing_path)
+
+
+@dataclass(frozen=True)
+class TrustedCandidateBuilderConfig:
+    policy_schema: str
+    repository: str
+    request_schema: str
+    response_schema: str
+    operation: str
+    readiness_operation: str
+    maximum_request_age_seconds: int
+    request_timeout_seconds: int
+    allowed_peer_uid: int
+    allowed_peer_cgroups: tuple[str, ...]
+    allowed_peer_executable_prefix: str
+    allowed_peer_runtime_prefix: str
+    allowed_peer_runtime_suffix: str
+    required_checks: tuple[str, ...]
+    evidence_layout: Mapping[str, str]
+    recipes: Mapping[str, Mapping[str, str]]
+    governed_recipes: Mapping[str, Mapping[str, str]]
+    policy_sha256: str
+
+    @classmethod
+    def from_policy(cls, policy: Mapping[str, Any]) -> "TrustedCandidateBuilderConfig":
+        required = {
+            "schema", "repository", "request_schema", "response_schema", "operation",
+            "readiness_operation", "maximum_request_age_seconds", "request_timeout_seconds",
+            "allowed_peer_uid", "allowed_peer_cgroups", "allowed_peer_executable_prefix",
+            "allowed_peer_runtime_prefix", "allowed_peer_runtime_suffix",
+            "required_checks", "evidence_layout", "recipes", "governed_recipes",
+        }
+        if set(policy) != required or policy.get("schema") != "aios.trusted-candidate-builder-policy.v1":
+            raise BuilderError("POLICY_SCHEMA_INVALID")
+        recipes = policy.get("recipes")
+        if not isinstance(recipes, Mapping) or set(recipes) != {"V2R2C2", "V2R2C3"}:
+            raise BuilderError("POLICY_RECIPES_INVALID")
+        c2, c3 = recipes["V2R2C2"], recipes["V2R2C3"]
+        if not isinstance(c2, Mapping) or set(c2) != {"mode", "task_id", "candidate_id", "base_oid", "source_commit", "version", "purpose"} or c2.get("mode") != "fixed":
+            raise BuilderError("POLICY_V2R2C2_INVALID")
+        if not isinstance(c3, Mapping) or set(c3) != {"mode", "derives_from", "version", "purpose"} or c3.get("mode") != "governed-derived-only" or c3.get("derives_from") != "V2R2C2":
+            raise BuilderError("POLICY_V2R2C3_INVALID")
+        if (
+            policy["repository"] != "aios/aios"
+            or policy["operation"] != "BuildTrustedCandidateV1"
+            or policy["readiness_operation"] != "ProbeTrustedCandidateBuilderV1"
+        ):
+            raise BuilderError("POLICY_SCOPE_INVALID")
+        if not isinstance(policy["maximum_request_age_seconds"], int) or not 1 <= policy["maximum_request_age_seconds"] <= 300:
+            raise BuilderError("POLICY_FRESHNESS_INVALID")
+        if not isinstance(policy["request_timeout_seconds"], int) or not 1 <= policy["request_timeout_seconds"] <= 120:
+            raise BuilderError("POLICY_TIMEOUT_INVALID")
+        if not isinstance(policy["allowed_peer_uid"], int) or policy["allowed_peer_uid"] != 1000:
+            raise BuilderError("POLICY_PEER_UID_INVALID")
+        cgroups, checks = policy["allowed_peer_cgroups"], policy["required_checks"]
+        if not isinstance(cgroups, list) or not cgroups or not all(isinstance(value, str) and value.startswith("/") for value in cgroups):
+            raise BuilderError("POLICY_CGROUPS_INVALID")
+        for field in (
+            "allowed_peer_executable_prefix",
+            "allowed_peer_runtime_prefix",
+            "allowed_peer_runtime_suffix",
+        ):
+            if not isinstance(policy[field], str) or not policy[field].startswith("/"):
+                raise BuilderError("POLICY_PROCESS_IDENTITY_INVALID")
+        if not isinstance(checks, list) or not checks or len(set(checks)) != len(checks) or not all(isinstance(value, str) for value in checks):
+            raise BuilderError("POLICY_CHECKS_INVALID")
+        layout = policy["evidence_layout"]
+        evidence_names = {
+            "artifact", "manifest", "package_closure", "tests", "canary",
+            "gitea", "rollback", "policy", "secret_scan",
+        }
+        if (
+            not isinstance(layout, Mapping)
+            or set(layout) != evidence_names
+            or not all(
+                isinstance(value, str)
+                and value.startswith("{version}/")
+                and ".." not in Path(value).parts
+                and not Path(value).is_absolute()
+                for value in layout.values()
+            )
+        ):
+            raise BuilderError("POLICY_EVIDENCE_LAYOUT_INVALID")
+        return cls(
+            policy_schema=str(policy["schema"]), repository=str(policy["repository"]),
+            request_schema=str(policy["request_schema"]), response_schema=str(policy["response_schema"]),
+            operation=str(policy["operation"]), readiness_operation=str(policy["readiness_operation"]),
+            maximum_request_age_seconds=policy["maximum_request_age_seconds"],
+            request_timeout_seconds=policy["request_timeout_seconds"],
+            allowed_peer_uid=policy["allowed_peer_uid"], allowed_peer_cgroups=tuple(cgroups),
+            allowed_peer_executable_prefix=policy["allowed_peer_executable_prefix"],
+            allowed_peer_runtime_prefix=policy["allowed_peer_runtime_prefix"],
+            allowed_peer_runtime_suffix=policy["allowed_peer_runtime_suffix"],
+            required_checks=tuple(checks), evidence_layout=dict(layout),
+            recipes={name: dict(value) for name, value in recipes.items()},
+            governed_recipes={name: dict(value) for name, value in policy["governed_recipes"].items()} if isinstance(policy["governed_recipes"], Mapping) else _invalid_governed_recipes(),
+            policy_sha256=hashlib.sha256(_canonical_json(dict(policy))).hexdigest(),
+        )
+
+    def fixed_recipe(self, name: str) -> Mapping[str, str]:
+        recipe = self.recipes.get(name)
+        if recipe is None:
+            raise BuilderError("RECIPE_UNKNOWN")
+        if recipe["mode"] == "governed-derived-only":
+            derived = self.governed_recipes.get(name)
+            required = {"mode", "task_id", "candidate_id", "base_oid", "source_commit", "version", "purpose", "derived_from_v2r2c2_attestation", "builder_engineering_oid"}
+            if not isinstance(derived, Mapping) or set(derived) != required or derived.get("mode") != "governed-derived" or derived.get("version") != name or derived.get("purpose") != recipe.get("purpose"):
+                raise BuilderError("GOVERNED_V2R2C3_DERIVATION_REQUIRED")
+            for name in ("base_oid", "source_commit", "builder_engineering_oid"):
+                value = derived.get(name)
+                if not isinstance(value, str) or len(value) != 40 or any(char not in "0123456789abcdef" for char in value.lower()):
+                    raise BuilderError("GOVERNED_V2R2C3_DERIVATION_REQUIRED")
+            return dict(derived)
+        return recipe
+
+    def resolve_recipe(self, version: str, candidate_id: str, source_commit: str) -> Mapping[str, str]:
+        recipe = self.fixed_recipe(version)
+        if {"version": recipe.get("version"), "candidate_id": recipe.get("candidate_id"), "source_commit": recipe.get("source_commit")} != {"version": version, "candidate_id": candidate_id, "source_commit": source_commit}:
+            raise BuilderError("REQUEST_RELEASE_BINDING_INVALID")
+        return recipe
+
+
+def _invalid_governed_recipes() -> Mapping[str, Mapping[str, str]]:
+    raise BuilderError("POLICY_GOVERNED_RECIPES_INVALID")
+
+
+def load_builder_policy(path: Path) -> Mapping[str, Any]:
+    try:
+        decoded = json.loads(Path(path).read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise BuilderError("POLICY_UNAVAILABLE") from exc
+    if not isinstance(decoded, Mapping):
+        raise BuilderError("POLICY_SCHEMA_INVALID")
+    TrustedCandidateBuilderConfig.from_policy(decoded)
+    return decoded
+
+
+class ServerOwnedEvidenceSource:
+    """Recomputes a fixed on-host evidence closure without caller-owned paths."""
+
+    def __init__(
+        self,
+        config: TrustedCandidateBuilderConfig,
+        authority_git_dir: Path,
+        evidence_root: Path,
+        *,
+        expected_owner_uid: int | None = 0,
+        git_runner: Callable[..., Any] = subprocess.run,
+    ):
+        self._config = config
+        self._authority_git_dir = Path(authority_git_dir)
+        self._evidence_root = Path(evidence_root)
+        self._expected_owner_uid = expected_owner_uid
+        self._git_runner = git_runner
+
+    def _regular_path(self, relative: str) -> Path:
+        root = self._evidence_root.resolve(strict=True)
+        candidate = self._evidence_root / relative
+        current = self._evidence_root
+        for part in Path(relative).parts:
+            current = current / part
+            if current.is_symlink():
+                raise BuilderError("COLLECTOR_SYMLINK_DENIED")
+        try:
+            resolved = candidate.resolve(strict=True)
+            resolved.relative_to(root)
+            metadata = resolved.stat(follow_symlinks=False)
+        except (OSError, ValueError) as exc:
+            raise BuilderError("COLLECTOR_EVIDENCE_UNAVAILABLE") from exc
+        if not stat.S_ISREG(metadata.st_mode):
+            raise BuilderError("COLLECTOR_SPECIAL_FILE_DENIED")
+        if os.name != "nt" and metadata.st_mode & 0o022:
+            raise BuilderError("COLLECTOR_EVIDENCE_MODE_DENIED")
+        if (
+            os.name != "nt"
+            and
+            self._expected_owner_uid is not None
+            and metadata.st_uid != self._expected_owner_uid
+        ):
+            raise BuilderError("COLLECTOR_EVIDENCE_OWNER_DENIED")
+        return resolved
+
+    def _digest(self, 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 _json(self, path: Path) -> Mapping[str, Any]:
+        try:
+            if path.stat().st_size > 2 * 1024 * 1024:
+                raise BuilderError("COLLECTOR_EVIDENCE_SIZE_INVALID")
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID") from exc
+        if not isinstance(value, Mapping):
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
+        return value
+
+    def __call__(self, recipe_name: str, recipe: Mapping[str, str]) -> Mapping[str, Any]:
+        try:
+            tree_result = self._git_runner(
+                ["git", f"--git-dir={self._authority_git_dir}", "rev-parse", f"{recipe['source_commit']}^{{tree}}"],
+                check=True, capture_output=True, text=True, timeout=15,
+                env={"PATH": os.environ.get("PATH", "")},
+            )
+            ancestry_result = self._git_runner(
+                ["git", f"--git-dir={self._authority_git_dir}", "merge-base", "--is-ancestor", recipe["base_oid"], recipe["source_commit"]],
+                check=False, capture_output=True, text=True, timeout=15,
+                env={"PATH": os.environ.get("PATH", "")},
+            )
+            tree_oid = tree_result.stdout.strip().lower()
+        except (OSError, subprocess.SubprocessError) as exc:
+            raise BuilderError("COLLECTOR_EVIDENCE_UNAVAILABLE") from exc
+        if getattr(ancestry_result, "returncode", 1) != 0:
+            raise BuilderError("COLLECTOR_GIT_MISMATCH")
+
+        receipt = self._json(self._regular_path(f"{recipe_name}/receipt.json"))
+        if set(receipt) != {"schema", "version", "candidate", "release_plan"}:
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
+        if (
+            receipt.get("schema") != "aios.trusted-candidate-builder.evidence.v1"
+            or receipt.get("version") != recipe_name
+        ):
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
+        candidate = receipt.get("candidate")
+        release_plan = receipt.get("release_plan")
+        if not isinstance(candidate, Mapping) or not isinstance(release_plan, Mapping):
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
+        if (
+            candidate.get("candidate_id") != recipe["candidate_id"]
+            or candidate.get("source_commit") != recipe["source_commit"]
+            or candidate.get("tree_oid") != tree_oid
+        ):
+            raise BuilderError("COLLECTOR_GIT_MISMATCH")
+        git_consistency = candidate.get("git_consistency")
+        if (
+            not isinstance(git_consistency, Mapping)
+            or git_consistency.get("consistent") is not True
+            or git_consistency.get("source_commit") != recipe["source_commit"]
+            or git_consistency.get("tree_oid") != tree_oid
+            or git_consistency.get("repository") != self._config.repository
+        ):
+            raise BuilderError("COLLECTOR_GIT_MISMATCH")
+
+        paths = {
+            name: self._regular_path(template.format(version=recipe_name))
+            for name, template in self._config.evidence_layout.items()
+        }
+        deployment = release_plan.get("deployment")
+        if not isinstance(deployment, Mapping):
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
+        digests = {name: self._digest(path) for name, path in paths.items()}
+        if (
+            digests["artifact"] != candidate.get("artifact_sha256")
+            or digests["manifest"] != candidate.get("manifest_sha256")
+            or digests["package_closure"] != deployment.get("package_closure_sha256")
+            or digests["policy"] != release_plan.get("policy_sha256")
+            or digests["policy"] != deployment.get("policy_sha256")
+        ):
+            raise BuilderError("COLLECTOR_DIGEST_MISMATCH")
+
+        tests = self._json(paths["tests"])
+        canary = self._json(paths["canary"])
+        gitea = self._json(paths["gitea"])
+        rollback = self._json(paths["rollback"])
+        secret_scan = self._json(paths["secret_scan"])
+        if tests != candidate.get("tests") or tests.get("passed") is not True:
+            raise BuilderError("COLLECTOR_TESTS_MISMATCH")
+        if canary != candidate.get("canary") or canary.get("passed") is not True:
+            raise BuilderError("COLLECTOR_CANARY_MISMATCH")
+        if (
+            digests["tests"] != hashlib.sha256(_canonical_json(dict(tests))).hexdigest()
+            or digests["canary"] != hashlib.sha256(_canonical_json(dict(canary))).hexdigest()
+        ):
+            raise BuilderError("COLLECTOR_EVIDENCE_CANONICALIZATION_MISMATCH")
+        if (
+            gitea != candidate.get("gitea_consistency")
+            or gitea.get("consistent") is not True
+            or gitea.get("repository") != self._config.repository
+            or gitea.get("source_commit") != recipe["source_commit"]
+            or gitea.get("tree_oid") != tree_oid
+            or gitea.get("remote_ref_oid") != recipe["source_commit"]
+        ):
+            raise BuilderError("COLLECTOR_GITEA_MISMATCH")
+        if (
+            candidate.get("rollback_ready") is not True
+            or rollback.get("ready") is not True
+            or rollback.get("source_commit") != recipe["source_commit"]
+            or rollback.get("rollback_sha256") != deployment.get("rollback_sha256")
+        ):
+            raise BuilderError("COLLECTOR_ROLLBACK_MISMATCH")
+        if (
+            digests["gitea"] != hashlib.sha256(_canonical_json(dict(gitea))).hexdigest()
+            or digests["rollback"] != hashlib.sha256(_canonical_json(dict(rollback))).hexdigest()
+        ):
+            raise BuilderError("COLLECTOR_EVIDENCE_CANONICALIZATION_MISMATCH")
+        if (
+            secret_scan.get("passed") is not True
+            or secret_scan.get("artifact_sha256") != candidate.get("artifact_sha256")
+        ):
+            raise BuilderError("COLLECTOR_SECRET_SCAN_MISMATCH")
+        checks = {name: True for name in self._config.required_checks}
+        return {
+            "candidate": dict(candidate),
+            "release_plan": dict(release_plan),
+            "checks": checks,
+            "evidence_digests": digests,
+        }
+
+
+class DeterministicEvidenceCollector:
+    """Validates a server-owned source; request fields never influence collection."""
+
+    _BUNDLE_FIELDS = {"candidate", "release_plan", "checks", "evidence_digests"}
+
+    def __init__(self, config: TrustedCandidateBuilderConfig, *, evidence_source: Callable[[str, Mapping[str, str]], Mapping[str, Any]] | None = None):
+        self._config = config
+        self._evidence_source = evidence_source
+
+    def collect(self, recipe_name: str, recipe: Mapping[str, str]) -> Mapping[str, Any]:
+        if self._evidence_source is None:
+            raise BuilderError("COLLECTOR_UNCONFIGURED")
+        bundle = self._evidence_source(recipe_name, recipe)
+        if not isinstance(bundle, Mapping) or set(bundle) != self._BUNDLE_FIELDS:
+            raise BuilderError("COLLECTOR_EVIDENCE_INVALID")
+        raw_checks = bundle.get("checks")
+        if (
+            not isinstance(raw_checks, Mapping)
+            or set(raw_checks) != set(self._config.required_checks)
+            or any(raw_checks.get(name) is not True for name in self._config.required_checks)
+        ):
+            raise BuilderError("COLLECTOR_CHECKS_INCOMPLETE")
+        evidence_digests = bundle.get("evidence_digests")
+        if (
+            not isinstance(evidence_digests, Mapping)
+            or set(evidence_digests) != set(self._config.evidence_layout)
+            or any(
+                not isinstance(value, str)
+                or len(value) != 64
+                or any(char not in "0123456789abcdef" for char in value.lower())
+                for value in evidence_digests.values()
+            )
+        ):
+            raise BuilderError("COLLECTOR_EVIDENCE_DIGESTS_INVALID")
+        try:
+            from silent_release_controller import ReleaseCandidate
+            from silent_release_hardening import ReleasePlan, TrustedCandidateBundle
+
+            candidate = ReleaseCandidate.from_dict(dict(bundle["candidate"]))
+            release_plan = ReleasePlan.from_dict(dict(bundle["release_plan"]))
+            checks = dict(raw_checks)
+            TrustedCandidateBundle(candidate, release_plan, checks).validate()
+        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"]:
+            raise BuilderError("COLLECTOR_SOURCE_MISMATCH")
+        return {
+            "candidate": candidate.to_dict(),
+            "release_plan": release_plan.to_dict(),
+            "checks": checks,
+            "evidence_digests": dict(evidence_digests),
+        }
+
+
+def _extract_peer_identity(connection: socket.socket) -> PeerIdentity:
+    if not hasattr(socket, "SO_PEERCRED"):
+        raise BuilderError("SO_PEERCRED_UNAVAILABLE")
+    raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))
+    if len(raw) != struct.calcsize("3i"):
+        raise BuilderError("SO_PEERCRED_INVALID")
+    return PeerIdentity(*struct.unpack("3i", raw))
+
+
+def _resolve_cgroups(pid: int) -> tuple[str, ...]:
+    try:
+        return tuple(
+            line.rsplit(":", 1)[1].strip()
+            for line in Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8").splitlines()
+            if ":" in line and line.rsplit(":", 1)[1].strip().startswith("/")
+        )
+    except OSError as exc:
+        raise BuilderError("PEER_CGROUP_UNAVAILABLE") from exc
+
+
+def _resolve_process_identity(pid: int) -> tuple[str, tuple[str, ...]]:
+    try:
+        executable = os.path.realpath(f"/proc/{pid}/exe")
+        argv = tuple(
+            item.decode("utf-8", "strict")
+            for item in Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")
+            if item
+        )
+    except (OSError, UnicodeDecodeError) as exc:
+        raise BuilderError("PEER_PROCESS_UNAVAILABLE") from exc
+    if not executable or not argv:
+        raise BuilderError("PEER_PROCESS_UNAVAILABLE")
+    return executable, argv
+
+
+class TrustedCandidateBuilder:
+    def __init__(
+        self, config: TrustedCandidateBuilderConfig, keyring: BuilderKeyring, *,
+        collector: DeterministicEvidenceCollector, signer: DomainPurposeSigner,
+        peer_identity_resolver: Callable[[socket.socket], PeerIdentity] = _extract_peer_identity,
+        cgroup_resolver: Callable[[int], tuple[str, ...] | str] = _resolve_cgroups,
+        process_identity_resolver: Callable[[int], tuple[str, tuple[str, ...]]] = _resolve_process_identity,
+        clock: Callable[[], float] = time.time, nonce_ledger_path: Path | None = None,
+    ) -> None:
+        self.config, self.keyring, self.collector, self.signer = config, keyring, collector, signer
+        self.peer_identity_resolver, self.cgroup_resolver, self.clock = peer_identity_resolver, cgroup_resolver, clock
+        self.process_identity_resolver = process_identity_resolver
+        self._nonce_ledger_path = Path(nonce_ledger_path) if nonce_ledger_path else None
+        self._nonce_ledger: dict[str, tuple[bytes, dict[str, Any]]] = self._load_nonce_ledger()
+
+    def _load_nonce_ledger(self) -> dict[str, tuple[bytes, dict[str, Any]]]:
+        if self._nonce_ledger_path is None or not self._nonce_ledger_path.exists():
+            return {}
+        if self._nonce_ledger_path.is_symlink():
+            raise BuilderError("NONCE_LEDGER_INVALID")
+        try:
+            raw = json.loads(self._nonce_ledger_path.read_text(encoding="utf-8"))
+            return {nonce: (bytes.fromhex(value["request_sha256"]), dict(value["result"])) for nonce, value in raw.items()}
+        except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
+            raise BuilderError("NONCE_LEDGER_INVALID") from exc
+
+    def _persist_nonce_ledger(self) -> None:
+        if self._nonce_ledger_path is None:
+            return
+        try:
+            self._nonce_ledger_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            temporary = self._nonce_ledger_path.with_suffix(".tmp")
+            payload = {nonce: {"request_sha256": digest.hex(), "result": result} for nonce, (digest, result) in self._nonce_ledger.items()}
+            with temporary.open("wb") as handle:
+                handle.write(_canonical_json(payload))
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.chmod(temporary, 0o600)
+            os.replace(temporary, self._nonce_ledger_path)
+            if os.name != "nt":
+                descriptor = os.open(self._nonce_ledger_path.parent, os.O_RDONLY)
+                try:
+                    os.fsync(descriptor)
+                finally:
+                    os.close(descriptor)
+        except OSError as exc:
+            raise BuilderError("NONCE_LEDGER_PERSIST_FAILED") from exc
+
+    def _verify_peer(self, connection: socket.socket) -> None:
+        peer = self.peer_identity_resolver(connection)
+        if peer.uid != self.config.allowed_peer_uid or peer.pid <= 0:
+            raise BuilderError("PEER_IDENTITY_DENIED")
+        resolved = self.cgroup_resolver(peer.pid)
+        cgroups = (resolved,) if isinstance(resolved, str) else tuple(resolved)
+        if not cgroups or not any(item in self.config.allowed_peer_cgroups for item in cgroups):
+            raise BuilderError("PEER_CGROUP_DENIED")
+        executable, argv = self.process_identity_resolver(peer.pid)
+        if not executable.startswith(self.config.allowed_peer_executable_prefix):
+            raise BuilderError("PEER_PROCESS_DENIED")
+        runtime_arguments = [
+            value
+            for value in argv
+            if value.startswith(self.config.allowed_peer_runtime_prefix)
+            and value.endswith(self.config.allowed_peer_runtime_suffix)
+        ]
+        if len(runtime_arguments) != 1:
+            raise BuilderError("PEER_PROCESS_DENIED")
+
+    def handle_request(self, request: Mapping[str, Any], *, connection: socket.socket) -> dict[str, Any]:
+        if not isinstance(request, Mapping):
+            raise BuilderError("REQUEST_SCHEMA_INVALID")
+        expected_keys = {"schema", "operation", "recipe", "repository", "task_id", "candidate_id", "base_oid", "source_commit", "version", "purpose", "issued_at", "nonce", "hmac"}
+        if (
+            set(request) != expected_keys
+            or request.get("schema") != self.config.request_schema
+            or request.get("operation") not in {
+                self.config.operation,
+                self.config.readiness_operation,
+            }
+        ):
+            raise BuilderError("REQUEST_SCHEMA_INVALID")
+        supplied_mac = request.get("hmac")
+        unsigned = {key: value for key, value in request.items() if key != "hmac"}
+        if not isinstance(supplied_mac, str) or not hmac.compare_digest(hmac.new(self.keyring.builder_hmac, _canonical_json(unsigned), hashlib.sha256).hexdigest(), supplied_mac):
+            raise BuilderError("REQUEST_HMAC_INVALID")
+        issued_at, nonce = request.get("issued_at"), request.get("nonce")
+        if not isinstance(issued_at, int) or not isinstance(nonce, str) or len(nonce) < 16 or len(nonce) > 256:
+            raise BuilderError("REQUEST_SCHEMA_INVALID")
+        self._verify_peer(connection)
+        request_bytes = _canonical_json(dict(request))
+        replay = self._nonce_ledger.get(nonce)
+        if replay is not None:
+            if hmac.compare_digest(replay[0], hashlib.sha256(request_bytes).digest()):
+                return dict(replay[1])
+            raise BuilderError("REPLAY_CONFLICT")
+        if abs(int(self.clock()) - issued_at) > self.config.maximum_request_age_seconds:
+            raise BuilderError("REQUEST_STALE")
+        recipe_name = request.get("recipe")
+        if not isinstance(recipe_name, str):
+            raise BuilderError("REQUEST_SCHEMA_INVALID")
+        recipe = self.config.resolve_recipe(str(request.get("version") or ""), str(request.get("candidate_id") or ""), str(request.get("source_commit") or ""))
+        if recipe_name != recipe.get("version"):
+            raise BuilderError("REQUEST_RELEASE_BINDING_INVALID")
+        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:
+            result = {
+                "schema": self.config.response_schema,
+                "status": "READY",
+                "repository": self.config.repository,
+                "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,
+                "signature_algorithm": "Ed25519",
+                "signature_purpose": "builder_attestation",
+                "policy_sha256": self.config.policy_sha256,
+                "collector_ready": True,
+                "signer_ready": True,
+            }
+            if len(self._nonce_ledger) >= 2048:
+                self._nonce_ledger.pop(next(iter(self._nonce_ledger)))
+            self._nonce_ledger[nonce] = (
+                hashlib.sha256(request_bytes).digest(),
+                dict(result),
+            )
+            self._persist_nonce_ledger()
+            return result
+        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],
+            "builder_key_id": self.signer.key_id,
+            "builder_key_generation": self.signer.key_generation,
+            "signature_algorithm": "Ed25519",
+            "signature_purpose": "builder_attestation",
+            "source_commit": candidate["source_commit"], "tree_oid": candidate["tree_oid"],
+            "artifact_sha256": candidate["artifact_sha256"], "manifest_sha256": candidate["manifest_sha256"],
+            "package_closure_sha256": release_plan["deployment"]["package_closure_sha256"],
+            "test_evidence_sha256": evidence_digests["tests"],
+            "canary_evidence_sha256": evidence_digests["canary"],
+            "gitea_evidence_sha256": evidence_digests["gitea"],
+            "rollback_evidence_sha256": evidence_digests["rollback"],
+            "policy_evidence_sha256": evidence_digests["policy"],
+            "secret_scan_evidence_sha256": evidence_digests["secret_scan"],
+            "issued_at": issued, "trust_domain": "aios.trusted-builder.attestation.v1",
+        }
+        canonical_attestation = _canonical_json(attestation)
+        fingerprint = hashlib.sha256(canonical_attestation).hexdigest()
+        signature = base64.b64encode(self.signer.sign_for_purpose("builder_attestation", canonical_attestation)).decode("ascii")
+        result = {
+            "schema": self.config.response_schema, "status": "BUILT", "recipe": recipe_name,
+            "repository": self.config.repository, "task_id": recipe["task_id"], "base_oid": recipe["base_oid"],
+            "purpose": recipe["purpose"], "checks": dict(checks),
+            "evidence": {"candidate_id": candidate["candidate_id"], "source_commit": candidate["source_commit"], "tree_oid": candidate["tree_oid"], "artifact_sha256": candidate["artifact_sha256"], "manifest_sha256": candidate["manifest_sha256"], "package_closure_sha256": release_plan["deployment"]["package_closure_sha256"], "test_evidence_sha256": attestation["test_evidence_sha256"], "canary_evidence_sha256": attestation["canary_evidence_sha256"]},
+            "attestation": {**attestation, "fingerprint": fingerprint, "signature": signature},
+        }
+        result["silent_release_bundle"] = {"candidate": candidate, "release_plan": release_plan, "checks": dict(checks), "attestation": result["attestation"]}
+        if len(self._nonce_ledger) >= 2048:
+            self._nonce_ledger.pop(next(iter(self._nonce_ledger)))
+        self._nonce_ledger[nonce] = (hashlib.sha256(request_bytes).digest(), dict(result))
+        self._persist_nonce_ledger()
+        return result
+
+    def serve_forever(self, listener: socket.socket) -> None:
+        while True:
+            connection, _ = listener.accept()
+            with connection:
+                try:
+                    raw = connection.makefile("rb").readline(65537)
+                    if not raw or len(raw) > 65536:
+                        raise BuilderError("REQUEST_SIZE_INVALID")
+                    result = self.handle_request(json.loads(raw), connection=connection)
+                except (BuilderError, UnicodeDecodeError, json.JSONDecodeError):
+                    result = {"schema": self.config.response_schema, "status": "REJECTED", "error_code": "REQUEST_REJECTED"}
+                connection.sendall(_canonical_json(result) + b"\n")
+
+
+class UnixTrustedCandidateBuilderClient:
+    def __init__(self, config: TrustedCandidateBuilderConfig, keyring: BuilderKeyring, *, socket_path: Path | None = None, clock: Callable[[], float] = time.time) -> None:
+        self.config, self.keyring, self.socket_path, self.clock = config, keyring, socket_path, clock
+
+    def create_request(self, version: str, candidate_id: str, source_commit: str, *, nonce: str, operation: str | None = None) -> dict[str, Any]:
+        selected = self.config.resolve_recipe(version, candidate_id, source_commit)
+        selected_operation = operation or self.config.operation
+        if selected_operation not in {self.config.operation, self.config.readiness_operation}:
+            raise BuilderError("REQUEST_OPERATION_INVALID")
+        request = {"schema": self.config.request_schema, "operation": selected_operation, "recipe": version, "repository": self.config.repository, "task_id": selected["task_id"], "candidate_id": selected["candidate_id"], "base_oid": selected["base_oid"], "source_commit": selected["source_commit"], "version": selected["version"], "purpose": selected["purpose"], "issued_at": int(self.clock()), "nonce": nonce}
+        request["hmac"] = hmac.new(self.keyring.builder_hmac, _canonical_json(request), hashlib.sha256).hexdigest()
+        return request
+
+    def _exchange(self, request: Mapping[str, Any]) -> Mapping[str, Any]:
+        if self.socket_path is None:
+            raise BuilderError("BUILDER_SOCKET_UNCONFIGURED")
+        try:
+            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+                connection.settimeout(self.config.request_timeout_seconds)
+                connection.connect(os.fspath(self.socket_path))
+                connection.sendall(_canonical_json(dict(request)) + b"\n")
+                raw = connection.makefile("rb").readline(65537)
+        except (OSError, TimeoutError) as exc:
+            raise BuilderError("BUILDER_TRANSPORT_UNAVAILABLE") from exc
+        try:
+            result = json.loads(raw)
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise BuilderError("BUILDER_RESPONSE_INVALID") from exc
+        if not isinstance(result, Mapping) or result.get("schema") != self.config.response_schema:
+            raise BuilderError("BUILDER_RESPONSE_INVALID")
+        return result
+
+    def build(self, version: str, candidate_id: str, source_commit: str, *, nonce: str) -> Mapping[str, Any]:
+        result = self._exchange(
+            self.create_request(version, candidate_id, source_commit, nonce=nonce)
+        )
+        if result.get("status") != "BUILT":
+            raise BuilderError("BUILDER_REQUEST_REJECTED")
+        return result
+
+    def probe(self, *, nonce: str | None = None) -> Mapping[str, Any]:
+        recipe = self.config.fixed_recipe("V2R2C2")
+        result = self._exchange(
+            self.create_request(
+                "V2R2C2",
+                recipe["candidate_id"],
+                recipe["source_commit"],
+                nonce=nonce or secrets.token_hex(32),
+                operation=self.config.readiness_operation,
+            )
+        )
+        if result.get("status") != "READY":
+            raise BuilderError("BUILDER_NOT_READY")
+        return result
+
+
+def readiness_probe(client: UnixTrustedCandidateBuilderClient) -> Mapping[str, Any]:
+    result = dict(client.probe())
+    required_true = {"collector_ready", "signer_ready"}
+    if any(result.get(name) is not True for name in required_true):
+        raise BuilderError("BUILDER_NOT_READY")
+    return result
+
+
+def _main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--socket-fd", required=True, type=int)
+    parser.add_argument("--policy", required=True, type=Path)
+    arguments = parser.parse_args()
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(arguments.policy))
+    credentials = os.environ.get("CREDENTIALS_DIRECTORY")
+    if not credentials:
+        raise SystemExit("CREDENTIALS_DIRECTORY is required")
+    keyring = BuilderKeyring.from_credentials_directory(Path(credentials))
+    signer = CredentialPathEd25519Signer(
+        keyring.signing_private_key_path,
+        key_id=os.environ["AIOS_TRUSTED_BUILDER_KEY_ID"],
+        key_generation=int(os.environ["AIOS_TRUSTED_BUILDER_KEY_GENERATION"]),
+        allowed_purpose="builder_attestation",
+    )
+    listener = socket.fromfd(arguments.socket_fd, socket.AF_UNIX, socket.SOCK_STREAM)
+    collector = DeterministicEvidenceCollector(
+        config,
+        evidence_source=ServerOwnedEvidenceSource(
+            config,
+            Path(os.environ["AIOS_TRUSTED_BUILDER_GIT_DIR"]),
+            Path(os.environ["AIOS_TRUSTED_BUILDER_EVIDENCE_ROOT"]),
+        ),
+    )
+    TrustedCandidateBuilder(config, keyring, collector=collector, signer=signer, nonce_ledger_path=Path(os.environ["AIOS_TRUSTED_BUILDER_NONCE_LEDGER"])).serve_forever(listener)
+
+
+if __name__ == "__main__":
+    _main()

active-runtime/app/silent_release_hardening.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import base64
 import hashlib
+import hmac
 import json
 import os
 import re
@@ -40,6 +41,8 @@ from silent_release_controller import (
 
 IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{2,255}$")
 NONCE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{15,255}$")
+BUILDER_ATTESTATION_TRUST_DOMAIN = "aios.trusted-builder.attestation.v1"
+BUILDER_ATTESTATION_PURPOSE = "builder_attestation"
 ALLOWED_CAPABILITIES = frozenset(
     {
         "runtime_switch",
@@ -62,6 +65,10 @@ class CapabilityAuthorizationError(RuntimeError):
     """The signed plan cannot authorize the requested operation."""
 
 
+class TrustedCandidateBuilderUnavailable(RuntimeError):
+    """The purpose-bound builder dependency cannot currently serve a build."""
+
+
 class StaleApprovalError(CapabilityAuthorizationError):
     """Live release evidence no longer matches the signed approval."""
 
@@ -222,6 +229,10 @@ JOURNAL_GENESIS_HASH = "0" * 64
 
 
 def _fsync_directory(path: Path) -> None:
+    if os.name == "nt":
+        # Windows does not permit opening a directory with os.open.  Durable
+        # Linux host execution retains the directory fsync below.
+        return
     descriptor = os.open(Path(path), os.O_RDONLY)
     try:
         os.fsync(descriptor)
@@ -715,6 +726,14 @@ class BuilderAttestation:
     canary_evidence_sha256: str
     issued_at: str
     signature: str
+    trust_domain: str = BUILDER_ATTESTATION_TRUST_DOMAIN
+    builder_key_generation: int | None = None
+    signature_algorithm: str | None = None
+    signature_purpose: str | None = None
+    gitea_evidence_sha256: str = ""
+    rollback_evidence_sha256: str = ""
+    policy_evidence_sha256: str = ""
+    secret_scan_evidence_sha256: str = ""
 
     def __post_init__(self) -> None:
         _validate_identifier(
@@ -731,7 +750,35 @@ class BuilderAttestation:
             "canary_evidence_sha256",
         ):
             _validate_sha(name, getattr(self, name))
+        for name in (
+            "gitea_evidence_sha256",
+            "rollback_evidence_sha256",
+            "policy_evidence_sha256",
+            "secret_scan_evidence_sha256",
+        ):
+            if getattr(self, name):
+                _validate_sha(name, getattr(self, name))
         _parse_utc(self.issued_at)
+        if self.trust_domain != BUILDER_ATTESTATION_TRUST_DOMAIN:
+            raise CapabilityAuthorizationError(
+                "builder attestation trust domain is invalid"
+            )
+        metadata = (
+            self.builder_key_generation,
+            self.signature_algorithm,
+            self.signature_purpose,
+        )
+        if any(value is not None for value in metadata):
+            if (
+                not isinstance(self.builder_key_generation, int)
+                or isinstance(self.builder_key_generation, bool)
+                or self.builder_key_generation < 0
+                or self.signature_algorithm != "Ed25519"
+                or self.signature_purpose != BUILDER_ATTESTATION_PURPOSE
+            ):
+                raise CapabilityAuthorizationError(
+                    "builder attestation signing metadata is invalid"
+                )
 
     @property
     def canonical_payload(self) -> dict[str, Any]:
@@ -754,20 +801,290 @@ class BuilderAttestation:
         }
 
     @classmethod
-    def from_dict(cls, payload: Mapping[str, Any]) -> "BuilderAttestation":
+    def from_dict(
+        cls,
+        payload: Mapping[str, Any],
+        *,
+        require_fingerprint: bool = False,
+    ) -> "BuilderAttestation":
         values = dict(payload)
-        values.pop("fingerprint", None)
-        return cls(**values)
+        supplied_fingerprint = values.pop("fingerprint", None)
+        # Older local records did not spell out the domain.  They are only
+        # usable when re-signed by the current builder; their original
+        # signatures cannot validate against this domain-separated payload.
+        values.setdefault("trust_domain", BUILDER_ATTESTATION_TRUST_DOMAIN)
+        value = cls(**values)
+        if supplied_fingerprint is None:
+            if require_fingerprint:
+                raise CapabilityAuthorizationError(
+                    "builder attestation digest is required"
+                )
+            return value
+        expected = _validate_sha(
+            "builder attestation digest", str(supplied_fingerprint)
+        )
+        if not hmac.compare_digest(expected, value.fingerprint):
+            raise CapabilityAuthorizationError(
+                "builder attestation digest is invalid"
+            )
+        return value
+
+
+@dataclass(frozen=True)
+class BuilderAttestationKeyRecord:
+    """One purpose-bound public verification key from the host keyring."""
+
+    key_id: str
+    public_key: bytes
+    algorithm: str
+    purpose: str
+    generation: int
+    not_before: str
+    expires_at: str
+    status: str
+
+    def __post_init__(self) -> None:
+        _validate_identifier("builder_key_id", self.key_id)
+        if self.algorithm != "Ed25519":
+            raise ValueError("unsupported builder signature algorithm")
+        if self.purpose != BUILDER_ATTESTATION_PURPOSE:
+            raise ValueError("builder key purpose is invalid")
+        if not isinstance(self.generation, int) or isinstance(
+            self.generation, bool
+        ) or self.generation < 0:
+            raise ValueError("builder key generation is invalid")
+        if self.status not in {"ACTIVE", "REVOKED"}:
+            raise ValueError("builder key status is invalid")
+        if len(self.public_key) != 32:
+            raise ValueError("Ed25519 public key must contain 32 bytes")
+        Ed25519PublicKey.from_public_bytes(self.public_key)
+        if _parse_utc(self.not_before) >= _parse_utc(self.expires_at):
+            raise ValueError("builder key validity window is invalid")
+
+    @property
+    def public_key_sha256(self) -> str:
+        return hashlib.sha256(self.public_key).hexdigest()
+
+    def validate_at(self, instant: datetime) -> None:
+        if self.status != "ACTIVE":
+            raise CapabilityAuthorizationError(
+                "builder attestation key is unavailable or revoked"
+            )
+        if instant.tzinfo is None:
+            raise CapabilityAuthorizationError(
+                "builder attestation verification time is invalid"
+            )
+        if not _parse_utc(self.not_before) <= instant <= _parse_utc(
+            self.expires_at
+        ):
+            raise CapabilityAuthorizationError(
+                "builder attestation key is outside its validity window"
+            )
+
+    @classmethod
+    def from_dict(
+        cls, payload: Mapping[str, Any]
+    ) -> "BuilderAttestationKeyRecord":
+        values = dict(payload)
+        encoded = values.pop("public_key", values.pop("public_key_b64", None))
+        if not isinstance(encoded, str):
+            raise ValueError("builder public key is missing")
+        try:
+            public_key = base64.b64decode(encoded, validate=True)
+        except (TypeError, ValueError) as exc:
+            raise ValueError("builder public key encoding is invalid") from exc
+        supplied_digest = values.pop(
+            "public_key_sha256",
+            values.pop("key_sha256", values.pop("digest", None)),
+        )
+        if not isinstance(supplied_digest, str):
+            raise ValueError("builder public key digest is missing")
+        digest = _validate_sha("builder public key digest", supplied_digest)
+        if not hmac.compare_digest(
+            digest, hashlib.sha256(public_key).hexdigest()
+        ):
+            raise ValueError("builder public key digest is invalid")
+        allowed = {
+            "algorithm",
+            "purpose",
+            "key_id",
+            "generation",
+            "not_before",
+            "expires_at",
+            "status",
+        }
+        unknown = set(values).difference(allowed)
+        if unknown:
+            raise ValueError("builder key record contains unknown fields")
+        return cls(public_key=public_key, **values)
+
+
+class BuilderAttestationKeyRing:
+    """Validated, public-only keyring for builder attestations.
+
+    The file format deliberately cannot carry Owner approval keys: both the
+    keyring and every record must declare the builder-attestation purpose.
+    """
+
+    def __init__(
+        self,
+        records: Iterable[BuilderAttestationKeyRecord],
+        *,
+        require_attestation_metadata: bool = False,
+    ):
+        self._records: dict[str, BuilderAttestationKeyRecord] = {}
+        self._require_attestation_metadata = require_attestation_metadata
+        for record in records:
+            if record.key_id in self._records:
+                raise ValueError("duplicate builder key id")
+            self._records[record.key_id] = record
+
+    @classmethod
+    def from_public_keys(
+        cls, public_keys: Mapping[str, bytes]
+    ) -> "BuilderAttestationKeyRing":
+        """Compatibility adapter for explicit in-process builder key maps.
+
+        This is intentionally not used for host configuration; host keyrings
+        must use :meth:`from_file` so purpose, digest, validity and status are
+        checked before the API accepts traffic.
+        """
+        return cls(
+            BuilderAttestationKeyRecord(
+                key_id=_validate_identifier("builder_key_id", key_id),
+                public_key=bytes(public_key),
+                algorithm="Ed25519",
+                purpose=BUILDER_ATTESTATION_PURPOSE,
+                generation=0,
+                not_before="1970-01-01T00:00:00Z",
+                expires_at="9999-12-31T23:59:59Z",
+                status="ACTIVE",
+            )
+            for key_id, public_key in public_keys.items()
+        )
+
+    @classmethod
+    def from_file(cls, path: Path) -> "BuilderAttestationKeyRing":
+        payload = json.loads(Path(path).read_text(encoding="utf-8"))
+        if payload.get("algorithm") != "Ed25519":
+            raise ValueError("unsupported builder keyring algorithm")
+        if payload.get("purpose") != BUILDER_ATTESTATION_PURPOSE:
+            raise ValueError("builder keyring purpose is invalid")
+        raw_records = payload.get("keys")
+        if not isinstance(raw_records, list) or not raw_records:
+            raise ValueError("builder keyring keys are invalid")
+        allowed = {"algorithm", "purpose", "keys", "keyring_sha256"}
+        if set(payload).difference(allowed):
+            raise ValueError("builder keyring contains unknown fields")
+        if "keyring_sha256" not in payload:
+            raise ValueError("builder keyring digest is missing")
+        expected = _validate_sha(
+            "builder keyring digest", str(payload["keyring_sha256"])
+        )
+        canonical = _canonical_json(
+            {
+                "algorithm": payload["algorithm"],
+                "purpose": payload["purpose"],
+                "keys": raw_records,
+            }
+        )
+        if not hmac.compare_digest(
+            expected, hashlib.sha256(canonical).hexdigest()
+        ):
+            raise ValueError("builder keyring digest is invalid")
+        if not all(isinstance(record, Mapping) for record in raw_records):
+            raise ValueError("builder keyring record is invalid")
+        return cls(
+            (
+                BuilderAttestationKeyRecord.from_dict(record)
+                for record in raw_records
+            ),
+            require_attestation_metadata=True,
+        )
+
+    def verify(
+        self,
+        key_id: str,
+        signature: bytes,
+        message: bytes,
+        *,
+        at: datetime,
+        generation: int | None,
+        algorithm: str | None,
+        purpose: str | None,
+    ) -> BuilderAttestationKeyRecord:
+        record = self._records.get(key_id)
+        if record is None:
+            raise CapabilityAuthorizationError(
+                "builder attestation key is not trusted"
+            )
+        record.validate_at(at)
+        metadata = (generation, algorithm, purpose)
+        if self._require_attestation_metadata and any(
+            value is None for value in metadata
+        ):
+            raise CapabilityAuthorizationError(
+                "builder attestation signing metadata is required"
+            )
+        if any(value is not None for value in metadata) and metadata != (
+            record.generation,
+            record.algorithm,
+            record.purpose,
+        ):
+            raise CapabilityAuthorizationError(
+                "builder attestation signing metadata does not match key"
+            )
+        try:
+            Ed25519PublicKey.from_public_bytes(record.public_key).verify(
+                signature, message
+            )
+        except InvalidSignature as exc:
+            raise CapabilityAuthorizationError(
+                "builder attestation signature is invalid"
+            ) from exc
+        return record
+
+    def require_active_identity(
+        self,
+        *,
+        key_id: str,
+        generation: int,
+        algorithm: str,
+        purpose: str,
+        at: datetime | None = None,
+    ) -> BuilderAttestationKeyRecord:
+        record = self._records.get(key_id)
+        if record is None:
+            raise CapabilityAuthorizationError(
+                "builder attestation key is not trusted"
+            )
+        if (
+            record.generation != generation
+            or record.algorithm != algorithm
+            or record.purpose != purpose
+        ):
+            raise CapabilityAuthorizationError(
+                "builder attestation key metadata is invalid"
+            )
+        record.validate_at(at or datetime.now(timezone.utc))
+        return record
 
 
 class BuilderAttestationVerifier:
     """Public-key-only verifier; no builder private material is accepted."""
 
-    def __init__(self, public_keys: Mapping[str, bytes]):
-        self.public_keys = {
-            _validate_identifier("builder_key_id", key_id): bytes(value)
-            for key_id, value in public_keys.items()
-        }
+    def __init__(
+        self,
+        public_keys: BuilderAttestationKeyRing | Mapping[str, bytes],
+    ):
+        if isinstance(public_keys, BuilderAttestationKeyRing):
+            self.keyring = public_keys
+        elif isinstance(public_keys, Mapping):
+            self.keyring = BuilderAttestationKeyRing.from_public_keys(
+                public_keys
+            )
+        else:
+            raise TypeError("builder verifier requires a builder keyring")
 
     def verify(
         self,
@@ -793,22 +1110,54 @@ class BuilderAttestationVerifier:
             raise CapabilityAuthorizationError(
                 "builder attestation does not bind candidate evidence"
             )
-        key = self.public_keys.get(attestation.builder_key_id)
-        if key is None:
-            raise CapabilityAuthorizationError(
-                "builder attestation key is not trusted"
-            )
         try:
             signature = base64.b64decode(
                 attestation.signature, validate=True
             )
-            Ed25519PublicKey.from_public_bytes(key).verify(
-                signature, attestation.canonical_bytes()
+            key_record = self.keyring.verify(
+                attestation.builder_key_id,
+                signature,
+                attestation.canonical_bytes(),
+                at=_parse_utc(attestation.issued_at),
+                generation=attestation.builder_key_generation,
+                algorithm=attestation.signature_algorithm,
+                purpose=attestation.signature_purpose,
             )
         except (InvalidSignature, ValueError, TypeError) as exc:
             raise CapabilityAuthorizationError(
                 "builder attestation signature is invalid"
             ) from exc
+        if key_record.generation >= 1:
+            expected_gitea = hashlib.sha256(
+                _canonical_json(candidate.gitea_consistency)
+            ).hexdigest()
+            expected_rollback = hashlib.sha256(
+                _canonical_json(
+                    {
+                        "ready": candidate.rollback_ready,
+                        "rollback_sha256": release_plan.deployment.rollback_sha256,
+                        "source_commit": candidate.source_commit,
+                    }
+                )
+            ).hexdigest()
+            production_evidence = (
+                attestation.gitea_evidence_sha256,
+                attestation.rollback_evidence_sha256,
+                attestation.policy_evidence_sha256,
+                attestation.secret_scan_evidence_sha256,
+            )
+            if any(not value for value in production_evidence):
+                raise CapabilityAuthorizationError(
+                    "builder attestation production evidence is incomplete"
+                )
+            if (
+                attestation.gitea_evidence_sha256 != expected_gitea
+                or attestation.rollback_evidence_sha256 != expected_rollback
+                or attestation.policy_evidence_sha256 != release_plan.policy_sha256
+            ):
+                raise CapabilityAuthorizationError(
+                    "builder attestation production evidence is invalid"
+                )
         return attestation
 
 
@@ -3181,9 +3530,12 @@ class SilentReleaseActionService:
         builder_attestation_verifier: BuilderAttestationVerifier | None = None,
         trusted_policy: TrustedReleasePolicy | None = None,
         orchestrator: ReleaseOrchestrator | None = None,
+        builder_readiness_probe: Callable[[], Mapping[str, Any]] | None = None,
         allow_legacy_candidate_input: bool = False,
         expected_repository: str = "",
         policy_sha256: str = "",
+        max_builder_attempts: int = 3,
+        builder_attempt_timeout_seconds: int = 180,
     ):
         self.root = Path(root)
         self.execution_enabled = bool(execution_enabled)
@@ -3194,11 +3546,28 @@ class SilentReleaseActionService:
         self.builder_attestation_verifier = builder_attestation_verifier
         self.trusted_policy = trusted_policy
         self.orchestrator = orchestrator
+        self.builder_readiness_probe = builder_readiness_probe
         self.allow_legacy_candidate_input = bool(
             allow_legacy_candidate_input
         )
         self.expected_repository = expected_repository
         self.policy_sha256 = policy_sha256
+        if (
+            not isinstance(max_builder_attempts, int)
+            or isinstance(max_builder_attempts, bool)
+            or not 1 <= max_builder_attempts <= 10
+        ):
+            raise ValueError("max_builder_attempts must be between 1 and 10")
+        self.max_builder_attempts = max_builder_attempts
+        if (
+            not isinstance(builder_attempt_timeout_seconds, int)
+            or isinstance(builder_attempt_timeout_seconds, bool)
+            or not 1 <= builder_attempt_timeout_seconds <= 3600
+        ):
+            raise ValueError(
+                "builder_attempt_timeout_seconds must be between 1 and 3600"
+            )
+        self.builder_attempt_timeout_seconds = builder_attempt_timeout_seconds
         self.candidates = ReleaseCandidateStore(
             self.root / "candidates"
         )
@@ -3235,10 +3604,26 @@ class SilentReleaseActionService:
                     + ", ".join(missing)
                 )
 
+    def get_builder_readiness(self) -> dict[str, Any]:
+        """Report dependency readiness without conflating it with API liveness."""
+        if self.candidate_builder is None:
+            return {"status": "UNAVAILABLE"}
+        if self.builder_readiness_probe is None:
+            # Explicit in-process test/adaptor construction remains usable;
+            # production wiring provides a probe backed by its credentials.
+            return {"status": "READY"}
+        try:
+            value = dict(self.builder_readiness_probe())
+        except Exception:
+            return {"status": "UNAVAILABLE"}
+        if value.get("status") != "READY":
+            return {"status": "UNAVAILABLE"}
+        return _sanitize(value)
+
     def create_release_intent(
         self, payload: Mapping[str, Any]
     ) -> dict[str, Any]:
-        allowed = {"event_id", "source_commit"}
+        allowed = {"event_id", "source_commit", "version"}
         unknown = set(payload).difference(allowed)
         if unknown:
             raise CapabilityAuthorizationError(
@@ -3250,12 +3635,17 @@ class SilentReleaseActionService:
         source_commit = _validate_oid(
             "source_commit", str(payload.get("source_commit") or "")
         )
+        version = str(payload.get("version") or "")
+        if version:
+            _validate_identifier("version", version)
         intent_id = "intent-" + hashlib.sha256(
             f"{event_id}\0{source_commit}".encode("utf-8")
         ).hexdigest()[:24]
         path = _safe_child(self.intents_root, intent_id)
+        was_existing = path.exists()
         record = {
             "created_at": utc_now(),
+            "attempt_count": 0,
             "event_id": event_id,
             "intent_id": intent_id,
             "source_commit": source_commit,
@@ -3265,42 +3655,153 @@ class SilentReleaseActionService:
                 else "BUILDER_UNAVAILABLE"
             ),
         }
-        if path.exists():
+        if version:
+            record["version"] = version
+        if was_existing:
             existing = json.loads(path.read_text(encoding="utf-8"))
             if (
                 existing.get("event_id") != event_id
                 or existing.get("source_commit") != source_commit
+                or str(existing.get("version") or "") != version
             ):
                 raise SagaConflict(
                     "release intent id is bound to different evidence"
                 )
-            return {
-                **_sanitize(existing),
-                "idempotent": True,
-                "execution_enabled": self.execution_enabled,
-                "shadow_mode": not self.execution_enabled,
-                "side_effects_executed": False,
+            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(
+                {
+                    "retryable": False,
+                    "status": "BUILDER_UNAVAILABLE",
+                }
+            )
+            _durable_json(path, record)
+            return self._intent_response(record, idempotent=was_existing)
+
+        record.update(
+            {
+                "attempt_count": int(record.get("attempt_count") or 0) + 1,
+                "attempt_started_at": utc_now(),
+                "retryable": False,
+                "status": "BUILDING",
+                "updated_at": utc_now(),
             }
+        )
         _durable_json(path, record)
-        result = dict(record)
-        if self.candidate_builder is not None:
+        try:
             candidate = self.create_release_candidate(
                 {
                     "event_id": event_id,
                     "intent_id": intent_id,
                     "source_commit": source_commit,
+                    **({"version": version} if version else {}),
                 }
             )
-            result.update(
+        except TrustedCandidateBuilderUnavailable:
+            record.update(
                 {
-                    "candidate_id": candidate["candidate_id"],
-                    "status": candidate["saga_state"],
+                    "failure_code": "TRUSTED_BUILDER_UNAVAILABLE",
+                    "retryable": record["attempt_count"]
+                    < self.max_builder_attempts,
+                    "status": (
+                        "BUILDER_RETRYABLE_FAILURE"
+                        if record["attempt_count"] < self.max_builder_attempts
+                        else "BUILDER_TERMINAL_FAILURE"
+                    ),
                 }
             )
-            _durable_json(path, result)
+            _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",
+                }
+            )
+            _durable_json(path, record)
+            return self._intent_response(record, idempotent=False)
+        except (ConnectionError, OSError, TimeoutError):
+            record.update(
+                {
+                    "failure_code": "TRUSTED_BUILDER_UNAVAILABLE",
+                    "retryable": record["attempt_count"]
+                    < self.max_builder_attempts,
+                    "status": (
+                        "BUILDER_RETRYABLE_FAILURE"
+                        if record["attempt_count"] < self.max_builder_attempts
+                        else "BUILDER_TERMINAL_FAILURE"
+                    ),
+                }
+            )
+            _durable_json(path, record)
+            return self._intent_response(record, idempotent=False)
+        except Exception:
+            # The durable intent is recoverable without serializing an
+            # arbitrary exception (which may contain a credential or host
+            # path).  Repeated requests consume the small fixed retry budget.
+            record.update(
+                {
+                    "failure_code": "TRUSTED_BUILDER_FAILURE",
+                    "retryable": record["attempt_count"]
+                    < self.max_builder_attempts,
+                    "status": (
+                        "BUILDER_RETRYABLE_FAILURE"
+                        if record["attempt_count"] < self.max_builder_attempts
+                        else "BUILDER_TERMINAL_FAILURE"
+                    ),
+                }
+            )
+            _durable_json(path, record)
+            return self._intent_response(record, idempotent=False)
+        result = dict(record)
+        result.update(
+            {
+                "candidate_id": candidate["candidate_id"],
+                "retryable": False,
+                "status": candidate["saga_state"],
+            }
+        )
+        _durable_json(path, result)
+        return self._intent_response(result, idempotent=False)
+
+    def _intent_response(
+        self, record: Mapping[str, Any], *, idempotent: bool
+    ) -> dict[str, Any]:
         return {
-            **_sanitize(result),
-            "idempotent": False,
+            **_sanitize(dict(record)),
+            "idempotent": idempotent,
             "execution_enabled": self.execution_enabled,
             "shadow_mode": not self.execution_enabled,
             "side_effects_executed": False,
@@ -3321,6 +3822,7 @@ class SilentReleaseActionService:
         self, payload: Mapping[str, Any]
     ) -> dict[str, Any]:
         if self.candidate_builder is not None:
+            values = dict(payload)
             forbidden_evidence = {
                 "artifact_sha256",
                 "canary",
@@ -3333,13 +3835,69 @@ class SilentReleaseActionService:
                 "tests",
                 "tree_oid",
             }
-            if forbidden_evidence.intersection(payload):
+            if forbidden_evidence.intersection(values):
                 raise CapabilityAuthorizationError(
                     "candidate evidence must come from the trusted builder"
                 )
+            worker_fields = {"candidate_id", "source_commit", "version"}
+            intent_fields = {"event_id", "intent_id", "source_commit", "version"}
+            supplied = set(values)
+            is_worker_request = worker_fields.issubset(supplied)
+            if is_worker_request:
+                if supplied != worker_fields:
+                    raise ValueError("release worker request contains unknown fields")
+                _validate_identifier(
+                    "candidate_id", str(values.get("candidate_id") or "")
+                )
+                _validate_identifier(
+                    "version", str(values.get("version") or "")
+                )
+            else:
+                if not supplied.issubset(intent_fields):
+                    raise ValueError("trusted builder intent contains unknown fields")
+                # Preserve the established internal intent contract
+                # (source_commit + version) while allowing newer callers to
+                # add an event/intent identifier.  Neither form can supply
+                # acceptance evidence; the host builder owns all collectors.
+                if values.get("event_id"):
+                    _validate_identifier(
+                        "event_id", str(values["event_id"])
+                    )
+                if values.get("intent_id"):
+                    _validate_identifier(
+                        "intent_id", str(values["intent_id"])
+                    )
+                if values.get("version"):
+                    _validate_identifier(
+                        "version", str(values["version"])
+                    )
+            _validate_oid(
+                "source_commit", str(values.get("source_commit") or "")
+            )
+            readiness = self.get_builder_readiness()
+            if readiness.get("status") != "READY":
+                raise TrustedCandidateBuilderUnavailable(
+                    "trusted builder dependency is unavailable"
+                )
             bundle = self.candidate_builder.build(dict(payload)).validate(
                 self.builder_attestation_verifier
             )
+            requested_source = str(payload.get("source_commit") or "")
+            if (
+                requested_source
+                and bundle.candidate.source_commit != requested_source
+            ):
+                raise CapabilityAuthorizationError(
+                    "trusted builder source commit does not match intent"
+                )
+            if (
+                is_worker_request
+                and bundle.candidate.candidate_id
+                != str(values["candidate_id"])
+            ):
+                raise CapabilityAuthorizationError(
+                    "trusted builder candidate id does not match request"
+                )
             if self.trusted_policy is None:
                 raise CapabilityAuthorizationError(
                     "trusted release policy is not configured"

config/gitops/trusted-candidate-builder-policy.json

@@ -0,0 +1,57 @@
+{
+  "schema": "aios.trusted-candidate-builder-policy.v1",
+  "repository": "aios/aios",
+  "request_schema": "aios.trusted-candidate-builder.request.v1",
+  "response_schema": "aios.trusted-candidate-builder.response.v1",
+  "operation": "BuildTrustedCandidateV1",
+  "readiness_operation": "ProbeTrustedCandidateBuilderV1",
+  "maximum_request_age_seconds": 60,
+  "request_timeout_seconds": 30,
+  "allowed_peer_uid": 1000,
+  "allowed_peer_cgroups": [
+    "/system.slice/aios-gitops-model-api.service"
+  ],
+  "allowed_peer_executable_prefix": "/usr/bin/python3",
+  "allowed_peer_runtime_prefix": "/opt/aios/gitops-model-api/",
+  "allowed_peer_runtime_suffix": "/runtime/app/aios_model_api_staging.py",
+  "required_checks": [
+    "authoritative_git",
+    "artifact",
+    "canary",
+    "gitea",
+    "manifest",
+    "package_closure",
+    "rollback",
+    "secret_scan",
+    "tests"
+  ],
+  "evidence_layout": {
+    "artifact": "{version}/artifact.tar.gz",
+    "manifest": "{version}/manifest.json",
+    "package_closure": "{version}/checksums.sha256",
+    "tests": "{version}/tests.json",
+    "canary": "{version}/canary.json",
+    "gitea": "{version}/gitea.json",
+    "rollback": "{version}/rollback.json",
+    "policy": "{version}/silent-release-policy.json",
+    "secret_scan": "{version}/secret-scan.json"
+  },
+  "recipes": {
+    "V2R2C2": {
+      "mode": "fixed",
+      "task_id": "r42-release-candidate-v2r2c2-live-20260819",
+      "candidate_id": "r42-release-candidate-v2r2c2-20260819",
+      "base_oid": "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+      "source_commit": "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+      "version": "V2R2C2",
+      "purpose": "release:candidate:trusted-build"
+    },
+    "V2R2C3": {
+      "mode": "governed-derived-only",
+      "derives_from": "V2R2C2",
+      "version": "V2R2C3",
+      "purpose": "release:candidate:trusted-build"
+    }
+  },
+  "governed_recipes": {}
+}

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

@@ -0,0 +1,44 @@
+[Unit]
+Description=AIOS trusted candidate builder
+Requires=aios-trusted-candidate-builder.socket
+After=aios-trusted-candidate-builder.socket
+
+[Service]
+Type=simple
+User=root
+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
+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_KEY_ID=builder-attestation-ed25519-g1
+Environment=AIOS_TRUSTED_BUILDER_KEY_GENERATION=1
+Environment=PYTHONDONTWRITEBYTECODE=1
+ExecStart=/usr/bin/python3 -B /opt/aios/trusted-candidate-builder/current/app/aios_trusted_candidate_builder.py --socket-fd 3 --policy ${AIOS_TRUSTED_BUILDER_POLICY}
+Restart=on-failure
+RestartSec=2
+StateDirectory=aios-trusted-candidate-builder
+StateDirectoryMode=0700
+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/trusted-candidate-builder /data/aios-gitops/repo.git /var/lib/aios-trusted-candidate-builder/evidence
+ReadWritePaths=/var/lib/aios-trusted-candidate-builder
+
+[Install]
+WantedBy=multi-user.target

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

@@ -0,0 +1,13 @@
+[Unit]
+Description=AIOS trusted candidate builder AF_UNIX socket
+
+[Socket]
+ListenStream=/run/aios-trusted-candidate-builder.sock
+SocketUser=root
+SocketGroup=gitops
+SocketMode=0660
+RemoveOnStop=yes
+Accept=no
+
+[Install]
+WantedBy=sockets.target

scripts/build_v2r2c1_model_api_staging.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 import argparse
+import gzip
 import hashlib
 import json
 import os
 from pathlib import Path
+import stat
 import tarfile
 import tempfile
 
@@ -17,8 +19,18 @@ REQUIRED_APP = (
     'gitea_tag_publication.py',
     'silent_release_controller.py',
     'silent_release_hardening.py',
+    'aios_trusted_candidate_builder.py',
     'aios_gitops_job_controller.py',
 )
+REQUIRED_RUNTIME = (
+    'orchestrator/__init__.py',
+    'orchestrator/action_monitor.py',
+)
+REQUIRED_BUILDER_ASSETS = (
+    Path('config/gitops/trusted-candidate-builder-policy.json'),
+    Path('deploy/systemd/aios-trusted-candidate-builder.socket'),
+    Path('deploy/systemd/aios-trusted-candidate-builder.service'),
+)
 OPTIONAL_APP = (
     'aios_chatgpt_actions_api.py',
     'aios_continuation_executor.py',
@@ -44,6 +56,38 @@ def private_marker(path: Path) -> bool:
     )
     return any(re.search(pattern, text) for pattern in patterns)
 
+def checked_regular_file(path: Path, label: str) -> Path:
+    forbidden_parts = {'.git', 'credential', 'credentials', 'secret', 'secrets'}
+    if forbidden_parts.intersection(part.lower() for part in Path(label).parts):
+        raise SystemExit(f'credential path rejected: {label}')
+    if path.is_symlink():
+        raise SystemExit(f'symlink rejected: {label}')
+    try:
+        mode = path.stat(follow_symlinks=False).st_mode
+    except FileNotFoundError:
+        raise SystemExit(f'required dependency missing: {label}') from None
+    if not stat.S_ISREG(mode):
+        raise SystemExit(f'non-regular file rejected: {label}')
+    return path
+
+def normalized_tar_info(path: Path, arcname: str) -> tarfile.TarInfo:
+    info = tarfile.TarInfo(arcname)
+    info.mtime = 0
+    info.uid = 0
+    info.gid = 0
+    info.uname = 'root'
+    info.gname = 'root'
+    if path.is_dir():
+        info.type = tarfile.DIRTYPE
+        info.mode = 0o755
+        info.size = 0
+    else:
+        info.type = tarfile.REGTYPE
+        source_mode = path.stat(follow_symlinks=False).st_mode
+        info.mode = 0o755 if source_mode & stat.S_IXUSR else 0o644
+        info.size = path.stat(follow_symlinks=False).st_size
+    return info
+
 def main() -> int:
     ap = argparse.ArgumentParser()
     ap.add_argument('--source-root', required=True, type=Path)
@@ -63,27 +107,48 @@ def main() -> int:
     paths = []
     for name in REQUIRED_APP:
         path = app / name
-        if not path.is_file():
-            raise SystemExit(f'required app dependency missing: {name}')
+        checked_regular_file(path, f'active-runtime/app/{name}')
         paths.append((Path('app') / name, path))
     for name in OPTIONAL_APP:
         path = app / name
+        if path.is_symlink():
+            raise SystemExit(f'symlink rejected: active-runtime/app/{name}')
         if path.is_file():
+            checked_regular_file(path, f'active-runtime/app/{name}')
             paths.append((Path('app') / name, path))
-    for rel in ('orchestrator/__init__.py', 'orchestrator/action_monitor.py'):
+    for rel in REQUIRED_RUNTIME:
         path = runtime / rel
-        if not path.is_file():
-            raise SystemExit(f'required runtime dependency missing: {rel}')
+        checked_regular_file(path, f'runtime/agent-os/{rel}')
         paths.append((Path('runtime-agent-os') / rel, path))
+    for rel in REQUIRED_BUILDER_ASSETS:
+        path = source / rel
+        checked_regular_file(path, str(rel).replace('\\', '/'))
+        paths.append((rel, path))
     openapi = docs / 'openapi-chatgpt-actions.yaml'
+    if openapi.is_symlink():
+        raise SystemExit('symlink rejected: docs/gitops/openapi-chatgpt-actions.yaml')
     if openapi.is_file():
+        checked_regular_file(openapi, 'docs/gitops/openapi-chatgpt-actions.yaml')
         paths.append((Path('docs') / 'gitops' / 'openapi-chatgpt-actions.yaml', openapi))
     for folder in ('config/model_registry', 'config/model_priority', 'patches', 'versions'):
         root = source / folder
         if root.is_dir():
             for path in sorted(root.rglob('*')):
+                if path.is_symlink():
+                    raise SystemExit(f'symlink rejected: {path.relative_to(source)}')
+                if not path.is_dir() and not path.is_file():
+                    raise SystemExit(f'special file rejected: {path.relative_to(source)}')
                 if path.is_file() and path.suffix.lower() in {'.json', '.yaml', '.yml', '.md'}:
+                    checked_regular_file(path, str(path.relative_to(source)).replace('\\', '/'))
                     paths.append((Path(folder) / path.relative_to(root), path))
+    seen: set[str] = set()
+    for rel, _ in paths:
+        normalized = str(rel).replace('\\', '/')
+        if normalized.startswith('/') or '..' in Path(normalized).parts:
+            raise SystemExit(f'path traversal rejected: {normalized}')
+        if normalized in seen:
+            raise SystemExit(f'duplicate archive member rejected: {normalized}')
+        seen.add(normalized)
     for rel, path in paths:
         if private_marker(path):
             raise SystemExit(f'private marker rejected: {rel}')
@@ -110,27 +175,25 @@ def main() -> int:
             target.parent.mkdir(parents=True, exist_ok=True)
             target.write_bytes(path.read_bytes())
         (root / 'manifest.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + '\n', encoding='utf-8')
-        with tarfile.open(archive, 'w:gz', format=tarfile.PAX_FORMAT) as tf:
-            for path in sorted(root.rglob('*')):
-                rel = path.relative_to(root.parent)
-                info = tf.gettarinfo(str(path), arcname=str(rel).replace('\\', '/'))
-                info.mtime = 0
-                info.uid = 0
-                info.gid = 0
-                info.uname = 'root'
-                info.gname = 'root'
-                if path.is_file():
-                    with path.open('rb') as fh:
-                        tf.addfile(info, fh)
-                else:
-                    tf.addfile(info)
+        with archive.open('wb') as raw:
+            with gzip.GzipFile(filename='', mode='wb', fileobj=raw, mtime=0) as compressed:
+                with tarfile.open(fileobj=compressed, mode='w', format=tarfile.PAX_FORMAT) as tf:
+                    for path in sorted(root.rglob('*')):
+                        rel = path.relative_to(root.parent)
+                        arcname = str(rel).replace('\\', '/')
+                        info = normalized_tar_info(path, arcname)
+                        if path.is_file():
+                            with path.open('rb') as fh:
+                                tf.addfile(info, fh)
+                        else:
+                            tf.addfile(info)
     digest = sha256(archive)
     receipt = {
         'schema': 'aios.v2r2c1.model-api-staging-build-receipt.v1',
         'target_version': 'V2R2C1',
         'source_oid': oid,
         'repository_identity': 'aios/aios',
-        'artifact_path': str(archive),
+        'artifact_path': archive.name,
         'artifact_sha256': digest,
         'deployment_state': 'STAGING',
         'manifest_schema': SCHEMA,

tests/test_model_api_immutable_staging.py

@@ -1,4 +1,13 @@
+import hashlib
+import importlib.util
+import json
 from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tarfile
+
+import pytest
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -33,3 +42,111 @@ def test_model_api_builder_manifest_contract():
     assert "'deployment_state': 'STAGING'" in builder
     assert "'legacy_global_writes_enabled': False" in builder
     assert 'private marker rejected' in builder
+
+
+def _load_builder_module():
+    path = ROOT / 'scripts' / 'build_v2r2c1_model_api_staging.py'
+    spec = importlib.util.spec_from_file_location('model_api_packager', path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _run_builder(source: Path, output: Path, oid: str = 'a' * 40):
+    return subprocess.run(
+        [
+            sys.executable,
+            str(ROOT / 'scripts' / 'build_v2r2c1_model_api_staging.py'),
+            '--source-root',
+            str(source),
+            '--output-root',
+            str(output),
+            '--source-oid',
+            oid,
+        ],
+        capture_output=True,
+        text=True,
+    )
+
+
+def _minimal_source(tmp_path: Path) -> Path:
+    module = _load_builder_module()
+    source = tmp_path / 'source'
+    for name in module.REQUIRED_APP:
+        target = source / 'active-runtime' / 'app' / name
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(ROOT / 'active-runtime' / 'app' / name, target)
+    for rel in module.REQUIRED_RUNTIME:
+        target = source / 'runtime' / 'agent-os' / rel
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(ROOT / 'runtime' / 'agent-os' / rel, target)
+    for rel in module.REQUIRED_BUILDER_ASSETS:
+        target = source / rel
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(ROOT / rel, target)
+    return source
+
+
+def test_model_api_package_contains_builder_runtime_and_is_deterministic(tmp_path):
+    first = tmp_path / 'first'
+    second = tmp_path / 'second'
+    one = _run_builder(ROOT, first)
+    two = _run_builder(ROOT, second)
+    assert one.returncode == 0, one.stderr
+    assert two.returncode == 0, two.stderr
+
+    receipt = json.loads(one.stdout)
+    peer_receipt = json.loads(two.stdout)
+    assert one.stdout == two.stdout
+    archive = first / receipt['artifact_path']
+    peer_archive = second / peer_receipt['artifact_path']
+    assert hashlib.sha256(archive.read_bytes()).hexdigest() == hashlib.sha256(
+        peer_archive.read_bytes()
+    ).hexdigest()
+
+    with tarfile.open(archive, 'r:gz') as bundle:
+        members = bundle.getmembers()
+        names = [member.name for member in members]
+        assert len(names) == len(set(names))
+        assert 'runtime/app/aios_trusted_candidate_builder.py' in names
+        assert 'runtime/config/gitops/trusted-candidate-builder-policy.json' in names
+        assert 'runtime/deploy/systemd/aios-trusted-candidate-builder.socket' in names
+        assert 'runtime/deploy/systemd/aios-trusted-candidate-builder.service' in names
+        assert all(member.isfile() or member.isdir() for member in members)
+        assert all('.git' not in Path(name).parts for name in names)
+        assert all(
+            member.mode == (0o755 if member.isdir() else 0o644)
+            for member in members
+        )
+
+
+def test_model_api_packager_rejects_symlinked_required_asset(tmp_path):
+    source = _minimal_source(tmp_path)
+    policy = source / 'config' / 'gitops' / 'trusted-candidate-builder-policy.json'
+    policy.unlink()
+    try:
+        policy.symlink_to(
+            source / 'active-runtime' / 'app' / 'silent_release_hardening.py'
+        )
+    except OSError:
+        pytest.skip('creating symlinks is not permitted by this host')
+    result = _run_builder(source, tmp_path / 'output')
+    assert result.returncode != 0
+    assert 'symlink' in (result.stderr + result.stdout).lower()
+
+
+def test_private_marker_rejects_real_pem_but_allows_prose(tmp_path):
+    module = _load_builder_module()
+    prose = tmp_path / 'prose.txt'
+    prose.write_text('Documentation says BEGIN PRIVATE KEY without a PEM fence.\n')
+    assert module.private_marker(prose) is False
+
+    private = tmp_path / 'private.pem'
+    pem_begin = '-----' + 'BEGIN PRIVATE KEY' + '-----\n'
+    pem_end = '-----' + 'END PRIVATE KEY' + '-----\n'
+    private.write_text(
+        pem_begin + 'c2Vuc2l0aXZlLWJ1aWxkZXIta2V5LW1hdGVyaWFs\n'
+        + pem_end
+    )
+    assert module.private_marker(private) is True

tests/test_model_api_trusted_candidate_builder.py

@@ -0,0 +1,484 @@
+from __future__ import annotations
+
+import base64
+import dataclasses
+import hashlib
+import importlib
+import importlib.util
+import json
+import socket
+import sys
+import threading
+import urllib.error
+import urllib.request
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+
+APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
+AGENT_OS = Path(__file__).resolve().parents[1] / "runtime" / "agent-os"
+for import_root in (APP, AGENT_OS):
+    if str(import_root) not in sys.path:
+        sys.path.insert(0, str(import_root))
+
+from silent_release_hardening import (  # noqa: E402
+    BuilderAttestation,
+    BuilderAttestationKeyRing,
+    BuilderAttestationVerifier,
+    CapabilityAuthorizationError,
+    SilentReleaseActionService,
+    TrustedCandidateBuilderUnavailable,
+)
+import silent_release_hardening as hardening  # noqa: E402
+
+
+COMMIT = "a" * 40
+TREE = "b" * 40
+SHA = "c" * 64
+GITEA = {"consistent": True}
+
+
+def _af_inet_available() -> bool:
+    try:
+        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    except OSError:
+        return False
+    probe.close()
+    return True
+
+
+def _canonical_digest(value: dict) -> str:
+    return hashlib.sha256(
+        json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest()
+
+
+def _instant(delta: timedelta = timedelta()) -> str:
+    return (datetime.now(timezone.utc) + delta).isoformat().replace(
+        "+00:00", "Z"
+    )
+
+
+def _keyring_payload(private: Ed25519PrivateKey, *, purpose: str) -> dict:
+    public = private.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw
+    )
+    record = {
+        "algorithm": "Ed25519",
+        "purpose": purpose,
+        "key_id": "builder-key-001",
+        "generation": 1,
+        "not_before": _instant(timedelta(minutes=-1)),
+        "expires_at": _instant(timedelta(days=1)),
+        "status": "ACTIVE",
+        "public_key": base64.b64encode(public).decode("ascii"),
+        "public_key_sha256": hashlib.sha256(public).hexdigest(),
+    }
+    payload = {
+        "algorithm": "Ed25519",
+        "purpose": "builder_attestation",
+        "keys": [record],
+    }
+    payload["keyring_sha256"] = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    return payload
+
+
+def _refresh_keyring_digest(payload: dict) -> dict:
+    canonical = {
+        "algorithm": payload["algorithm"],
+        "purpose": payload["purpose"],
+        "keys": payload["keys"],
+    }
+    payload["keyring_sha256"] = hashlib.sha256(
+        json.dumps(canonical, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    return payload
+
+
+def _attestation(private: Ed25519PrivateKey) -> BuilderAttestation:
+    unsigned = BuilderAttestation(
+        builder_attestation_id="builder-attestation-001",
+        builder_key_id="builder-key-001",
+        source_commit=COMMIT,
+        tree_oid=TREE,
+        artifact_sha256=SHA,
+        manifest_sha256=SHA,
+        package_closure_sha256=SHA,
+        test_evidence_sha256=SHA,
+        canary_evidence_sha256=SHA,
+        issued_at=_instant(),
+        signature="",
+        builder_key_generation=1,
+        signature_algorithm="Ed25519",
+        signature_purpose="builder_attestation",
+        gitea_evidence_sha256=_canonical_digest(GITEA),
+        rollback_evidence_sha256=_canonical_digest({
+            "ready": True,
+            "rollback_sha256": SHA,
+            "source_commit": COMMIT,
+        }),
+        policy_evidence_sha256=SHA,
+        secret_scan_evidence_sha256=SHA,
+    )
+    return BuilderAttestation(
+        **{
+            **unsigned.canonical_payload,
+            "signature": base64.b64encode(
+                private.sign(unsigned.canonical_bytes())
+            ).decode("ascii"),
+        }
+    )
+
+
+def test_builder_verifier_rejects_owner_key_record_cross_purpose(tmp_path):
+    private = Ed25519PrivateKey.generate()
+    payload = _keyring_payload(private, purpose="owner_approval")
+    path = tmp_path / "builder-keyring.json"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="purpose"):
+        BuilderAttestationKeyRing.from_file(path)
+
+
+def test_builder_verifier_checks_domain_key_window_digest_and_signature(tmp_path):
+    private = Ed25519PrivateKey.generate()
+    path = tmp_path / "builder-keyring.json"
+    path.write_text(
+        json.dumps(_keyring_payload(private, purpose="builder_attestation")),
+        encoding="utf-8",
+    )
+    verifier = BuilderAttestationVerifier(
+        BuilderAttestationKeyRing.from_file(path)
+    )
+    candidate = SimpleNamespace(
+        source_commit=COMMIT,
+        tree_oid=TREE,
+        artifact_sha256=SHA,
+        manifest_sha256=SHA,
+        gitea_consistency=GITEA,
+        rollback_ready=True,
+    )
+    plan = SimpleNamespace(
+        deployment=SimpleNamespace(
+            package_closure_sha256=SHA,
+            rollback_sha256=SHA,
+        ),
+        policy_sha256=SHA,
+    )
+    assert verifier.verify(_attestation(private), candidate, plan).builder_key_id == "builder-key-001"
+
+
+@pytest.mark.parametrize("state", ["REVOKED", "EXPIRED", "FUTURE"])
+def test_builder_verifier_rejects_unavailable_key_lifecycle_states(tmp_path, state):
+    private = Ed25519PrivateKey.generate()
+    payload = _keyring_payload(private, purpose="builder_attestation")
+    record = payload["keys"][0]
+    if state == "REVOKED":
+        record["status"] = "REVOKED"
+    elif state == "EXPIRED":
+        record["not_before"] = _instant(timedelta(days=-2))
+        record["expires_at"] = _instant(timedelta(days=-1))
+    else:
+        record["not_before"] = _instant(timedelta(days=1))
+        record["expires_at"] = _instant(timedelta(days=2))
+    path = tmp_path / "builder-keyring.json"
+    path.write_text(json.dumps(_refresh_keyring_digest(payload)), encoding="utf-8")
+    verifier = BuilderAttestationVerifier(BuilderAttestationKeyRing.from_file(path))
+    candidate = SimpleNamespace(
+        source_commit=COMMIT, tree_oid=TREE,
+        artifact_sha256=SHA, manifest_sha256=SHA,
+        gitea_consistency=GITEA, rollback_ready=True,
+    )
+    plan = SimpleNamespace(deployment=SimpleNamespace(package_closure_sha256=SHA, rollback_sha256=SHA), policy_sha256=SHA)
+    with pytest.raises(CapabilityAuthorizationError, match="validity|revoked|unavailable"):
+        verifier.verify(_attestation(private), candidate, plan)
+
+
+def test_builder_verifier_rejects_unknown_key_invalid_signature_and_keyring_digest(tmp_path):
+    private = Ed25519PrivateKey.generate()
+    payload = _keyring_payload(private, purpose="builder_attestation")
+    path = tmp_path / "builder-keyring.json"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    verifier = BuilderAttestationVerifier(BuilderAttestationKeyRing.from_file(path))
+    candidate = SimpleNamespace(
+        source_commit=COMMIT, tree_oid=TREE,
+        artifact_sha256=SHA, manifest_sha256=SHA,
+        gitea_consistency=GITEA, rollback_ready=True,
+    )
+    plan = SimpleNamespace(deployment=SimpleNamespace(package_closure_sha256=SHA, rollback_sha256=SHA), policy_sha256=SHA)
+    with pytest.raises(CapabilityAuthorizationError, match="not trusted"):
+        verifier.verify(
+            dataclasses.replace(_attestation(private), builder_key_id="unknown-builder-key"),
+            candidate,
+            plan,
+        )
+    with pytest.raises(CapabilityAuthorizationError, match="signature"):
+        verifier.verify(
+            dataclasses.replace(_attestation(private), signature=base64.b64encode(b"x" * 64).decode()),
+            candidate,
+            plan,
+        )
+
+    payload["keyring_sha256"] = "0" * 64
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    with pytest.raises(ValueError, match="keyring digest"):
+        BuilderAttestationKeyRing.from_file(path)
+
+
+def test_builder_attestation_rejects_tampered_digest_and_signing_metadata():
+    private = Ed25519PrivateKey.generate()
+    raw = _attestation(private).to_dict()
+    raw["fingerprint"] = "0" * 64
+    with pytest.raises(CapabilityAuthorizationError, match="digest is invalid"):
+        BuilderAttestation.from_dict(raw, require_fingerprint=True)
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("builder_key_generation", 2),
+        ("signature_algorithm", "RSA"),
+        ("signature_purpose", "owner_approval"),
+    ],
+)
+def test_builder_keyring_rejects_attestation_metadata_mismatch(tmp_path, field, value):
+    private = Ed25519PrivateKey.generate()
+    path = tmp_path / "builder-keyring.json"
+    path.write_text(
+        json.dumps(_keyring_payload(private, purpose="builder_attestation")),
+        encoding="utf-8",
+    )
+    verifier = BuilderAttestationVerifier(
+        BuilderAttestationKeyRing.from_file(path)
+    )
+    candidate = SimpleNamespace(
+        source_commit=COMMIT, tree_oid=TREE,
+        artifact_sha256=SHA, manifest_sha256=SHA,
+        gitea_consistency=GITEA, rollback_ready=True,
+    )
+    plan = SimpleNamespace(deployment=SimpleNamespace(package_closure_sha256=SHA, rollback_sha256=SHA), policy_sha256=SHA)
+    with pytest.raises(CapabilityAuthorizationError, match="metadata"):
+        verifier.verify(
+            dataclasses.replace(_attestation(private), **{field: value}),
+            candidate,
+            plan,
+        )
+    raw = _attestation(private).to_dict()
+    raw["builder_key_generation"] = 2
+    with pytest.raises(CapabilityAuthorizationError, match="digest is invalid"):
+        BuilderAttestation.from_dict(raw, require_fingerprint=True)
+
+
+def test_builder_unavailable_intent_is_bounded_retryable_and_recovers_without_secret(tmp_path, monkeypatch):
+    class UnavailableBuilder:
+        def build(self, intent):
+            raise TrustedCandidateBuilderUnavailable("socket token=should-not-leak")
+
+    monkeypatch.setattr(hardening, "_fsync_directory", lambda _path: None)
+    service = SilentReleaseActionService(
+        tmp_path,
+        candidate_builder=UnavailableBuilder(),
+        max_builder_attempts=2,
+    )
+    payload = {"event_id": "event-builder-001", "source_commit": COMMIT}
+
+    first = service.create_release_intent(payload)
+    second = service.create_release_intent(payload)
+
+    assert first["status"] == "BUILDER_RETRYABLE_FAILURE"
+    assert first["retryable"] is True
+    assert first["attempt_count"] == 1
+    assert "should-not-leak" not in json.dumps(first)
+    assert second["status"] == "BUILDER_TERMINAL_FAILURE"
+    assert second["retryable"] is False
+    assert second["attempt_count"] == 2
+
+
+def test_stale_building_intent_recovers_after_restart_without_double_running_active_attempt(tmp_path, monkeypatch):
+    class CountingUnavailableBuilder:
+        calls = 0
+
+        def build(self, intent):
+            self.calls += 1
+            raise TrustedCandidateBuilderUnavailable("nonsecret")
+
+    monkeypatch.setattr(hardening, "_fsync_directory", lambda _path: None)
+    builder = CountingUnavailableBuilder()
+    service = SilentReleaseActionService(
+        tmp_path,
+        candidate_builder=builder,
+        max_builder_attempts=3,
+        builder_attempt_timeout_seconds=1,
+    )
+    payload = {"event_id": "event-builder-crash-001", "source_commit": COMMIT}
+    intent_id = "intent-" + hashlib.sha256(
+        f"{payload['event_id']}\0{COMMIT}".encode()
+    ).hexdigest()[:24]
+    intent_path = tmp_path / "intents" / f"{intent_id}.json"
+    intent_path.parent.mkdir(parents=True)
+    active_record = {
+        "attempt_count": 1,
+        "attempt_started_at": _instant(),
+        "created_at": _instant(timedelta(minutes=-1)),
+        "event_id": payload["event_id"],
+        "intent_id": intent_id,
+        "source_commit": COMMIT,
+        "status": "BUILDING",
+    }
+    intent_path.write_text(json.dumps(active_record), encoding="utf-8")
+
+    active = service.create_release_intent(payload)
+    assert active["status"] == "BUILDING"
+    assert active["idempotent"] is True
+    assert builder.calls == 0
+
+    active_record["attempt_started_at"] = _instant(timedelta(minutes=-5))
+    intent_path.write_text(json.dumps(active_record), encoding="utf-8")
+    recovered = service.create_release_intent(payload)
+    assert recovered["status"] == "BUILDER_RETRYABLE_FAILURE"
+    assert recovered["attempt_count"] == 2
+    assert recovered["failure_code"] == "TRUSTED_BUILDER_UNAVAILABLE"
+    assert recovered["last_recovery_code"] == "TRUSTED_BUILDER_CRASH_RECOVERED"
+    assert recovered["recovery_count"] == 1
+    assert builder.calls == 1
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="model API runtime requires POSIX fcntl")
+@pytest.mark.skipif(
+    not _af_inet_available(),
+    reason="isolated executor permits AF_UNIX only; live HTTP probe runs at activation",
+)
+def test_model_api_keeps_process_liveness_separate_from_builder_readiness(monkeypatch):
+    api = importlib.import_module("aios_model_api_staging")
+
+    class FakeService:
+        ready = False
+
+        def get_builder_readiness(self):
+            return {"status": "READY" if self.ready else "UNAVAILABLE"}
+
+        def create_release_candidate(self, payload):
+            if not payload:
+                raise ValueError("event_id is required")
+            return {"status": "READY_FOR_APPROVAL"}
+
+    service = FakeService()
+    monkeypatch.setattr(api, "silent_release_service", lambda: service)
+    monkeypatch.setattr(
+        api,
+        "authenticate",
+        lambda token: (
+            {"account": "test", "role": "owner", "scopes": ["release:candidate"]}
+            if token == "test-token"
+            else None
+        ),
+    )
+    server = api.ThreadingHTTPServer(("127.0.0.1", 0), api.H)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    base = f"http://127.0.0.1:{server.server_port}"
+
+    def request(path, body=None):
+        data = None if body is None else json.dumps(body).encode("utf-8")
+        value = urllib.request.Request(
+            base + path,
+            data=data,
+            headers={"Authorization": "Bearer test-token"},
+            method="POST" if body is not None else "GET",
+        )
+        try:
+            with urllib.request.urlopen(value, timeout=5) as response:
+                return response.status, json.load(response)
+        except urllib.error.HTTPError as error:
+            return error.code, json.load(error)
+
+    try:
+        health_code, health = request("/health")
+        unavailable_code, unavailable = request("/createReleaseCandidate", {})
+        service.ready = True
+        invalid_code, invalid = request("/createReleaseCandidate", {})
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+    assert health_code == 200
+    assert health["process_liveness"] == "PASS"
+    assert health["builder_readiness"]["status"] == "UNAVAILABLE"
+    assert unavailable_code == 503
+    assert unavailable["error"] == "TRUSTED_BUILDER_UNAVAILABLE"
+    assert invalid_code == 400
+    assert invalid["error"] == "event_id is required"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="model API runtime requires POSIX fcntl")
+def test_host_adapter_accepts_a_real_builder_bundle_response():
+    spec = importlib.util.spec_from_file_location(
+        "trusted_builder_public_seam",
+        APP.parents[1] / "tests" / "test_trusted_candidate_builder.py",
+    )
+    assert spec and spec.loader
+    backend_seam = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(backend_seam)
+    server, client, _ = backend_seam.make_seam()
+    candidate_id = "r42-release-candidate-v2r2c2-20260819"
+    source_commit = "71e736b05fab13efc06af7a4fe2dc022c1cf9a85"
+    response = server.handle_request(
+        client.create_request(
+            "V2R2C2", candidate_id, source_commit, nonce="z" * 32
+        ),
+        connection=object(),
+    )
+
+    class ResponseClient:
+        def build(self, version, supplied_candidate_id, supplied_source, *, nonce):
+            assert version == "V2R2C2"
+            assert supplied_candidate_id == candidate_id
+            assert supplied_source == source_commit
+            assert len(nonce) >= 32
+            return response
+
+    api = importlib.import_module("aios_model_api_staging")
+    bundle = api._HostTrustedCandidateBuilderAdapter(ResponseClient()).build(
+        {
+            "candidate_id": candidate_id,
+            "source_commit": source_commit,
+            "version": "V2R2C2",
+        }
+    )
+
+    assert bundle.validate().candidate.source_commit == response["base_oid"]
+    assert bundle.attestation.builder_key_id == response["attestation"]["builder_key_id"]
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="model API runtime requires POSIX fcntl")
+def test_host_adapter_forwards_v2r2c3_to_the_policy_and_fails_closed_when_unavailable():
+    from aios_trusted_candidate_builder import BuilderError
+    api = importlib.import_module("aios_model_api_staging")
+    calls = []
+
+    class DerivedOnlyClient:
+        def build(self, version, candidate_id, source_commit, *, nonce):
+            calls.append((version, candidate_id, source_commit, nonce))
+            raise BuilderError("GOVERNED_V2R2C3_DERIVATION_REQUIRED")
+
+    with pytest.raises(TrustedCandidateBuilderUnavailable):
+        api._HostTrustedCandidateBuilderAdapter(DerivedOnlyClient()).build(
+            {
+                "candidate_id": "r42-release-candidate-v2r2c3-20260820",
+                "source_commit": COMMIT,
+                "version": "V2R2C3",
+            }
+        )
+    assert calls[0][:3] == (
+        "V2R2C3",
+        "r42-release-candidate-v2r2c3-20260820",
+        COMMIT,
+    )

tests/test_trusted_candidate_builder.py

@@ -0,0 +1,592 @@
+from __future__ import annotations
+
+import dataclasses
+import hashlib
+import hmac
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
+
+import aios_trusted_candidate_builder as builder_runtime  # noqa: E402
+from aios_trusted_candidate_builder import (  # noqa: E402
+    BuilderError,
+    BuilderKeyring,
+    DeterministicEvidenceCollector,
+    PeerIdentity,
+    ServerOwnedEvidenceSource,
+    TrustedCandidateBuilder,
+    TrustedCandidateBuilderConfig,
+    UnixTrustedCandidateBuilderClient,
+    load_builder_policy,
+)
+from silent_release_controller import ReleaseCandidate, ReleaseState  # noqa: E402
+from silent_release_hardening import (  # noqa: E402
+    BuilderAttestation,
+    BuilderAttestationKeyRecord,
+    BuilderAttestationKeyRing,
+    BuilderAttestationVerifier,
+    DeploymentPlan,
+    FinalizationPlan,
+    ReleasePlan,
+    TagPlan,
+    TrustedCandidateBundle,
+)
+
+
+POLICY_PATH = ROOT / "config" / "gitops" / "trusted-candidate-builder-policy.json"
+NOW = 1_725_000_000
+
+
+class RecordingDomainSigner:
+    key_id = "test-builder-key"
+    key_generation = 1
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, bytes]] = []
+
+    def sign_for_purpose(self, purpose: str, payload: bytes) -> bytes:
+        self.calls.append((purpose, payload))
+        return hashlib.sha256(b"test-signature\0" + payload).digest()
+
+
+def make_seam(*, peer: PeerIdentity | None = None, cgroup: str = "/system.slice/aios-gitops-model-api.service", process=("/usr/bin/python3.12", ("/usr/bin/python3", "/opt/aios/gitops-model-api/test/runtime/app/aios_model_api_staging.py")), server_clock=lambda: NOW, client_clock=lambda: NOW, collector=None, nonce_ledger_path=None, signer_override=None) -> tuple[TrustedCandidateBuilder, UnixTrustedCandidateBuilderClient, RecordingDomainSigner]:
+    policy = load_builder_policy(POLICY_PATH)
+    config = TrustedCandidateBuilderConfig.from_policy(policy)
+    keyring = BuilderKeyring(builder_hmac=b"h" * 32)
+    signer = signer_override or RecordingDomainSigner()
+    server = TrustedCandidateBuilder(
+        config,
+        keyring,
+        collector=collector or DeterministicEvidenceCollector(config, evidence_source=passing_bundle),
+        signer=signer,
+        peer_identity_resolver=lambda _connection: peer or PeerIdentity(pid=77, uid=1000, gid=1000),
+        cgroup_resolver=lambda _pid: cgroup,
+        process_identity_resolver=lambda _pid: process,
+        clock=server_clock,
+        nonce_ledger_path=nonce_ledger_path,
+    )
+    return server, UnixTrustedCandidateBuilderClient(config, keyring, clock=client_clock), signer
+
+
+def passing_bundle(recipe_name, recipe):
+    base_oid = recipe["base_oid"]
+    digest = lambda label: hashlib.sha256(f"{label}\0{base_oid}".encode()).hexdigest()
+    candidate = ReleaseCandidate(
+        candidate_id=recipe["candidate_id"], source_commit=base_oid,
+        tree_oid=digest("tree")[:40], artifact_sha256=digest("artifact"),
+        manifest_sha256=digest("manifest"), tests={"passed": True, "receipt": digest("tests")},
+        canary={"passed": True, "receipt": digest("canary")}, rollback_ready=True,
+        git_consistency={"consistent": True}, gitea_consistency={"consistent": True},
+        state=ReleaseState.READY_FOR_APPROVAL,
+    )
+    policy_sha = digest("policy")
+    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=digest("rollback"),
+            source_commit=base_oid, tree_oid=candidate.tree_oid,
+            artifact_sha256=candidate.artifact_sha256, manifest_sha256=candidate.manifest_sha256,
+            package_closure_sha256=digest("closure"), authoritative_git_dir="/data/aios-gitops/repo.git",
+            repository_identity="aios/aios", 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=digest("tag")[:40],
+            expected_target_oid=base_oid, evidence_closure_oid=digest("closure-oid")[:40],
+            approval_plan_id="owner-plan-trusted-v2r2c2", repository_identity="aios/aios",
+            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=digest("tag")[:40], expected_target_oid=base_oid,
+            expected_previous_baseline_sha256=digest("baseline"), registry_expected_record={"base": base_oid},
+            baseline_expected_record={"base": base_oid}, final_consistency_contract="trusted-builder/v1",
+        ), policy_sha256=policy_sha,
+    )
+    checks = {name: True for name in [
+        "authoritative_git", "artifact", "canary", "gitea", "manifest",
+        "package_closure", "rollback", "secret_scan", "tests",
+    ]}
+    TrustedCandidateBundle(candidate, plan, checks).validate()
+    canonical_digest = lambda value: hashlib.sha256(
+        json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest()
+    evidence_digests = {
+        "artifact": candidate.artifact_sha256,
+        "manifest": candidate.manifest_sha256,
+        "package_closure": plan.deployment.package_closure_sha256,
+        "tests": canonical_digest(candidate.tests),
+        "canary": canonical_digest(candidate.canary),
+        "gitea": canonical_digest(candidate.gitea_consistency),
+        "rollback": canonical_digest({
+            "ready": True,
+            "rollback_sha256": plan.deployment.rollback_sha256,
+            "source_commit": candidate.source_commit,
+        }),
+        "policy": plan.policy_sha256,
+        "secret_scan": digest("secret-scan"),
+    }
+    return {
+        "candidate": candidate.to_dict(),
+        "release_plan": plan.to_dict(),
+        "checks": checks,
+        "evidence_digests": evidence_digests,
+    }
+
+
+def rehmac(request: dict) -> None:
+    unsigned = {key: value for key, value in request.items() if key != "hmac"}
+    request["hmac"] = hmac.new(
+        b"h" * 32,
+        json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode(),
+        hashlib.sha256,
+    ).hexdigest()
+
+
+def test_public_build_seam_accepts_only_the_fixed_v2r2c2_recipe_and_returns_server_owned_evidence() -> None:
+    server, client, signer = make_seam()
+
+    request = client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="a" * 32)
+    result = server.handle_request(request, connection=object())
+
+    assert result["schema"] == "aios.trusted-candidate-builder.response.v1"
+    assert result["status"] == "BUILT"
+    assert result["recipe"] == "V2R2C2"
+    assert result["repository"] == "aios/aios"
+    assert result["task_id"] == "r42-release-candidate-v2r2c2-live-20260819"
+    assert result["silent_release_bundle"]["candidate"]["candidate_id"] == "r42-release-candidate-v2r2c2-20260819"
+    assert result["base_oid"] == "71e736b05fab13efc06af7a4fe2dc022c1cf9a85"
+    assert set(result["checks"]) == set(server.config.required_checks)
+    assert all(result["checks"].values())
+    assert result["attestation"]["trust_domain"] == "aios.trusted-builder.attestation.v1"
+    assert signer.calls and signer.calls[0][0] == "builder_attestation"
+    assert result["attestation"]["fingerprint"] == hashlib.sha256(signer.calls[0][1]).hexdigest()
+    assert BuilderAttestation.from_dict(result["attestation"]).canonical_bytes() == signer.calls[0][1]
+    silent_bundle = result["silent_release_bundle"]
+    round_trip = TrustedCandidateBundle(
+        ReleaseCandidate.from_dict(silent_bundle["candidate"]),
+        ReleasePlan.from_dict(silent_bundle["release_plan"]),
+        silent_bundle["checks"],
+    )
+    assert round_trip.validate().candidate.candidate_id == result["evidence"]["candidate_id"]
+
+
+@pytest.mark.parametrize(
+    "mutate, error",
+    [
+        (lambda request: request.update({"base_oid": "f" * 40}), "REQUEST_HMAC_INVALID"),
+        (lambda request: request.update({"artifact_sha256": "0" * 64}), "REQUEST_SCHEMA_INVALID"),
+    ],
+)
+def test_public_build_seam_rejects_tampering_and_caller_owned_evidence(mutate, error) -> None:
+    server, client, _ = make_seam()
+    request = client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="b" * 32)
+    mutate(request)
+
+    with pytest.raises(BuilderError, match=error):
+        server.handle_request(request, connection=object())
+
+
+@pytest.mark.parametrize("field,value", [
+    ("repository", "other/repository"),
+    ("task_id", "other-task"),
+    ("base_oid", "f" * 40),
+    ("purpose", "owner_approval"),
+])
+def test_public_build_seam_rejects_reauthenticated_scope_rebinding(field, value) -> None:
+    server, client, _ = make_seam()
+    request = client.create_request(
+        "V2R2C2",
+        "r42-release-candidate-v2r2c2-20260819",
+        "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+        nonce="s" * 32,
+    )
+    request[field] = value
+    rehmac(request)
+    with pytest.raises(BuilderError, match="REQUEST_SCOPE_INVALID"):
+        server.handle_request(request, connection=object())
+
+
+def test_public_build_seam_rejects_wrong_hmac() -> None:
+    server, client, _ = make_seam()
+    request = client.create_request(
+        "V2R2C2",
+        "r42-release-candidate-v2r2c2-20260819",
+        "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+        nonce="w" * 32,
+    )
+    request["hmac"] = "0" * 64
+    with pytest.raises(BuilderError, match="REQUEST_HMAC_INVALID"):
+        server.handle_request(request, connection=object())
+
+
+def test_public_build_seam_returns_the_original_result_for_exact_nonce_replay_and_rejects_conflicts() -> None:
+    server, client, _ = make_seam()
+    request = client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="c" * 32)
+    first = server.handle_request(request, connection=object())
+    assert server.handle_request(request, connection=object()) == first
+
+    conflict = client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="c" * 32)
+    conflict["issued_at"] += 1
+    unsigned = {key: value for key, value in conflict.items() if key != "hmac"}
+    conflict["hmac"] = hmac.new(b"h" * 32, json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode(), hashlib.sha256).hexdigest()
+
+    with pytest.raises(BuilderError, match="REPLAY_CONFLICT"):
+        server.handle_request(conflict, connection=object())
+
+
+def test_exact_replay_survives_restart_and_the_original_freshness_window(tmp_path) -> None:
+    ledger = tmp_path / "nonce-ledger.json"
+    server, client, _ = make_seam(nonce_ledger_path=ledger)
+    request = client.create_request(
+        "V2R2C2",
+        "r42-release-candidate-v2r2c2-20260819",
+        "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+        nonce="r" * 32,
+    )
+    first = server.handle_request(request, connection=object())
+
+    restarted, _, _ = make_seam(
+        server_clock=lambda: NOW + 3600,
+        nonce_ledger_path=ledger,
+    )
+    assert restarted.handle_request(request, connection=object()) == first
+
+
+def test_readiness_probe_checks_collector_and_signer_without_signing() -> None:
+    server, client, signer = make_seam()
+    request = client.create_request(
+        "V2R2C2",
+        "r42-release-candidate-v2r2c2-20260819",
+        "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+        nonce="q" * 32,
+        operation=server.config.readiness_operation,
+    )
+    result = server.handle_request(request, connection=object())
+
+    assert result["status"] == "READY"
+    assert result["collector_ready"] is True
+    assert result["signer_ready"] is True
+    assert result["builder_key_id"] == signer.key_id
+    assert result["policy_sha256"] == server.config.policy_sha256
+    assert "signature" not in result
+    assert "silent_release_bundle" not in result
+    assert signer.calls == []
+
+
+def test_readiness_fails_closed_when_signer_is_unavailable() -> None:
+    class UnavailableSigner:
+        @property
+        def key_id(self):
+            raise BuilderError("SIGNER_UNAVAILABLE")
+
+        @property
+        def key_generation(self):
+            return 1
+
+        def sign_for_purpose(self, purpose, payload):
+            raise AssertionError("readiness must not attempt signing")
+
+    server, client, _ = make_seam(signer_override=UnavailableSigner())
+    request = client.create_request(
+        "V2R2C2",
+        "r42-release-candidate-v2r2c2-20260819",
+        "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+        nonce="u" * 32,
+        operation=server.config.readiness_operation,
+    )
+    with pytest.raises(BuilderError, match="SIGNER_UNAVAILABLE"):
+        server.handle_request(request, connection=object())
+
+
+def test_unix_client_applies_bounded_timeout_and_fails_closed(monkeypatch) -> None:
+    observed = {}
+
+    class TimeoutSocket:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            return False
+
+        def settimeout(self, seconds):
+            observed["timeout"] = seconds
+
+        def connect(self, _path):
+            raise TimeoutError("bounded test timeout")
+
+    monkeypatch.setattr(builder_runtime.socket, "socket", lambda *_args: TimeoutSocket())
+    monkeypatch.setattr(builder_runtime.socket, "AF_UNIX", 1, raising=False)
+    monkeypatch.setattr(builder_runtime.socket, "SOCK_STREAM", 1, raising=False)
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
+    client = UnixTrustedCandidateBuilderClient(
+        config,
+        BuilderKeyring(builder_hmac=b"h" * 32),
+        socket_path=Path("/run/aios-trusted-candidate-builder.sock"),
+        clock=lambda: NOW,
+    )
+    with pytest.raises(BuilderError, match="BUILDER_TRANSPORT_UNAVAILABLE"):
+        client.build(
+            "V2R2C2",
+            "r42-release-candidate-v2r2c2-20260819",
+            "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+            nonce="t" * 32,
+        )
+    assert observed["timeout"] == config.request_timeout_seconds
+
+
+def test_backend_attestation_round_trip_verifies_with_production_keyring() -> None:
+    private = Ed25519PrivateKey.generate()
+
+    class Ed25519Signer:
+        key_id = "builder-key-001"
+        key_generation = 1
+
+        def sign_for_purpose(self, purpose, payload):
+            assert purpose == "builder_attestation"
+            return private.sign(payload)
+
+    server, client, _ = make_seam(signer_override=Ed25519Signer())
+    response = server.handle_request(
+        client.create_request(
+            "V2R2C2",
+            "r42-release-candidate-v2r2c2-20260819",
+            "71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+            nonce="v" * 32,
+        ),
+        connection=object(),
+    )
+    public_key = private.public_key().public_bytes(
+        serialization.Encoding.Raw,
+        serialization.PublicFormat.Raw,
+    )
+    verifier = BuilderAttestationVerifier(
+        BuilderAttestationKeyRing(
+            [BuilderAttestationKeyRecord(
+                key_id="builder-key-001",
+                public_key=public_key,
+                algorithm="Ed25519",
+                purpose="builder_attestation",
+                generation=1,
+                not_before="2020-01-01T00:00:00Z",
+                expires_at="2099-01-01T00:00:00Z",
+                status="ACTIVE",
+            )],
+            require_attestation_metadata=True,
+        )
+    )
+    raw = response["silent_release_bundle"]
+    bundle = TrustedCandidateBundle(
+        ReleaseCandidate.from_dict(raw["candidate"]),
+        ReleasePlan.from_dict(raw["release_plan"]),
+        raw["checks"],
+        BuilderAttestation.from_dict(raw["attestation"], require_fingerprint=True),
+    )
+    assert bundle.validate(verifier).candidate.candidate_id == response["evidence"]["candidate_id"]
+
+
+def test_public_build_seam_rejects_direct_unbound_v2r2c3_and_authentication_boundaries() -> None:
+    server, client, _ = make_seam()
+    with pytest.raises(BuilderError, match="GOVERNED_V2R2C3_DERIVATION_REQUIRED"):
+        client.create_request("V2R2C3", "unknown", "f" * 40, nonce="d" * 32)
+    denied_peer, denied_client, _ = make_seam(peer=PeerIdentity(pid=77, uid=0, gid=0))
+    with pytest.raises(BuilderError, match="PEER_IDENTITY_DENIED"):
+        denied_peer.handle_request(denied_client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="e" * 32), connection=object())
+    denied_cgroup, denied_client, _ = make_seam(cgroup="/system.slice/untrusted.service")
+    with pytest.raises(BuilderError, match="PEER_CGROUP_DENIED"):
+        denied_cgroup.handle_request(denied_client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="f" * 32), connection=object())
+    denied_process, denied_client, _ = make_seam(
+        process=("/usr/bin/python3.12", ("/usr/bin/python3", "/tmp/aios_model_api_staging.py"))
+    )
+    with pytest.raises(BuilderError, match="PEER_PROCESS_DENIED"):
+        denied_process.handle_request(denied_client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="p" * 32), connection=object())
+    stale_server, stale_client, _ = make_seam(client_clock=lambda: NOW - 61)
+    with pytest.raises(BuilderError, match="REQUEST_STALE"):
+        stale_server.handle_request(stale_client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="g" * 32), connection=object())
+
+
+def test_public_build_seam_fails_closed_when_server_owned_collector_is_incomplete() -> None:
+    def incomplete_evidence(recipe_name, recipe):
+        value = passing_bundle(recipe_name, recipe)
+        value["checks"]["canary"] = False
+        return value
+
+    policy = load_builder_policy(POLICY_PATH)
+    config = TrustedCandidateBuilderConfig.from_policy(policy)
+    server, client, _ = make_seam(
+        collector=DeterministicEvidenceCollector(config, evidence_source=incomplete_evidence)
+    )
+    with pytest.raises(BuilderError, match="COLLECTOR_CHECKS_INCOMPLETE"):
+        server.handle_request(client.create_request("V2R2C2", "r42-release-candidate-v2r2c2-20260819", "71e736b05fab13efc06af7a4fe2dc022c1cf9a85", nonce="h" * 32), connection=object())
+
+
+def test_server_owned_source_rehashes_every_fixed_evidence_file_and_rejects_tampering(tmp_path) -> None:
+    config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
+    recipe = config.fixed_recipe("V2R2C2")
+    raw = passing_bundle("V2R2C2", recipe)
+    candidate = ReleaseCandidate.from_dict(raw["candidate"])
+    plan = ReleasePlan.from_dict(raw["release_plan"])
+    evidence_root = tmp_path / "evidence"
+    version_root = evidence_root / "V2R2C2"
+    version_root.mkdir(parents=True)
+
+    def write_bytes(name: str, value: bytes) -> str:
+        path = version_root / name
+        path.write_bytes(value)
+        path.chmod(0o600)
+        return hashlib.sha256(value).hexdigest()
+
+    def canonical(value: dict) -> bytes:
+        return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+    tests = {"passed": True, "receipt": "tests-fixed"}
+    canary = {"passed": True, "receipt": "canary-fixed"}
+    gitea = {
+        "consistent": True,
+        "remote_ref_oid": recipe["source_commit"],
+        "repository": "aios/aios",
+        "source_commit": recipe["source_commit"],
+        "tree_oid": candidate.tree_oid,
+    }
+    git_consistency = {
+        "consistent": True,
+        "repository": "aios/aios",
+        "source_commit": recipe["source_commit"],
+        "tree_oid": candidate.tree_oid,
+    }
+    rollback = {
+        "ready": True,
+        "rollback_sha256": plan.deployment.rollback_sha256,
+        "source_commit": recipe["source_commit"],
+    }
+    artifact_bytes = b"immutable-artifact"
+    manifest_bytes = b'{"manifest":"fixed"}\n'
+    closure_bytes = b"closure  artifact.tar.gz\n"
+    policy_bytes = b'{"policy":"fixed"}\n'
+    artifact_sha = write_bytes("artifact.tar.gz", artifact_bytes)
+    manifest_sha = write_bytes("manifest.json", manifest_bytes)
+    closure_sha = write_bytes("checksums.sha256", closure_bytes)
+    tests_sha = write_bytes("tests.json", canonical(tests))
+    canary_sha = write_bytes("canary.json", canonical(canary))
+    gitea_sha = write_bytes("gitea.json", canonical(gitea))
+    rollback_sha = write_bytes("rollback.json", canonical(rollback))
+    policy_sha = write_bytes("silent-release-policy.json", policy_bytes)
+    secret_scan = {"artifact_sha256": artifact_sha, "passed": True}
+    secret_sha = write_bytes("secret-scan.json", canonical(secret_scan))
+
+    candidate.artifact_sha256 = artifact_sha
+    candidate.manifest_sha256 = manifest_sha
+    candidate.tests = tests
+    candidate.canary = canary
+    candidate.git_consistency = git_consistency
+    candidate.gitea_consistency = gitea
+    deployment = dataclasses.replace(
+        plan.deployment,
+        artifact_sha256=artifact_sha,
+        manifest_sha256=manifest_sha,
+        package_closure_sha256=closure_sha,
+        policy_sha256=policy_sha,
+    )
+    plan = dataclasses.replace(plan, deployment=deployment, policy_sha256=policy_sha)
+    receipt = {
+        "schema": "aios.trusted-candidate-builder.evidence.v1",
+        "version": "V2R2C2",
+        "candidate": candidate.to_dict(),
+        "release_plan": plan.to_dict(),
+    }
+    write_bytes("receipt.json", canonical(receipt))
+
+    git_state = {"tree": candidate.tree_oid, "ancestry": 0}
+
+    def git_runner(argv, **_kwargs):
+        if "rev-parse" in argv:
+            return SimpleNamespace(stdout=git_state["tree"] + "\n", returncode=0)
+        return SimpleNamespace(stdout="", returncode=git_state["ancestry"])
+
+    source = ServerOwnedEvidenceSource(
+        config,
+        tmp_path / "repo.git",
+        evidence_root,
+        expected_owner_uid=None,
+        git_runner=git_runner,
+    )
+    collected = source("V2R2C2", recipe)
+    assert collected["candidate"]["artifact_sha256"] == artifact_sha
+    assert collected["evidence_digests"] == {
+        "artifact": artifact_sha,
+        "manifest": manifest_sha,
+        "package_closure": closure_sha,
+        "tests": tests_sha,
+        "canary": canary_sha,
+        "gitea": gitea_sha,
+        "rollback": rollback_sha,
+        "policy": policy_sha,
+        "secret_scan": secret_sha,
+    }
+
+    (version_root / "manifest.json").write_bytes(b"tampered")
+    with pytest.raises(BuilderError, match="COLLECTOR_DIGEST_MISMATCH"):
+        source("V2R2C2", recipe)
+    (version_root / "manifest.json").write_bytes(manifest_bytes)
+
+    for name, tampered, expected in (
+        ("artifact.tar.gz", b"tampered", "COLLECTOR_DIGEST_MISMATCH"),
+        ("checksums.sha256", b"tampered", "COLLECTOR_DIGEST_MISMATCH"),
+        ("silent-release-policy.json", b"tampered", "COLLECTOR_DIGEST_MISMATCH"),
+        ("tests.json", canonical({"passed": True, "receipt": "wrong"}), "COLLECTOR_TESTS_MISMATCH"),
+        ("canary.json", canonical({"passed": True, "receipt": "wrong"}), "COLLECTOR_CANARY_MISMATCH"),
+        ("gitea.json", canonical({**gitea, "consistent": False}), "COLLECTOR_GITEA_MISMATCH"),
+        ("rollback.json", canonical({**rollback, "ready": False}), "COLLECTOR_ROLLBACK_MISMATCH"),
+        ("secret-scan.json", canonical({**secret_scan, "passed": False}), "COLLECTOR_SECRET_SCAN_MISMATCH"),
+    ):
+        path = version_root / name
+        original = path.read_bytes()
+        path.write_bytes(tampered)
+        with pytest.raises(BuilderError, match=expected):
+            source("V2R2C2", recipe)
+        path.write_bytes(original)
+
+    git_state["tree"] = "f" * 40
+    with pytest.raises(BuilderError, match="COLLECTOR_GIT_MISMATCH"):
+        source("V2R2C2", recipe)
+    git_state["tree"] = candidate.tree_oid
+    git_state["ancestry"] = 1
+    with pytest.raises(BuilderError, match="COLLECTOR_GIT_MISMATCH"):
+        source("V2R2C2", recipe)
+
+
+def test_systemd_socket_and_service_are_af_unix_hardened_and_credential_bound() -> None:
+    socket_unit = (ROOT / "deploy/systemd/aios-trusted-candidate-builder.socket").read_text()
+    service_unit = (ROOT / "deploy/systemd/aios-trusted-candidate-builder.service").read_text()
+
+    assert "ListenStream=/run/aios-trusted-candidate-builder.sock" in socket_unit
+    assert "SocketMode=0660" in socket_unit
+    assert "LoadCredentialEncrypted=signing_private_key:" in service_unit
+    assert "LoadCredentialEncrypted=builder_hmac:" in service_unit
+    assert "NoNewPrivileges=true" in service_unit
+    assert "ProtectSystem=strict" in service_unit
+    assert "PrivateNetwork=true" in service_unit
+    assert "RestrictAddressFamilies=AF_UNIX" in service_unit
+    assert "StateDirectory=aios-trusted-candidate-builder" in service_unit
+    assert "AIOS_TRUSTED_BUILDER_KEY_GENERATION=1" in service_unit
+    assert "aios-approved-release-executor" not in service_unit
+    assert "--socket-fd 3" in service_unit
+    policy = load_builder_policy(POLICY_PATH)
+    assert policy["allowed_peer_uid"] == 1000
+    assert policy["allowed_peer_cgroups"] == [
+        "/system.slice/aios-gitops-model-api.service"
+    ]