fix(governance): classify release tag target mismatches

active-runtime/app/aios_chatgpt_actions_api.py

@@ -3,6 +3,12 @@ from __future__ import annotations
 import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, urllib.request, uuid
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
+from version_registry import (
+    configured_authoritative_git_dir,
+    configured_repository_identity,
+    list_registry_versions,
+    list_registry_versions_report,
+)
 from typing import Any
 
 try:
@@ -72,8 +78,8 @@ HOST = os.getenv('AIOS_GITOPS_HOST', '127.0.0.1')
 PORT = int(os.getenv('AIOS_GITOPS_PORT', '18890'))
 OPENAPI = WS / 'docs' / 'gitops' / 'openapi-chatgpt-actions.yaml'
 BARE_REPO = Path(os.getenv('AIOS_GITOPS_BARE_REPO', str(ROOT / 'runtime' / 'gitea-bin' / 'data' / 'repositories' / 'aios' / 'aios.git')))
-AUTHORITATIVE_GIT_DIR = Path(os.getenv('AIOS_GITOPS_AUTHORITATIVE_GIT_DIR')) if os.getenv('AIOS_GITOPS_AUTHORITATIVE_GIT_DIR') else None
-REPOSITORY_IDENTITY = os.getenv('AIOS_GITOPS_REPOSITORY_IDENTITY', '')
+AUTHORITATIVE_GIT_DIR = configured_authoritative_git_dir()
+REPOSITORY_IDENTITY = configured_repository_identity()
 SILENT_RELEASE_STATE_ROOT = Path(os.getenv(
     'AIOS_SILENT_RELEASE_STATE_ROOT',
     '/var/lib/aios-gitops/silent-release',
@@ -466,6 +472,14 @@ class H(BaseHTTPRequestHandler):
                 ))
                 return self.sendj(data)
             if path in ('/api/ver','/api/versions','/ver'):
+                registry_report = list_registry_versions_report(
+                    WS,
+                    baseline_file=Path(os.getenv('AIOS_GITOPS_BASELINE_AUTHORITY', '/var/lib/aios-gitops-runtime-anchor/baseline/current.json')),
+                    # Do not invent an authoritative repository when authority
+                    # discovery is disabled/omitted. Preserve the unavailable
+                    # authority path (503) instead of turning it into a 404.
+                    repo_git_dir=AUTHORITATIVE_GIT_DIR,
+                )
                 snapshot = repository_snapshot(
                     git_dir=AUTHORITATIVE_GIT_DIR,
                     repository_identity=REPOSITORY_IDENTITY,
@@ -474,7 +488,8 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({
                         'repository_available': False,
                         'query_error': snapshot['query_error'],
-                        'registry_versions': [p.name for p in (WS/'versions').glob('*')],
+                        'registry_versions': registry_report['registry_versions'],
+                        **{key: registry_report[key] for key in ('effective_registry_versions', 'runtime_registry_versions', 'attested_projection_versions', 'attested_projection_version', 'source_registry_converged', 'runtime_registry_converged', 'runtime_bundle_contains_current_version', 'runtime_bundle_contains_version', 'registry_consistency_status')},
                         'local_authoritative_tags': [],
                     }, 503)
                 try:
@@ -483,10 +498,11 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({
                         'repository_available': False,
                         'query_error': str(exc),
-                        'registry_versions': [p.name for p in (WS/'versions').glob('*')],
+                        'registry_versions': registry_report['registry_versions'],
+                        **{key: registry_report[key] for key in ('effective_registry_versions', 'runtime_registry_versions', 'attested_projection_versions', 'attested_projection_version', 'source_registry_converged', 'runtime_registry_converged', 'runtime_bundle_contains_current_version', 'runtime_bundle_contains_version', 'registry_consistency_status')},
                         'local_authoritative_tags': [],
                     }, 503)
-                registry = [p.name for p in (WS/'versions').glob('*')]
+                registry = registry_report['registry_versions']
                 return self.sendj({
                     'registry_versions': registry,
                     'local_authoritative_tags': tags,
@@ -496,6 +512,7 @@ class H(BaseHTTPRequestHandler):
                     'versions_dir': registry,
                     'head': snapshot['authoritative_head_oid'],
                     'authoritative_head_oid': snapshot['authoritative_head_oid'],
+                    **{key: registry_report[key] for key in ('effective_registry_versions', 'runtime_registry_versions', 'attested_projection_versions', 'attested_projection_version', 'source_registry_converged', 'runtime_registry_converged', 'runtime_bundle_contains_current_version', 'runtime_bundle_contains_version', 'registry_consistency_status')},
                 })
             if path in ('/api/diff','/diff'):
                 frm=(q.get('from') or [''])[0]
@@ -950,4 +967,3 @@ def main():
     print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
     ThreadingHTTPServer((HOST,PORT), H).serve_forever()
 if __name__ == '__main__': main()
-

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -15,6 +15,7 @@ BRIDGE_OPERATIONS = (
     "acquireWriterLease",
     "commitFileGoverned",
     "commitFilesGoverned",
+    "publishExistingBranchGoverned",
     "getGitOpsJob",
     "resumeGitOpsJob",
     "reconcileGitOpsJob",
@@ -33,6 +34,7 @@ OPERATION_REQUIRED_FIELDS = {
     "acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
     "commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
     "commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
+    "publishExistingBranchGoverned": ("repository", "branch", "expected_branch_oid", "owner_approval_reference", "idempotency_key"),
     "getGitOpsJob": ("job_id",),
     "resumeGitOpsJob": ("job_id", "expected_revision"),
     "reconcileGitOpsJob": ("job_id", "expected_revision"),
@@ -153,6 +155,7 @@ class BridgeV2Adapter:
     def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
     def commitFileGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFileGoverned", request)
     def commitFilesGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFilesGoverned", request)
+    def publishExistingBranchGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("publishExistingBranchGoverned", request)
     def getGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("getGitOpsJob", request)
     def resumeGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("resumeGitOpsJob", request)
     def reconcileGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("reconcileGitOpsJob", request)

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -19,6 +19,7 @@ from aios_continuation_lease import WriterLeaseRequestV2
 from aios_control_plane_snapshot import build_control_plane_snapshot
 from aios_gitops_job_controller import GitOpsJobController, JobStore
 from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
+from aios_existing_branch_publisher import ExistingBranchPublisher
 from governed_production_release import build_executor_config
 
 MAX_BODY = 2 * 1024 * 1024
@@ -29,6 +30,7 @@ ROUTES = {
     "/v2/writer-lease": "acquireWriterLease",
     "/v2/commit-file": "commitFileGoverned",
     "/v2/commit-files": "commitFilesGoverned",
+    "/v2/branch/publish": "publishExistingBranchGoverned",
     "/v2/job": "getGitOpsJob",
     "/v2/job/resume": "resumeGitOpsJob",
     "/v2/job/reconcile": "reconcileGitOpsJob",
@@ -150,6 +152,36 @@ class GovernedRuntime:
         executor.workspace.materialize_files(commit.files)
         return executor.execute(binding, commit)
 
+    def publish_existing_branch(self, request: Mapping[str, Any]) -> dict[str, Any]:
+        expected = str(request["expected_branch_oid"])
+        key = _safe("existing-branch-" + str(request["idempotency_key"]))
+        worktree_root = self.config.state_root / "worktrees"
+        worktree_path = worktree_root / key
+        if not worktree_path.exists():
+            self.authority.create_detached_worktree(expected, key, worktree_root)
+        workspace = DetachedGitWorkspace(worktree_path)
+        credential = ServerSideCredentialProjection(self.config.ssh_key_path, self.config.token_path)
+        remote = GiteaContinuationRemote(
+            workspace=worktree_path,
+            credential=credential,
+            remote_name=self.config.remote_name,
+            repository=self.config.repository,
+            api_base=self.config.api_base,
+        )
+        publisher = ExistingBranchPublisher(
+            self.authority,
+            remote,
+            self.config.state_root / "existing-branch-publications",
+        )
+        projected = False
+        try:
+            credential.project()
+            projected = True
+            return publisher.publish(request)
+        finally:
+            if projected:
+                credential.revoke()
+
     def _controller_for_job(self, job_id: str) -> GitOpsJobController:
         for controller in self.controllers.values():
             try:
@@ -345,6 +377,7 @@ class GovernedRuntime:
             "acquireWriterLease": self.acquire_lease,
             "commitFileGoverned": lambda request: self.commit_files({**request, "files": {request["path"]: request["content_base64"]}}),
             "commitFilesGoverned": self.commit_files,
+            "publishExistingBranchGoverned": self.publish_existing_branch,
             "getGitOpsJob": lambda request: self.job(request, "get"),
             "resumeGitOpsJob": lambda request: self.job(request, "resume"),
             "reconcileGitOpsJob": lambda request: self.job(request, "reconcile"),

active-runtime/app/aios_continuation_executor.py

@@ -258,7 +258,7 @@ class GiteaContinuationRemote:
             raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
         return oid
 
-    def push_exact(self, branch: str, expected: str, new: str) -> None:
+    def push_exact(self, branch: str, expected: str | None, new: str) -> None:
         self._require_projected()
         self.repository_identity()
         if self.read_git_ref(branch) != expected:

active-runtime/app/aios_existing_branch_publisher.py

@@ -0,0 +1,252 @@
+"""Fail-closed governed publication of an existing Authority branch at an exact OID."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import asdict, dataclass
+from enum import Enum
+import hashlib
+import json
+import os
+import re
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Protocol
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+BRANCH_RE = re.compile(r"^(?!refs/)(?!tags/)[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
+
+
+class ExistingBranchPublicationError(RuntimeError):
+    """Stable, non-secret error for the governed existing-branch publisher."""
+
+
+class ExistingBranchPublicationConflict(ExistingBranchPublicationError):
+    pass
+
+
+class ExistingBranchPublicationState(str, Enum):
+    QUEUED = "QUEUED"
+    PUBLISHING = "PUBLISHING"
+    REMOTE_PUSH_PENDING = "REMOTE_PUSH_PENDING"
+    RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED"
+    MANUAL_OWNER_REVIEW_REQUIRED = "MANUAL_OWNER_REVIEW_REQUIRED"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+
+
+@dataclass(frozen=True, slots=True)
+class ExistingBranchPublicationRequest:
+    repository: str
+    branch: str
+    expected_branch_oid: str
+    owner_approval_reference: str
+    idempotency_key: str
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "ExistingBranchPublicationRequest":
+        required = ("repository", "branch", "expected_branch_oid", "owner_approval_reference", "idempotency_key")
+        if not isinstance(value, Mapping):
+            raise ExistingBranchPublicationConflict("REQUEST_OBJECT_REQUIRED")
+        if any(not isinstance(value.get(field), str) or not value[field].strip() for field in required):
+            raise ExistingBranchPublicationConflict("REQUEST_FIELDS_MISSING")
+        repository = value["repository"].strip()
+        branch = value["branch"].strip()
+        expected = value["expected_branch_oid"].strip().lower()
+        approval = value["owner_approval_reference"].strip()
+        idempotency = value["idempotency_key"].strip()
+        if repository != "aios/aios":
+            raise ExistingBranchPublicationConflict("REPOSITORY_IDENTITY_MISMATCH")
+        if not BRANCH_RE.fullmatch(branch) or branch in {"main", "master"}:
+            raise ExistingBranchPublicationConflict("BRANCH_NAMESPACE_REJECTED")
+        if not OID_RE.fullmatch(expected):
+            raise ExistingBranchPublicationConflict("EXPECTED_BRANCH_OID_INVALID")
+        if len(approval) > 512:
+            raise ExistingBranchPublicationConflict("OWNER_APPROVAL_REFERENCE_INVALID")
+        if not IDEMPOTENCY_RE.fullmatch(idempotency):
+            raise ExistingBranchPublicationConflict("IDEMPOTENCY_KEY_INVALID")
+        return cls(repository, branch, expected, approval, idempotency)
+
+    @property
+    def publication_job_id(self) -> str:
+        return "publication-" + hashlib.sha256(self.idempotency_key.encode()).hexdigest()[:32]
+
+    @property
+    def digest(self) -> str:
+        raw = json.dumps(asdict(self), sort_keys=True, separators=(",", ":")).encode()
+        return hashlib.sha256(raw).hexdigest()
+
+
+class ExistingBranchAuthority(Protocol):
+    def read_ref(self, ref: str) -> str | None: ...
+
+
+class ExistingBranchRemote(Protocol):
+    def repository_identity(self) -> str: ...
+    def read_git_ref(self, branch: str) -> str | None: ...
+    def read_api_branch(self, branch: str) -> str | None: ...
+    def commit_visible(self, oid: str) -> bool: ...
+    def push_exact(self, branch: str, expected: str | None, new: str) -> None: ...
+
+
+class ExistingBranchPublicationStore:
+    def __init__(self, root: Path):
+        self.root = Path(root)
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, request: ExistingBranchPublicationRequest) -> Path:
+        return self.root / f"{request.publication_job_id}.json"
+
+    def load(self, request: ExistingBranchPublicationRequest) -> dict[str, Any]:
+        try:
+            value = json.loads(self._path(request).read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            return {}
+        if not isinstance(value, dict):
+            raise ExistingBranchPublicationConflict("PUBLICATION_STATE_INVALID")
+        return value
+
+    def record(self, request: ExistingBranchPublicationRequest, state: ExistingBranchPublicationState, **evidence: Any) -> dict[str, Any]:
+        payload = {
+            "schema": "aios.existing-branch-publication.v1",
+            "publication_job_id": request.publication_job_id,
+            "repository": request.repository,
+            "branch": request.branch,
+            "expected_branch_oid": request.expected_branch_oid,
+            "owner_approval_reference": request.owner_approval_reference,
+            "idempotency_key": request.idempotency_key,
+            "request_digest": request.digest,
+            "publication_state": state.value,
+            "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            **evidence,
+        }
+        path = self._path(request)
+        fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                json.dump(payload, handle, ensure_ascii=False, sort_keys=True)
+                handle.write("\n")
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temporary, path)
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
+        return payload
+
+
+class ExistingBranchPublisher:
+    """Publish only the exact, already-authorized Authority branch tip."""
+
+    def __init__(self, authority: ExistingBranchAuthority, remote: ExistingBranchRemote, state_root: Path):
+        self.authority = authority
+        self.remote = remote
+        self.store = ExistingBranchPublicationStore(state_root)
+
+    @staticmethod
+    def _safe_error(exc: BaseException) -> str:
+        return type(exc).__name__
+
+    def _observe(self, request: ExistingBranchPublicationRequest) -> dict[str, Any]:
+        return {
+            "authority_branch_oid": self.authority.read_ref(f"refs/heads/{request.branch}"),
+            "gitea_git_branch_oid": self.remote.read_git_ref(request.branch),
+            "gitea_api_branch_oid": self.remote.read_api_branch(request.branch),
+            "commit_visible": bool(self.remote.commit_visible(request.expected_branch_oid)),
+        }
+
+    @staticmethod
+    def _has_remote_conflict(request: ExistingBranchPublicationRequest, observation: Mapping[str, Any]) -> bool:
+        return any(
+            observation.get(field) not in (None, request.expected_branch_oid)
+            for field in ("gitea_git_branch_oid", "gitea_api_branch_oid")
+        )
+
+    def _manual(self, request: ExistingBranchPublicationRequest, reason: str, observation: Mapping[str, Any]) -> dict[str, Any]:
+        return self.store.record(
+            request,
+            ExistingBranchPublicationState.MANUAL_OWNER_REVIEW_REQUIRED,
+            error_code=reason,
+            **dict(observation),
+        )
+
+    def _finish(self, request: ExistingBranchPublicationRequest) -> dict[str, Any]:
+        try:
+            observation = self._observe(request)
+        except Exception as exc:
+            return self.store.record(
+                request,
+                ExistingBranchPublicationState.RECONCILIATION_REQUIRED,
+                error_code=self._safe_error(exc),
+            )
+        if observation.get("authority_branch_oid") != request.expected_branch_oid:
+            return self._manual(request, "AUTHORITY_BRANCH_OID_CHANGED", observation)
+        if self._has_remote_conflict(request, observation):
+            return self._manual(request, "REMOTE_BRANCH_OID_CONFLICT", observation)
+        if (
+            observation.get("gitea_git_branch_oid") == request.expected_branch_oid
+            and observation.get("gitea_api_branch_oid") == request.expected_branch_oid
+            and observation.get("commit_visible") is True
+        ):
+            return self.store.record(request, ExistingBranchPublicationState.COMPLETED, **dict(observation))
+        return self.store.record(
+            request,
+            ExistingBranchPublicationState.RECONCILIATION_REQUIRED,
+            error_code="REMOTE_READBACK_INCOMPLETE",
+            **dict(observation),
+        )
+
+    def publish(self, value: Mapping[str, Any]) -> dict[str, Any]:
+        request = ExistingBranchPublicationRequest.from_mapping(value)
+        previous = self.store.load(request)
+        if previous:
+            if previous.get("request_digest") != request.digest:
+                raise ExistingBranchPublicationConflict("IDEMPOTENCY_KEY_REUSED")
+            if previous.get("publication_state") == ExistingBranchPublicationState.COMPLETED.value:
+                return {**previous, "idempotent": True}
+        authority_oid = self.authority.read_ref(f"refs/heads/{request.branch}")
+        if authority_oid is None:
+            raise ExistingBranchPublicationConflict("AUTHORITY_BRANCH_MISSING")
+        if authority_oid != request.expected_branch_oid:
+            raise ExistingBranchPublicationConflict("AUTHORITY_BRANCH_OID_MISMATCH")
+        if self.remote.repository_identity() != request.repository:
+            raise ExistingBranchPublicationConflict("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
+        self.store.record(request, ExistingBranchPublicationState.QUEUED)
+        try:
+            observation = self._observe(request)
+        except Exception as exc:
+            return self.store.record(
+                request,
+                ExistingBranchPublicationState.RECONCILIATION_REQUIRED,
+                error_code=self._safe_error(exc),
+            )
+        if observation.get("authority_branch_oid") != request.expected_branch_oid:
+            return self._manual(request, "AUTHORITY_BRANCH_OID_CHANGED", observation)
+        if self._has_remote_conflict(request, observation):
+            return self._manual(request, "REMOTE_BRANCH_OID_CONFLICT", observation)
+        if (
+            observation.get("gitea_git_branch_oid") == request.expected_branch_oid
+            and observation.get("gitea_api_branch_oid") == request.expected_branch_oid
+            and observation.get("commit_visible") is True
+        ):
+            return self.store.record(request, ExistingBranchPublicationState.COMPLETED, **dict(observation))
+        if observation.get("gitea_git_branch_oid") is not None or observation.get("gitea_api_branch_oid") is not None:
+            return self.store.record(
+                request,
+                ExistingBranchPublicationState.RECONCILIATION_REQUIRED,
+                error_code="REMOTE_READBACK_INCOMPLETE",
+                **dict(observation),
+            )
+        self.store.record(request, ExistingBranchPublicationState.PUBLISHING, **dict(observation))
+        try:
+            self.remote.push_exact(request.branch, None, request.expected_branch_oid)
+        except Exception as exc:
+            return self.store.record(
+                request,
+                ExistingBranchPublicationState.REMOTE_PUSH_PENDING,
+                error_code=self._safe_error(exc),
+                desired_remote_oid=request.expected_branch_oid,
+                **dict(observation),
+            )
+        return self._finish(request)

active-runtime/app/aios_model_api_staging.py

@@ -3,6 +3,12 @@ from __future__ import annotations
 import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, urllib.request, uuid
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
+from version_registry import (
+    configured_authoritative_git_dir,
+    configured_repository_identity,
+    list_registry_versions,
+    list_registry_versions_report,
+)
 from baseline_authority import BaselineAuthority, BaselineAuthorityError
 from typing import Any
 
@@ -15,6 +21,9 @@ except ModuleNotFoundError:
 from gitea_tag_publication import (
     GiteaTagClient,
     GitTagBackend,
+    GiteaConfigurationError,
+    InvalidReleaseVersion,
+    PublicationError,
     PublicationConflict,
     PublicationPending,
     PublicationManualOwnerReviewRequired,
@@ -73,10 +82,13 @@ HOST = os.getenv('AIOS_GITOPS_HOST', '127.0.0.1')
 PORT = int(os.getenv('AIOS_GITOPS_PORT', '18890'))
 OPENAPI = WS / 'docs' / 'gitops' / 'openapi-chatgpt-actions.yaml'
 BARE_REPO = Path(os.getenv('AIOS_GITOPS_BARE_REPO', str(ROOT / 'runtime' / 'gitea-bin' / 'data' / 'repositories' / 'aios' / 'aios.git')))
-AUTHORITATIVE_GIT_DIR = Path(os.getenv('AIOS_GITOPS_AUTHORITATIVE_GIT_DIR')) if os.getenv('AIOS_GITOPS_AUTHORITATIVE_GIT_DIR') else None
-REPOSITORY_IDENTITY = os.getenv('AIOS_GITOPS_REPOSITORY_IDENTITY', '')
+AUTHORITATIVE_GIT_DIR = configured_authoritative_git_dir()
+REPOSITORY_IDENTITY = configured_repository_identity()
 BASELINE_AUTHORITY = BaselineAuthority()
-GIT_DIR = Path(os.getenv('AIOS_GITOPS_AUTHORITATIVE_GIT_DIR') or os.getenv('AIOS_GITOPS_REPO', '/data/aios-gitops/repo.git'))
+# Retain the generic Git directory only for legacy/staging compatibility
+# commands.  Source-registry convergence receives the separate authority
+# below and never falls back to this path.
+GIT_DIR = Path(os.getenv('AIOS_GITOPS_REPO', '/data/aios-gitops/repo.git'))
 LEGACY_GIT_WRITE_COMMANDS = frozenset({'add','commit','checkout','branch','tag','push','fetch','reset','merge','rebase','update-ref','worktree'})
 SILENT_RELEASE_STATE_ROOT = Path(os.getenv(
     'AIOS_SILENT_RELEASE_STATE_ROOT',
@@ -358,12 +370,7 @@ def publication_publisher() -> TagPublisher:
     return TagPublisher(
         WS,
         GitTagBackend(WS, authoritative_repo=authoritative_repo),
-        GiteaTagClient(
-            os.getenv('AIOS_GITOPS_GITEA_API_URL',''),
-            os.getenv('AIOS_GITOPS_GITEA_REPOSITORY',''),
-            Path(os.getenv('AIOS_GITOPS_GITEA_TOKEN_FILE','')),
-            token_line=int(os.getenv('AIOS_GITOPS_GITEA_TOKEN_LINE','1')),
-        ),
+        GiteaTagClient.from_authoritative_environment(),
         os.getenv('AIOS_GITOPS_GITEA_REMOTE','gitea'),
     )
 
@@ -495,6 +502,11 @@ class H(BaseHTTPRequestHandler):
                 data['legacy_global_writes_enabled'] = False
                 return self.sendj(data)
             if path in ('/api/ver','/api/versions','/ver'):
+                registry_report = list_registry_versions_report(
+                    WS,
+                    baseline_file=BASELINE_AUTHORITY.path,
+                    repo_git_dir=AUTHORITATIVE_GIT_DIR,
+                )
                 try:
                     baseline = dict(BASELINE_AUTHORITY.read())
                 except BaselineAuthorityError as exc:
@@ -507,7 +519,8 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({
                         'repository_available': False,
                         'query_error': snapshot['query_error'],
-                        'registry_versions': [p.name for p in (WS/'versions').glob('*')],
+                        'registry_versions': registry_report['registry_versions'],
+                        **{key: registry_report[key] for key in ('effective_registry_versions', 'runtime_registry_versions', 'attested_projection_versions', 'attested_projection_version', 'source_registry_converged', 'runtime_registry_converged', 'runtime_bundle_contains_current_version', 'runtime_bundle_contains_version', 'registry_consistency_status')},
                         'local_authoritative_tags': [],
                         'baseline_version': baseline.get('current_baseline'),
                         'baseline_generation': baseline.get('generation'),
@@ -519,13 +532,14 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({
                         'repository_available': False,
                         'query_error': str(exc),
-                        'registry_versions': [p.name for p in (WS/'versions').glob('*')],
+                        'registry_versions': registry_report['registry_versions'],
+                        **{key: registry_report[key] for key in ('effective_registry_versions', 'runtime_registry_versions', 'attested_projection_versions', 'attested_projection_version', 'source_registry_converged', 'runtime_registry_converged', 'runtime_bundle_contains_current_version', 'runtime_bundle_contains_version', 'registry_consistency_status')},
                         'local_authoritative_tags': [],
                         'baseline_version': baseline.get('current_baseline'),
                         'baseline_generation': baseline.get('generation'),
                         'baseline_sha256': baseline.get('authority_sha256'),
                     }, 503)
-                registry = [p.name for p in (WS/'versions').glob('*')]
+                registry = registry_report['registry_versions']
                 return self.sendj({
                     'registry_versions': registry,
                     'local_authoritative_tags': tags,
@@ -539,6 +553,7 @@ class H(BaseHTTPRequestHandler):
                     'baseline_generation': baseline.get('generation'),
                     'baseline_sha256': baseline.get('authority_sha256'),
                     'baseline': baseline,
+                    **{key: registry_report[key] for key in ('effective_registry_versions', 'runtime_registry_versions', 'attested_projection_versions', 'attested_projection_version', 'source_registry_converged', 'runtime_registry_converged', 'runtime_bundle_contains_current_version', 'runtime_bundle_contains_version', 'registry_consistency_status')},
                 })
             if path in ('/api/diff','/diff'):
                 frm=(q.get('from') or [''])[0]
@@ -634,12 +649,22 @@ class H(BaseHTTPRequestHandler):
                 try:
                     version=validate_version(version)
                     status=publication_publisher().get_tag_publication_status(version)
-                except ValueError:
-                    return self.sendj({'error':'invalid release version'},400)
+                except InvalidReleaseVersion as exc:
+                    return self.sendj({'error': exc.code},400)
+                except GiteaConfigurationError as exc:
+                    return self.sendj({'error': exc.code},503)
+                except PublicationError as exc:
+                    return self.sendj({'error': getattr(exc, 'code', str(exc))},503)
                 return self.sendj(status)
             if path == '/api/release-consistency':
                 request=publication_request_from_query(q)
-                return self.sendj(publication_publisher().get_release_consistency(request))
+                try:
+                    result=publication_publisher().get_release_consistency(request)
+                except GiteaConfigurationError as exc:
+                    return self.sendj({'error': exc.code},503)
+                except PublicationError as exc:
+                    return self.sendj({'error': getattr(exc, 'code', str(exc))},503)
+                return self.sendj(result)
             if path == '/api/release-acceptance':
                 request=publication_request_from_query(q)
                 return self.sendj(publication_publisher().run_release_acceptance(request))
@@ -1019,4 +1044,3 @@ def main():
     print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
     ThreadingHTTPServer((HOST,PORT), H).serve_forever()
 if __name__ == '__main__': main()
-

active-runtime/app/aios_signer_broker.py

@@ -0,0 +1,86 @@
+"""Narrow, non-root client contract for the privileged signer broker."""
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+from pathlib import Path
+import re
+import socket
+from typing import Any, Protocol
+
+DEFAULT_SIGNER_BROKER_SOCKET = Path("/run/aios/signer-broker.sock")
+MAX_SIGNING_PAYLOAD = 1024 * 1024
+CODE_RE = re.compile(r"^[A-Z0-9_.:-]{3,96}$")
+
+
+class SignerBrokerError(RuntimeError):
+    def __init__(self, code: str):
+        self.code = code
+        super().__init__(code)
+
+
+class SignerBroker(Protocol):
+    def self_test(self) -> None: ...
+    def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]: ...
+
+
+class UnixSignerBroker:
+    """AF_UNIX request client; it never accepts commands, paths, or shell text."""
+
+    def __init__(self, socket_path: Path = DEFAULT_SIGNER_BROKER_SOCKET, timeout: float = 15.0):
+        path = Path(socket_path)
+        if path != DEFAULT_SIGNER_BROKER_SOCKET or not path.is_absolute():
+            raise SignerBrokerError("SIGNER_BROKER_SOCKET_INVALID")
+        self.socket_path = path
+        self.timeout = timeout
+
+    def _request(self, operation: str, **payload: Any) -> dict[str, Any]:
+        if operation not in {"self_test", "sign"}:
+            raise SignerBrokerError("SIGNER_OPERATION_REJECTED")
+        request = {
+            "schema": "aios.signer-broker.request.v1",
+            "operation": operation,
+            **payload,
+        }
+        try:
+            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel:
+                channel.settimeout(self.timeout)
+                channel.connect(str(self.socket_path))
+                channel.sendall((json.dumps(request, sort_keys=True, separators=(",", ":")) + "\n").encode())
+                raw = channel.makefile("rb").readline()
+        except (OSError, TimeoutError) as exc:
+            raise SignerBrokerError("SIGNER_BROKER_UNAVAILABLE") from exc
+        try:
+            response = json.loads(raw.decode())
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID") from exc
+        if not isinstance(response, dict) or response.get("schema") != "aios.signer-broker.response.v1":
+            raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID")
+        if response.get("status") != "PASS":
+            code = str(response.get("error_code") or "SIGNER_BROKER_REJECTED")
+            raise SignerBrokerError(code if CODE_RE.fullmatch(code) else "SIGNER_BROKER_REJECTED")
+        return response
+
+    def self_test(self) -> None:
+        self._request("self_test")
+
+    def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]:
+        if payload_kind not in {"approval_plan", "grant_payload"}:
+            raise SignerBrokerError("SIGNER_PAYLOAD_KIND_REJECTED")
+        if not isinstance(payload, bytes) or not payload or len(payload) > MAX_SIGNING_PAYLOAD:
+            raise SignerBrokerError("SIGNER_PAYLOAD_INVALID")
+        response = self._request(
+            "sign",
+            payload_kind=payload_kind,
+            payload_sha256=hashlib.sha256(payload).hexdigest(),
+            payload_base64=base64.b64encode(payload).decode("ascii"),
+        )
+        try:
+            signature = base64.b64decode(str(response["signature_base64"]), validate=True)
+            key_id = str(response["key_id"]).strip()
+        except (KeyError, ValueError) as exc:
+            raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID") from exc
+        if not signature or not key_id or len(key_id) > 256:
+            raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID")
+        return signature, key_id

active-runtime/app/aios_signer_broker_service.py

@@ -0,0 +1,289 @@
+"""Privileged, fixed-function signer broker service."""
+from __future__ import annotations
+
+import argparse
+import base64
+import binascii
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import socketserver
+import stat
+import struct
+import subprocess
+import tempfile
+import threading
+from typing import Any, Callable, Mapping, Protocol, Sequence
+
+REQUEST_SCHEMA = "aios.signer-broker.request.v1"
+RESPONSE_SCHEMA = "aios.signer-broker.response.v1"
+AUDIT_SCHEMA = "aios.signer-broker.audit.v1"
+DEFAULT_SOCKET_PATH = Path("/run/aios/signer-broker.sock")
+DEFAULT_SIGNER_PATH = Path("/usr/local/libexec/aios-grant-signer-adapter")
+DEFAULT_AUDIT_PATH = Path("/var/lib/aios-gitops/audit/signer-broker.jsonl")
+DEFAULT_STATE_ROOT = Path("/var/lib/aios-gitops/signer-broker")
+ALLOWED_OPERATIONS = frozenset({"self_test", "sign"})
+ALLOWED_PAYLOAD_KINDS = frozenset({"approval_plan", "grant_payload"})
+MAX_SIGNING_PAYLOAD = 1024 * 1024
+MAX_REQUEST_BYTES = 2 * MAX_SIGNING_PAYLOAD + 8192
+CODE_RE = re.compile(r"^[A-Z0-9_.:-]{3,96}$")
+SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+
+
+class SignerBrokerServiceError(RuntimeError):
+    def __init__(self, code: str):
+        self.code = code if CODE_RE.fullmatch(code) else "SIGNER_INTERNAL_ERROR"
+        super().__init__(self.code)
+
+
+class SignerRunner(Protocol):
+    def self_test(self) -> None: ...
+    def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class SignerBrokerPolicy:
+    socket_path: Path = DEFAULT_SOCKET_PATH
+    signer_path: Path = DEFAULT_SIGNER_PATH
+    audit_path: Path = DEFAULT_AUDIT_PATH
+    state_root: Path = DEFAULT_STATE_ROOT
+    allowed_uids: frozenset[int] = frozenset()
+    signer_timeout: float = 60.0
+    max_payload_bytes: int = MAX_SIGNING_PAYLOAD
+
+    def __post_init__(self) -> None:
+        for name in ("socket_path", "signer_path", "audit_path", "state_root"):
+            if not Path(getattr(self, name)).is_absolute():
+                raise SignerBrokerServiceError("SIGNER_POLICY_PATH_INVALID")
+        if not self.allowed_uids or any(not isinstance(uid, int) or uid < 0 for uid in self.allowed_uids):
+            raise SignerBrokerServiceError("SIGNER_POLICY_CALLER_INVALID")
+        if not 0 < self.signer_timeout <= 300:
+            raise SignerBrokerServiceError("SIGNER_POLICY_TIMEOUT_INVALID")
+        if not 0 < self.max_payload_bytes <= MAX_SIGNING_PAYLOAD:
+            raise SignerBrokerServiceError("SIGNER_POLICY_PAYLOAD_LIMIT_INVALID")
+
+
+class FixedSignerRunner:
+    """Use one configured signer executable and broker-owned temporary paths."""
+
+    def __init__(self, policy: SignerBrokerPolicy):
+        self.policy = policy
+
+    def _run(self, argv: Sequence[str]) -> None:
+        try:
+            result = subprocess.run(list(argv), capture_output=True, text=True, check=False, timeout=self.policy.signer_timeout)
+        except FileNotFoundError as exc:
+            raise SignerBrokerServiceError("SIGNER_UNAVAILABLE") from exc
+        except subprocess.TimeoutExpired as exc:
+            raise SignerBrokerServiceError("SIGNER_TIMEOUT") from exc
+        except OSError as exc:
+            raise SignerBrokerServiceError("SIGNER_UNAVAILABLE") from exc
+        if result.returncode != 0:
+            raise SignerBrokerServiceError("SIGNER_FAILED")
+
+    def self_test(self) -> None:
+        self._run((str(self.policy.signer_path), "self-test"))
+
+    def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]:
+        if payload_kind not in ALLOWED_PAYLOAD_KINDS:
+            raise SignerBrokerServiceError("SIGNER_PAYLOAD_KIND_REJECTED")
+        try:
+            self.policy.state_root.mkdir(parents=True, exist_ok=True)
+            os.chmod(self.policy.state_root, 0o700)
+            with tempfile.TemporaryDirectory(prefix="request-", dir=self.policy.state_root) as directory:
+                root = Path(directory)
+                input_path = root / "payload.canonical"
+                signature_path = root / "payload.sig"
+                key_id_path = root / "payload.key-id"
+                input_path.write_bytes(payload)
+                os.chmod(input_path, 0o600)
+                self._run((str(self.policy.signer_path), "sign", "--namespace", payload_kind, "--input", str(input_path), "--signature-out", str(signature_path), "--key-id-out", str(key_id_path)))
+                signature = signature_path.read_bytes()
+                key_id = key_id_path.read_text(encoding="utf-8").strip()
+        except SignerBrokerServiceError:
+            raise
+        except (OSError, UnicodeError) as exc:
+            raise SignerBrokerServiceError("SIGNER_OUTPUT_INVALID") from exc
+        if not signature or not key_id or len(key_id) > 256 or "\n" in key_id or "\r" in key_id:
+            raise SignerBrokerServiceError("SIGNER_OUTPUT_INVALID")
+        return signature, key_id
+
+
+class SignerBrokerService:
+    def __init__(self, policy: SignerBrokerPolicy, runner: SignerRunner | None = None, *, now: Callable[[], datetime] | None = None) -> None:
+        self.policy = policy
+        self.runner = runner or FixedSignerRunner(policy)
+        self.now = now or (lambda: datetime.now(timezone.utc))
+        self._audit_lock = threading.Lock()
+
+    @staticmethod
+    def _operation(value: Any) -> str:
+        return value if isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,64}", value) else "INVALID"
+
+    def _audit(self, *, peer_uid: int, operation: str, status: str, error_code: str | None = None, payload_sha256: str | None = None) -> None:
+        event: dict[str, Any] = {"schema": AUDIT_SCHEMA, "timestamp": self.now().astimezone(timezone.utc).isoformat(), "peer_uid": peer_uid, "operation": self._operation(operation), "status": status}
+        if isinstance(payload_sha256, str) and SHA256_RE.fullmatch(payload_sha256):
+            event["payload_sha256"] = payload_sha256
+        if error_code:
+            event["error_code"] = error_code if CODE_RE.fullmatch(error_code) else "SIGNER_INTERNAL_ERROR"
+        try:
+            with self._audit_lock:
+                self.policy.audit_path.parent.mkdir(parents=True, exist_ok=True)
+                with self.policy.audit_path.open("a", encoding="utf-8") as handle:
+                    handle.write(json.dumps(event, sort_keys=True, separators=(",", ":")) + "\n")
+                    handle.flush()
+                    os.fsync(handle.fileno())
+                os.chmod(self.policy.audit_path, 0o600)
+        except OSError as exc:
+            raise SignerBrokerServiceError("SIGNER_AUDIT_UNAVAILABLE") from exc
+
+    @staticmethod
+    def _failure(code: str) -> dict[str, Any]:
+        return {"schema": RESPONSE_SCHEMA, "status": "FAIL", "error_code": code if CODE_RE.fullmatch(code) else "SIGNER_INTERNAL_ERROR"}
+
+    def _dispatch(self, request: Mapping[str, Any], peer_uid: int) -> tuple[dict[str, Any], str, str | None]:
+        operation = self._operation(request.get("operation"))
+        if peer_uid not in self.policy.allowed_uids:
+            raise SignerBrokerServiceError("CALLER_UNAUTHORIZED")
+        if request.get("schema") != REQUEST_SCHEMA:
+            raise SignerBrokerServiceError("REQUEST_SCHEMA_INVALID")
+        if operation not in ALLOWED_OPERATIONS:
+            raise SignerBrokerServiceError("SIGNER_OPERATION_REJECTED")
+        if operation == "self_test":
+            if set(request) != {"schema", "operation"}:
+                raise SignerBrokerServiceError("REQUEST_FIELDS_REJECTED")
+            self.runner.self_test()
+            return {"schema": RESPONSE_SCHEMA, "status": "PASS", "operation": operation}, operation, None
+        if set(request) != {"schema", "operation", "payload_kind", "payload_sha256", "payload_base64"}:
+            raise SignerBrokerServiceError("REQUEST_FIELDS_REJECTED")
+        payload_kind = request.get("payload_kind")
+        encoded = request.get("payload_base64")
+        claimed_hash = request.get("payload_sha256")
+        if payload_kind not in ALLOWED_PAYLOAD_KINDS or not isinstance(encoded, str) or not isinstance(claimed_hash, str) or not SHA256_RE.fullmatch(claimed_hash):
+            raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID")
+        if len(encoded) > self.policy.max_payload_bytes * 2 + 8:
+            raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID")
+        try:
+            payload = base64.b64decode(encoded.encode("ascii"), validate=True)
+        except (ValueError, UnicodeError, binascii.Error) as exc:
+            raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID") from exc
+        if not payload or len(payload) > self.policy.max_payload_bytes:
+            raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID")
+        actual_hash = hashlib.sha256(payload).hexdigest()
+        if actual_hash != claimed_hash:
+            raise SignerBrokerServiceError("SIGNER_PAYLOAD_HASH_MISMATCH")
+        signature, key_id = self.runner.sign(payload_kind, payload)
+        if not isinstance(signature, bytes) or not signature or not isinstance(key_id, str) or not key_id.strip() or len(key_id) > 256:
+            raise SignerBrokerServiceError("SIGNER_OUTPUT_INVALID")
+        return {"schema": RESPONSE_SCHEMA, "status": "PASS", "operation": operation, "signature_base64": base64.b64encode(signature).decode("ascii"), "key_id": key_id.strip()}, operation, actual_hash
+
+    def handle_request(self, request: Mapping[str, Any], peer_uid: int) -> dict[str, Any]:
+        operation = self._operation(request.get("operation")) if isinstance(request, Mapping) else "INVALID"
+        payload_hash = request.get("payload_sha256") if isinstance(request, Mapping) else None
+        status = "FAIL"
+        error_code: str | None = None
+        try:
+            if not isinstance(request, Mapping):
+                raise SignerBrokerServiceError("REQUEST_OBJECT_REQUIRED")
+            response, operation, payload_hash = self._dispatch(request, peer_uid)
+            status = "PASS"
+        except SignerBrokerServiceError as exc:
+            error_code = exc.code
+            response = self._failure(exc.code)
+        except Exception:
+            error_code = "SIGNER_INTERNAL_ERROR"
+            response = self._failure(error_code)
+        try:
+            self._audit(peer_uid=peer_uid, operation=operation, status=status, error_code=error_code, payload_sha256=payload_hash if isinstance(payload_hash, str) else None)
+        except SignerBrokerServiceError as exc:
+            response = self._failure(exc.code)
+        return response
+
+    def handle_json_line(self, raw: bytes, peer_uid: int) -> dict[str, Any]:
+        if len(raw) > MAX_REQUEST_BYTES:
+            return self.handle_request({"operation": "invalid"}, peer_uid) | {"error_code": "REQUEST_TOO_LARGE", "status": "FAIL"}
+        try:
+            request = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            return self.handle_request({"operation": "invalid"}, peer_uid) | {"error_code": "REQUEST_JSON_INVALID", "status": "FAIL"}
+        return self.handle_request(request, peer_uid)
+
+
+class _UnixRequestHandler(socketserver.StreamRequestHandler):
+    def handle(self) -> None:
+        service: SignerBrokerService = self.server.service  # type: ignore[attr-defined]
+        response = service.handle_json_line(self.rfile.readline(MAX_REQUEST_BYTES + 1), _peer_uid(self.request))
+        self.wfile.write((json.dumps(response, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8"))
+        self.wfile.flush()
+
+
+class SignerBrokerUnixServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+    def __init__(self, socket_path: Path, service: SignerBrokerService):
+        self.socket_path = Path(socket_path)
+        self.service = service
+        super().__init__(str(self.socket_path), _UnixRequestHandler)
+
+    def server_close(self) -> None:
+        super().server_close()
+        try:
+            if self.socket_path.exists() and stat.S_ISSOCK(self.socket_path.stat().st_mode):
+                self.socket_path.unlink()
+        except OSError:
+            pass
+
+
+def _peer_uid(channel: socket.socket) -> int:
+    if not hasattr(socket, "SO_PEERCRED"):
+        return -1
+    try:
+        _pid, uid, _gid = struct.unpack("3i", channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i")))
+        return int(uid)
+    except (OSError, struct.error):
+        return -1
+
+
+def build_server(policy: SignerBrokerPolicy, runner: SignerRunner | None = None) -> SignerBrokerUnixServer:
+    socket_path = policy.socket_path
+    socket_path.parent.mkdir(parents=True, exist_ok=True)
+    if socket_path.exists():
+        if not stat.S_ISSOCK(socket_path.stat().st_mode):
+            raise SignerBrokerServiceError("SIGNER_BROKER_SOCKET_CONFLICT")
+        socket_path.unlink()
+    server = SignerBrokerUnixServer(socket_path, SignerBrokerService(policy, runner))
+    os.chmod(socket_path, 0o660)
+    return server
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="AIOS fixed-function signer broker")
+    parser.add_argument("command", choices=("serve",))
+    parser.add_argument("--socket", default=str(DEFAULT_SOCKET_PATH))
+    parser.add_argument("--signer", default=str(DEFAULT_SIGNER_PATH))
+    parser.add_argument("--audit", default=str(DEFAULT_AUDIT_PATH))
+    parser.add_argument("--state-root", default=str(DEFAULT_STATE_ROOT))
+    parser.add_argument("--allowed-uid", action="append", type=int, required=True)
+    parser.add_argument("--timeout", type=float, default=60.0)
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    policy = SignerBrokerPolicy(socket_path=Path(args.socket), signer_path=Path(args.signer), audit_path=Path(args.audit), state_root=Path(args.state_root), allowed_uids=frozenset(args.allowed_uid), signer_timeout=args.timeout)
+    server = build_server(policy)
+    try:
+        server.serve_forever()
+    finally:
+        server.server_close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/approved_release_executor.py

@@ -31,6 +31,8 @@ import time
 from pathlib import Path
 from typing import Any, Mapping, Sequence
 
+from version_registry import configured_authoritative_git_dir, verify_authoritative_repository
+
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA_RE = re.compile(r"^[0-9a-f]{64}$")
 PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
@@ -49,6 +51,11 @@ REQUIRED_V2R1C7_MEMBERS = {
     "payload/ARTIFACT-MANIFEST.json",
     "payload/ROLLBACK-METADATA.json",
 }
+REQUIRED_V2R2C1_MEMBERS = {
+    "payload/versions/V2R2C1.json",
+}
+V2R2C1_TAG_OID = "07fc2746b688f5f243f29b39de39f32edc3dd9d2"
+DEFAULT_AUTHORITATIVE_GIT_DIR = Path("/data/aios-gitops/repo.git")
 
 class ReleaseBlocked(RuntimeError):
     pass
@@ -72,6 +79,13 @@ class Binding:
     production_service: str
     production_port: int
     repository_identity: str = "aios/aios"
+    # Legacy V2R2C1 receipts intentionally omit tag_oid. Future release
+    # bindings must provide it explicitly and may use a versioned schema.
+    tag_oid: str | None = None
+    binding_schema: str | None = None
+    source_registry_main_oid: str | None = None
+    source_registry_path: str | None = None
+    source_registry_sha256: str | None = None
 
     def __post_init__(self) -> None:
         if not self.version or not OID_RE.fullmatch(self.target_oid):
@@ -90,6 +104,32 @@ class Binding:
             raise ValueError("invalid production port")
         if self.repository_identity != "aios/aios":
             raise ValueError("unexpected repository identity")
+        if self.tag_oid is not None and not OID_RE.fullmatch(self.tag_oid):
+            raise ValueError("invalid annotated tag object OID")
+        if self.binding_schema is not None and not self.binding_schema.startswith(
+            "aios.release-binding."
+        ):
+            raise ValueError("invalid release binding schema")
+        provenance = (
+            self.source_registry_main_oid,
+            self.source_registry_sha256,
+        )
+        if any(item is not None for item in provenance):
+            if not all(item is not None for item in provenance):
+                raise ValueError("source registry provenance must be complete")
+            if not OID_RE.fullmatch(str(self.source_registry_main_oid)):
+                raise ValueError("invalid source registry main OID")
+            if not SHA_RE.fullmatch(str(self.source_registry_sha256)):
+                raise ValueError("invalid source registry SHA-256")
+            path = self.source_registry_path or f"versions/{self.version}.json"
+            candidate = Path(path)
+            if (
+                candidate.is_absolute()
+                or ".." in candidate.parts
+                or "\\" in path
+                or not path.endswith(".json")
+            ):
+                raise ValueError("invalid source registry path")
 
     @property
     def tag(self) -> str:
@@ -108,10 +148,22 @@ class Binding:
             "production_port": self.production_port,
             "repository_identity": self.repository_identity,
         }
+        if self.tag_oid is not None:
+            payload["tag_oid"] = self.tag_oid
+            payload["binding_schema"] = self.binding_schema or "aios.release-binding.v2"
+        if self.source_registry_main_oid is not None:
+            payload.update(
+                {
+                    "source_registry_main_oid": self.source_registry_main_oid,
+                    "source_registry_path": self.source_registry_path
+                    or f"versions/{self.version}.json",
+                    "source_registry_sha256": self.source_registry_sha256,
+                }
+            )
         return hashlib.sha256(canonical_json(payload)).hexdigest()
 
     def request(self, operation: str) -> dict[str, Any]:
-        return {
+        request = {
             "schema": "aios.production-executor-request.v1",
             "operation": operation,
             "binding_fingerprint": self.fingerprint,
@@ -127,6 +179,19 @@ class Binding:
             "production_port": self.production_port,
             "repository_identity": self.repository_identity,
         }
+        if self.tag_oid is not None:
+            request["tag_oid"] = self.tag_oid
+            request["binding_schema"] = self.binding_schema or "aios.release-binding.v2"
+        if self.source_registry_main_oid is not None:
+            request.update(
+                {
+                    "source_registry_main_oid": self.source_registry_main_oid,
+                    "source_registry_path": self.source_registry_path
+                    or f"versions/{self.version}.json",
+                    "source_registry_sha256": self.source_registry_sha256,
+                }
+            )
+        return request
 
 
 def canonical_json(value: object) -> bytes:
@@ -168,6 +233,23 @@ def run_git(git_dir: Path, *args: str, check: bool = True) -> subprocess.Complet
     )
 
 
+def run_git_bytes(
+    git_dir: Path,
+    *args: str,
+    check: bool = True,
+) -> subprocess.CompletedProcess[bytes]:
+    completed = subprocess.run(
+        ["git", f"--git-dir={git_dir}", *args],
+        check=False,
+        capture_output=True,
+        timeout=30,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if check and completed.returncode != 0:
+        raise ReleaseBlocked(completed.stderr.decode("utf-8", "replace").strip())
+    return completed
+
+
 def verify_git_lineage(git_dir: Path, binding: Binding) -> dict[str, Any]:
     authority_main = run_git(git_dir, "rev-parse", "refs/heads/main").stdout.strip()
     if not OID_RE.fullmatch(authority_main):
@@ -209,7 +291,45 @@ def verify_git_lineage(git_dir: Path, binding: Binding) -> dict[str, Any]:
     if not target_in_evidence:
         raise ReleaseBlocked("approved target is not an ancestor of evidence closure")
 
-    return {
+    source_main = None
+    if binding.source_registry_main_oid is not None:
+        source_main = binding.source_registry_main_oid
+        source_type = run_git(git_dir, "cat-file", "-t", source_main).stdout.strip()
+        if source_type != "commit":
+            raise ReleaseBlocked("source_registry_main_oid is not a commit")
+        target_in_source_main = run_git(
+            git_dir,
+            "merge-base",
+            "--is-ancestor",
+            binding.target_oid,
+            source_main,
+            check=False,
+        ).returncode == 0
+        if not target_in_source_main:
+            raise ManualOwnerReviewRequired(
+                "approved target is not an ancestor of source registry main"
+            )
+    expected_tag_oid = binding.tag_oid
+    if expected_tag_oid is None and binding.version == "V2R2C1":
+        expected_tag_oid = V2R2C1_TAG_OID
+    if expected_tag_oid is not None:
+        local_tag = run_git(
+            git_dir,
+            "rev-parse",
+            "--verify",
+            f"refs/tags/{binding.version}",
+        ).stdout.strip()
+        if local_tag != expected_tag_oid:
+            raise ManualOwnerReviewRequired(
+                "future release binding tag object differs from local tag"
+            )
+        if run_git(git_dir, "cat-file", "-t", local_tag).stdout.strip() != "tag":
+            raise ReleaseBlocked("release tag is not annotated")
+        peeled = run_git(git_dir, "rev-parse", "--verify", f"{local_tag}^{{commit}}").stdout.strip()
+        if peeled != binding.target_oid:
+            raise ManualOwnerReviewRequired("release tag peeled target mismatch")
+
+    result = {
         "authority_main_oid": authority_main,
         "target_oid": binding.target_oid,
         "target_is_ancestor_of_main": True,
@@ -217,9 +337,231 @@ def verify_git_lineage(git_dir: Path, binding: Binding) -> dict[str, Any]:
         "evidence_object_type": evidence_type,
         "target_is_ancestor_of_evidence": True,
     }
+    if source_main is not None:
+        result["source_registry_main_oid"] = source_main
+        result["target_is_ancestor_of_source_registry_main"] = True
+    return result
 
 
-def verify_artifact(binding: Binding) -> dict[str, Any]:
+def _source_registry_git_provenance(
+    git_dir: Path,
+    binding: Binding,
+) -> dict[str, Any] | None:
+    """Read M:path and C:path independently of the artifact metadata."""
+    if binding.source_registry_main_oid is None:
+        return None
+    verify_authoritative_repository(git_dir)
+    source_main = binding.source_registry_main_oid
+    source_path = binding.source_registry_path or f"versions/{binding.version}.json"
+    authority_main = run_git(git_dir, "rev-parse", "--verify", "refs/heads/main").stdout.strip()
+    if run_git(git_dir, "cat-file", "-t", source_main).stdout.strip() != "commit":
+        raise ReleaseBlocked("source_registry_main_oid is not a commit")
+    if run_git(git_dir, "cat-file", "-t", authority_main).stdout.strip() != "commit":
+        raise ReleaseBlocked("Authority main OID is not a commit")
+    if run_git(
+        git_dir,
+        "merge-base",
+        "--is-ancestor",
+        source_main,
+        authority_main,
+        check=False,
+    ).returncode != 0:
+        raise ManualOwnerReviewRequired(
+            "source registry main is not an ancestor of current Authority main"
+        )
+    source_blob = run_git_bytes(git_dir, "show", f"{source_main}:{source_path}").stdout
+    current_blob = run_git_bytes(git_dir, "show", f"{authority_main}:{source_path}").stdout
+    if source_blob != current_blob:
+        raise ManualOwnerReviewRequired(
+            "current Authority main source registry entry differs from pinned M"
+        )
+    try:
+        source = json.loads(source_blob.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ReleaseBlocked("Git source registry entry is invalid") from exc
+    expected_tag_oid = binding.tag_oid
+    if expected_tag_oid is None and binding.version == "V2R2C1":
+        expected_tag_oid = V2R2C1_TAG_OID
+    expected = {
+        "version": binding.version,
+        "repository": "aios/aios",
+        "tag_oid": expected_tag_oid,
+        "target_oid": binding.target_oid,
+        "peeled_target_oid": binding.target_oid,
+        "evidence_closure_oid": binding.evidence_closure_oid,
+        "approval_plan_id": binding.approval_plan_id,
+    }
+    if not isinstance(source, dict) or source.get("schema") not in {
+        "aios.source-registry.v1",
+        "aios.production-registry.v2",
+    }:
+        raise ReleaseBlocked("Git source registry schema is invalid")
+    if expected_tag_oid is None or any(source.get(key) != value for key, value in expected.items()):
+        raise ReleaseBlocked("Git source registry identity mismatch")
+    if source.get("tag_object") != expected_tag_oid:
+        raise ReleaseBlocked("Git source registry tag object mismatch")
+    digest = hashlib.sha256(source_blob).hexdigest()
+    if digest != binding.source_registry_sha256:
+        raise ReleaseBlocked("Git source registry SHA-256 mismatch")
+    return {
+        "source_registry_main_oid": source_main,
+        "current_main_oid": authority_main,
+        "source_registry_path": source_path,
+        "source_registry_bytes": source_blob,
+        "source_registry_sha256": digest,
+        "source_registry_identity_verified": True,
+        "source_registry_main_converged": True,
+    }
+
+
+def _verify_artifact_members(
+    path: Path,
+    required: set[str],
+    binding: Binding | None = None,
+    git_provenance: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Verify archive safety and independently verify packaged release data."""
+    # Python tarfile may not have zstd support on the target host, so
+    # decompress into a private temporary tar and inspect it there.
+    with tempfile.TemporaryDirectory(prefix="aios-artifact-verify-") as temp_dir:
+        tar_path = Path(temp_dir) / "release.tar"
+        with tar_path.open("wb") as output:
+            completed = subprocess.run(
+                ["zstd", "-dc", str(path)],
+                check=False,
+                stdout=output,
+                stderr=subprocess.PIPE,
+                timeout=60,
+                env={"PATH": os.environ.get("PATH", "")},
+            )
+        if completed.returncode != 0:
+            raise ReleaseBlocked("artifact zstd decompression failed")
+        import tarfile
+
+        with tarfile.open(tar_path, "r:") as archive:
+            members = archive.getmembers()
+            names: set[str] = set()
+            for member in members:
+                candidate = Path(member.name)
+                if (
+                    candidate.is_absolute()
+                    or ".." in candidate.parts
+                    or member.issym()
+                    or member.islnk()
+                    or not (member.isfile() or member.isdir())
+                ):
+                    raise ReleaseBlocked(f"unsafe artifact member: {member.name}")
+                if member.name in names:
+                    raise ReleaseBlocked(f"duplicate artifact member: {member.name}")
+                names.add(member.name)
+            source_path = None
+            if binding is not None and binding.version != "V2R1C7":
+                source_path = binding.source_registry_path or f"versions/{binding.version}.json"
+            required_members = set(required)
+            if binding is not None and binding.source_registry_main_oid is not None:
+                required_members.add("payload/BUILD-METADATA.json")
+            if source_path is not None and binding is not None:
+                required_members.add(f"payload/{source_path}")
+            missing = sorted(required_members - names)
+            if missing:
+                raise ReleaseBlocked(
+                    "artifact missing required members: " + ",".join(missing)
+                )
+            verification: dict[str, Any] = {}
+            if binding is not None and source_path is not None:
+                member_name = f"payload/{source_path}"
+                member = archive.getmember(member_name)
+                stream = archive.extractfile(member)
+                if stream is None:
+                    raise ReleaseBlocked("source registry member is unreadable")
+                raw = stream.read()
+                try:
+                    source = json.loads(raw.decode("utf-8"))
+                except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+                    raise ReleaseBlocked("source registry member is invalid") from exc
+                canonical_source = (
+                    json.dumps(source, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
+                ).encode("utf-8")
+                if raw != canonical_source:
+                    raise ReleaseBlocked("source registry member is not canonical JSON")
+                expected_tag_oid = binding.tag_oid
+                if expected_tag_oid is None and binding.version == "V2R2C1":
+                    expected_tag_oid = V2R2C1_TAG_OID
+                expected = {
+                    "version": binding.version,
+                    "repository": "aios/aios",
+                    "tag_oid": expected_tag_oid,
+                    "target_oid": binding.target_oid,
+                    "peeled_target_oid": binding.target_oid,
+                    "evidence_closure_oid": binding.evidence_closure_oid,
+                    "approval_plan_id": binding.approval_plan_id,
+                }
+                if expected_tag_oid is None:
+                    raise ReleaseBlocked("future release binding tag object is required")
+                if not isinstance(source, dict) or any(
+                    source.get(key) != value for key, value in expected.items()
+                ):
+                    raise ReleaseBlocked("packaged source registry identity mismatch")
+                if source.get("tag_object") != expected_tag_oid:
+                    raise ReleaseBlocked("packaged source registry tag object mismatch")
+                source_digest = hashlib.sha256(raw).hexdigest()
+                if git_provenance is not None:
+                    if raw != git_provenance["source_registry_bytes"]:
+                        raise ReleaseBlocked(
+                            "packaged source registry differs from Git M:path"
+                        )
+                    if source_digest != git_provenance["source_registry_sha256"]:
+                        raise ReleaseBlocked(
+                            "packaged source registry SHA-256 differs from Git M:path"
+                        )
+                verification.update(
+                    {
+                        "source_registry_path": source_path,
+                        "source_registry_sha256": source_digest,
+                        "source_registry_identity_verified": True,
+                    }
+                )
+                if binding.source_registry_sha256 is not None:
+                    if source_digest != binding.source_registry_sha256:
+                        raise ReleaseBlocked("packaged source registry SHA-256 mismatch")
+                    metadata_member = archive.getmember("payload/BUILD-METADATA.json")
+                    metadata_stream = archive.extractfile(metadata_member)
+                    if metadata_stream is None:
+                        raise ReleaseBlocked("BUILD-METADATA.json is unreadable")
+                    try:
+                        metadata = json.loads(metadata_stream.read().decode("utf-8"))
+                    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+                        raise ReleaseBlocked("BUILD-METADATA.json is invalid") from exc
+                    metadata_expected = {
+                        "release_target_oid": binding.target_oid,
+                        "source_registry_main_oid": binding.source_registry_main_oid,
+                        "source_registry_path": source_path,
+                        "source_registry_sha256": binding.source_registry_sha256,
+                    }
+                    if not isinstance(metadata, dict) or any(
+                        metadata.get(key) != value
+                        for key, value in metadata_expected.items()
+                    ):
+                        raise ReleaseBlocked("BUILD-METADATA source provenance mismatch")
+                    verification["build_metadata_provenance_verified"] = True
+                if git_provenance is not None:
+                    verification.update(
+                        {
+                            "git_source_registry_member_verified": True,
+                            "git_source_registry_main_converged": True,
+                            "git_source_registry_sha256": git_provenance[
+                                "source_registry_sha256"
+                            ],
+                        }
+                    )
+            return verification
+
+
+def verify_artifact(
+    binding: Binding,
+    *,
+    repository: Path | str | None = None,
+) -> dict[str, Any]:
     path = Path(binding.artifact_path)
     if path.stat().st_size != binding.artifact_size:
         raise ReleaseBlocked("artifact size mismatch")
@@ -227,46 +569,32 @@ def verify_artifact(binding: Binding) -> dict[str, Any]:
     if actual != binding.artifact_sha256:
         raise ReleaseBlocked("artifact SHA-256 mismatch")
 
-    if binding.version == "V2R1C7":
-        # Python tarfile may not have zstd support on the target host, so
-        # decompress into a private temporary tar and inspect it there.
-        with tempfile.TemporaryDirectory(prefix="aios-artifact-verify-") as temp_dir:
-            tar_path = Path(temp_dir) / "release.tar"
-            with tar_path.open("wb") as output:
-                completed = subprocess.run(
-                    ["zstd", "-dc", str(path)],
-                    check=False,
-                    stdout=output,
-                    stderr=subprocess.PIPE,
-                    timeout=60,
-                    env={"PATH": os.environ.get("PATH", "")},
-                )
-            if completed.returncode != 0:
-                raise ReleaseBlocked("artifact zstd decompression failed")
-            import tarfile
-            with tarfile.open(tar_path, "r:") as archive:
-                names: set[str] = set()
-                for member in archive.getmembers():
-                    candidate = Path(member.name)
-                    if (
-                        candidate.is_absolute()
-                        or ".." in candidate.parts
-                        or member.issym()
-                        or member.islnk()
-                        or not (member.isfile() or member.isdir())
-                    ):
-                        raise ReleaseBlocked(f"unsafe artifact member: {member.name}")
-                    names.add(member.name)
-            missing = sorted(REQUIRED_V2R1C7_MEMBERS - names)
-            if missing:
-                raise ReleaseBlocked(
-                    "artifact missing required members: " + ",".join(missing)
-                )
+    required = {
+        "V2R1C7": REQUIRED_V2R1C7_MEMBERS,
+        "V2R2C1": REQUIRED_V2R2C1_MEMBERS,
+    }.get(binding.version)
+    member_verification: dict[str, Any] = {}
+    git_provenance = None
+    if binding.source_registry_main_oid is not None:
+        git_provenance = _source_registry_git_provenance(
+            Path(repository) if repository is not None else Path(
+                configured_authoritative_git_dir()
+            ),
+            binding,
+        )
+    if required or binding.source_registry_main_oid is not None:
+        member_verification = _verify_artifact_members(
+            path,
+            required or set(),
+            binding,
+            git_provenance,
+        )
     return {
         "artifact_path": str(path),
         "artifact_sha256": actual,
         "artifact_size": binding.artifact_size,
         "verified": True,
+        **member_verification,
     }
 
 
@@ -282,6 +610,18 @@ def verify_authorization(receipt: Mapping[str, Any], binding: Binding) -> dict[s
         "artifact_sha256": binding.artifact_sha256,
         "approval_plan_id": binding.approval_plan_id,
     }
+    if binding.tag_oid is not None:
+        expected["tag_oid"] = binding.tag_oid
+        expected["binding_schema"] = binding.binding_schema or "aios.release-binding.v2"
+    if binding.source_registry_main_oid is not None:
+        expected.update(
+            {
+                "source_registry_main_oid": binding.source_registry_main_oid,
+                "source_registry_path": binding.source_registry_path
+                or f"versions/{binding.version}.json",
+                "source_registry_sha256": binding.source_registry_sha256,
+            }
+        )
     for key, value in expected.items():
         if receipt.get(key) != value:
             raise ManualOwnerReviewRequired(f"Owner authorization mismatch: {key}")
@@ -433,13 +773,29 @@ def tag_step(command: Sequence[str], binding: Binding) -> dict[str, Any]:
     tag_oid = probe.get("tag_oid")
     if not isinstance(tag_oid, str) or not OID_RE.fullmatch(tag_oid):
         raise ReleaseBlocked("verified annotated tag object OID missing")
+    if binding.tag_oid is not None and tag_oid != binding.tag_oid:
+        raise ManualOwnerReviewRequired(
+            "same-name tag object differs from the future release binding"
+        )
     return dict(probe)
 
 
 def deploy_step(command: Sequence[str], binding: Binding) -> dict[str, Any]:
+    def require_runtime_registry(result: Mapping[str, Any]) -> None:
+        if binding.source_registry_main_oid is None:
+            return
+        if (
+            result.get("runtime_bundle_contains_current_version") is not True
+            or result.get("runtime_registry_converged") is not True
+        ):
+            raise ReconciliationRequired(
+                "runtime bundle/runtime registry convergence was not independently verified"
+            )
+
     probe = run_adapter(command, binding.request("probe"))
     require_exact_response(probe, binding)
     if probe.get("status") == "ACTIVE_VERIFIED":
+        require_runtime_registry(probe)
         return dict(probe)
     if probe.get("status") == "CONFLICT":
         raise ManualOwnerReviewRequired("production runtime has conflicting binding")
@@ -458,9 +814,30 @@ def deploy_step(command: Sequence[str], binding: Binding) -> dict[str, Any]:
         or probe.get("production_deployed") is not True
     ):
         raise ReconciliationRequired("production health contract is not ACTIVE")
+    require_runtime_registry(probe)
     return dict(probe)
 
 
+def verify_post_deploy_source_registry(
+    repository: Path,
+    binding: Binding,
+) -> dict[str, Any]:
+    """Re-read authoritative Git after runtime deployment and before closure."""
+    if binding.source_registry_main_oid is None:
+        return {
+            "source_registry_converged": "not_required",
+            "source_registry_check": "not_required_for_legacy_binding",
+        }
+    provenance = _source_registry_git_provenance(repository, binding)
+    if provenance is None:
+        raise ReconciliationRequired("authoritative source registry was not checked")
+    return {
+        **provenance,
+        "source_registry_converged": True,
+        "source_registry_check": "post_deploy_authoritative_git_read",
+    }
+
+
 def finalize_step(
     command: Sequence[str],
     binding: Binding,
@@ -501,7 +878,7 @@ def execute(
 ) -> dict[str, Any]:
     authorization = verify_authorization(secure_json(authorization_path), binding)
     lineage = verify_git_lineage(repository, binding)
-    artifact = verify_artifact(binding)
+    artifact = verify_artifact(binding, repository=repository)
     state = DurableState(state_root, binding)
     with FileLock(state_root / "executor.lock"):
         state.record(
@@ -518,6 +895,13 @@ def execute(
             runtime = deploy_step(deploy_command, binding)
             state.record("RUNTIME_VERIFIED", "runtime", runtime)
 
+            # The runtime adapter proves only runtime facts.  Re-read M:path
+            # and C:path from authoritative Git immediately before Registry
+            # and Baseline closure so a post-preflight main change cannot be
+            # mistaken for source convergence.
+            source_registry = verify_post_deploy_source_registry(repository, binding)
+            state.record("SOURCE_REGISTRY_VERIFIED", "source_registry", source_registry)
+
             registry = finalize_step(
                 registry_command,
                 binding,
@@ -547,10 +931,21 @@ def execute(
                 require_exact_response(value, binding)
             if final_runtime.get("status") != "ACTIVE_VERIFIED":
                 raise ReconciliationRequired("final runtime readback failed")
+            if binding.source_registry_main_oid is not None and not all(
+                final_runtime.get(name) is True
+                for name in (
+                    "runtime_bundle_contains_current_version",
+                    "runtime_registry_converged",
+                )
+            ):
+                raise ReconciliationRequired(
+                    "final runtime registry convergence readback failed"
+                )
             if final_registry.get("status") != "VERIFIED":
                 raise ReconciliationRequired("final Registry readback failed")
             if final_baseline.get("status") != "VERIFIED":
                 raise ReconciliationRequired("final Baseline readback failed")
+            final_source_registry = verify_post_deploy_source_registry(repository, binding)
             final = {
                 "status": "PRODUCTION_ACCEPTED",
                 "version": binding.version,
@@ -561,6 +956,7 @@ def execute(
                 "tag_oid": tag["tag_oid"],
                 "tag": final_tag,
                 "runtime": final_runtime,
+                "source_registry": final_source_registry,
                 "registry": final_registry,
                 "baseline": final_baseline,
                 "force_push_used": False,
@@ -600,6 +996,11 @@ def parser() -> argparse.ArgumentParser:
     p.add_argument("--evidence-closure-oid", required=True)
     p.add_argument("--artifact-sha256", required=True)
     p.add_argument("--approval-plan-id", required=True)
+    p.add_argument("--tag-oid")
+    p.add_argument("--binding-schema")
+    p.add_argument("--source-registry-main-oid")
+    p.add_argument("--source-registry-path")
+    p.add_argument("--source-registry-sha256")
     p.add_argument("--production-service", required=True)
     p.add_argument("--production-port", required=True, type=int)
     p.add_argument("--repository-identity", default="aios/aios")
@@ -623,6 +1024,11 @@ def main() -> int:
         production_service=args.production_service,
         production_port=args.production_port,
         repository_identity=args.repository_identity,
+        tag_oid=args.tag_oid,
+        binding_schema=args.binding_schema,
+        source_registry_main_oid=args.source_registry_main_oid,
+        source_registry_path=args.source_registry_path,
+        source_registry_sha256=args.source_registry_sha256,
     )
     try:
         result = execute(

active-runtime/app/gitea_tag_publication.py

@@ -64,6 +64,40 @@ class PublicationError(RuntimeError):
     pass
 
 
+class PublicationValueError(PublicationError, ValueError):
+    def __init__(self, code: str, message: str):
+        self.code = code
+        super().__init__(message)
+
+
+class InvalidReleaseVersion(PublicationValueError):
+    def __init__(self):
+        super().__init__("INVALID_RELEASE_VERSION", "invalid release version")
+
+
+class GiteaConfigurationError(PublicationValueError):
+    def __init__(self, code: str):
+        super().__init__(code, code)
+
+
+class GiteaUnavailableError(PublicationError):
+    def __init__(self, code: str = "GITEA_UNAVAILABLE"):
+        self.code = code
+        super().__init__(code)
+
+
+class GiteaTagNotFound(PublicationError):
+    def __init__(self):
+        self.code = "TAG_NOT_FOUND"
+        super().__init__(self.code)
+
+
+class TagTargetMismatch(PublicationError):
+    def __init__(self):
+        self.code = "TAG_TARGET_MISMATCH"
+        super().__init__(self.code)
+
+
 class PublicationConflict(PublicationError):
     pass
 
@@ -85,7 +119,7 @@ class PublicationManualOwnerReviewRequired(
 def validate_version(version: str) -> str:
     value = str(version or "")
     if not VERSION_RE.fullmatch(value):
-        raise ValueError("invalid release version")
+        raise InvalidReleaseVersion()
     return value
 
 
@@ -402,6 +436,40 @@ class GitTagBackend:
         )
 
 
+@dataclass(frozen=True)
+class GiteaClientConfig:
+    api_url: str
+    repository: str
+    token_file: Path
+    remote: str = "gitea"
+    token_line: int = 1
+
+    @classmethod
+    def from_authoritative_environment(cls, environ: dict[str, str] | None = None) -> "GiteaClientConfig":
+        values = os.environ if environ is None else environ
+        api_url = str(values.get("AIOS_GITOPS_GITEA_API_URL", "")).strip()
+        repository = str(values.get("AIOS_GITOPS_GITEA_REPOSITORY", "")).strip()
+        token_ref = str(values.get("AIOS_GITOPS_GITEA_TOKEN_FILE", "")).strip()
+        if not api_url:
+            raise GiteaConfigurationError("GITEA_API_URL_MISSING")
+        if not api_url.startswith(("https://", "http://")):
+            raise GiteaConfigurationError("GITEA_API_URL_INVALID")
+        if not repository:
+            raise GiteaConfigurationError("GITEA_REPOSITORY_MISSING")
+        if repository.count("/") != 1 or repository.startswith("/"):
+            raise GiteaConfigurationError("GITEA_REPOSITORY_INVALID")
+        if not token_ref:
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_MISSING")
+        token_file = Path(token_ref)
+        if not token_file.is_absolute():
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_INVALID")
+        remote = str(values.get("AIOS_GITOPS_GITEA_REMOTE", "gitea")).strip() or "gitea"
+        token_line = int(str(values.get("AIOS_GITOPS_GITEA_TOKEN_LINE", "1")))
+        if token_line < 1:
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_INVALID")
+        return cls(api_url, repository, token_file, remote, token_line)
+
+
 class GiteaTagClient:
     def __init__(
         self,
@@ -411,23 +479,37 @@ class GiteaTagClient:
         timeout: float = 15,
         token_line: int = 1,
     ):
-        self.api_url = api_url.rstrip("/")
-        self.repository = repository.strip("/")
-        self.token_file = token_file
+        self.api_url = str(api_url or "").rstrip("/")
+        self.repository = str(repository or "").strip("/")
+        self.token_file = Path(token_file)
         self.timeout = timeout
         self.token_line = token_line
+        if not self.api_url:
+            raise GiteaConfigurationError("GITEA_API_URL_MISSING")
         if not self.api_url.startswith(("https://", "http://")):
-            raise ValueError("Gitea API URL must be HTTP(S)")
+            raise GiteaConfigurationError("GITEA_API_URL_INVALID")
+        if not self.repository:
+            raise GiteaConfigurationError("GITEA_REPOSITORY_MISSING")
         if self.repository.count("/") != 1:
-            raise ValueError("Gitea repository must be owner/name")
+            raise GiteaConfigurationError("GITEA_REPOSITORY_INVALID")
+        if not str(token_file) or not self.token_file.is_absolute():
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_MISSING")
+
+    @classmethod
+    def from_authoritative_environment(cls, environ: dict[str, str] | None = None) -> "GiteaTagClient":
+        config = GiteaClientConfig.from_authoritative_environment(environ)
+        return cls(config.api_url, config.repository, config.token_file, token_line=config.token_line)
 
     def _get(self, path: str) -> Any:
-        lines = self.token_file.read_text(encoding="utf-8").splitlines()
+        try:
+            lines = self.token_file.read_text(encoding="utf-8").splitlines()
+        except (OSError, UnicodeError) as exc:
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_UNAVAILABLE") from exc
         if self.token_line < 1 or self.token_line > len(lines):
-            raise RuntimeError("Gitea credential reference has no configured token line")
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_INVALID")
         token = lines[self.token_line - 1].strip()
         if not token:
-            raise RuntimeError("Gitea token file is empty")
+            raise GiteaConfigurationError("GITEA_TOKEN_REFERENCE_INVALID")
         request = urllib.request.Request(
             f"{self.api_url}{path}",
             headers={
@@ -439,10 +521,12 @@ class GiteaTagClient:
         try:
             with urllib.request.urlopen(request, timeout=self.timeout) as response:
                 return json.loads(response.read().decode("utf-8"))
-        except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError) as exc:
-            status = getattr(exc, "code", None)
-            detail = f" HTTP {status}" if status else ""
-            raise RuntimeError(f"Gitea API unavailable{detail}") from exc
+        except urllib.error.HTTPError as exc:
+            if exc.code == 404:
+                raise GiteaTagNotFound() from exc
+            raise GiteaUnavailableError() from exc
+        except (urllib.error.URLError, TimeoutError, OSError) as exc:
+            raise GiteaUnavailableError() from exc
 
     def repository_identity(self) -> str:
         repository = self._get(f"/repos/{self.repository}")
@@ -950,6 +1034,12 @@ class TagPublisher:
                 listing = state.get("gitea_listing_visible") is True
             expected_tag_oid = state.get("tag_oid")
             registry_tag_oid = registry.get("tag_oid") or registry.get("tag_object")
+            target_mismatch = any(
+                isinstance(value, tuple)
+                and len(value) > 1
+                and value[1] not in (None, request.target_oid)
+                for value in (remote, api)
+            )
             checks = {
                 "fingerprint": all(
                     baseline.get(field) == getattr(request, field)
@@ -974,7 +1064,7 @@ class TagPublisher:
                 and registry.get("enabled") is True,
             }
             status = "PASS" if all(checks.values()) else "FAIL"
-            return {
+            result = {
                 "version": request.version,
                 "status": status,
                 "checks": checks,
@@ -986,8 +1076,19 @@ class TagPublisher:
                 "gitea_listing_visible": listing,
                 "repository_identity": repository_identity,
             }
+            if target_mismatch:
+                result["fail_closed"] = True
+                result["error"] = TagTargetMismatch().code
+            return result
         except PublicationManualOwnerReviewRequired:
             raise
+        except PublicationError as exc:
+            return {
+                "version": request.version,
+                "status": "UNKNOWN",
+                "fail_closed": True,
+                "error": str(getattr(exc, "code", exc)),
+            }
         except Exception as exc:
             return {
                 "version": request.version,

active-runtime/app/governed_production_release.py

@@ -29,9 +29,18 @@ from gitea_tag_publication import (
     PublicationRequest,
     TagPublisher,
 )
+from version_registry import (
+    RegistryConvergenceError,
+    assert_source_registry_convergence,
+    resolve_source_registry_main_oid,
+    source_registry_snapshot,
+    verify_authoritative_repository,
+    verify_source_registry_git_identity,
+)
 
 
 VERSION = "V2R2C1"
+TAG_OID = "07fc2746b688f5f243f29b39de39f32edc3dd9d2"
 TARGET_OID = "998d30304787e31e5ae607c0bd86b5edb2b0e5ad"
 EVIDENCE_CLOSURE_OID = "a8b906d1bc24088c19647e98de6a679268f6d0a2"
 APPROVAL_PLAN_ID = "sha256:c0df52008ae7f64cec5ee2d28947aa7b892066c670ab8295e4549140b1a7500e"
@@ -79,6 +88,7 @@ def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o600) -> None
 @dataclass(frozen=True)
 class ReleaseBinding:
     version: str = VERSION
+    tag_oid: str = TAG_OID
     target_oid: str = TARGET_OID
     evidence_closure_oid: str = EVIDENCE_CLOSURE_OID
     approval_plan_id: str = APPROVAL_PLAN_ID
@@ -89,6 +99,8 @@ class ReleaseBinding:
     def validate(self) -> None:
         if self.version != VERSION or self.repository != REPOSITORY:
             raise ManualOwnerReviewRequired("release binding is outside V2R2C1 scope")
+        if not OID_RE.fullmatch(self.tag_oid) or self.tag_oid != TAG_OID:
+            raise ManualOwnerReviewRequired("V2R2C1 tag_oid mismatch")
         if not OID_RE.fullmatch(self.target_oid) or self.target_oid != TARGET_OID:
             raise ManualOwnerReviewRequired("V2R2C1 target_oid mismatch")
         if not OID_RE.fullmatch(self.evidence_closure_oid) or self.evidence_closure_oid != EVIDENCE_CLOSURE_OID:
@@ -192,19 +204,48 @@ def _ancestor(git_dir: Path, ancestor: str, descendant: str) -> bool:
     ).returncode == 0
 
 
-def verify_lineage(git_dir: Path, binding: ReleaseBinding) -> dict[str, Any]:
+def verify_lineage(
+    git_dir: Path,
+    binding: ReleaseBinding,
+    *,
+    authority_main_oid: str | None = None,
+) -> dict[str, Any]:
     binding.validate()
+    main_oid = authority_main_oid or _git(git_dir, "rev-parse", "refs/heads/main")
+    if not OID_RE.fullmatch(main_oid) or _git(git_dir, "cat-file", "-t", main_oid) != "commit":
+        raise ReleaseBlocked("authoritative main OID is not a commit")
     if _git(git_dir, "cat-file", "-t", binding.target_oid) != "commit":
         raise ReleaseBlocked("approved target_oid is not a commit")
     if _git(git_dir, "cat-file", "-t", binding.evidence_closure_oid) != "commit":
         raise ReleaseBlocked("evidence closure is not a commit")
     if not _ancestor(git_dir, binding.target_oid, binding.evidence_closure_oid):
         raise ManualOwnerReviewRequired("approved target is not an ancestor of evidence closure")
+    if not _ancestor(git_dir, binding.target_oid, main_oid):
+        raise ManualOwnerReviewRequired("approved target is not an ancestor of source registry main")
     return {
         "target_oid": binding.target_oid,
         "evidence_closure_oid": binding.evidence_closure_oid,
         "target_is_ancestor_of_evidence": True,
-        "main_oid": _git(git_dir, "rev-parse", "refs/heads/main"),
+        "main_oid": main_oid,
+        "target_is_ancestor_of_main": True,
+    }
+
+
+def verify_local_annotated_tag(git_dir: Path, binding: ReleaseBinding) -> dict[str, Any]:
+    """Verify the local authoritative tag object without creating or rewriting it."""
+    tag_oid = _git(git_dir, "rev-parse", "--verify", f"refs/tags/{binding.version}")
+    if tag_oid != binding.tag_oid:
+        raise ManualOwnerReviewRequired("local annotated tag object mismatch")
+    if _git(git_dir, "cat-file", "-t", tag_oid) != "tag":
+        raise ReleaseBlocked("release tag is not annotated")
+    peeled = _git(git_dir, "rev-parse", "--verify", f"{tag_oid}^{{commit}}")
+    if peeled != binding.target_oid:
+        raise ManualOwnerReviewRequired("local annotated tag target mismatch")
+    return {
+        "tag_oid": tag_oid,
+        "tag_object_type": "tag",
+        "peeled_target_oid": peeled,
+        "target_oid": binding.target_oid,
     }
 
 
@@ -248,9 +289,41 @@ def build_production_artifact(
     binding: ReleaseBinding,
     output_root: Path,
 ) -> dict[str, Any]:
-    """Build a deterministic executable V2R2C1 health artifact from target_oid."""
+    """Build a deterministic artifact from composite immutable provenance.
+
+    Release payload files are read from target ``T``.  The source registry is
+    read once from the pinned main/convergence commit ``M`` and its exact blob
+    bytes are reused for packaging and all recorded digests.
+    """
     binding.validate()
-    verify_lineage(git_dir, binding)
+    verify_authoritative_repository(git_dir)
+    source_registry_main_oid = resolve_source_registry_main_oid(
+        git_dir,
+        main_ref="refs/heads/main",
+    )
+    verify_lineage(git_dir, binding, authority_main_oid=source_registry_main_oid)
+    tag_lineage = verify_local_annotated_tag(git_dir, binding)
+    try:
+        source_snapshot = source_registry_snapshot(
+            git_dir,
+            VERSION,
+            source_registry_main_oid=source_registry_main_oid,
+            expected_tag_oid=binding.tag_oid,
+            expected_target_oid=binding.target_oid,
+            expected_evidence_closure_oid=binding.evidence_closure_oid,
+            expected_approval_plan_id=binding.approval_plan_id,
+        )
+        source_registry = source_snapshot["record"]
+        source_registry_bytes = source_snapshot["bytes"]
+        verify_source_registry_git_identity(git_dir, source_registry)
+    except RegistryConvergenceError as exc:
+        raise ReleaseBlocked(f"SOURCE_REGISTRY_NOT_CONVERGED:{exc}") from exc
+    if source_snapshot["source_registry_main_oid"] != source_registry_main_oid:
+        raise ReconciliationRequired("source registry main OID changed during build")
+    if not _ancestor(git_dir, binding.target_oid, source_registry_main_oid):
+        raise ManualOwnerReviewRequired("release target is not an ancestor of source registry main")
+    source_registry_path = source_snapshot["source_registry_path"]
+    source_registry_sha256 = source_snapshot["source_registry_sha256"]
     epoch = _commit_epoch(git_dir, binding.target_oid)
     output_root.mkdir(parents=True, exist_ok=True)
     stem = f"aios-{VERSION}-{binding.target_oid[:12]}"
@@ -261,11 +334,15 @@ def build_production_artifact(
         source_paths = {
             "active-runtime/app/deployment_state.py": "active-runtime/app/deployment_state.py",
             "active-runtime/app/v2r1c6_privacy_tools.py": "active-runtime/app/v2r1c6_privacy_tools.py",
+            "versions/V2R2C1.json": "versions/V2R2C1.json",
         }
         for destination, source in source_paths.items():
             target = root / destination
             target.parent.mkdir(parents=True, exist_ok=True)
-            target.write_bytes(_payload_file(git_dir, binding.target_oid, source))
+            if destination == source_registry_path:
+                target.write_bytes(source_registry_bytes)
+            else:
+                target.write_bytes(_payload_file(git_dir, binding.target_oid, source))
         generated: dict[str, str] = {
             "VERSION": VERSION + "\n",
             "active-runtime/app/v2r2c1_health_service.py": (
@@ -307,16 +384,6 @@ def build_production_artifact(
                 "with urllib.request.urlopen(url, timeout=5) as r: print(json.dumps(json.load(r), sort_keys=True))\n"
             ),
             "scripts/rollback_v2r2c1_release.sh": "#!/bin/sh\nset -eu\nprintf '%s\\n' 'rollback requires governed owner-approved executor'\nexit 2\n",
-            "release-manifests/V2R2C1.json": json.dumps({
-                "schema": "aios.release-source.v2",
-                "version": VERSION,
-                "repository": REPOSITORY,
-                "target_oid": binding.target_oid,
-                "evidence_closure_oid": binding.evidence_closure_oid,
-                "production_deployed": False,
-                "deployment_state": "STAGING",
-                "artifact_kind": "executable-production-health",
-            }, sort_keys=True, separators=(",", ":")) + "\n",
         }
         for rel, text in generated.items():
             target = root / rel
@@ -328,6 +395,10 @@ def build_production_artifact(
             "schema": "aios.release-artifact-build.v3",
             "version": VERSION,
             "repository": REPOSITORY,
+            "release_target_oid": binding.target_oid,
+            "source_registry_main_oid": source_registry_main_oid,
+            "source_registry_path": source_registry_path,
+            "source_registry_sha256": source_registry_sha256,
             "source_head_oid": binding.target_oid,
             "evidence_closure_oid": binding.evidence_closure_oid,
             "source_date_epoch": epoch,
@@ -337,6 +408,31 @@ def build_production_artifact(
             "tag_created": False,
             "baseline_changed": False,
             "secrets_included": False,
+            "source_registry": {
+                "path": source_registry_path,
+                "version": source_registry["version"],
+                "tag_oid": source_registry["tag_oid"],
+                "target_oid": source_registry["target_oid"],
+                "peeled_target_oid": source_registry["peeled_target_oid"],
+                "evidence_closure_oid": source_registry["evidence_closure_oid"],
+                "approval_plan_id": source_registry["approval_plan_id"],
+                "source_registry_main_oid": source_registry_main_oid,
+                "source_registry_sha256": source_registry_sha256,
+            },
+            "provenance": {
+                "release_payload_source": {
+                    "commit_oid": binding.target_oid,
+                    "role": "release_target",
+                },
+                "source_registry_source": {
+                    "commit_oid": source_registry_main_oid,
+                    "path": source_registry_path,
+                    "sha256": source_registry_sha256,
+                    "role": "source_registry_convergence",
+                },
+                "release_target_is_ancestor_of_source_registry_main": True,
+                "local_annotated_tag": tag_lineage,
+            },
         }
         (root / "BUILD-METADATA.json").write_bytes(canonical(metadata))
         (root / "ROLLBACK-METADATA.json").write_bytes(canonical(json.loads((root / "deploy/aios-v2/v2r2c1-rollback.json").read_text(encoding="utf-8"))))
@@ -344,6 +440,10 @@ def build_production_artifact(
         (root / "ARTIFACT-MANIFEST.json").write_bytes(canonical({
             "schema": "aios.release-artifact-manifest.v3",
             "version": VERSION,
+            "release_target_oid": binding.target_oid,
+            "source_registry_main_oid": source_registry_main_oid,
+            "source_registry_path": source_registry_path,
+            "source_registry_sha256": source_registry_sha256,
             "source_head_oid": binding.target_oid,
             "evidence_closure_oid": binding.evidence_closure_oid,
             "repository": REPOSITORY,
@@ -375,6 +475,10 @@ def build_production_artifact(
         "schema": "aios.release-artifact-receipt.v3",
         "version": VERSION,
         "repository": REPOSITORY,
+        "release_target_oid": binding.target_oid,
+        "source_registry_main_oid": source_registry_main_oid,
+        "source_registry_path": source_registry_path,
+        "source_registry_sha256": source_registry_sha256,
         "source_head_oid": binding.target_oid,
         "evidence_closure_oid": binding.evidence_closure_oid,
         "artifact": str(output),
@@ -385,6 +489,14 @@ def build_production_artifact(
         "production_deployed": False,
         "tag_created": False,
         "baseline_changed": False,
+        "source_registry_path": source_registry_path,
+        "source_registry_identity": source_registry,
+        "provenance": {
+            "release_payload_source": binding.target_oid,
+            "source_registry_source": source_registry_main_oid,
+            "release_target_is_ancestor_of_source_registry_main": True,
+            "local_annotated_tag": tag_lineage,
+        },
     }
     atomic_json(output.with_suffix(output.suffix + ".receipt.json"), receipt, mode=0o640)
     return receipt
@@ -458,6 +570,12 @@ class GovernedProductionReleaseExecutor:
                 self.state.fail("RECONCILIATION_REQUIRED", str(exc))
                 raise ReconciliationRequired(str(exc)) from exc
             result = dict(verified)
+            if result.get("tag_oid") != self.binding.tag_oid:
+                self.state.fail(
+                    "MANUAL_OWNER_REVIEW_REQUIRED",
+                    "same-name tag object differs from the approved V2R2C1 tag",
+                )
+                raise ManualOwnerReviewRequired("approved tag object mismatch")
             self.state.record("TAG_VERIFIED", "tag", result)
             return {"status": "REMOTE_VERIFIED", **result}
 
@@ -478,6 +596,8 @@ class GovernedProductionReleaseExecutor:
         tag_state = self.state.value.get("completed", {}).get("tag")
         if not isinstance(tag_state, Mapping) or tag_state.get("state") not in {"GITEA_API_CONFIRMED", "REMOTE_VERIFIED"}:
             raise ReleaseBlocked("REMOTE_TAG_PROOF_REQUIRED")
+        if tag_state.get("tag_oid") != self.binding.tag_oid:
+            raise ManualOwnerReviewRequired("approved tag object mismatch")
         artifact = self.state.value.get("completed", {}).get("production_artifact")
         if not isinstance(artifact, Mapping):
             artifact = self.build_artifact(request)
@@ -488,6 +608,13 @@ class GovernedProductionReleaseExecutor:
             "artifact_size": int(artifact["artifact_size"]),
             "artifact_sha256": artifact["artifact_sha256"],
         }
+        for name in (
+            "source_registry_main_oid",
+            "source_registry_path",
+            "source_registry_sha256",
+        ):
+            if name in artifact:
+                value[name] = artifact[name]
         # The production adapter is privileged and is run through the root
         # approved-release service boundary by the caller.  This handler is
         # still useful for a root-owned staging invocation and remains
@@ -507,6 +634,21 @@ class GovernedProductionReleaseExecutor:
         result["production_health"] = _health(self.config.health_url, self.binding, self.state.value.get("completed", {}).get("production_artifact", {}))
         result["registry"] = _read_json(Path(self.config.registry_root) / f"{VERSION}.json")
         result["baseline"] = _read_json(Path(self.config.baseline_path))
+        try:
+            assert_source = Path(self.config.registry_root) / f"{VERSION}.json"
+            if not assert_source.is_file():
+                raise RegistryConvergenceError("runtime_registry_entry_missing")
+            assert_source_registry_convergence(
+                assert_source,
+                repo_git_dir=self.config.authoritative_git_dir,
+            )
+            source_check = {"status": "PASS", "path": str(assert_source)}
+        except RegistryConvergenceError as exc:
+            source_check = {"status": "FAIL", "error": str(exc)}
+        result["source_registry"] = source_check
+        result.setdefault("checks", {})["source_registry"] = source_check["status"] == "PASS"
+        if result.get("status") == "PASS" and source_check["status"] != "PASS":
+            result["status"] = "FAIL"
         return result
 
     def acceptance(self, request: Mapping[str, Any]) -> dict[str, Any]:

active-runtime/app/release_metadata_adapter.py

@@ -7,7 +7,7 @@ Registry and authoritative runtime-anchor Baseline only.
 Registry is finalized before Baseline. Baseline apply independently verifies:
 - exact Authority annotated tag and peeled target
 - exact ACTIVE production health
-- exact V2R1C7 Registry record
+- exact runtime Registry record and source-registry convergence
 - exact previous Baseline file SHA-256 (CAS)
 """
 from __future__ import annotations
@@ -25,6 +25,7 @@ import time
 import urllib.error
 import urllib.request
 from pathlib import Path
+from version_registry import assert_source_registry_convergence
 from typing import Any, Mapping
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
@@ -97,11 +98,40 @@ def load_request() -> dict[str, Any]:
         raise ValueError("approval_plan_id_invalid")
     if "tag_oid" in value and not OID_RE.fullmatch(str(value["tag_oid"])):
         raise ValueError("tag_oid_invalid")
+    if "source_registry_main_oid" in value and not OID_RE.fullmatch(
+        str(value["source_registry_main_oid"])
+    ):
+        raise ValueError("source_registry_main_oid_invalid")
+    if "source_registry_sha256" in value and not SHA_RE.fullmatch(
+        str(value["source_registry_sha256"])
+    ):
+        raise ValueError("source_registry_sha256_invalid")
+    if "source_registry_path" in value:
+        source_path = str(value["source_registry_path"])
+        if (
+            not source_path.startswith("versions/")
+            or ".." in Path(source_path).parts
+            or "\\" in source_path
+            or not source_path.endswith(".json")
+        ):
+            raise ValueError("source_registry_path_invalid")
+    provenance = tuple(
+        value.get(name)
+        for name in (
+            "source_registry_main_oid",
+            "source_registry_path",
+            "source_registry_sha256",
+        )
+    )
+    if any(item is not None for item in provenance) and not all(
+        item is not None for item in provenance
+    ):
+        raise ValueError("source_registry_provenance_incomplete")
     return value
 
 
 def exact_response(value: Mapping[str, Any], status: str, **extra: Any) -> dict[str, Any]:
-    return {
+    result = {
         "status": status,
         "version": value["version"],
         "target_oid": value["target_oid"],
@@ -110,6 +140,14 @@ def exact_response(value: Mapping[str, Any], status: str, **extra: Any) -> dict[
         "approval_plan_id": value["approval_plan_id"],
         **extra,
     }
+    for name in (
+        "source_registry_main_oid",
+        "source_registry_path",
+        "source_registry_sha256",
+    ):
+        if name in value:
+            result[name] = value[name]
+    return result
 
 
 def git(repository: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
@@ -154,7 +192,7 @@ def health(url: str, value: Mapping[str, Any]) -> dict[str, Any]:
 
 
 def registry_record(value: Mapping[str, Any], tag_oid: str, production_health: Mapping[str, Any]) -> dict[str, Any]:
-    return {
+    result = {
         "schema": "aios.production-registry.v2",
         "version": value["version"],
         "repository": "aios/aios",
@@ -175,10 +213,18 @@ def registry_record(value: Mapping[str, Any], tag_oid: str, production_health: M
         "deployment_state": production_health["deployment_state"],
         "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
     }
+    for name in (
+        "source_registry_main_oid",
+        "source_registry_path",
+        "source_registry_sha256",
+    ):
+        if name in value:
+            result[name] = value[name]
+    return result
 
 
 def registry_matches(record: Mapping[str, Any], value: Mapping[str, Any], tag_oid: str) -> bool:
-    return all(
+    matches = all(
         record.get(key) == expected
         for key, expected in {
             "version": value["version"],
@@ -192,11 +238,16 @@ def registry_matches(record: Mapping[str, Any], value: Mapping[str, Any], tag_oi
             "production_deployed": True,
         }.items()
     ) and record.get("registered") is True and record.get("enabled") is True
+    return matches and all(
+        record.get(name) == value.get(name)
+        for name in ("source_registry_main_oid", "source_registry_path", "source_registry_sha256")
+        if name in value
+    )
 
 
 def baseline_matches(record: Mapping[str, Any], value: Mapping[str, Any], tag_oid: str) -> bool:
     current = record.get("current_baseline") or record.get("version")
-    return current == value["version"] and all(
+    matches = current == value["version"] and all(
         record.get(key) == expected
         for key, expected in {
             "target_oid": value["target_oid"],
@@ -209,6 +260,11 @@ def baseline_matches(record: Mapping[str, Any], value: Mapping[str, Any], tag_oi
             "status": "PRODUCTION_DEPLOYED",
         }.items()
     )
+    return matches and all(
+        record.get(name) == value.get(name)
+        for name in ("source_registry_main_oid", "source_registry_path", "source_registry_sha256")
+        if name in value
+    )
 
 
 def registry_probe(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[str, Any]:
@@ -305,6 +361,11 @@ def baseline_apply(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[s
         int(previous.get("generation", 0) or 0),
         int(previous.get("authority_generation", 0) or 0),
     ) + 1
+    assert_source_registry_convergence(
+        registry_path,
+        repo_git_dir=args.repository,
+        source_registry_main_oid=value.get("source_registry_main_oid"),
+    )
     record = {
         "schema": "aios.production-baseline.v2",
         "source_identity": "aios-gitops-runtime-anchor-baseline-authority-v1",
@@ -336,6 +397,13 @@ def baseline_apply(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[s
         "promoted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
         "promoted_by": "aios-approved-production-release-executor",
     }
+    for name in (
+        "source_registry_main_oid",
+        "source_registry_path",
+        "source_registry_sha256",
+    ):
+        if name in value:
+            record[name] = value[name]
     atomic_json(args.baseline_path, record)
     return baseline_probe(args, value)
 

active-runtime/app/signer_contract_compatibility.py

@@ -15,6 +15,8 @@ import subprocess
 from pathlib import Path
 from typing import Any, Mapping, Sequence
 
+from aios_signer_broker import SignerBroker, SignerBrokerError, UnixSignerBroker
+
 
 _SIGNER_FLAGS = (
     "--namespace",
@@ -155,6 +157,36 @@ def _sign(
         raise SignerContractError("signer_output_empty")
 
 
+def run_signer_self_test(broker: SignerBroker | None = None) -> dict[str, str]:
+    active = broker or UnixSignerBroker()
+    try:
+        active.self_test()
+    except SignerBrokerError as exc:
+        raise SignerContractError(exc.code) from exc
+    return {"status": "PASS", "contract": "aios.signer-broker.v1"}
+
+
+def _sign_with_broker(
+    broker: SignerBroker,
+    payload_kind: str,
+    input_path: Path,
+    signature_path: Path,
+    key_id_path: Path,
+) -> None:
+    if signature_path.exists() or key_id_path.exists():
+        raise SignerContractError("contract_output_exists")
+    try:
+        signature, key_id = broker.sign(payload_kind, input_path.read_bytes())
+    except SignerBrokerError as exc:
+        raise SignerContractError(exc.code) from exc
+    if not signature or not key_id.strip():
+        raise SignerContractError("signer_output_empty")
+    signature_path.write_bytes(signature)
+    key_id_path.write_text(key_id.strip() + "\n", encoding="utf-8")
+    os.chmod(signature_path, 0o600)
+    os.chmod(key_id_path, 0o600)
+
+
 def _manifest(grant_dir: Path, names: Sequence[str]) -> None:
     rows = []
     for name in names:
@@ -284,6 +316,8 @@ def run_compatible_signer(
     signer_command: Sequence[str],
     grant_dir: Path,
     result_path: Path,
+    *,
+    broker: SignerBroker | None = None,
 ) -> dict[str, str]:
     try:
         if not _supports_deployed_signer(signer_command):
@@ -318,22 +352,21 @@ def run_compatible_signer(
                 "signing_key_identity": signer,
             }
         prepared = _prepare(grant_dir)
-        config_path = Path(
-            os.environ.get("AIOS_GRANT_ROOT_CONFIG", "/etc/aios/grant-root/config.json")
-        )
-        config = _read(config_path)
-        grant_namespace = str(config["grant_namespace"])
-        approval_namespace = str(config["approval_namespace"])
-        _sign(
-            signer_command,
-            approval_namespace,
+        active_broker = broker or UnixSignerBroker()
+        try:
+            active_broker.self_test()
+        except SignerBrokerError as exc:
+            raise SignerContractError(exc.code) from exc
+        _sign_with_broker(
+            active_broker,
+            "approval_plan",
             grant_dir / "approval-plan.canonical.json",
             grant_dir / "approval-plan.sig",
             grant_dir / "approval-plan.key-id",
         )
-        _sign(
-            signer_command,
-            grant_namespace,
+        _sign_with_broker(
+            active_broker,
+            "grant_payload",
             grant_dir / "grant-payload.canonical.json",
             grant_dir / "signed-grant.sig",
             grant_dir / "signed-grant.key-id",

active-runtime/app/v2r2c1_production_adapter.py

@@ -7,6 +7,7 @@ previous V2R1C7 release available for an explicit, owner-approved rollback.
 from __future__ import annotations
 
 import json
+import hashlib
 import os
 import shutil
 import sys
@@ -29,6 +30,7 @@ ROLLBACK_ROOT = Path("/var/lib/aios/v2r2c1/rollback")
 
 REQUIRED_MEMBERS = {
     "payload/VERSION",
+    "payload/versions/V2R2C1.json",
     "payload/active-runtime/app/deployment_state.py",
     "payload/active-runtime/app/v2r1c6_privacy_tools.py",
     "payload/active-runtime/app/v2r2c1_health_service.py",
@@ -87,12 +89,65 @@ def execute_request(value: Mapping[str, Any]) -> dict[str, Any]:
         _backup(Path("/var/lib/aios/v2r1c7/deployment-state.json"), "v2r1c7-deployment-state.previous")
         return base.install_release(_args(), value)
     if value.get("operation") == "probe":
-        return base.probe(_args(), value)
+        result = base.probe(_args(), value)
+        if result.get("status") == "ACTIVE_VERIFIED":
+            result.update(_runtime_registry_gate(value))
+        return result
     if value.get("operation") == "rollback":
         return rollback(value)
     raise ValueError("unsupported_operation")
 
 
+def _runtime_registry_gate(value: Mapping[str, Any]) -> dict[str, Any]:
+    """Prove the deployed immutable runtime contains the exact M:path blob."""
+    source_digest = value.get("source_registry_sha256")
+    source_main = value.get("source_registry_main_oid")
+    if not source_digest or not source_main:
+        return {}
+    source_path = str(value.get("source_registry_path") or f"versions/{VERSION}.json")
+    member = CURRENT_LINK / source_path
+    if member.is_symlink() or not member.is_file():
+        raise RuntimeError("RUNTIME_SOURCE_REGISTRY_MEMBER_MISSING")
+    raw = member.read_bytes()
+    digest = hashlib.sha256(raw).hexdigest()
+    if digest != source_digest:
+        raise RuntimeError("RUNTIME_SOURCE_REGISTRY_DIGEST_MISMATCH")
+    try:
+        source = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError("RUNTIME_SOURCE_REGISTRY_INVALID") from exc
+    expected = {
+        "version": VERSION,
+        "repository": "aios/aios",
+        "tag_oid": value.get("tag_oid"),
+        "target_oid": value.get("target_oid"),
+        "peeled_target_oid": value.get("target_oid"),
+        "evidence_closure_oid": value.get("evidence_closure_oid"),
+        "approval_plan_id": value.get("approval_plan_id"),
+    }
+    if not isinstance(source, dict) or any(source.get(key) != expected_value for key, expected_value in expected.items()):
+        raise RuntimeError("RUNTIME_SOURCE_REGISTRY_IDENTITY_MISMATCH")
+    if source.get("tag_object") != expected["tag_oid"]:
+        raise RuntimeError("RUNTIME_SOURCE_REGISTRY_TAG_OBJECT_MISMATCH")
+    version_name = f"{VERSION}.json"
+    return {
+        "runtime_registry_versions": [version_name],
+        "attested_projection_versions": [version_name],
+        "effective_registry_versions": [version_name],
+        # The adapter can only inspect the deployed immutable runtime.  It
+        # must not claim convergence with mutable authoritative Git; the
+        # executor performs that independent M/C read after deployment.
+        "source_registry_converged": "not_checked",
+        "source_registry_convergence_error": "executor_must_recheck_authoritative_git",
+        "runtime_registry_converged": True,
+        "runtime_bundle_contains_current_version": True,
+        "runtime_bundle_contains_version": True,
+        "registry_consistency_status": "RUNTIME_ONLY_VERIFIED",
+        "runtime_source_registry_main_oid": source_main,
+        "runtime_source_registry_sha256": digest,
+    }
+
+
 def rollback(value: Mapping[str, Any]) -> dict[str, Any]:
     """Explicit rollback only; this adapter never invokes it automatically."""
     import subprocess

active-runtime/app/version_registry.py

@@ -0,0 +1,982 @@
+"""Release registry visibility and source-convergence gates.
+
+Two invariants are enforced:
+
+1. listVersions may overlay exactly the current Production Baseline-attested
+   projection onto the immutable source bundle. Arbitrary projection files
+   are not surfaced.
+
+2. release_metadata_adapter.baseline_apply may not finalize a production
+   baseline unless the governed repository/main already contains a source
+   versions/<version>.json entry matching the release identity.
+
+The source convergence gate intentionally checks release identity, not
+post-deployment runtime fields. This allows the source entry to be committed
+before deployment while still preventing PRODUCTION_DEPLOYED from being
+written for an uncommitted release identity.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import stat
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+_VERSION_RE = re.compile(r"^V[0-9]+R[0-9]+C[0-9]+(?:-[A-Za-z0-9_.-]+)?$")
+_OID_RE = re.compile(r"^[0-9a-f]{40}$")
+_PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
+VERSION_RE = _VERSION_RE
+OID_RE = _OID_RE
+SOURCE_REGISTRY_SCHEMAS = frozenset(
+    {"aios.source-registry.v1", "aios.production-registry.v2"}
+)
+SOURCE_REGISTRY_FIELDS = (
+    "version",
+    "tag_oid",
+    "target_oid",
+    "peeled_target_oid",
+    "evidence_closure_oid",
+    "approval_plan_id",
+)
+REPOSITORY_IDENTITY = "aios/aios"
+
+DEFAULT_PROJECTION_ROOT = Path(
+    os.getenv(
+        "AIOS_GITOPS_REGISTRY_PROJECTION_ROOT",
+        "/var/lib/aios-gitops/runtime/registry-projection",
+    )
+)
+DEFAULT_BASELINE_FILE = Path(
+    os.getenv(
+        "AIOS_GITOPS_BASELINE_FILE",
+        "/var/lib/aios-gitops-runtime-anchor/baseline/current.json",
+    )
+)
+DEFAULT_MAIN_REF = os.getenv("AIOS_GITOPS_MAIN_REF", "refs/heads/main")
+DEFAULT_AUTHORITATIVE_GIT_DIR = Path("/data/aios-gitops/repo.git")
+DEFAULT_APPROVAL_AUTHORITY_ROOT = Path("/var/lib/aios-gitops/approval-authority")
+TEST_MODE_ENV = "AIOS_TEST_MODE"
+AUTHORITATIVE_GIT_ENV = "AIOS_GITOPS_AUTHORITATIVE_GIT_DIR"
+APPROVAL_AUTHORITY_ROOT_ENV = "AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT"
+REPOSITORY_IDENTITY_ENV = "AIOS_GITOPS_REPOSITORY_IDENTITY"
+
+
+class RegistryConvergenceError(RuntimeError):
+    """Release registry/source convergence is not proven."""
+
+
+def _load_object(path: Path) -> dict[str, Any]:
+    obj = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(obj, dict):
+        raise RegistryConvergenceError(f"json_not_object:{path}")
+    return obj
+
+
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _git_text(repo_git_dir: Path | str, *args: str) -> str:
+    completed = subprocess.run(
+        ["git", f"--git-dir={Path(repo_git_dir)}", *args],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=45,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if completed.returncode != 0:
+        raise RegistryConvergenceError(
+            completed.stderr.strip() or "authoritative_git_query_failed"
+        )
+    return completed.stdout.strip()
+
+
+def _git_bytes(repo_git_dir: Path | str, *args: str) -> bytes:
+    completed = subprocess.run(
+        ["git", f"--git-dir={Path(repo_git_dir)}", *args],
+        capture_output=True,
+        check=False,
+        timeout=45,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if completed.returncode != 0:
+        raise RegistryConvergenceError(
+            completed.stderr.decode("utf-8", "replace").strip()
+            or "authoritative_git_query_failed"
+        )
+    return completed.stdout
+
+
+def test_mode_enabled() -> bool:
+    """Return whether explicit dependency injection is allowed for tests."""
+    return os.environ.get(TEST_MODE_ENV) == "1"
+
+
+def _configured_override(env_name: str, default: str, label: str) -> Path:
+    if env_name in os.environ:
+        if not test_mode_enabled():
+            raise RegistryConvergenceError(f"{label}_env_override_forbidden")
+        value = os.environ.get(env_name, "")
+        if not value:
+            raise RegistryConvergenceError(f"{label}_env_override_invalid")
+        return Path(value)
+    return Path(default)
+
+
+def configured_authoritative_git_dir() -> Path:
+    return _configured_override(
+        AUTHORITATIVE_GIT_ENV,
+        str(DEFAULT_AUTHORITATIVE_GIT_DIR),
+        "authoritative_repository",
+    )
+
+
+def configured_repository_identity() -> str:
+    if REPOSITORY_IDENTITY_ENV in os.environ:
+        if not test_mode_enabled():
+            raise RegistryConvergenceError("repository_identity_env_override_forbidden")
+        value = os.environ.get(REPOSITORY_IDENTITY_ENV, "")
+        if not value:
+            raise RegistryConvergenceError("repository_identity_env_override_invalid")
+        return value
+    return REPOSITORY_IDENTITY
+
+
+def verify_authoritative_repository(
+    repo_git_dir: Path | str,
+    *,
+    repository_identity: str = REPOSITORY_IDENTITY,
+    expected_path: Path | str | None = None,
+) -> dict[str, str]:
+    """Prove that a Git directory is the configured AIOS authority.
+
+    A repository that merely contains internally consistent objects is not
+    sufficient.  The resolved path must be the configured authority and the
+    bare repository must advertise the expected Gitea repository identity.
+    """
+    if repository_identity != REPOSITORY_IDENTITY:
+        raise RegistryConvergenceError("authoritative_repository_identity_invalid")
+    candidate = Path(repo_git_dir)
+    configured = Path(expected_path) if expected_path is not None else configured_authoritative_git_dir()
+    if test_mode_enabled() and expected_path is None and AUTHORITATIVE_GIT_ENV not in os.environ:
+        # Explicit test mode may inject a temporary bare repository directly.
+        # Production mode always compares against the fixed authority path.
+        configured = candidate
+    try:
+        candidate_real = candidate.resolve(strict=True)
+        configured_real = configured.resolve(strict=True)
+    except OSError as exc:
+        raise RegistryConvergenceError("authoritative_repository_path_unavailable") from exc
+    if candidate_real != configured_real:
+        raise RegistryConvergenceError("authoritative_repository_path_mismatch")
+    if not candidate_real.is_dir():
+        raise RegistryConvergenceError("authoritative_repository_not_directory")
+    if _git_text(candidate_real, "rev-parse", "--is-bare-repository") != "true":
+        raise RegistryConvergenceError("authoritative_repository_not_bare")
+    remotes = subprocess.run(
+        [
+            "git",
+            f"--git-dir={candidate_real}",
+            "config",
+            "--get-regexp",
+            r"^remote\..*\.url$",
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=45,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    urls = [line.split(None, 1)[1] for line in remotes.stdout.splitlines() if " " in line]
+    identity_suffix = repository_identity.removesuffix(".git")
+    identity_verified = any(
+        url.rstrip("/").removesuffix(".git").replace("\\", "/").split(":")[-1]
+        in {identity_suffix, "/" + identity_suffix}
+        or url.rstrip("/").removesuffix(".git").replace("\\", "/").split(":")[-1].endswith(
+            "/" + identity_suffix
+        )
+        for url in urls
+    )
+    if not test_mode_enabled() and not identity_verified:
+        raise RegistryConvergenceError("authoritative_repository_identity_unverified")
+    return {
+        "repository_identity": repository_identity,
+        "configured_path": str(configured_real),
+        "resolved_path": str(candidate_real),
+    }
+
+
+def resolve_source_registry_main_oid(
+    repo_git_dir: Path | str,
+    *,
+    main_ref: str = DEFAULT_MAIN_REF,
+) -> str:
+    """Resolve the authoritative main ref to one immutable commit OID."""
+    oid = _git_text(repo_git_dir, "rev-parse", "--verify", f"{main_ref}^{{commit}}")
+    if not _OID_RE.fullmatch(oid):
+        raise RegistryConvergenceError("source_registry_main_oid_invalid")
+    if _git_text(repo_git_dir, "cat-file", "-t", oid) != "commit":
+        raise RegistryConvergenceError("source_registry_main_oid_not_commit")
+    return oid
+
+
+def source_registry_snapshot(
+    repo_git_dir: Path | str,
+    version: str,
+    *,
+    main_ref: str = DEFAULT_MAIN_REF,
+    source_registry_main_oid: str | None = None,
+    expected_tag_oid: str | None = None,
+    expected_target_oid: str | None = None,
+    expected_evidence_closure_oid: str | None = None,
+    expected_approval_plan_id: str | None = None,
+) -> dict[str, Any]:
+    """Read one source entry and its exact bytes from one immutable commit.
+
+    Callers building an artifact should resolve ``source_registry_main_oid``
+    once and pass it back here.  The function then never consults the mutable
+    ref again, so validation and packaging use the same Git blob.
+    """
+    if not isinstance(version, str) or not _VERSION_RE.fullmatch(version):
+        raise RegistryConvergenceError("source_registry_version_invalid")
+    main_oid = source_registry_main_oid or resolve_source_registry_main_oid(
+        repo_git_dir,
+        main_ref=main_ref,
+    )
+    if not _OID_RE.fullmatch(main_oid):
+        raise RegistryConvergenceError("source_registry_main_oid_invalid")
+    if _git_text(repo_git_dir, "cat-file", "-t", main_oid) != "commit":
+        raise RegistryConvergenceError("source_registry_main_oid_not_commit")
+    path = f"versions/{version}.json"
+    try:
+        raw = _git_bytes(repo_git_dir, "show", f"{main_oid}:{path}")
+    except RegistryConvergenceError as exc:
+        if "path" in str(exc).lower() or "does not exist" in str(exc).lower():
+            raise RegistryConvergenceError(
+                f"source_registry_entry_missing:{main_oid}:{path}"
+            ) from exc
+        raise
+    try:
+        source = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RegistryConvergenceError("source_registry_json_invalid") from exc
+    record = validate_source_registry_record(
+        source,
+        expected_version=version,
+        expected_tag_oid=expected_tag_oid,
+        expected_target_oid=expected_target_oid,
+        expected_evidence_closure_oid=expected_evidence_closure_oid,
+        expected_approval_plan_id=expected_approval_plan_id,
+    )
+    return {
+        "record": record,
+        "bytes": raw,
+        "source_registry_main_oid": main_oid,
+        "source_registry_path": path,
+        "source_registry_sha256": hashlib.sha256(raw).hexdigest(),
+    }
+
+
+def validate_source_registry_record(
+    value: Any,
+    *,
+    expected_version: str | None = None,
+    expected_tag_oid: str | None = None,
+    expected_target_oid: str | None = None,
+    expected_evidence_closure_oid: str | None = None,
+    expected_approval_plan_id: str | None = None,
+) -> dict[str, Any]:
+    """Validate the immutable identity carried by versions/<version>.json."""
+    if not isinstance(value, dict):
+        raise RegistryConvergenceError("source_registry_not_object")
+    if value.get("schema") not in SOURCE_REGISTRY_SCHEMAS:
+        raise RegistryConvergenceError("source_registry_schema_invalid")
+    version = value.get("version")
+    if not isinstance(version, str) or not _VERSION_RE.fullmatch(version):
+        raise RegistryConvergenceError("source_registry_version_invalid")
+    if expected_version is not None and version != expected_version:
+        raise RegistryConvergenceError("source_registry_mismatch:version")
+    if value.get("repository") != REPOSITORY_IDENTITY:
+        raise RegistryConvergenceError("source_registry_repository_invalid")
+    for field in ("tag_oid", "target_oid", "peeled_target_oid", "evidence_closure_oid"):
+        item = value.get(field)
+        if not isinstance(item, str) or not _OID_RE.fullmatch(item):
+            raise RegistryConvergenceError(f"source_registry_{field}_invalid")
+    if value["peeled_target_oid"] != value["target_oid"]:
+        raise RegistryConvergenceError("source_registry_peeled_target_mismatch")
+    approval = value.get("approval_plan_id")
+    if not isinstance(approval, str) or not _PLAN_RE.fullmatch(approval):
+        raise RegistryConvergenceError("source_registry_approval_plan_invalid")
+    if value.get("tag_object", value["tag_oid"]) != value["tag_oid"]:
+        raise RegistryConvergenceError("source_registry_tag_object_mismatch")
+    expected = {
+        "tag_oid": expected_tag_oid,
+        "target_oid": expected_target_oid,
+        "evidence_closure_oid": expected_evidence_closure_oid,
+        "approval_plan_id": expected_approval_plan_id,
+    }
+    for field, wanted in expected.items():
+        if wanted is not None and value.get(field) != wanted:
+            raise RegistryConvergenceError(f"source_registry_mismatch:{field}")
+    return dict(value)
+
+
+def build_source_registry_entry(
+    *,
+    version: str,
+    tag_oid: str,
+    target_oid: str,
+    peeled_target_oid: str,
+    evidence_closure_oid: str,
+    approval_plan_id: str,
+    repository: str = REPOSITORY_IDENTITY,
+) -> dict[str, Any]:
+    """Build the pre-promotion source entry; it performs no Git or runtime write."""
+    value = {
+        "schema": "aios.source-registry.v1",
+        "version": version,
+        "repository": repository,
+        "tag": version,
+        "tag_oid": tag_oid,
+        "tag_object": tag_oid,
+        "target_oid": target_oid,
+        "peeled_target_oid": peeled_target_oid,
+        "evidence_closure_oid": evidence_closure_oid,
+        "approval_plan_id": approval_plan_id,
+    }
+    return validate_source_registry_record(
+        value,
+        expected_version=version,
+        expected_tag_oid=tag_oid,
+        expected_target_oid=target_oid,
+        expected_evidence_closure_oid=evidence_closure_oid,
+        expected_approval_plan_id=approval_plan_id,
+    )
+
+
+def verify_source_registry_git_identity(
+    repo_git_dir: Path | str,
+    record: dict[str, Any],
+) -> dict[str, Any]:
+    """Prove that every Git identity in a source entry exists and is related."""
+    record = validate_source_registry_record(record)
+    version = record["version"]
+    tag_oid = record["tag_oid"]
+    target_oid = record["target_oid"]
+    evidence_oid = record["evidence_closure_oid"]
+    if _git_text(repo_git_dir, "cat-file", "-t", tag_oid) != "tag":
+        raise RegistryConvergenceError("source_registry_tag_not_annotated")
+    named_tag = _git_text(
+        repo_git_dir,
+        "rev-parse",
+        "--verify",
+        f"refs/tags/{version}",
+    )
+    if named_tag != tag_oid:
+        raise RegistryConvergenceError("source_registry_named_tag_mismatch")
+    if _git_text(repo_git_dir, "cat-file", "-t", target_oid) != "commit":
+        raise RegistryConvergenceError("source_registry_target_not_commit")
+    peeled = _git_text(
+        repo_git_dir,
+        "rev-parse",
+        "--verify",
+        f"{tag_oid}^{{commit}}",
+    )
+    if peeled != target_oid or record["peeled_target_oid"] != target_oid:
+        raise RegistryConvergenceError("source_registry_tag_target_mismatch")
+    if _git_text(repo_git_dir, "cat-file", "-t", evidence_oid) != "commit":
+        raise RegistryConvergenceError("source_registry_evidence_not_commit")
+    ancestor = subprocess.run(
+        [
+            "git",
+            f"--git-dir={Path(repo_git_dir)}",
+            "merge-base",
+            "--is-ancestor",
+            target_oid,
+            evidence_oid,
+        ],
+        check=False,
+        capture_output=True,
+        timeout=45,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if ancestor.returncode != 0:
+        raise RegistryConvergenceError("source_registry_evidence_lineage_invalid")
+    return {
+        "repository": REPOSITORY_IDENTITY,
+        "version": version,
+        "tag_oid": tag_oid,
+        "tag_object_type": "tag",
+        "target_oid": target_oid,
+        "peeled_target_oid": peeled,
+        "evidence_closure_oid": evidence_oid,
+        "target_is_ancestor_of_evidence": True,
+    }
+
+
+def verify_source_registry_main_convergence(
+    repo_git_dir: Path | str,
+    version: str,
+    *,
+    main_ref: str = DEFAULT_MAIN_REF,
+    source_registry_main_oid: str | None = None,
+    expected_tag_oid: str | None = None,
+    expected_target_oid: str | None = None,
+    expected_evidence_closure_oid: str | None = None,
+    expected_approval_plan_id: str | None = None,
+) -> dict[str, Any]:
+    """Prove that pinned M is still converged with current authoritative main C."""
+    current_main_oid = resolve_source_registry_main_oid(repo_git_dir, main_ref=main_ref)
+    source_main_oid = source_registry_main_oid or current_main_oid
+    if not _OID_RE.fullmatch(source_main_oid):
+        raise RegistryConvergenceError("source_registry_main_oid_invalid")
+    ancestor = subprocess.run(
+        [
+            "git",
+            f"--git-dir={Path(repo_git_dir)}",
+            "merge-base",
+            "--is-ancestor",
+            source_main_oid,
+            current_main_oid,
+        ],
+        check=False,
+        capture_output=True,
+        timeout=45,
+        env={"PATH": os.environ.get("PATH", "")},
+    )
+    if ancestor.returncode != 0:
+        raise RegistryConvergenceError("source_registry_main_not_ancestor_of_current_main")
+    pinned = source_registry_snapshot(
+        repo_git_dir,
+        version,
+        source_registry_main_oid=source_main_oid,
+        expected_tag_oid=expected_tag_oid,
+        expected_target_oid=expected_target_oid,
+        expected_evidence_closure_oid=expected_evidence_closure_oid,
+        expected_approval_plan_id=expected_approval_plan_id,
+    )
+    current = source_registry_snapshot(
+        repo_git_dir,
+        version,
+        source_registry_main_oid=current_main_oid,
+        expected_tag_oid=expected_tag_oid,
+        expected_target_oid=expected_target_oid,
+        expected_evidence_closure_oid=expected_evidence_closure_oid,
+        expected_approval_plan_id=expected_approval_plan_id,
+    )
+    if pinned["bytes"] != current["bytes"]:
+        raise RegistryConvergenceError("source_registry_main_entry_changed_after_pin")
+    return {
+        "source_registry_main_oid": source_main_oid,
+        "current_main_oid": current_main_oid,
+        "source_registry_path": pinned["source_registry_path"],
+        "source_registry_sha256": pinned["source_registry_sha256"],
+        "current_main_source_registry_sha256": current["source_registry_sha256"],
+        "source_registry_main_is_ancestor_of_current_main": True,
+        "source_registry_entry_matches_current_main": True,
+    }
+
+
+_APPROVAL_AUTHORITY_STATES = frozenset(
+    {
+        "APPROVED",
+        "GITEA_API_CONFIRMED",
+        "REMOTE_VERIFIED",
+        "PRODUCTION_DEPLOYED",
+    }
+)
+_APPROVAL_AUTHORITY_SCHEMAS = frozenset(
+    {
+        "aios.v2r2c1.tag-approval-request.v1",
+        "aios.owner-release-approval.v1",
+        "aios.release-binding-approval.v2",
+    }
+)
+
+
+def configured_approval_authority_root() -> Path:
+    return _configured_override(
+        APPROVAL_AUTHORITY_ROOT_ENV,
+        str(DEFAULT_APPROVAL_AUTHORITY_ROOT),
+        "approval_authority_root",
+    )
+
+
+def _secure_approval_authority_path(
+    authority_path: Path | str,
+    version: str,
+) -> tuple[Path, Path]:
+    root = configured_approval_authority_root()
+    path = Path(authority_path)
+    if path.is_dir():
+        path = path / f"{version}.json"
+    try:
+        root_real = root.resolve(strict=True)
+        path_real = path.resolve(strict=True)
+    except OSError as exc:
+        raise RegistryConvergenceError("approval_authority_trusted_root_missing") from exc
+    try:
+        path_real.relative_to(root_real)
+    except ValueError as exc:
+        raise RegistryConvergenceError("approval_authority_untrusted_path") from exc
+    metadata = path.lstat()
+    if not stat.S_ISREG(metadata.st_mode) or path.is_symlink():
+        raise RegistryConvergenceError("approval_authority_not_regular")
+    if metadata.st_mode & 0o022:
+        raise RegistryConvergenceError("approval_authority_permissions_invalid")
+    if APPROVAL_AUTHORITY_ROOT_ENV not in os.environ and metadata.st_uid != 0:
+        raise RegistryConvergenceError("approval_authority_owner_invalid")
+    return path, root_real
+
+
+def _canonical_json(value: object) -> bytes:
+    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode(
+        "utf-8"
+    )
+
+
+def _approval_is_revoked(root: Path, approval_plan_id: str, authority: dict[str, Any]) -> bool:
+    if any(
+        authority.get(name) is True
+        for name in ("revoked", "invalidated", "revocation")
+    ):
+        return True
+    state = authority.get("state") or authority.get("status")
+    if state in {"REVOKED", "INVALIDATED", "EXPIRED", "CANCELLED"}:
+        return True
+    for marker in (
+        root / "revocations" / approval_plan_id,
+        root / "revocations" / approval_plan_id.replace(":", "_"),
+    ):
+        if marker.exists():
+            return True
+    expires_at = authority.get("expires_at")
+    if isinstance(expires_at, str) and expires_at:
+        try:
+            instant = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
+        except ValueError as exc:
+            raise RegistryConvergenceError("approval_expiry_invalid") from exc
+        if instant <= datetime.now(timezone.utc):
+            return True
+    return False
+
+
+def verify_approval_authority(
+    authority_path: Path | str,
+    record: dict[str, Any],
+) -> dict[str, Any]:
+    """Verify the existing release approval/publication receipt association.
+
+    The authority document is intentionally supplied by the governed caller;
+    this helper never creates or rewrites an approval.  The path must be under
+    the configured canonical Owner-authority root and the document must use a
+    known release-approval schema.  A caller-supplied JSON file elsewhere is
+    never an authority, even when its fields are internally consistent.
+    """
+    path, trusted_root = _secure_approval_authority_path(authority_path, record["version"])
+    authority = _load_object(path)
+    schema = authority.get("schema")
+    if schema not in _APPROVAL_AUTHORITY_SCHEMAS:
+        raise RegistryConvergenceError("approval_authority_schema_invalid")
+    if _approval_is_revoked(trusted_root, record["approval_plan_id"], authority):
+        raise RegistryConvergenceError("approval_plan_revoked_or_invalidated")
+    nested = authority.get("release_binding") or authority.get("release_request")
+    if not isinstance(nested, dict):
+        nested = authority.get("binding")
+    if not isinstance(nested, dict):
+        raise RegistryConvergenceError("approval_authority_binding_missing")
+    expected = {
+        "version": record["version"],
+        "repository": REPOSITORY_IDENTITY,
+        "tag_oid": record["tag_oid"],
+        "target_oid": record["target_oid"],
+        "peeled_target_oid": record["peeled_target_oid"],
+        "evidence_closure_oid": record["evidence_closure_oid"],
+        "approval_plan_id": record["approval_plan_id"],
+    }
+    if authority.get("approval_plan_id") != expected["approval_plan_id"]:
+        raise RegistryConvergenceError("approval_plan_not_found_or_mismatched")
+    state = authority.get("state") or authority.get("status")
+    if state not in _APPROVAL_AUTHORITY_STATES and authority.get("approved") is not True:
+        raise RegistryConvergenceError("approval_plan_not_approved")
+    owner_approval = authority.get("owner_approval")
+    if schema == "aios.v2r2c1.tag-approval-request.v1":
+        if not isinstance(owner_approval, dict):
+            raise RegistryConvergenceError("canonical_owner_approval_missing")
+        if (
+            owner_approval.get("approved") is not True
+            or owner_approval.get("status") != "APPROVED"
+            or owner_approval.get("approved_by") != "aiman"
+            or owner_approval.get("approval_source") != "explicit_owner_approval_in_chat"
+        ):
+            raise RegistryConvergenceError("canonical_owner_approval_invalid")
+        expected_digest = hashlib.sha256(_canonical_json(owner_approval)).hexdigest()
+        if authority.get("owner_approval_digest_sha256") != expected_digest:
+            raise RegistryConvergenceError("canonical_owner_approval_digest_invalid")
+    for field, wanted in expected.items():
+        observed = nested.get(field)
+        if field == "repository" and observed is None:
+            observed = nested.get("repository_identity") or authority.get(
+                "repository_identity"
+            )
+        if field == "peeled_target_oid" and observed is None:
+            observed = (
+                nested.get("gitea_peeled_target_oid")
+                or authority.get("gitea_peeled_target_oid")
+                or nested.get("remote_peeled_target_oid")
+                or authority.get("remote_peeled_target_oid")
+            )
+        if observed != wanted:
+            raise RegistryConvergenceError(f"approval_plan_release_binding_mismatch:{field}")
+    if state == "GITEA_API_CONFIRMED":
+        gitea_tag = authority.get("gitea_tag_oid")
+        gitea_peeled = authority.get("gitea_peeled_target_oid")
+        if gitea_tag != expected["tag_oid"] or gitea_peeled != expected["target_oid"]:
+            raise RegistryConvergenceError("approval_plan_remote_tag_binding_mismatch")
+    return {
+        "path": str(path),
+        "schema": schema,
+        "state": state or "APPROVED",
+        "approval_plan_id": expected["approval_plan_id"],
+        "owner_approval_verified": schema == "aios.v2r2c1.tag-approval-request.v1",
+        "release_binding_verified": True,
+    }
+
+
+def source_registry_json(value: dict[str, Any]) -> bytes:
+    """Return the canonical bytes used by the governed file writer."""
+    return (json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode(
+        "utf-8"
+    )
+
+
+def _validated_current_projection(
+    projection_root: Path,
+    baseline_file: Path,
+) -> Path | None:
+    """Return only the exact projection attested by the current Baseline."""
+    if not baseline_file.is_file():
+        return None
+    try:
+        baseline = _load_object(baseline_file)
+    except Exception:
+        return None
+
+    version = baseline.get("version")
+    if not isinstance(version, str) or not _VERSION_RE.fullmatch(version):
+        return None
+    if baseline.get("status") != "PRODUCTION_DEPLOYED":
+        return None
+    if baseline.get("production_deployed") is not True:
+        return None
+
+    source = baseline.get("registry_source")
+    digest = baseline.get("registry_digest")
+    if not isinstance(source, str) or not isinstance(digest, str):
+        return None
+
+    candidate = projection_root / f"{version}.json"
+    try:
+        if candidate.resolve() != Path(source).resolve():
+            return None
+    except Exception:
+        return None
+    if not candidate.is_file() or _sha256(candidate) != digest:
+        return None
+
+    try:
+        record = _load_object(candidate)
+    except Exception:
+        return None
+
+    try:
+        validate_source_registry_record(
+            record,
+            expected_version=version,
+            expected_tag_oid=baseline.get("tag_oid"),
+            expected_target_oid=baseline.get("target_oid"),
+            expected_evidence_closure_oid=baseline.get("evidence_closure_oid"),
+            expected_approval_plan_id=baseline.get("approval_plan_id"),
+        )
+    except RegistryConvergenceError:
+        return None
+    if record.get("schema") != "aios.production-registry.v2":
+        return None
+    if record.get("status") != "PRODUCTION_DEPLOYED":
+        return None
+    if record.get("version") != version:
+        return None
+    if record.get("repository") != REPOSITORY_IDENTITY:
+        return None
+    if record.get("production_deployed") is not True:
+        return None
+
+    tag_oid = record.get("tag_oid")
+    target_oid = record.get("target_oid")
+    if not isinstance(tag_oid, str) or not _OID_RE.fullmatch(tag_oid):
+        return None
+    if not isinstance(target_oid, str) or not _OID_RE.fullmatch(target_oid):
+        return None
+    if record.get("peeled_target_oid") != target_oid:
+        return None
+
+    return candidate
+
+
+def list_registry_versions(
+    workspace: Path | str,
+    *,
+    projection_root: Path = DEFAULT_PROJECTION_ROOT,
+    baseline_file: Path = DEFAULT_BASELINE_FILE,
+) -> list[str]:
+    """Return the legacy display list; use the report for provenance."""
+    return list_registry_versions_report(
+        workspace,
+        projection_root=projection_root,
+        baseline_file=baseline_file,
+    )["registry_versions"]
+
+
+def list_registry_versions_report(
+    workspace: Path | str,
+    *,
+    projection_root: Path = DEFAULT_PROJECTION_ROOT,
+    baseline_file: Path = DEFAULT_BASELINE_FILE,
+    repo_git_dir: Path | str | None = None,
+    main_ref: str = DEFAULT_MAIN_REF,
+) -> dict[str, Any]:
+    """Return version visibility with runtime/projection provenance.
+
+    The display list intentionally retains the historical projection overlay
+    for API compatibility.  The explicit booleans make that overlay
+    non-authoritative: a projection-only version has
+    ``runtime_bundle_contains_version == false`` and can never be reported as
+    converged.
+    """
+    ws = Path(workspace)
+    versions_root = ws / "versions"
+    runtime_names = {
+        path.name
+        for path in versions_root.iterdir()
+        if path.is_file()
+        and not path.is_symlink()
+        and path.suffix == ".json"
+        and _VERSION_RE.fullmatch(path.stem)
+    } if versions_root.is_dir() else set()
+    current = _validated_current_projection(projection_root, baseline_file)
+    attested_version = current.stem if current is not None else None
+    runtime_bundle_contains = bool(
+        attested_version is not None
+        and f"{attested_version}.json" in runtime_names
+    )
+    consistency_status = "NO_ATTESTED_PROJECTION"
+    source_converged: bool | str = "unknown"
+    source_error: str | None = None
+    runtime_converged = False
+    projection_record: dict[str, Any] | None = None
+    if current is not None:
+        try:
+            projection_record = _load_object(current)
+        except (OSError, RegistryConvergenceError, json.JSONDecodeError) as exc:
+            source_error = f"projection_invalid:{type(exc).__name__}"
+        if repo_git_dir is None:
+            source_converged = "unknown"
+            source_error = "authoritative_repository_unavailable"
+        elif projection_record is not None:
+            try:
+                verify_authoritative_repository(repo_git_dir)
+                verify_source_registry_main_convergence(
+                    repo_git_dir,
+                    attested_version or "",
+                    main_ref=main_ref,
+                    expected_tag_oid=projection_record.get("tag_oid"),
+                    expected_target_oid=projection_record.get("target_oid"),
+                    expected_evidence_closure_oid=projection_record.get(
+                        "evidence_closure_oid"
+                    ),
+                    expected_approval_plan_id=projection_record.get("approval_plan_id"),
+                )
+                source_converged = True
+            except RegistryConvergenceError as exc:
+                detail = str(exc)
+                if detail.startswith("authoritative_repository_"):
+                    source_converged = "unknown"
+                    source_error = (
+                        "authoritative_repository_unavailable"
+                        if detail.endswith("path_unavailable")
+                        else detail
+                    )
+                else:
+                    source_converged = False
+                    source_error = detail
+    if current is not None and not runtime_bundle_contains:
+        consistency_status = "RUNTIME_ARTIFACT_RECONCILIATION_REQUIRED"
+    elif current is not None and runtime_bundle_contains:
+        runtime_path = versions_root / current.name
+        try:
+            projection_record = projection_record or _load_object(current)
+            runtime_record = _load_object(runtime_path)
+            identity_fields = (
+                "version",
+                "repository",
+                "tag_oid",
+                "target_oid",
+                "peeled_target_oid",
+                "evidence_closure_oid",
+                "approval_plan_id",
+            )
+            runtime_converged = all(
+                runtime_record.get(field) == projection_record.get(field)
+                for field in identity_fields
+            )
+            if not runtime_converged:
+                consistency_status = "RUNTIME_IDENTITY_MISMATCH"
+        except (OSError, RegistryConvergenceError, json.JSONDecodeError):
+            consistency_status = "RUNTIME_ENTRY_INVALID"
+            runtime_converged = False
+    if current is not None and runtime_bundle_contains and runtime_converged:
+        if source_converged is True:
+            consistency_status = "FULLY_CONSISTENT"
+        elif source_converged == "unknown":
+            consistency_status = "SOURCE_GIT_NOT_CHECKED"
+        else:
+            consistency_status = "SOURCE_REGISTRY_NOT_CONVERGED"
+    elif current is not None and not runtime_bundle_contains:
+        if source_converged is False:
+            consistency_status = "SOURCE_REGISTRY_NOT_CONVERGED"
+        else:
+            consistency_status = "RUNTIME_ARTIFACT_RECONCILIATION_REQUIRED"
+    names = set(runtime_names)
+    if current is not None:
+        names.add(current.name)
+    effective = sorted(names)
+    projection_versions = [attested_version + ".json"] if attested_version else []
+    return {
+        "registry_versions": effective,
+        "effective_registry_versions": effective,
+        "runtime_registry_versions": sorted(runtime_names),
+        "attested_projection_versions": projection_versions,
+        "attested_projection_version": attested_version,
+        "source_registry_converged": source_converged,
+        "source_registry_convergence_error": source_error,
+        "runtime_registry_converged": runtime_converged,
+        "runtime_bundle_contains_current_version": runtime_bundle_contains,
+        "runtime_bundle_contains_version": runtime_bundle_contains,
+        "registry_consistency_status": consistency_status,
+    }
+
+
+def assert_runtime_registry_convergence(
+    workspace: Path | str,
+    version: str,
+    *,
+    projection_root: Path = DEFAULT_PROJECTION_ROOT,
+    baseline_file: Path = DEFAULT_BASELINE_FILE,
+) -> dict[str, Any]:
+    """Fail closed when a projection-only version is visible but unpackaged."""
+    report = list_registry_versions_report(
+        workspace,
+        projection_root=projection_root,
+        baseline_file=baseline_file,
+    )
+    expected_name = f"{version}.json"
+    if (
+        report.get("attested_projection_version") == version
+        and report.get("runtime_bundle_contains_version") is not True
+    ):
+        raise RegistryConvergenceError("runtime_bundle_missing_source_registry")
+    if expected_name not in report.get("runtime_registry_versions", []):
+        raise RegistryConvergenceError("runtime_bundle_missing_source_registry")
+    if report.get("source_registry_converged") is not True:
+        raise RegistryConvergenceError("source_registry_not_converged_or_not_checked")
+    if report.get("registry_consistency_status") not in {"FULLY_CONSISTENT", "PASS"}:
+        raise RegistryConvergenceError(
+            f"runtime_registry_consistency:{report.get('registry_consistency_status')}"
+        )
+    return report
+
+
+def source_registry_from_main(
+    repo_git_dir: Path | str,
+    version: str,
+    *,
+    main_ref: str = DEFAULT_MAIN_REF,
+    source_registry_main_oid: str | None = None,
+    expected_tag_oid: str | None = None,
+    expected_target_oid: str | None = None,
+    expected_evidence_closure_oid: str | None = None,
+    expected_approval_plan_id: str | None = None,
+) -> dict[str, Any]:
+    """Read and validate the canonical source entry from the supplied main ref."""
+    try:
+        return source_registry_snapshot(
+            repo_git_dir,
+            version,
+            main_ref=main_ref,
+            source_registry_main_oid=source_registry_main_oid,
+            expected_tag_oid=expected_tag_oid,
+            expected_target_oid=expected_target_oid,
+            expected_evidence_closure_oid=expected_evidence_closure_oid,
+            expected_approval_plan_id=expected_approval_plan_id,
+        )["record"]
+    except RegistryConvergenceError as exc:
+        if str(exc).startswith("fatal:") or "path" in str(exc).lower():
+            raise RegistryConvergenceError(
+                f"source_registry_entry_missing:{source_registry_main_oid or main_ref}:versions/{version}.json"
+            ) from exc
+        raise
+
+
+def assert_source_registry_convergence(
+    registry_path: Path | str,
+    *,
+    repo_git_dir: Path | str,
+    main_ref: str = DEFAULT_MAIN_REF,
+    source_registry_main_oid: str | None = None,
+) -> dict[str, Any]:
+    """Require governed main to contain a source entry for the release identity.
+
+    The source entry is intentionally required to match only fields that are
+    knowable and governable before production deployment. Post-deployment
+    fields such as production_service/port/status timestamps are not required.
+    """
+    registry_path = Path(registry_path)
+    repo_git_dir = Path(repo_git_dir)
+    verify_authoritative_repository(repo_git_dir)
+    record = _load_object(registry_path)
+
+    version = record.get("version")
+    tag_oid = record.get("tag_oid")
+    target_oid = record.get("target_oid")
+
+    try:
+        validate_source_registry_record(
+            record,
+            expected_version=version,
+            expected_tag_oid=tag_oid,
+            expected_target_oid=target_oid,
+            expected_evidence_closure_oid=record.get("evidence_closure_oid"),
+            expected_approval_plan_id=record.get("approval_plan_id"),
+        )
+    except RegistryConvergenceError as exc:
+        raise RegistryConvergenceError(f"registry_identity_invalid:{exc}") from exc
+    convergence = verify_source_registry_main_convergence(
+        repo_git_dir,
+        version,
+        main_ref=main_ref,
+        source_registry_main_oid=source_registry_main_oid,
+        expected_tag_oid=tag_oid,
+        expected_target_oid=target_oid,
+        expected_evidence_closure_oid=record["evidence_closure_oid"],
+        expected_approval_plan_id=record["approval_plan_id"],
+    )
+    return {
+        **convergence,
+    }

reports/V2R2C1-source-registry-reconciliation-20260810.json

@@ -0,0 +1,24 @@
+{
+  "schema": "aios.source-registry-reconciliation.v1",
+  "version": "V2R2C1",
+  "root_cause": "PROMOTION_STATE_MACHINE_FINALIZED_BEFORE_SOURCE_REGISTRY_CONVERGENCE",
+  "main_before": "3244f0f17f8845592ac94d585f5ebd65693c9f29",
+  "tag_oid": "07fc2746b688f5f243f29b39de39f32edc3dd9d2",
+  "target_oid": "998d30304787e31e5ae607c0bd86b5edb2b0e5ad",
+  "projection_sha256": "924dfc0ec58afc4e31638f6538e551c5bc8858f804cdbee5af7dce6eba7c90fe",
+  "evidence_closure_oid": "a8b906d1bc24088c19647e98de6a679268f6d0a2",
+  "approval_plan_id": "sha256:c0df52008ae7f64cec5ee2d28947aa7b892066c670ab8295e4549140b1a7500e",
+  "remote_tag_verified": true,
+  "remote_target_verified": true,
+  "gitea_api_verified_by_owner_acceptance": true,
+  "source_entry_before": "missing",
+  "source_entry_added": "versions/V2R2C1.json",
+  "source_entry_is_exact_attested_projection": true,
+  "release_manifest_added": false,
+  "release_manifest_reason": "Historical release-manifest schema differs; no canonical V2R2C1 source-manifest generator was proven. Do not synthesize one by copying old release values.",
+  "runtime_backing": "immutable WS/versions bundle",
+  "runtime_mutated": false,
+  "tag_rewritten": false,
+  "baseline_modified": false,
+  "production_modified": false
+}

scripts/autonomous-release/build_artifact.py

@@ -32,18 +32,36 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--tree-oid", required=True)
     parser.add_argument("--output-directory", required=True, type=Path)
     parser.add_argument("--name", default="aios-autonomous-release-candidate")
+    parser.add_argument("--release-version")
+    parser.add_argument("--release-binding-json", type=Path)
+    parser.add_argument("--artifact-mode", choices=("release", "general"))
     return parser
 
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
+    release_binding = (
+        json.loads(args.release_binding_json.read_text(encoding="utf-8"))
+        if args.release_binding_json is not None
+        else None
+    )
+    if release_binding is not None and not isinstance(release_binding, dict):
+        raise ValueError("release binding JSON must be an object")
     output = args.output_directory
     output.mkdir(parents=True, exist_ok=True, mode=0o750)
     if output.is_symlink():
         raise RuntimeError("artifact output directory symlink rejected")
     os.chmod(output, output.stat().st_mode & 0o750)
     builder = ArtifactBuilder()
-    built = builder.build_from_directory(args.source_oid, args.tree_oid, args.source)
+    built = builder.build_from_directory(
+        args.source_oid,
+        args.tree_oid,
+        args.source,
+        release_version=args.release_version,
+        release_binding=release_binding,
+        release_mode=args.release_version is not None,
+        artifact_mode=args.artifact_mode,
+    )
     builder.verify(built)
     artifact_path = output / f"{args.name}.tar"
     manifest_path = output / f"{args.name}.manifest.json"

scripts/generate_source_registry.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Generate one canonical source entry for a governed non-main change.
+
+This utility only creates the file in the caller's worktree. It never commits,
+pushes, changes main, rewrites a tag, or updates a production projection. The
+caller must submit the resulting file through the existing governed writer and
+approval flow.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
+sys.path.insert(0, str(APP))
+
+from version_registry import (  # noqa: E402
+    build_source_registry_entry,
+    configured_authoritative_git_dir,
+    configured_repository_identity,
+    source_registry_json,
+    validate_source_registry_record,
+    verify_approval_authority,
+    verify_authoritative_repository,
+    verify_source_registry_git_identity,
+)
+
+
+def parser() -> argparse.ArgumentParser:
+    value = argparse.ArgumentParser()
+    value.add_argument("--output", required=True, type=Path)
+    value.add_argument("--version", required=True)
+    value.add_argument("--tag-oid", required=True)
+    value.add_argument("--target-oid", required=True)
+    value.add_argument("--peeled-target-oid", required=True)
+    value.add_argument("--evidence-closure-oid", required=True)
+    value.add_argument("--approval-plan-id", required=True)
+    value.add_argument(
+        "--repository",
+        type=Path,
+        default=configured_authoritative_git_dir(),
+        help="authoritative bare repository; required for release generation",
+    )
+    value.add_argument(
+        "--approval-authority",
+        type=Path,
+        default=(
+            Path(os.environ["AIOS_GITOPS_APPROVAL_AUTHORITY"])
+            if os.environ.get("AIOS_GITOPS_APPROVAL_AUTHORITY")
+            else None
+        ),
+        help="canonical approval/publication receipt for this release request",
+    )
+    value.add_argument(
+        "--repository-identity",
+        default=configured_repository_identity(),
+    )
+    return value
+
+
+def main() -> int:
+    args = parser().parse_args()
+    if args.approval_authority is None:
+        raise ValueError("canonical approval authority is required")
+    verify_authoritative_repository(
+        args.repository,
+        repository_identity=args.repository_identity,
+    )
+    tag_oid = args.tag_oid.lower()
+    target_oid = args.target_oid.lower()
+    peeled_target_oid = args.peeled_target_oid.lower()
+    evidence_closure_oid = args.evidence_closure_oid.lower()
+    expected = build_source_registry_entry(
+        version=args.version,
+        tag_oid=tag_oid,
+        target_oid=target_oid,
+        peeled_target_oid=peeled_target_oid,
+        evidence_closure_oid=evidence_closure_oid,
+        approval_plan_id=args.approval_plan_id,
+        repository=args.repository_identity,
+    )
+    verify_source_registry_git_identity(args.repository, expected)
+    verify_approval_authority(args.approval_authority, expected)
+    destination = args.output.resolve()
+    if destination.name != f"{args.version}.json" or destination.parent.name != "versions":
+        raise ValueError("output must be versions/<version>.json")
+    payload = source_registry_json(expected)
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    if destination.is_symlink():
+        raise ValueError("existing source registry path must not be a symlink")
+    if destination.exists() and not destination.is_file():
+        raise ValueError("existing source registry path must be a regular file")
+    if destination.exists():
+        current = json.loads(destination.read_text(encoding="utf-8"))
+        validate_source_registry_record(
+            current,
+            expected_version=args.version,
+            expected_tag_oid=tag_oid,
+            expected_target_oid=target_oid,
+            expected_evidence_closure_oid=evidence_closure_oid,
+            expected_approval_plan_id=args.approval_plan_id,
+        )
+        print(json.dumps({"status": "EXISTING_IDENTITY_VERIFIED", "path": str(destination)}))
+        return 0
+    descriptor = os.open(
+        destination,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+        0o640,
+    )
+    try:
+        os.write(descriptor, payload)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    print(json.dumps({"status": "GENERATED", "path": str(destination)}))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/artifact.py

@@ -8,6 +8,7 @@ import io
 import json
 import os
 from pathlib import Path, PurePosixPath
+import re
 import stat
 import tarfile
 from typing import Mapping
@@ -75,6 +76,30 @@ class BuiltArtifact:
     manifest: ArtifactManifest
 
 
+@dataclass(frozen=True, slots=True)
+class ReleaseIdentityBinding:
+    """Immutable release identity required for a release artifact build."""
+
+    version: str
+    repository: str
+    tag_oid: str
+    target_oid: str
+    evidence_closure_oid: str
+    approval_plan_id: str
+    peeled_target_oid: str | None = None
+
+    def as_dict(self) -> dict[str, str]:
+        return {
+            "version": self.version,
+            "repository": self.repository,
+            "tag_oid": self.tag_oid,
+            "target_oid": self.target_oid,
+            "peeled_target_oid": self.peeled_target_oid or self.target_oid,
+            "evidence_closure_oid": self.evidence_closure_oid,
+            "approval_plan_id": self.approval_plan_id,
+        }
+
+
 class ArtifactBuilder:
     MAX_FILE_BYTES = 8 * 1024 * 1024
     MAX_ARTIFACT_BYTES = 64 * 1024 * 1024
@@ -92,14 +117,188 @@ class ArtifactBuilder:
         if not is_git_oid(value):
             raise ValueError(f"{name} must be a full lowercase Git OID")
 
+    @staticmethod
+    def _binding_dict(
+        binding: ReleaseIdentityBinding | Mapping[str, str],
+    ) -> dict[str, str]:
+        if isinstance(binding, ReleaseIdentityBinding):
+            return binding.as_dict()
+        values = {str(key): str(value) for key, value in binding.items()}
+        values.setdefault("peeled_target_oid", values.get("target_oid", ""))
+        return values
+
+    @classmethod
+    def _validate_source_registry(
+        cls,
+        root: Path,
+        release_version: str,
+        release_binding: ReleaseIdentityBinding | Mapping[str, str] | None = None,
+        *,
+        release_mode: bool = False,
+    ) -> dict[str, str]:
+        if not re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", release_version):
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_INVALID", "release version is not canonical"
+            )
+        path = root / "versions" / f"{release_version}.json"
+        if path.is_symlink() or not path.is_file():
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_MISSING",
+                f"artifact source is missing versions/{release_version}.json",
+            )
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_INVALID",
+                f"artifact source registry cannot be read: {path}",
+            ) from exc
+        if not isinstance(value, dict):
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_INVALID", "source registry is not an object"
+            )
+        if value.get("version") != release_version or value.get("repository") != "aios/aios":
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_IDENTITY_MISMATCH",
+                "source registry version or repository does not match the artifact",
+            )
+        if value.get("schema") not in {
+            "aios.source-registry.v1",
+            "aios.production-registry.v2",
+        }:
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_INVALID", "source registry schema is invalid"
+            )
+        for field in ("tag_oid", "target_oid", "peeled_target_oid", "evidence_closure_oid"):
+            item = value.get(field)
+            if not isinstance(item, str) or not re.fullmatch(r"[0-9a-f]{40}", item):
+                raise ReleaseBlocked(
+                    "ARTIFACT_SOURCE_REGISTRY_INVALID",
+                    f"source registry field is invalid: {field}",
+                )
+        if value["peeled_target_oid"] != value["target_oid"]:
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_IDENTITY_MISMATCH",
+                "source registry peeled target differs from target",
+            )
+        if not isinstance(value.get("approval_plan_id"), str) or not re.fullmatch(
+            r"sha256:[0-9a-f]{64}", value["approval_plan_id"]
+        ):
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_REGISTRY_INVALID", "source registry approval plan is invalid"
+            )
+        if release_mode and release_binding is None:
+            raise ReleaseBlocked(
+                "ARTIFACT_RELEASE_BINDING_REQUIRED",
+                "release artifact builds require an expected release identity",
+            )
+        if release_binding is not None:
+            expected = cls._binding_dict(release_binding)
+            required = (
+                "version",
+                "repository",
+                "tag_oid",
+                "target_oid",
+                "peeled_target_oid",
+                "evidence_closure_oid",
+                "approval_plan_id",
+            )
+            if any(not expected.get(name) for name in required):
+                raise ReleaseBlocked(
+                    "ARTIFACT_RELEASE_BINDING_INVALID",
+                    "expected release identity is incomplete",
+                )
+            for name in ("tag_oid", "target_oid", "peeled_target_oid", "evidence_closure_oid"):
+                if not re.fullmatch(r"[0-9a-f]{40}", expected[name]):
+                    raise ReleaseBlocked(
+                        "ARTIFACT_RELEASE_BINDING_INVALID",
+                        f"expected release identity field is invalid: {name}",
+                    )
+            if not re.fullmatch(r"sha256:[0-9a-f]{64}", expected["approval_plan_id"]):
+                raise ReleaseBlocked(
+                    "ARTIFACT_RELEASE_BINDING_INVALID",
+                    "expected release identity approval plan is invalid",
+                )
+            for name in required:
+                if value.get(name) != expected[name]:
+                    raise ReleaseBlocked(
+                        "ARTIFACT_SOURCE_REGISTRY_IDENTITY_MISMATCH",
+                        f"source registry does not match expected release identity: {name}",
+                    )
+        return {
+            "version": str(value["version"]),
+            "repository": str(value["repository"]),
+            "tag_oid": str(value["tag_oid"]),
+            "target_oid": str(value["target_oid"]),
+            "peeled_target_oid": str(value["peeled_target_oid"]),
+            "evidence_closure_oid": str(value["evidence_closure_oid"]),
+            "approval_plan_id": str(value["approval_plan_id"]),
+        }
+
     def build_from_directory(
-        self, source_oid: str, tree_oid: str, source: str | Path
+        self,
+        source_oid: str,
+        tree_oid: str,
+        source: str | Path,
+        release_version: str | None = None,
+        release_binding: ReleaseIdentityBinding | Mapping[str, str] | None = None,
+        release_mode: bool | None = None,
+        artifact_mode: str | None = None,
     ) -> BuiltArtifact:
         root = Path(source)
         if root.is_symlink() or not root.is_dir():
             raise ReleaseBlocked(
                 "ARTIFACT_SOURCE_LINK_REJECTED", "artifact source must be a real directory"
             )
+        if artifact_mode not in {None, "release", "general"}:
+            raise ValueError("artifact_mode must be release, general, or omitted")
+        if artifact_mode == "release" and release_mode is False:
+            raise ValueError("artifact_mode=release conflicts with release_mode=False")
+        if artifact_mode == "general" and release_mode is True:
+            raise ValueError("artifact_mode=general conflicts with release_mode=True")
+
+        explicit_release_version = release_version is not None
+        detected_canonical_version = False
+        if release_version is None:
+            version_file = root / "VERSION"
+            if version_file.is_file() and not version_file.is_symlink():
+                release_version = version_file.read_text(encoding="utf-8").strip()
+                detected_canonical_version = bool(
+                    re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", release_version)
+                )
+        elif re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", release_version):
+            detected_canonical_version = True
+
+        binding_expected: dict[str, str] | None = None
+        if release_binding is not None:
+            binding_expected = self._binding_dict(release_binding)
+            if not binding_expected.get("version"):
+                raise ReleaseBlocked(
+                    "ARTIFACT_RELEASE_BINDING_INVALID",
+                    "release identity binding must include version",
+                )
+            release_version = release_version or binding_expected["version"]
+
+        release_requested = bool(
+            explicit_release_version
+            or detected_canonical_version
+            or release_binding is not None
+            or release_mode is True
+            or artifact_mode == "release"
+        )
+        if artifact_mode == "general" and release_requested:
+            raise ReleaseBlocked(
+                "ARTIFACT_GENERAL_MODE_RELEASE_VERSION",
+                "general artifacts cannot contain canonical release metadata",
+            )
+        if release_requested:
+            release_mode = True
+            self._validate_source_registry(
+                root,
+                release_version or "",
+                release_binding,
+                release_mode=True,
+            )
         files: list[ArtifactFile] = []
         for directory, dirnames, filenames in os.walk(root, followlinks=False):
             current = Path(directory)

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/cli.py

@@ -61,6 +61,9 @@ def build_parser() -> argparse.ArgumentParser:
     artifact_build.add_argument("--source-oid", required=True)
     artifact_build.add_argument("--tree-oid", required=True)
     artifact_build.add_argument("--output-directory", required=True, type=Path)
+    artifact_build.add_argument("--release-version")
+    artifact_build.add_argument("--release-binding-json", type=Path)
+    artifact_build.add_argument("--artifact-mode", choices=("release", "general"))
     artifact_verify = artifact_sub.add_parser("verify")
     artifact_verify.add_argument("--artifact", required=True, type=Path)
     artifact_verify.add_argument("--manifest", required=True, type=Path)
@@ -145,7 +148,20 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "artifact":
         builder = ArtifactBuilder()
         if args.artifact_command == "build":
-            built = builder.build_from_directory(args.source_oid, args.tree_oid, args.source)
+            release_binding = (
+                load_json(args.release_binding_json)
+                if args.release_binding_json is not None
+                else None
+            )
+            built = builder.build_from_directory(
+                args.source_oid,
+                args.tree_oid,
+                args.source,
+                release_version=args.release_version,
+                release_binding=release_binding,
+                release_mode=args.release_version is not None,
+                artifact_mode=args.artifact_mode,
+            )
             builder.verify(built)
             artifact_path = args.output_directory / "candidate.tar"
             manifest_path = args.output_directory / "manifest.json"

tests/autonomous_release/test_artifact_closure.py

@@ -1,11 +1,35 @@
+import json
 from pathlib import Path
 
 import pytest
 
-from aios_autonomous_release_candidate.artifact import ArtifactBuilder, ArtifactFile, BuiltArtifact
+from aios_autonomous_release_candidate.artifact import (
+    ArtifactBuilder,
+    ArtifactFile,
+    BuiltArtifact,
+    ReleaseIdentityBinding,
+)
 from aios_autonomous_release_candidate.errors import EvidenceIncomplete, ReleaseBlocked
 
 
+def _source_registry(root: Path) -> None:
+    target = "2" * 40
+    value = {
+        "schema": "aios.source-registry.v1",
+        "version": "V2R2C1",
+        "repository": "aios/aios",
+        "tag_oid": "1" * 40,
+        "tag_object": "1" * 40,
+        "target_oid": target,
+        "peeled_target_oid": target,
+        "evidence_closure_oid": "3" * 40,
+        "approval_plan_id": "sha256:" + "4" * 64,
+    }
+    path = root / "versions" / "V2R2C1.json"
+    path.parent.mkdir(parents=True)
+    path.write_text(json.dumps(value) + "\n", encoding="utf-8")
+
+
 def test_artifact_is_deterministic_and_reverified() -> None:
     builder = ArtifactBuilder()
     files = (
@@ -32,6 +56,143 @@ def test_source_tree_rejects_symlink_and_hardlink(tmp_path: Path) -> None:
         ArtifactBuilder().build_from_directory("a" * 40, "b" * 40, root)
 
 
+def test_release_source_tree_requires_source_registry_entry(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_SOURCE_REGISTRY_MISSING"):
+        ArtifactBuilder().build_from_directory("a" * 40, "b" * 40, root)
+
+
+def test_canonical_version_without_binding_fails_closed(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    _source_registry(root)
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_RELEASE_BINDING_REQUIRED"):
+        ArtifactBuilder().build_from_directory("a" * 40, "b" * 40, root)
+
+
+def test_release_source_tree_requires_expected_release_identity(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    _source_registry(root)
+    binding = ReleaseIdentityBinding(
+        version="V2R2C1",
+        repository="aios/aios",
+        tag_oid="1" * 40,
+        target_oid="2" * 40,
+        evidence_closure_oid="3" * 40,
+        approval_plan_id="sha256:" + "4" * 64,
+    )
+    built = ArtifactBuilder().build_from_directory(
+        "a" * 40,
+        "b" * 40,
+        root,
+        release_binding=binding,
+    )
+    ArtifactBuilder().verify(built)
+
+
+def test_canonical_version_with_malformed_binding_fails_closed(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    _source_registry(root)
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_RELEASE_BINDING_INVALID"):
+        ArtifactBuilder().build_from_directory(
+            "a" * 40,
+            "b" * 40,
+            root,
+            release_binding={
+                "version": "V2R2C1",
+                "repository": "aios/aios",
+                "tag_oid": "not-an-oid",
+            },
+        )
+
+
+def test_explicit_general_artifact_without_release_version_passes(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "README.txt").write_text("general candidate\n", encoding="utf-8")
+    built = ArtifactBuilder().build_from_directory(
+        "a" * 40,
+        "b" * 40,
+        root,
+        artifact_mode="general",
+    )
+    assert "README.txt" in built.manifest.files
+
+
+def test_explicit_general_mode_rejects_canonical_version(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    _source_registry(root)
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_GENERAL_MODE_RELEASE_VERSION"):
+        ArtifactBuilder().build_from_directory(
+            "a" * 40,
+            "b" * 40,
+            root,
+            artifact_mode="general",
+        )
+
+
+@pytest.mark.parametrize(
+    "field,wrong",
+    [
+        ("repository", "other/repository"),
+        ("version", "V2R2C2"),
+        ("tag_oid", "9" * 40),
+        ("target_oid", "8" * 40),
+        ("peeled_target_oid", "7" * 40),
+        ("evidence_closure_oid", "6" * 40),
+        ("approval_plan_id", "sha256:" + "5" * 64),
+    ],
+)
+def test_release_source_tree_rejects_every_identity_mismatch(
+    tmp_path: Path, field: str, wrong: str
+) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    _source_registry(root)
+    binding = {
+        "version": "V2R2C1",
+        "repository": "aios/aios",
+        "tag_oid": "1" * 40,
+        "target_oid": "2" * 40,
+        "peeled_target_oid": "2" * 40,
+        "evidence_closure_oid": "3" * 40,
+        "approval_plan_id": "sha256:" + "4" * 64,
+    }
+    binding[field] = wrong
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_SOURCE_REGISTRY"):
+        ArtifactBuilder().build_from_directory(
+            "a" * 40,
+            "b" * 40,
+            root,
+            release_binding=binding,
+        )
+
+
+def test_release_mode_without_binding_fails_closed(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    (root / "VERSION").write_text("V2R2C1\n", encoding="utf-8")
+    _source_registry(root)
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_RELEASE_BINDING_REQUIRED"):
+        ArtifactBuilder().build_from_directory(
+            "a" * 40,
+            "b" * 40,
+            root,
+            release_version="V2R2C1",
+            release_mode=True,
+        )
+
+
 def test_artifact_rejects_secret_markers() -> None:
     with pytest.raises(ReleaseBlocked, match="ARTIFACT_SECRET_MARKER"):
         ArtifactBuilder().build(

tests/test_approved_release_executor.py

@@ -158,7 +158,7 @@ def test_execute_orders_registry_before_baseline_and_baseline_last(
     )
     monkeypatch.setattr(
         "approved_release_executor.verify_artifact",
-        lambda binding: {"verified": True},
+        lambda binding, **kwargs: {"verified": True},
     )
 
     calls = []

tests/test_authoritative_git_context.py

@@ -73,6 +73,7 @@ class AuthoritativeGitContextTests(unittest.TestCase):
             json.dumps({"branch": "patch/context"}), encoding="utf-8"
         )
         self.env = {
+            "AIOS_TEST_MODE": "1",
             "AIOS_GITOPS_AUTHORITATIVE_GIT_DIR": str(self.bare),
             "AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios",
         }

tests/test_bridge_v2_governed_http_bootstrap.py

@@ -76,6 +76,7 @@ def test_http_route_contract_is_complete():
     expected = {
         "/v2/runtime", "/v2/capabilities", "/v2/governed-task",
         "/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
+        "/v2/branch/publish",
         "/v2/job", "/v2/job/resume", "/v2/job/reconcile",
         "/v2/tag/approval-request",
         "/v2/tag/publish",

tests/test_control_plane_gate_closure.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import tempfile
+from unittest.mock import patch
+
+import pytest
+
+from aios_existing_branch_publisher import ExistingBranchPublicationState, ExistingBranchPublisher
+from gitea_tag_publication import GiteaConfigurationError, GiteaTagClient, InvalidReleaseVersion, PublicationRequest, TagPublisher, validate_version
+from signer_contract_compatibility import run_compatible_signer
+
+
+class Authority:
+    def __init__(self, oid: str):
+        self.oid = oid
+    def read_ref(self, ref: str) -> str | None:
+        return self.oid if ref.startswith("refs/heads/") else None
+
+
+class Remote:
+    def __init__(self, oid: str | None = None, api_oid: str | None = None, visible: bool = True):
+        self.oid = oid
+        self.api_oid = api_oid
+        self.visible = visible
+        self.pushes: list[tuple[str, str | None, str]] = []
+    def repository_identity(self) -> str:
+        return "aios/aios"
+    def read_git_ref(self, branch: str) -> str | None:
+        return self.oid
+    def read_api_branch(self, branch: str) -> str | None:
+        return self.api_oid
+    def commit_visible(self, oid: str) -> bool:
+        return self.visible
+    def push_exact(self, branch: str, expected: str | None, new: str) -> None:
+        assert expected is None
+        self.pushes.append((branch, expected, new))
+        self.oid = new
+        self.api_oid = new
+        self.visible = True
+
+
+def request(oid: str, key: str = "gate-closure-test-0001") -> dict[str, str]:
+    return {
+        "repository": "aios/aios",
+        "branch": "agent/gate-closure-test",
+        "expected_branch_oid": oid,
+        "owner_approval_reference": "owner-authorization:test",
+        "idempotency_key": key,
+    }
+
+
+def test_existing_branch_publisher_creates_exact_missing_remote_and_replays_idempotently(tmp_path: Path):
+    oid = "a" * 40
+    remote = Remote()
+    publisher = ExistingBranchPublisher(Authority(oid), remote, tmp_path / "state")
+    first = publisher.publish(request(oid))
+    assert first["publication_state"] == ExistingBranchPublicationState.COMPLETED.value
+    assert remote.pushes == [("agent/gate-closure-test", None, oid)]
+    second = publisher.publish(request(oid))
+    assert second["publication_state"] == ExistingBranchPublicationState.COMPLETED.value
+    assert second["idempotent"] is True
+
+
+def test_existing_branch_publisher_conflict_is_fail_closed(tmp_path: Path):
+    oid = "a" * 40
+    remote = Remote(oid="b" * 40, api_oid="b" * 40)
+    result = ExistingBranchPublisher(Authority(oid), remote, tmp_path / "state").publish(request(oid, "gate-closure-test-0002"))
+    assert result["publication_state"] == ExistingBranchPublicationState.MANUAL_OWNER_REVIEW_REQUIRED.value
+    assert remote.pushes == []
+
+
+def test_existing_branch_publisher_rejects_main_and_wrong_authority_oid(tmp_path: Path):
+    oid = "a" * 40
+    with pytest.raises(Exception, match="BRANCH_NAMESPACE_REJECTED"):
+        ExistingBranchPublisher(Authority(oid), Remote(), tmp_path / "state").publish({**request(oid), "branch": "main", "idempotency_key": "gate-closure-test-0003"})
+    with pytest.raises(Exception, match="AUTHORITY_BRANCH_OID_MISMATCH"):
+        ExistingBranchPublisher(Authority("b" * 40), Remote(), tmp_path / "state").publish(request(oid, "gate-closure-test-0004"))
+
+
+def test_signer_broker_path_has_no_direct_root_signer_call():
+    class FakeBroker:
+        def self_test(self):
+            return None
+        def sign(self, payload_kind, payload):
+            return b"signature", "test-broker-key"
+    request_doc = {
+        "schema": "aios.task-scoped-writer-grant-request.v1",
+        "task_id": "gate-closure-signer-unit",
+        "repository": "aios/aios",
+        "base": {"ref": "refs/heads/main", "oid": "a" * 40},
+        "target": {"branch": "agent/gate-closure-signer-unit"},
+        "owner_authorization": {"identity": "aiman", "reference": "owner-authorization:test"},
+        "authorization": {
+            "production_change_authorized": False,
+            "tag_change_authorized": False,
+            "baseline_change_authorized": False,
+            "main_write_authorized": False,
+        },
+        "allowed_operations": ["create_worktree"],
+        "denied_operations": ["direct_main_push", "force_push"],
+        "one_time_use": True,
+        "not_before": "2026-08-10T00:00:00Z",
+        "expires_at": "2026-08-10T00:00:00Z",
+        "nonce": "unit-nonce",
+    }
+    with tempfile.TemporaryDirectory() as td:
+        root = Path(td)
+        grant = root / "grant"
+        grant.mkdir()
+        (grant / "grant-request.json").write_text(json.dumps(request_doc), encoding="utf-8")
+        (grant / "approval-plan.json").write_text(json.dumps({"approval_plan_id": "sha256:" + "a" * 64}), encoding="utf-8")
+        with patch("signer_contract_compatibility._supports_deployed_signer", return_value=True):
+            result = run_compatible_signer(["unused-signer-command"], grant, root / "result.json", broker=FakeBroker())
+        assert result["signing_key_identity"] == "test-broker-key"
+
+
+def test_gitea_configuration_has_explicit_error_codes():
+    with pytest.raises(GiteaConfigurationError, match="GITEA_API_URL_MISSING"):
+        GiteaTagClient.from_authoritative_environment({})
+    with pytest.raises(GiteaConfigurationError, match="GITEA_REPOSITORY_MISSING"):
+        GiteaTagClient.from_authoritative_environment({"AIOS_GITOPS_GITEA_API_URL": "https://gitea.invalid/api/v1"})
+    with pytest.raises(GiteaConfigurationError, match="GITEA_TOKEN_REFERENCE_MISSING"):
+        GiteaTagClient.from_authoritative_environment({"AIOS_GITOPS_GITEA_API_URL": "https://gitea.invalid/api/v1", "AIOS_GITOPS_GITEA_REPOSITORY": "aios/aios"})
+    assert validate_version("V2R2C1") == "V2R2C1"
+    with pytest.raises(InvalidReleaseVersion, match="invalid release version"):
+        validate_version("not-a-release")
+
+
+
+def test_release_consistency_reports_tag_target_mismatch(tmp_path: Path):
+    target = "a" * 40
+    evidence = "b" * 40
+    tag_oid = "c" * 40
+    mismatched_target = "d" * 40
+    request = PublicationRequest("V2R2C1", target, evidence, "approval-plan:test")
+    (tmp_path / "baseline").mkdir()
+    (tmp_path / "versions").mkdir()
+    common = {
+        "version": request.version,
+        "target_oid": request.target_oid,
+        "evidence_closure_oid": request.evidence_closure_oid,
+        "approval_plan_id": request.approval_plan_id,
+        "tag_oid": tag_oid,
+    }
+    (tmp_path / "baseline/current.json").write_text(
+        json.dumps({**common, "current_baseline": request.version}), encoding="utf-8"
+    )
+    (tmp_path / "versions/V2R2C1.json").write_text(
+        json.dumps({**common, "registered": True, "enabled": True}), encoding="utf-8"
+    )
+    verified = {
+        **common,
+        "remote_tag_oid": tag_oid,
+        "remote_peeled_target_oid": mismatched_target,
+        "gitea_tag_oid": tag_oid,
+        "gitea_peeled_target_oid": target,
+        "gitea_listing_visible": True,
+        "repository_identity": "aios/aios",
+    }
+    result = TagPublisher(tmp_path, object(), object(), "gitea").get_release_consistency(
+        request, verified=verified
+    )
+    assert result["status"] == "FAIL"
+    assert result["error"] == "TAG_TARGET_MISMATCH"

tests/test_governed_release_provenance.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import io
+import json
+import shutil
+import subprocess
+import tarfile
+from pathlib import Path
+import sys
+
+import pytest
+
+APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
+sys.path.insert(0, str(APP))
+
+import governed_production_release as release  # noqa: E402
+from version_registry import build_source_registry_entry, source_registry_json  # noqa: E402
+
+
+def _git(work: Path, *args: str) -> str:
+    return subprocess.run(
+        ["git", "-C", str(work), *args],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+
+
+def _commit(work: Path, message: str) -> str:
+    subprocess.run(["git", "-C", str(work), "add", "-A"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", message], check=True)
+    return _git(work, "rev-parse", "HEAD")
+
+
+def test_artifact_uses_pinned_source_registry_commit_when_main_advances(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    if shutil.which("zstd") is None:
+        pytest.skip("zstd is required for governed artifact integration")
+    work = tmp_path / "work"
+    bare = tmp_path / "repo.git"
+    subprocess.run(["git", "init", "-q", "-b", "main", str(work)], check=True)
+    _git(work, "config", "user.name", "test")
+    _git(work, "config", "user.email", "test@example.invalid")
+    (work / "active-runtime/app").mkdir(parents=True)
+    (work / "active-runtime/app/deployment_state.py").write_text("state = 'candidate'\n")
+    (work / "active-runtime/app/v2r1c6_privacy_tools.py").write_text("def main(): return 0\n")
+    target_oid = _commit(work, "release target")
+    subprocess.run(
+        ["git", "-C", str(work), "tag", "-a", "V2R2C1", target_oid, "-m", "release"],
+        check=True,
+    )
+    tag_oid = _git(work, "rev-parse", "refs/tags/V2R2C1")
+    (work / "evidence.txt").write_text("evidence\n")
+    evidence_oid = _commit(work, "evidence closure")
+    record = build_source_registry_entry(
+        version="V2R2C1",
+        tag_oid=tag_oid,
+        target_oid=target_oid,
+        peeled_target_oid=target_oid,
+        evidence_closure_oid=evidence_oid,
+        approval_plan_id="sha256:" + "4" * 64,
+    )
+    source_path = work / "versions/V2R2C1.json"
+    source_path.parent.mkdir(parents=True)
+    source_path.write_bytes(source_registry_json(record))
+    main_oid = _commit(work, "source registry convergence")
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    subprocess.run(["git", "-C", str(work), "remote", "add", "authority", str(bare)], check=True)
+    monkeypatch.setenv("AIOS_TEST_MODE", "1")
+    monkeypatch.setenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", str(bare))
+
+    monkeypatch.setattr(release, "TAG_OID", tag_oid)
+    monkeypatch.setattr(release, "TARGET_OID", target_oid)
+    monkeypatch.setattr(release, "EVIDENCE_CLOSURE_OID", evidence_oid)
+    monkeypatch.setattr(release, "APPROVAL_PLAN_ID", record["approval_plan_id"])
+    binding = release.ReleaseBinding(
+        tag_oid=tag_oid,
+        target_oid=target_oid,
+        evidence_closure_oid=evidence_oid,
+        approval_plan_id=record["approval_plan_id"],
+    )
+
+    original_payload = release._payload_file
+    advanced = False
+
+    def advance_after_first_payload(git_dir: Path, oid: str, path: str) -> bytes:
+        nonlocal advanced
+        result = original_payload(git_dir, oid, path)
+        if not advanced and path == "active-runtime/app/deployment_state.py":
+            updated = dict(record)
+            updated["review_marker"] = "main-advanced-after-validation"
+            source_path.write_bytes(source_registry_json(updated))
+            new_main = _commit(work, "advance main after source validation")
+            subprocess.run(["git", "-C", str(work), "push", "-q", "authority", "main"], check=True)
+            assert new_main != main_oid
+            advanced = True
+        return result
+
+    monkeypatch.setattr(release, "_payload_file", advance_after_first_payload)
+    receipt = release.build_production_artifact(bare, binding, tmp_path / "artifacts")
+    assert advanced is True
+    assert receipt["release_target_oid"] == target_oid
+    assert receipt["source_registry_main_oid"] == main_oid
+    assert receipt["source_registry_path"] == "versions/V2R2C1.json"
+
+    expected_bytes = subprocess.run(
+        ["git", f"--git-dir={bare}", "show", f"{main_oid}:versions/V2R2C1.json"],
+        capture_output=True,
+        check=True,
+    ).stdout
+    assert receipt["source_registry_sha256"] == __import__("hashlib").sha256(expected_bytes).hexdigest()
+    compressed = Path(receipt["artifact"]).read_bytes()
+    raw_tar = subprocess.run(["zstd", "-dc", "-"], input=compressed, capture_output=True, check=True).stdout
+    with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r:") as archive:
+        packaged = archive.extractfile("payload/versions/V2R2C1.json").read()
+        metadata = json.loads(archive.extractfile("payload/BUILD-METADATA.json").read())
+    assert packaged == expected_bytes
+    assert metadata["source_registry_main_oid"] == main_oid
+    assert metadata["source_registry_sha256"] == receipt["source_registry_sha256"]

tests/test_model_api_immutable_staging.py

@@ -25,6 +25,18 @@ def test_model_api_staging_unit_is_non_production_and_read_only():
     assert 'production' not in unit.lower().replace('model-api-v2r2c1-staging', '')
 
 
+def test_model_api_source_convergence_never_falls_back_to_generic_git_dir():
+    source = (ROOT / 'active-runtime' / 'app' / 'aios_model_api_staging.py').read_text(encoding='utf-8')
+    assert "repo_git_dir=AUTHORITATIVE_GIT_DIR or GIT_DIR" not in source
+    assert "repo_git_dir=AUTHORITATIVE_GIT_DIR" in source
+
+
+def test_production_adapter_never_claims_source_git_convergence():
+    source = (ROOT / 'active-runtime' / 'app' / 'v2r2c1_production_adapter.py').read_text(encoding='utf-8')
+    assert '"source_registry_converged": True' not in source
+    assert '"source_registry_converged": "not_checked"' in source
+
+
 def test_model_api_builder_manifest_contract():
     builder = (ROOT / 'scripts' / 'build_v2r2c1_model_api_staging.py').read_text(encoding='utf-8')
     assert 'aios.v2r2c1.model-api-staging-manifest.v1' in builder

tests/test_release_metadata_adapter.py

@@ -47,6 +47,8 @@ def old_baseline(path: Path) -> str:
 
 
 def test_baseline_record_uses_real_previous_production_baseline(tmp_path, monkeypatch):
+    # This test isolates previous-baseline semantics; convergence is tested separately.
+    monkeypatch.setattr(adapter, "assert_source_registry_convergence", lambda *a, **kw: None)
     baseline = tmp_path / "baseline.json"
     previous_sha = old_baseline(baseline)
     registry_root = tmp_path / "versions"

tests/test_signer_broker_service.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import base64
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+import json
+from pathlib import Path
+import threading
+
+from aios_signer_broker_service import (
+    REQUEST_SCHEMA,
+    RESPONSE_SCHEMA,
+    SignerBrokerPolicy,
+    SignerBrokerService,
+    SignerBrokerServiceError,
+)
+
+
+class FakeRunner:
+    def __init__(self, failure: str | None = None):
+        self.failure = failure
+        self.calls: list[tuple[str, str]] = []
+        self.lock = threading.Lock()
+
+    def self_test(self) -> None:
+        with self.lock:
+            self.calls.append(("self_test", ""))
+        if self.failure:
+            raise SignerBrokerServiceError(self.failure)
+
+    def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]:
+        with self.lock:
+            self.calls.append(("sign", payload_kind))
+        if self.failure:
+            raise SignerBrokerServiceError(self.failure)
+        return b"signature:" + payload, "test-broker-key"
+
+
+def make_policy(tmp_path: Path, *, allowed_uids: frozenset[int] = frozenset({1000})) -> SignerBrokerPolicy:
+    return SignerBrokerPolicy(
+        socket_path=tmp_path / "signer.sock",
+        signer_path=Path("/usr/local/libexec/aios-grant-signer-adapter"),
+        audit_path=tmp_path / "audit.jsonl",
+        state_root=tmp_path / "state",
+        allowed_uids=allowed_uids,
+    )
+
+
+def sign_request(payload: bytes = b"canonical payload") -> dict[str, str]:
+    return {
+        "schema": REQUEST_SCHEMA,
+        "operation": "sign",
+        "payload_kind": "approval_plan",
+        "payload_sha256": hashlib.sha256(payload).hexdigest(),
+        "payload_base64": base64.b64encode(payload).decode("ascii"),
+    }
+
+
+def read_audit(path: Path) -> list[dict[str, object]]:
+    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+
+
+def test_authorized_request_and_self_test_emit_metadata_only_audit(tmp_path: Path):
+    service = SignerBrokerService(make_policy(tmp_path), FakeRunner())
+    self_test = service.handle_request({"schema": REQUEST_SCHEMA, "operation": "self_test"}, 1000)
+    signed = service.handle_request(sign_request(b"secret-shaped canonical input"), 1000)
+    assert self_test == {"schema": RESPONSE_SCHEMA, "status": "PASS", "operation": "self_test"}
+    assert signed["status"] == "PASS"
+    assert base64.b64decode(signed["signature_base64"]) == b"signature:secret-shaped canonical input"
+    assert signed["key_id"] == "test-broker-key"
+    audit_text = (tmp_path / "audit.jsonl").read_text(encoding="utf-8")
+    assert "secret-shaped canonical input" not in audit_text
+    assert "payload_base64" not in audit_text
+    allowed = {"schema", "timestamp", "peer_uid", "operation", "status", "error_code", "payload_sha256"}
+    assert all(set(event) <= allowed for event in read_audit(tmp_path / "audit.jsonl"))
+
+
+def test_unauthorized_caller_is_rejected_before_signer_use(tmp_path: Path):
+    runner = FakeRunner()
+    service = SignerBrokerService(make_policy(tmp_path), runner)
+    result = service.handle_request(sign_request(), 1001)
+    assert result["status"] == "FAIL"
+    assert result["error_code"] == "CALLER_UNAUTHORIZED"
+    assert runner.calls == []
+
+
+def test_arbitrary_command_and_invalid_operation_are_rejected(tmp_path: Path):
+    runner = FakeRunner()
+    service = SignerBrokerService(make_policy(tmp_path), runner)
+    injected = {"schema": REQUEST_SCHEMA, "operation": "sign", "command": ["/bin/sh", "-c", "id"]}
+    invalid = {"schema": REQUEST_SCHEMA, "operation": "run", "command": ["id"]}
+    assert service.handle_request(injected, 1000)["error_code"] == "REQUEST_FIELDS_REJECTED"
+    assert service.handle_request(invalid, 1000)["error_code"] == "SIGNER_OPERATION_REJECTED"
+    assert runner.calls == []
+
+
+def test_invalid_payload_and_hash_fail_closed(tmp_path: Path):
+    service = SignerBrokerService(make_policy(tmp_path), FakeRunner())
+    invalid = sign_request()
+    invalid["payload_base64"] = "not-base64"
+    assert service.handle_request(invalid, 1000)["error_code"] == "SIGNER_PAYLOAD_INVALID"
+    mismatch = sign_request()
+    mismatch["payload_sha256"] = "0" * 64
+    assert service.handle_request(mismatch, 1000)["error_code"] == "SIGNER_PAYLOAD_HASH_MISMATCH"
+
+
+def test_signer_unavailable_and_timeout_are_stable_failures(tmp_path: Path):
+    for code in ("SIGNER_UNAVAILABLE", "SIGNER_TIMEOUT"):
+        result = SignerBrokerService(make_policy(tmp_path / code), FakeRunner(code)).handle_request(
+            {"schema": REQUEST_SCHEMA, "operation": "self_test"}, 1000
+        )
+        assert result == {"schema": RESPONSE_SCHEMA, "status": "FAIL", "error_code": code}
+
+
+def test_concurrent_calls_are_safe_and_audited(tmp_path: Path):
+    service = SignerBrokerService(make_policy(tmp_path), FakeRunner())
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        results = list(pool.map(lambda index: service.handle_request(sign_request(f"payload-{index}".encode()), 1000), range(24)))
+    assert all(result["status"] == "PASS" for result in results)
+    assert len(read_audit(tmp_path / "audit.jsonl")) == 24

tests/test_signer_contract_compatibility.py

@@ -20,6 +20,7 @@ from signer_contract_compatibility import (  # noqa: E402
     canonical_json_bytes,
     deployed_signer_help,
     run_compatible_signer,
+    run_signer_self_test,
     resolve_ref_oid,
     run_compatible_validator,
     sha256_json,
@@ -113,9 +114,13 @@ class SignerContractCompatibilityTests(unittest.TestCase):
         help_result = subprocess.run([signer, "sign", "--help"], capture_output=True, text=True, check=False)
         self.assertEqual(help_result.returncode, 0)
         self.assertTrue(deployed_signer_help(help_result.stdout, help_result.stderr))
-        self_test = subprocess.run([signer, "self-test"], capture_output=True, text=True, check=False)
-        self.assertEqual(self_test.returncode, 0)
-        self.assertIn("SIGNER_SELF_TEST=PASS", self_test.stdout)
+        try:
+            broker_result = run_signer_self_test()
+        except SignerContractError as exc:
+            if str(exc) == "SIGNER_BROKER_UNAVAILABLE":
+                self.skipTest("signer broker is not deployed in this engineering phase")
+            raise
+        self.assertEqual(broker_result["status"], "PASS")
 
     def test_resolve_ref_oid_uses_full_synthetic_commit_and_fails_closed(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:

tests/test_v2r2c1_pre_push_review.py

@@ -0,0 +1,392 @@
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import tarfile
+
+import pytest
+
+from approved_release_executor import (
+    Binding,
+    ManualOwnerReviewRequired,
+    ReleaseBlocked,
+    ReconciliationRequired,
+    deploy_step,
+    execute,
+    tag_step,
+    verify_artifact,
+    verify_authorization,
+)
+
+
+def _binding(tmp_path: Path, **overrides) -> Binding:
+    artifact = tmp_path / "artifact.tar.zst"
+    artifact.write_bytes(b"x")
+    values = {
+        "version": "VTEST",
+        "target_oid": "a" * 40,
+        "evidence_closure_oid": "b" * 40,
+        "artifact_sha256": "c" * 64,
+        "approval_plan_id": "sha256:" + "d" * 64,
+        "artifact_path": str(artifact),
+        "artifact_size": 1,
+        "production_service": "test.service",
+        "production_port": 18927,
+    }
+    values.update(overrides)
+    return Binding(**values)
+
+
+def _authorization(binding: Binding) -> dict[str, object]:
+    result: dict[str, object] = {
+        "accepted": True,
+        "binding_replaced": False,
+        "version": binding.version,
+        "target_oid": binding.target_oid,
+        "evidence_closure_oid": binding.evidence_closure_oid,
+        "artifact_sha256": binding.artifact_sha256,
+        "approval_plan_id": binding.approval_plan_id,
+    }
+    if binding.tag_oid is not None:
+        result.update(
+            {
+                "binding_schema": binding.binding_schema or "aios.release-binding.v2",
+                "tag_oid": binding.tag_oid,
+            }
+        )
+    if binding.source_registry_main_oid is not None:
+        result.update(
+            {
+                "source_registry_main_oid": binding.source_registry_main_oid,
+                "source_registry_path": binding.source_registry_path
+                or f"versions/{binding.version}.json",
+                "source_registry_sha256": binding.source_registry_sha256,
+            }
+        )
+    return result
+
+
+def test_legacy_v2r2c1_style_authorization_remains_compatible(tmp_path: Path) -> None:
+    binding = _binding(tmp_path)
+    assert verify_authorization(_authorization(binding), binding)["accepted"] is True
+
+
+def test_future_binding_requires_explicit_tag_oid(tmp_path: Path) -> None:
+    binding = _binding(
+        tmp_path,
+        version="V2R2C2",
+        tag_oid="1" * 40,
+        binding_schema="aios.release-binding.v2",
+    )
+    assert verify_authorization(_authorization(binding), binding)["tag_oid"] == "1" * 40
+    wrong = _authorization(binding)
+    wrong["tag_oid"] = "2" * 40
+    with pytest.raises(ManualOwnerReviewRequired, match="tag_oid"):
+        verify_authorization(wrong, binding)
+
+
+def test_future_same_name_different_tag_object_requires_manual_review(monkeypatch, tmp_path: Path) -> None:
+    binding = _binding(
+        tmp_path,
+        version="V2R2C2",
+        tag_oid="1" * 40,
+        binding_schema="aios.release-binding.v2",
+    )
+
+    def fake(_command, _request):
+        return {
+            "status": "APPLIED_VERIFIED",
+            "version": binding.version,
+            "target_oid": binding.target_oid,
+            "artifact_sha256": binding.artifact_sha256,
+            "tag_oid": "2" * 40,
+            "peeled_target_oid": binding.target_oid,
+        }
+
+    monkeypatch.setattr("approved_release_executor.run_adapter", fake)
+    with pytest.raises(ManualOwnerReviewRequired, match="same-name tag object"):
+        tag_step(["/tag"], binding)
+
+
+def test_executor_rechecks_source_registry_digest_and_build_provenance(
+    monkeypatch, tmp_path: Path
+) -> None:
+    if shutil.which("zstd") is None:
+        pytest.skip("zstd is required for archive verification")
+    source = {
+        "schema": "aios.source-registry.v1",
+        "version": "V2R2C2",
+        "repository": "aios/aios",
+        "tag": "V2R2C2",
+        "tag_oid": "1" * 40,
+        "tag_object": "1" * 40,
+        "target_oid": "2" * 40,
+        "peeled_target_oid": "2" * 40,
+        "evidence_closure_oid": "3" * 40,
+        "approval_plan_id": "sha256:" + "4" * 64,
+    }
+    source_bytes = (json.dumps(source, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode()
+    source_digest = hashlib.sha256(source_bytes).hexdigest()
+    repo_work = tmp_path / "authority-work"
+    repo_bare = tmp_path / "authority.git"
+    (repo_work / "versions").mkdir(parents=True)
+    (repo_work / "versions" / "V2R2C2.json").write_bytes(source_bytes)
+    subprocess.run(["git", "init", "-q", str(repo_work)], check=True)
+    subprocess.run(["git", "-C", str(repo_work), "branch", "-M", "main"], check=True)
+    subprocess.run(["git", "-C", str(repo_work), "config", "user.name", "test"], check=True)
+    subprocess.run(
+        ["git", "-C", str(repo_work), "config", "user.email", "test@example.invalid"],
+        check=True,
+    )
+    subprocess.run(["git", "-C", str(repo_work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(repo_work), "commit", "-qm", "source"], check=True)
+    source_main_oid = subprocess.run(
+        ["git", "-C", str(repo_work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(repo_work), str(repo_bare)], check=True)
+    subprocess.run(
+        ["git", f"--git-dir={repo_bare}", "config", "remote.gitea.url", "https://g.cnnetmkt.top/aios/aios.git"],
+        check=True,
+    )
+    monkeypatch.setenv("AIOS_TEST_MODE", "1")
+    monkeypatch.setenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", str(repo_bare))
+    metadata = {
+        "release_target_oid": "2" * 40,
+        "source_registry_main_oid": source_main_oid,
+        "source_registry_path": "versions/V2R2C2.json",
+        "source_registry_sha256": source_digest,
+    }
+    tar_path = tmp_path / "release.tar"
+    with tarfile.open(tar_path, "w") as archive:
+        for name, content in (
+            ("payload/versions/V2R2C2.json", source_bytes),
+            ("payload/BUILD-METADATA.json", (json.dumps(metadata, sort_keys=True) + "\n").encode()),
+        ):
+            info = tarfile.TarInfo(name)
+            info.size = len(content)
+            archive.addfile(info, io.BytesIO(content))
+    artifact = tmp_path / "release.tar.zst"
+    subprocess.run(["zstd", "-q", "-19", "-f", str(tar_path), "-o", str(artifact)], check=True)
+    binding = _binding(
+        tmp_path,
+        version="V2R2C2",
+        tag_oid="1" * 40,
+        binding_schema="aios.release-binding.v2",
+        target_oid="2" * 40,
+        evidence_closure_oid="3" * 40,
+        approval_plan_id="sha256:" + "4" * 64,
+        source_registry_main_oid=source_main_oid,
+        source_registry_path="versions/V2R2C2.json",
+        source_registry_sha256=source_digest,
+        artifact_path=str(artifact),
+        artifact_size=artifact.stat().st_size,
+        artifact_sha256=hashlib.sha256(artifact.read_bytes()).hexdigest(),
+    )
+    result = verify_artifact(binding)
+    assert result["source_registry_sha256"] == source_digest
+    assert result["build_metadata_provenance_verified"] is True
+
+    bad_metadata = dict(metadata)
+    bad_metadata["source_registry_sha256"] = "0" * 64
+    bad_tar = tmp_path / "bad-metadata.tar"
+    with tarfile.open(bad_tar, "w") as archive:
+        for name, content in (
+            ("payload/versions/V2R2C2.json", source_bytes),
+            ("payload/BUILD-METADATA.json", (json.dumps(bad_metadata, sort_keys=True) + "\n").encode()),
+        ):
+            info = tarfile.TarInfo(name)
+            info.size = len(content)
+            archive.addfile(info, io.BytesIO(content))
+    bad_artifact = tmp_path / "bad-metadata.tar.zst"
+    subprocess.run(["zstd", "-q", "-19", "-f", str(bad_tar), "-o", str(bad_artifact)], check=True)
+    bad_binding = _binding(
+        tmp_path,
+        version="V2R2C2",
+        tag_oid="1" * 40,
+        binding_schema="aios.release-binding.v2",
+        target_oid="2" * 40,
+        evidence_closure_oid="3" * 40,
+        approval_plan_id="sha256:" + "4" * 64,
+        source_registry_main_oid=source_main_oid,
+        source_registry_path="versions/V2R2C2.json",
+        source_registry_sha256=source_digest,
+        artifact_path=str(bad_artifact),
+        artifact_size=bad_artifact.stat().st_size,
+        artifact_sha256=hashlib.sha256(bad_artifact.read_bytes()).hexdigest(),
+    )
+    with pytest.raises(ReleaseBlocked, match="BUILD-METADATA"):
+        verify_artifact(bad_binding)
+
+    subprocess.run(["git", "-C", str(repo_work), "remote", "add", "origin", str(repo_bare)], check=True)
+    changed = dict(source)
+    changed["review_marker"] = "changed-after-M"
+    changed_bytes = (json.dumps(changed, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode()
+    (repo_work / "versions" / "V2R2C2.json").write_bytes(changed_bytes)
+    subprocess.run(["git", "-C", str(repo_work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(repo_work), "commit", "-qm", "change-source"], check=True)
+    subprocess.run(["git", "-C", str(repo_work), "push", "-q", "origin", "HEAD:main"], check=True)
+    with pytest.raises(ManualOwnerReviewRequired, match="current Authority main source registry"):
+        verify_artifact(binding)
+
+
+def test_executor_requires_runtime_bundle_convergence_for_provenance_binding(tmp_path: Path):
+    binding = _binding(
+        tmp_path,
+        version="V2R2C2",
+        tag_oid="1" * 40,
+        binding_schema="aios.release-binding.v2",
+        source_registry_main_oid="5" * 40,
+        source_registry_path="versions/V2R2C2.json",
+        source_registry_sha256="6" * 64,
+    )
+
+    def fake_missing(_command, _request):
+        return {
+            "status": "ACTIVE_VERIFIED",
+            "version": binding.version,
+            "target_oid": binding.target_oid,
+            "evidence_closure_oid": binding.evidence_closure_oid,
+            "artifact_sha256": binding.artifact_sha256,
+            "approval_plan_id": binding.approval_plan_id,
+            "deployment_state": "ACTIVE",
+            "production_deployed": True,
+        }
+
+    monkeypatch = pytest.MonkeyPatch()
+    monkeypatch.setattr("approved_release_executor.run_adapter", fake_missing)
+    try:
+        with pytest.raises(ReconciliationRequired, match="runtime bundle/runtime registry"):
+            deploy_step(["/deploy"], binding)
+    finally:
+        monkeypatch.undo()
+
+    def fake_converged(_command, _request):
+        return {
+            "status": "ACTIVE_VERIFIED",
+            "version": binding.version,
+            "target_oid": binding.target_oid,
+            "evidence_closure_oid": binding.evidence_closure_oid,
+            "artifact_sha256": binding.artifact_sha256,
+            "approval_plan_id": binding.approval_plan_id,
+            "deployment_state": "ACTIVE",
+            "production_deployed": True,
+            "runtime_bundle_contains_current_version": True,
+            "runtime_registry_converged": True,
+        }
+
+    monkeypatch.setattr("approved_release_executor.run_adapter", fake_converged)
+    try:
+        assert deploy_step(["/deploy"], binding)["runtime_registry_converged"] is True
+    finally:
+        monkeypatch.undo()
+
+
+def test_execute_rechecks_authoritative_git_after_runtime_before_registry(
+    monkeypatch, tmp_path: Path
+):
+    work = tmp_path / "authority-work"
+    bare = tmp_path / "authority.git"
+    (work / "versions").mkdir(parents=True)
+    subprocess.run(["git", "init", "-q", "-b", "main", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(
+        ["git", "-C", str(work), "config", "user.email", "test@example.invalid"],
+        check=True,
+    )
+    source = {
+        "schema": "aios.source-registry.v1",
+        "version": "VTEST",
+        "repository": "aios/aios",
+        "tag_oid": "1" * 40,
+        "tag_object": "1" * 40,
+        "target_oid": "a" * 40,
+        "peeled_target_oid": "a" * 40,
+        "evidence_closure_oid": "b" * 40,
+        "approval_plan_id": "sha256:" + "d" * 64,
+    }
+    source_path = work / "versions" / "VTEST.json"
+    source_path.write_bytes((json.dumps(source, sort_keys=True, indent=2) + "\n").encode())
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "source"], check=True)
+    pinned_main = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    source_sha = hashlib.sha256(source_path.read_bytes()).hexdigest()
+    binding = _binding(
+        tmp_path,
+        source_registry_main_oid=pinned_main,
+        source_registry_path="versions/VTEST.json",
+        source_registry_sha256=source_sha,
+    )
+    monkeypatch.setenv("AIOS_TEST_MODE", "1")
+    monkeypatch.setenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", str(bare))
+    monkeypatch.setattr("approved_release_executor.secure_json", lambda _path: _authorization(binding))
+    monkeypatch.setattr("approved_release_executor.verify_git_lineage", lambda *_args: {"verified": True})
+    monkeypatch.setattr("approved_release_executor.verify_artifact", lambda *_args, **_kwargs: {"verified": True})
+
+    deployed = False
+    calls = []
+
+    def fake_adapter(command, request):
+        nonlocal deployed
+        name = command[0]
+        operation = request["operation"]
+        calls.append((name, operation))
+        common = {
+            "version": binding.version,
+            "target_oid": binding.target_oid,
+            "evidence_closure_oid": binding.evidence_closure_oid,
+            "artifact_sha256": binding.artifact_sha256,
+            "approval_plan_id": binding.approval_plan_id,
+        }
+        if name == "/tag":
+            return {
+                **common,
+                "status": "APPLIED_VERIFIED",
+                "tag_oid": "1" * 40,
+                "peeled_target_oid": binding.target_oid,
+            }
+        if name == "/deploy":
+            if operation == "probe" and not deployed:
+                changed = dict(source)
+                changed["review_marker"] = "changed-after-runtime"
+                source_path.write_bytes((json.dumps(changed, sort_keys=True, indent=2) + "\n").encode())
+                subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+                subprocess.run(["git", "-C", str(work), "commit", "-qm", "advance-main"], check=True)
+                subprocess.run(["git", "-C", str(work), "push", "-q", str(bare), "HEAD:main"], check=True)
+                deployed = True
+            return {
+                **common,
+                "status": "ACTIVE_VERIFIED",
+                "deployment_state": "ACTIVE",
+                "production_deployed": True,
+                "runtime_bundle_contains_current_version": True,
+                "runtime_registry_converged": True,
+            }
+        raise AssertionError(f"unexpected adapter call: {name}")
+
+    monkeypatch.setattr("approved_release_executor.run_adapter", fake_adapter)
+    with pytest.raises(ManualOwnerReviewRequired, match="current Authority main source registry"):
+        execute(
+            binding=binding,
+            authorization_path=tmp_path / "authorization.json",
+            repository=bare,
+            state_root=tmp_path / "state",
+            tag_command=["/tag"],
+            deploy_command=["/deploy"],
+            registry_command=["/registry"],
+            baseline_command=["/baseline"],
+        )
+    assert deployed is True
+    assert not any(name in {"/registry", "/baseline"} for name, _ in calls)

tests/test_version_registry_convergence.py

@@ -0,0 +1,850 @@
+from __future__ import annotations
+
+import hashlib
+import inspect
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
+sys.path.insert(0, str(APP))
+
+import release_metadata_adapter as adapter  # noqa: E402
+import version_registry as registry_module  # noqa: E402
+from version_registry import (  # noqa: E402
+    RegistryConvergenceError,
+    assert_runtime_registry_convergence,
+    assert_source_registry_convergence,
+    build_source_registry_entry,
+    configured_approval_authority_root,
+    configured_authoritative_git_dir,
+    list_registry_versions,
+    list_registry_versions_report,
+    source_registry_snapshot,
+    verify_authoritative_repository,
+    verify_approval_authority,
+    verify_source_registry_main_convergence,
+    verify_source_registry_git_identity,
+)
+
+
+def _write_json(path: Path, obj: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(obj, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def _record(
+    *,
+    version: str = "V2R2C1",
+    tag_oid: str = "1" * 40,
+    target_oid: str = "2" * 40,
+) -> dict:
+    return {
+        "schema": "aios.production-registry.v2",
+        "version": version,
+        "repository": "aios/aios",
+        "status": "PRODUCTION_DEPLOYED",
+        "production_deployed": True,
+        "tag_oid": tag_oid,
+        "target_oid": target_oid,
+        "peeled_target_oid": target_oid,
+        "evidence_closure_oid": "3" * 40,
+        "approval_plan_id": "sha256:" + "4" * 64,
+    }
+
+
+def _canonical_approval(root: Path, record: dict) -> Path:
+    owner = {
+        "approved": True,
+        "status": "APPROVED",
+        "approved_by": "aiman",
+        "approval_source": "explicit_owner_approval_in_chat",
+    }
+    binding = {
+        key: record[key]
+        for key in (
+            "version",
+            "repository",
+            "tag_oid",
+            "target_oid",
+            "peeled_target_oid",
+            "evidence_closure_oid",
+            "approval_plan_id",
+        )
+    }
+    authority = {
+        "schema": "aios.v2r2c1.tag-approval-request.v1",
+        "state": "APPROVED",
+        "status": "APPROVED",
+        "execute": False,
+        "approval_plan_id": record["approval_plan_id"],
+        "binding": binding,
+        "owner_approval": owner,
+        "owner_approval_digest_sha256": hashlib.sha256(
+            (json.dumps(owner, sort_keys=True, separators=(",", ":")) + "\n").encode()
+        ).hexdigest(),
+    }
+    path = root / f"{record['version']}.json"
+    _write_json(path, authority)
+    path.chmod(0o640)
+    return path
+
+
+@pytest.fixture(autouse=True)
+def _explicit_test_authority_mode(monkeypatch: pytest.MonkeyPatch):
+    """Permit only these unit tests to inject temporary authority roots."""
+    monkeypatch.setenv("AIOS_TEST_MODE", "1")
+
+
+def test_list_registry_versions_adds_only_baseline_attested_projection(tmp_path: Path):
+    ws = tmp_path / "ws"
+    (ws / "versions").mkdir(parents=True)
+    (ws / "versions" / "V2R1C7.json").write_text("{}\n")
+
+    projection_root = tmp_path / "projection"
+    projection = projection_root / "V2R2C1.json"
+    _write_json(projection, _record())
+    digest = hashlib.sha256(projection.read_bytes()).hexdigest()
+
+    baseline = tmp_path / "baseline" / "current.json"
+    _write_json(
+        baseline,
+        {
+            "version": "V2R2C1",
+            "status": "PRODUCTION_DEPLOYED",
+            "production_deployed": True,
+            "registry_source": str(projection),
+            "registry_digest": digest,
+        },
+    )
+
+    assert list_registry_versions(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+    ) == ["V2R1C7.json", "V2R2C1.json"]
+
+
+def test_list_registry_versions_rejects_projection_digest_drift(tmp_path: Path):
+    ws = tmp_path / "ws"
+    (ws / "versions").mkdir(parents=True)
+    projection_root = tmp_path / "projection"
+    projection = projection_root / "V2R2C1.json"
+    _write_json(projection, _record())
+
+    baseline = tmp_path / "baseline.json"
+    _write_json(
+        baseline,
+        {
+            "version": "V2R2C1",
+            "status": "PRODUCTION_DEPLOYED",
+            "production_deployed": True,
+            "registry_source": str(projection),
+            "registry_digest": "0" * 64,
+        },
+    )
+
+    assert list_registry_versions(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+    ) == []
+
+
+def test_projection_only_version_is_not_runtime_artifact_convergence(tmp_path: Path):
+    ws = tmp_path / "ws"
+    (ws / "versions").mkdir(parents=True)
+    projection_root = tmp_path / "projection"
+    projection = projection_root / "V2R2C1.json"
+    _write_json(projection, _record())
+    baseline = tmp_path / "baseline.json"
+    _write_json(
+        baseline,
+        {
+            "version": "V2R2C1",
+            "status": "PRODUCTION_DEPLOYED",
+            "production_deployed": True,
+            "registry_source": str(projection),
+            "registry_digest": hashlib.sha256(projection.read_bytes()).hexdigest(),
+        },
+    )
+
+    report = list_registry_versions_report(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+    )
+    assert report["registry_versions"] == ["V2R2C1.json"]
+    assert report["attested_projection_version"] == "V2R2C1"
+    assert report["runtime_bundle_contains_version"] is False
+    assert report["source_registry_converged"] == "unknown"
+    assert report["runtime_registry_converged"] is False
+    assert report["runtime_bundle_contains_current_version"] is False
+    assert report["registry_consistency_status"] == "RUNTIME_ARTIFACT_RECONCILIATION_REQUIRED"
+    with pytest.raises(RegistryConvergenceError, match="runtime_bundle_missing"):
+        assert_runtime_registry_convergence(
+            ws,
+            "V2R2C1",
+            projection_root=projection_root,
+            baseline_file=baseline,
+        )
+
+
+def _attested_projection(tmp_path: Path, record: dict) -> tuple[Path, Path]:
+    projection_root = tmp_path / "projection"
+    projection = projection_root / f"{record['version']}.json"
+    _write_json(projection, record)
+    baseline = tmp_path / "baseline.json"
+    _write_json(
+        baseline,
+        {
+            "version": record["version"],
+            "status": "PRODUCTION_DEPLOYED",
+            "production_deployed": True,
+            "registry_source": str(projection),
+            "registry_digest": hashlib.sha256(projection.read_bytes()).hexdigest(),
+        },
+    )
+    return projection_root, baseline
+
+
+def test_list_report_source_good_runtime_missing_is_not_fully_consistent(tmp_path: Path):
+    record = _record()
+    bare, branch = _init_bare_repo(tmp_path, record)
+    projection_root, baseline = _attested_projection(tmp_path, record)
+    ws = tmp_path / "runtime"
+    (ws / "versions").mkdir(parents=True)
+    report = list_registry_versions_report(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+        repo_git_dir=bare,
+        main_ref=f"refs/heads/{branch}",
+    )
+    assert report["source_registry_converged"] is True
+    assert report["runtime_registry_converged"] is False
+    assert report["runtime_bundle_contains_current_version"] is False
+    assert report["registry_consistency_status"] == "RUNTIME_ARTIFACT_RECONCILIATION_REQUIRED"
+
+
+def test_list_report_source_runtime_projection_good_is_fully_consistent(tmp_path: Path):
+    record = _record()
+    bare, branch = _init_bare_repo(tmp_path, record)
+    projection_root, baseline = _attested_projection(tmp_path, record)
+    ws = tmp_path / "runtime"
+    versions = ws / "versions"
+    versions.mkdir(parents=True)
+    _write_json(versions / f"{record['version']}.json", record)
+    report = list_registry_versions_report(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+        repo_git_dir=bare,
+        main_ref=f"refs/heads/{branch}",
+    )
+    assert report["source_registry_converged"] is True
+    assert report["runtime_registry_converged"] is True
+    assert report["runtime_bundle_contains_current_version"] is True
+    assert report["registry_consistency_status"] == "FULLY_CONSISTENT"
+
+
+def test_list_report_source_missing_is_not_converged(tmp_path: Path):
+    record = _record()
+    work = tmp_path / "no-source-work"
+    bare = tmp_path / "no-source.git"
+    subprocess.run(["git", "init", "-q", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(
+        ["git", "-C", str(work), "config", "user.email", "test@example.invalid"],
+        check=True,
+    )
+    (work / "README").write_text("no source\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "README"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "no source"], check=True)
+    branch = subprocess.run(
+        ["git", "-C", str(work), "symbolic-ref", "--short", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    projection_root, baseline = _attested_projection(tmp_path, record)
+    ws = tmp_path / "runtime"
+    (ws / "versions").mkdir(parents=True)
+    _write_json(ws / "versions" / f"{record['version']}.json", record)
+    report = list_registry_versions_report(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+        repo_git_dir=bare,
+        main_ref=f"refs/heads/{branch}",
+    )
+    assert report["source_registry_converged"] is False
+    assert report["runtime_registry_converged"] is True
+    assert report["registry_consistency_status"] == "SOURCE_REGISTRY_NOT_CONVERGED"
+
+
+def test_list_registry_versions_excludes_non_version_entries(tmp_path: Path):
+    ws = tmp_path / "ws"
+    versions = ws / "versions"
+    versions.mkdir(parents=True)
+    (versions / "README.txt").write_text("do not expose\n", encoding="utf-8")
+    (versions / "scratch.json").write_text("{}\n", encoding="utf-8")
+    (versions / "proposals").mkdir()
+    (versions / "V2R1C7.json").write_text("{}\n", encoding="utf-8")
+    assert list_registry_versions(
+        ws,
+        projection_root=tmp_path / "projection",
+        baseline_file=tmp_path / "baseline.json",
+    ) == ["V2R1C7.json"]
+
+
+def _init_bare_repo(tmp_path: Path, record: dict) -> tuple[Path, str]:
+    work = tmp_path / "work"
+    bare = tmp_path / "repo.git"
+    subprocess.run(["git", "init", "-q", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.email", "test@example.invalid"], check=True)
+    _write_json(work / "versions" / f"{record['version']}.json", record)
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "registry"], check=True)
+    branch = subprocess.run(
+        ["git", "-C", str(work), "symbolic-ref", "--short", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    return bare, branch
+
+
+def _init_bare_repo_with_work(tmp_path: Path, record: dict) -> tuple[Path, Path, str]:
+    work = tmp_path / "history-work"
+    bare = tmp_path / "history.git"
+    subprocess.run(["git", "init", "-q", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(
+        ["git", "-C", str(work), "config", "user.email", "test@example.invalid"],
+        check=True,
+    )
+    _write_json(work / "versions" / f"{record['version']}.json", record)
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "registry"], check=True)
+    branch = subprocess.run(
+        ["git", "-C", str(work), "symbolic-ref", "--short", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    subprocess.run(["git", "-C", str(work), "remote", "add", "origin", str(bare)], check=True)
+    return work, bare, branch
+
+
+def test_assert_source_registry_convergence_passes_matching_identity(tmp_path: Path):
+    record = _record()
+    bare, branch = _init_bare_repo(tmp_path, record)
+    projection = tmp_path / "projection.json"
+    _write_json(projection, record)
+
+    assert_source_registry_convergence(
+        projection,
+        repo_git_dir=bare,
+        main_ref=f"refs/heads/{branch}",
+    )
+
+
+def test_assert_source_registry_convergence_fails_missing_entry(tmp_path: Path):
+    record = _record()
+    work = tmp_path / "work"
+    bare = tmp_path / "repo.git"
+    subprocess.run(["git", "init", "-q", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.email", "test@example.invalid"], check=True)
+    (work / "README").write_text("x\n")
+    subprocess.run(["git", "-C", str(work), "add", "README"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "init"], check=True)
+    branch = subprocess.run(
+        ["git", "-C", str(work), "symbolic-ref", "--short", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+
+    projection = tmp_path / "projection.json"
+    _write_json(projection, record)
+
+    with pytest.raises(RegistryConvergenceError, match="source_registry_entry_missing"):
+        assert_source_registry_convergence(
+            projection,
+            repo_git_dir=bare,
+            main_ref=f"refs/heads/{branch}",
+        )
+
+
+def test_assert_source_registry_convergence_fails_identity_mismatch(tmp_path: Path):
+    source = _record(tag_oid="5" * 40)
+    projection_record = _record(tag_oid="1" * 40)
+    bare, branch = _init_bare_repo(tmp_path, source)
+    projection = tmp_path / "projection.json"
+    _write_json(projection, projection_record)
+
+    with pytest.raises(RegistryConvergenceError, match="source_registry_mismatch:tag_oid"):
+        assert_source_registry_convergence(
+            projection,
+            repo_git_dir=bare,
+            main_ref=f"refs/heads/{branch}",
+        )
+
+
+def test_assert_source_registry_convergence_requires_all_identity_fields(tmp_path: Path):
+    source = _record()
+    source.pop("peeled_target_oid")
+    bare, branch = _init_bare_repo(tmp_path, source)
+    projection = tmp_path / "projection.json"
+    _write_json(projection, _record())
+    with pytest.raises(RegistryConvergenceError, match="peeled_target_oid"):
+        assert_source_registry_convergence(
+            projection,
+            repo_git_dir=bare,
+            main_ref=f"refs/heads/{branch}",
+        )
+
+
+def test_assert_source_registry_convergence_rejects_wrong_repository(tmp_path: Path):
+    source = _record()
+    source["repository"] = "other/repository"
+    bare, branch = _init_bare_repo(tmp_path, source)
+    projection = tmp_path / "projection.json"
+    _write_json(projection, _record())
+    with pytest.raises(RegistryConvergenceError, match="repository"):
+        assert_source_registry_convergence(
+            projection,
+            repo_git_dir=bare,
+            main_ref=f"refs/heads/{branch}",
+        )
+
+
+def test_source_registry_main_convergence_accepts_unrelated_later_main_commit(tmp_path: Path):
+    record = _record()
+    work, bare, branch = _init_bare_repo_with_work(tmp_path, record)
+    source_main = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    (work / "README").write_text("unrelated later main commit\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "README"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "unrelated"], check=True)
+    subprocess.run(["git", "-C", str(work), "push", "-q", "origin", branch], check=True)
+    result = verify_source_registry_main_convergence(
+        bare,
+        record["version"],
+        main_ref=f"refs/heads/{branch}",
+        source_registry_main_oid=source_main,
+        expected_tag_oid=record["tag_oid"],
+        expected_target_oid=record["target_oid"],
+        expected_evidence_closure_oid=record["evidence_closure_oid"],
+        expected_approval_plan_id=record["approval_plan_id"],
+    )
+    assert result["source_registry_main_is_ancestor_of_current_main"] is True
+
+
+def test_source_registry_main_convergence_rejects_side_branch_only_commit(tmp_path: Path):
+    record = _record()
+    work, bare, branch = _init_bare_repo_with_work(tmp_path, record)
+    base = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "-C", str(work), "checkout", "-qb", "side"], check=True)
+    (work / "side.txt").write_text("side\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "side.txt"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "side"], check=True)
+    side = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "-C", str(work), "checkout", "-q", branch], check=True)
+    (work / "main.txt").write_text("main\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "main.txt"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "main"], check=True)
+    subprocess.run(["git", "-C", str(work), "push", "-q", "origin", branch], check=True)
+    assert base != side
+    with pytest.raises(RegistryConvergenceError, match="not_ancestor"):
+        verify_source_registry_main_convergence(
+            bare,
+            record["version"],
+            main_ref=f"refs/heads/{branch}",
+            source_registry_main_oid=side,
+        )
+
+
+def test_source_registry_main_convergence_rejects_removed_entry(tmp_path: Path):
+    record = _record()
+    work, bare, branch = _init_bare_repo_with_work(tmp_path, record)
+    source_main = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    (work / "versions" / f"{record['version']}.json").unlink()
+    subprocess.run(["git", "-C", str(work), "add", "-u", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "remove"], check=True)
+    subprocess.run(["git", "-C", str(work), "push", "-q", "origin", branch], check=True)
+    with pytest.raises(RegistryConvergenceError, match="entry_missing"):
+        verify_source_registry_main_convergence(
+            bare,
+            record["version"],
+            main_ref=f"refs/heads/{branch}",
+            source_registry_main_oid=source_main,
+        )
+
+
+def test_source_registry_main_convergence_rejects_altered_entry(tmp_path: Path):
+    record = _record()
+    work, bare, branch = _init_bare_repo_with_work(tmp_path, record)
+    source_main = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    altered = dict(record)
+    altered["approval_plan_id"] = "sha256:" + "9" * 64
+    _write_json(work / "versions" / f"{record['version']}.json", altered)
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "alter"], check=True)
+    subprocess.run(["git", "-C", str(work), "push", "-q", "origin", branch], check=True)
+    with pytest.raises(RegistryConvergenceError, match="entry_changed"):
+        verify_source_registry_main_convergence(
+            bare,
+            record["version"],
+            main_ref=f"refs/heads/{branch}",
+            source_registry_main_oid=source_main,
+        )
+
+
+def test_source_registry_generator_predeclares_release_identity():
+    value = build_source_registry_entry(
+        version="V2R2C1",
+        tag_oid="1" * 40,
+        target_oid="2" * 40,
+        peeled_target_oid="2" * 40,
+        evidence_closure_oid="3" * 40,
+        approval_plan_id="sha256:" + "4" * 64,
+    )
+    assert set(
+        (
+            "version",
+            "tag_oid",
+            "target_oid",
+            "peeled_target_oid",
+            "evidence_closure_oid",
+            "approval_plan_id",
+        )
+    ).issubset(value)
+
+
+def test_authoritative_repository_rejects_arbitrary_repository(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    record = _record()
+    bare, _ = _init_bare_repo(tmp_path, record)
+    monkeypatch.delenv("AIOS_TEST_MODE")
+    with pytest.raises(RegistryConvergenceError, match="path_mismatch"):
+        verify_authoritative_repository(bare)
+
+
+def test_production_authority_defaults_are_fixed_and_env_overrides_fail(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    monkeypatch.delenv("AIOS_TEST_MODE")
+    monkeypatch.delenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", raising=False)
+    monkeypatch.delenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", raising=False)
+    assert configured_authoritative_git_dir() == Path("/data/aios-gitops/repo.git")
+    assert configured_approval_authority_root() == Path(
+        "/var/lib/aios-gitops/approval-authority"
+    )
+    monkeypatch.setenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", str(tmp_path / "fake.git"))
+    with pytest.raises(RegistryConvergenceError, match="override_forbidden"):
+        configured_authoritative_git_dir()
+    monkeypatch.setenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", str(tmp_path / "fake-root"))
+    with pytest.raises(RegistryConvergenceError, match="override_forbidden"):
+        configured_approval_authority_root()
+
+
+def test_production_canonical_repository_and_test_injected_authority(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    record = _record()
+    bare, _ = _init_bare_repo(tmp_path, record)
+    subprocess.run(
+        [
+            "git",
+            f"--git-dir={bare}",
+            "config",
+            "remote.gitea.url",
+            "https://g.cnnetmkt.top/aios/aios.git",
+        ],
+        check=True,
+    )
+    monkeypatch.delenv("AIOS_TEST_MODE")
+    monkeypatch.setattr(registry_module, "DEFAULT_AUTHORITATIVE_GIT_DIR", bare)
+    assert verify_authoritative_repository(bare)["repository_identity"] == "aios/aios"
+
+    monkeypatch.setenv("AIOS_TEST_MODE", "1")
+    monkeypatch.setenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", str(bare))
+    assert verify_authoritative_repository(bare)["configured_path"] == str(bare.resolve())
+
+
+def test_authoritative_git_unavailable_stays_unknown_and_closure_fails(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    record = _record()
+    projection_root, baseline = _attested_projection(tmp_path, record)
+    ws = tmp_path / "runtime"
+    (ws / "versions").mkdir(parents=True)
+    _write_json(ws / "versions" / f"{record['version']}.json", record)
+    missing = tmp_path / "missing-authority.git"
+    report = list_registry_versions_report(
+        ws,
+        projection_root=projection_root,
+        baseline_file=baseline,
+        repo_git_dir=missing,
+    )
+    assert report["source_registry_converged"] == "unknown"
+    assert report["source_registry_convergence_error"] == "authoritative_repository_unavailable"
+    monkeypatch.delenv("AIOS_TEST_MODE")
+    with pytest.raises(RegistryConvergenceError, match="path_unavailable"):
+        assert_source_registry_convergence(
+            projection_root / f"{record['version']}.json",
+            repo_git_dir=missing,
+        )
+
+
+def test_source_registry_snapshot_stays_pinned_after_main_advances(tmp_path: Path):
+    work = tmp_path / "work"
+    bare = tmp_path / "repo.git"
+    subprocess.run(["git", "init", "-q", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(
+        ["git", "-C", str(work), "config", "user.email", "test@example.invalid"],
+        check=True,
+    )
+    record = _record()
+    _write_json(work / "versions" / "V2R2C1.json", record)
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "registry-1"], check=True)
+    branch = subprocess.run(
+        ["git", "-C", str(work), "branch", "--show-current"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    pinned = source_registry_snapshot(
+        bare,
+        "V2R2C1",
+        main_ref=f"refs/heads/{branch}",
+    )
+
+    updated = dict(record)
+    updated["review_marker"] = "main-advanced"
+    _write_json(work / "versions" / "V2R2C1.json", updated)
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "registry-2"], check=True)
+    subprocess.run(["git", "-C", str(work), "remote", "add", "authority", str(bare)], check=True)
+    subprocess.run(["git", "-C", str(work), "push", "-q", "authority", branch], check=True)
+
+    repinned = source_registry_snapshot(
+        bare,
+        "V2R2C1",
+        main_ref=f"refs/heads/{branch}",
+        source_registry_main_oid=pinned["source_registry_main_oid"],
+    )
+    assert repinned["source_registry_main_oid"] == pinned["source_registry_main_oid"]
+    assert repinned["bytes"] == pinned["bytes"]
+    assert repinned["source_registry_sha256"] == pinned["source_registry_sha256"]
+    assert repinned["record"].get("review_marker") is None
+
+
+def _git_commit_with_release_tag(tmp_path: Path, *, annotated: bool = True):
+    work = tmp_path / "authority-work"
+    bare = tmp_path / "authority.git"
+    subprocess.run(["git", "init", "-q", str(work)], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "test"], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.email", "test@example.invalid"], check=True)
+    (work / "README").write_text("release\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "README"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "target"], check=True)
+    target = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    tag_args = ["git", "-C", str(work), "tag"]
+    if annotated:
+        tag_args += ["-a", "V2R2C1", target, "-m", "release"]
+    else:
+        tag_args += ["V2R2C1", target]
+    subprocess.run(tag_args, check=True)
+    tag_oid = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "refs/tags/V2R2C1"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    subprocess.run(["git", "-C", str(work), "commit", "--allow-empty", "-qm", "evidence"], check=True)
+    evidence = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    record = _record(tag_oid=tag_oid, target_oid=target)
+    record["evidence_closure_oid"] = evidence
+    _write_json(work / "versions" / "V2R2C1.json", record)
+    subprocess.run(["git", "-C", str(work), "add", "versions"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-qm", "source registry"], check=True)
+    branch = subprocess.run(
+        ["git", "-C", str(work), "branch", "--show-current"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(work), str(bare)], check=True)
+    return work, bare, branch, record, target, evidence, tag_oid
+
+
+def test_source_registry_git_identity_rejects_lightweight_tag(tmp_path: Path):
+    _, bare, _, record, _, _, _ = _git_commit_with_release_tag(tmp_path, annotated=False)
+    with pytest.raises(RegistryConvergenceError, match="tag_not_annotated"):
+        verify_source_registry_git_identity(bare, record)
+
+
+def test_source_registry_git_identity_proves_annotated_tag_and_lineage(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    _, bare, _, record, target, evidence, tag_oid = _git_commit_with_release_tag(tmp_path)
+    result = verify_source_registry_git_identity(bare, record)
+    assert result["tag_oid"] == tag_oid
+    assert result["target_oid"] == target
+    assert result["evidence_closure_oid"] == evidence
+
+    authority_root = tmp_path / "authority-root"
+    authority_root.mkdir()
+    monkeypatch.setenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", str(authority_root))
+    authority = _canonical_approval(authority_root, record)
+    assert verify_approval_authority(authority, record)["release_binding_verified"] is True
+
+
+def test_source_registry_generator_requires_authoritative_git_and_approval(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+):
+    _, bare, _, record, _, _, _ = _git_commit_with_release_tag(tmp_path)
+    subprocess.run(
+        ["git", f"--git-dir={bare}", "config", "remote.gitea.url", "https://g.cnnetmkt.top/aios/aios.git"],
+        check=True,
+    )
+    authority_root = tmp_path / "authority-root"
+    authority_root.mkdir()
+    monkeypatch.setenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", str(bare))
+    monkeypatch.setenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", str(authority_root))
+    authority = _canonical_approval(authority_root, record)
+    output = tmp_path / "generated" / "versions" / "V2R2C1.json"
+    script = Path(__file__).resolve().parents[1] / "scripts" / "generate_source_registry.py"
+    command = [
+        sys.executable,
+        str(script),
+        "--output",
+        str(output),
+        "--version",
+        record["version"],
+        "--tag-oid",
+        record["tag_oid"],
+        "--target-oid",
+        record["target_oid"],
+        "--peeled-target-oid",
+        record["peeled_target_oid"],
+        "--evidence-closure-oid",
+        record["evidence_closure_oid"],
+        "--approval-plan-id",
+        record["approval_plan_id"],
+        "--repository",
+        str(bare),
+        "--approval-authority",
+        str(authority),
+    ]
+    completed = subprocess.run(command, capture_output=True, text=True, check=False)
+    assert completed.returncode == 0, completed.stderr
+    assert json.loads(output.read_text(encoding="utf-8"))["tag_oid"] == record["tag_oid"]
+
+    wrong = list(command)
+    wrong[wrong.index("--approval-plan-id") + 1] = "sha256:" + "f" * 64
+    failed = subprocess.run(wrong, capture_output=True, text=True, check=False)
+    assert failed.returncode != 0
+    other = tmp_path / "other.git"
+    subprocess.run(["git", "clone", "-q", "--bare", str(bare), str(other)], check=True)
+    arbitrary = list(command)
+    arbitrary[arbitrary.index("--repository") + 1] = str(other)
+    failed_authority = subprocess.run(arbitrary, capture_output=True, text=True, check=False)
+    assert failed_authority.returncode != 0
+
+
+def test_approval_authority_rejects_unassociated_plan(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    record = _record()
+    authority_root = tmp_path / "authority-root"
+    authority_root.mkdir()
+    monkeypatch.setenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", str(authority_root))
+    authority = authority_root / f"{record['version']}.json"
+    _write_json(
+        authority,
+        {
+            "schema": "aios.v2r2c1.tag-approval-request.v1",
+            "state": "APPROVED",
+            "approval_plan_id": record["approval_plan_id"],
+        },
+    )
+    authority.chmod(0o640)
+    with pytest.raises(RegistryConvergenceError, match="binding_missing"):
+        verify_approval_authority(authority, record)
+
+
+def test_approval_authority_rejects_untrusted_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    authority_root = tmp_path / "authority-root"
+    authority_root.mkdir()
+    monkeypatch.setenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", str(authority_root))
+    authority = tmp_path / "approval.json"
+    _write_json(authority, {"schema": "aios.v2r2c1.tag-approval-request.v1"})
+    with pytest.raises(RegistryConvergenceError, match="untrusted_path"):
+        verify_approval_authority(authority, _record())
+
+
+def test_approval_authority_rejects_revoked_receipt(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    authority_root = tmp_path / "authority-root"
+    authority_root.mkdir()
+    monkeypatch.setenv("AIOS_GITOPS_APPROVAL_AUTHORITY_ROOT", str(authority_root))
+    record = _record()
+    authority = _canonical_approval(authority_root, record)
+    value = json.loads(authority.read_text(encoding="utf-8"))
+    value["revoked"] = True
+    authority.write_text(json.dumps(value) + "\n", encoding="utf-8")
+    authority.chmod(0o640)
+    with pytest.raises(RegistryConvergenceError, match="revoked"):
+        verify_approval_authority(authority, record)
+
+
+def test_baseline_apply_wires_convergence_to_caller_repository():
+    source = inspect.getsource(adapter.baseline_apply)
+    assert "assert_source_registry_convergence(" in source
+    assert "repo_git_dir=args.repository" in source
+    # The gate must execute before the dict that writes final production status.
+    assert source.index("assert_source_registry_convergence(") < source.index(
+        '"status": "PRODUCTION_DEPLOYED"'
+    )

versions/V2R2C1.json

@@ -0,0 +1,21 @@
+{
+  "approval_plan_id": "sha256:c0df52008ae7f64cec5ee2d28947aa7b892066c670ab8295e4549140b1a7500e",
+  "artifact_sha256": "ab4ae257346c82863f05abf5e309381a1c86612957608c3a4f4ab5cc9233f052",
+  "deployment_state": "ACTIVE",
+  "enabled": true,
+  "evidence_closure_oid": "a8b906d1bc24088c19647e98de6a679268f6d0a2",
+  "peeled_target_oid": "998d30304787e31e5ae607c0bd86b5edb2b0e5ad",
+  "production_deployed": true,
+  "production_port": 18927,
+  "production_service": "aios-v2r1c7-production.service",
+  "registered": true,
+  "repository": "aios/aios",
+  "schema": "aios.production-registry.v2",
+  "status": "PRODUCTION_DEPLOYED",
+  "tag": "V2R2C1",
+  "tag_object": "07fc2746b688f5f243f29b39de39f32edc3dd9d2",
+  "tag_oid": "07fc2746b688f5f243f29b39de39f32edc3dd9d2",
+  "target_oid": "998d30304787e31e5ae607c0bd86b5edb2b0e5ad",
+  "updated_at": "2026-08-09T09:13:13Z",
+  "version": "V2R2C1"
+}