docs(release): align V2R2C1 review docs with latest mechanism state

README.md

@@ -27,3 +27,36 @@ docs/CODEX_FULL_EXECUTION_PROMPT_V1R1C0.md
 ```
 
 本包是 V1R1C0 全量工程部署包，包含文档、脚本、迁移、接口适配器、测试用例。部署时应由 Codex 在服务器上定位当前生产包和运行目录，按增量方式叠加，不得破坏现有 WebOps/NPM/Casdoor/OpenClaw/Hermes。
+
+## AIOS V2R2C1 Pre-Push Review 当前状态（语义一致性 follow-up）
+
+本段针对最新 mechanism branch HEAD 重新核对，不把早期 b660e2c 文档快照当作当前代码状态。文档只写入非 main 的 Owner-review 分支，不能替代 Production Baseline、生产运行时或 main 合并。
+
+| 项目 | 当前值 | 语义 |
+| --- | --- | --- |
+| Repository | aios/aios | 权威仓库身份 |
+| Authoritative Git | /data/aios-gitops/repo.git | 权威 bare repository |
+| Authoritative main HEAD | 3244f0f17f8845592ac94d585f5ebd65693c9f29 | 当前工程主线 |
+| Production Baseline | V2R2C1 / generation 4 | PRODUCTION_DEPLOYED / ACTIVE |
+| Baseline target | 998d30304787e31e5ae607c0bd86b5edb2b0e5ad | 当前生产发布目标 |
+| V2R2C1 tag object | 07fc2746b688f5f243f29b39de39f32edc3dd9d2 | annotated tag object |
+| Latest mechanism HEAD | 52b74ba61040246fffa8b10011c1199830e3d665 | 最新待审机制分支状态 |
+| Full audit scope | 17 paths, +1403/-79 | b660e2c → 51b5f0e → 52b74ba |
+| Latest-tip delta | 12 paths, +334/-39 | 51b5f0e → 52b74ba |
+| Documentation branch | agent/docs/v2r2c1-provenance-mainline-refresh-20260810 | 非 main Owner-review 分支 |
+| Branch HEAD before semantic repair | 869cc007649a39035f5b74513cecc55bb1ae673f | 本次 follow-up 的父提交 |
+| Documentation branch write/readback | COMPLETE after this follow-up | 仅针对文档分支 |
+| PR #65 | open / unmerged | Owner reviewer remains required |
+| Owner review | REQUIRED | review 后停止，不自动合并 |
+
+文档分支状态与生产状态分开报告。mechanism/release production write、main merge、Production release、Baseline update、tag rewrite 和 force push 均为 NOT_RUN。legacy_global_writes_enabled 必须保持 false。
+
+### 最新机制与验收事实
+
+初始 b660e2c 修复覆盖 14 个文件；完整 b660e2c → 51b5f0e → 52b74ba 审计覆盖 17 个路径，净增量为 +1403/-79；最新 52b74ba 自身覆盖 12 个路径，净增量为 +334/-39。README、requirements、architecture、acceptance 和 version differences 已按这三个范围分别表述。
+
+当前测试事实为 760 passed、2 skipped、1 failed、154 subtests passed。唯一失败为 tests/test_signer_contract_compatibility.py::test_real_signer_help_and_self_test_are_available。/usr/local/libexec/aios-grant-signer-adapter self-test 返回 42，两个独立重跑仍返回 42。因此 Full external gate 为 PARTIAL，Signer gate 为 FAIL / EXTERNAL，Production ready 保持 false。
+
+当前架构模型必须同时表达 R、A、T、E、M、C、P：权威仓库身份、annotated tag object、peeled release target、evidence closure、pinned source-registry main commit、当前权威 main/reference commit 和 Owner-approved approval plan。还必须分别说明 source-registry 精确字节与 SHA-256、authority/approval binding、release/general artifact binding、executor 独立复核、runtime projection 与 Git convergence、post-deploy source-registry recheck 以及 fail-closed 状态。
+
+五个文档均需通过 UTF-8 strict decode、CJK > 0、U+FFFD = 0、异常连续问号 = 0，并在提交后证明 source bytes SHA-256 与 Git blob 内容一致。任何生产 consequential action 仍需要独立 authority mechanism 与 Owner review。

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_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
 
@@ -73,10 +79,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',
@@ -495,6 +504,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 +521,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 +534,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 +555,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]
@@ -1019,4 +1036,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/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/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/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,
+    }

docs/acceptance/V2R2C1-pre-push-review-acceptance.md

@@ -0,0 +1,107 @@
+# AIOS V2R2C1 Pre-Push Review Acceptance Record（最新机制语义版）
+
+## 1. 验收对象
+
+- Repository: aios/aios
+- Authoritative Git: /data/aios-gitops/repo.git
+- Documentation branch: agent/docs/v2r2c1-provenance-mainline-refresh-20260810
+- Branch HEAD before semantic repair: 869cc007649a39035f5b74513cecc55bb1ae673f
+- Follow-up parent: 869cc007649a39035f5b74513cecc55bb1ae673f
+- Latest mechanism HEAD: 52b74ba61040246fffa8b10011c1199830e3d665
+- Authoritative main HEAD: 3244f0f17f8845592ac94d585f5ebd65693c9f29
+- Owner review: REQUIRED
+
+文档 follow-up 不是 mechanism production release。当前 Production Baseline 仍为 V2R2C1 generation 4、PRODUCTION_DEPLOYED / ACTIVE，tag object 为 07fc2746b688f5f243f29b39de39f32edc3dd9d2，peeled target 为 998d30304787e31e5ae607c0bd86b5edb2b0e5ad。
+
+## 2. 最新机制差异审计
+
+差异范围分三层记录：
+
+| 审计层 | 范围 | 结果 |
+| --- | --- | --- |
+| Initial review fix | 6ece3525 → b660e2c | 14 个文件 |
+| Full mechanism audit | b660e2c → 51b5f0e → 52b74ba | 17 个路径，+1403/-79 |
+| Latest tip delta | 51b5f0e → 52b74ba | 12 个路径，+334/-39 |
+
+因此 README、requirements、architecture、acceptance 与 version differences 必须引用 52b74ba 的最新状态，而不能把 b660e2c 当作当前机制分支 tip。
+
+## 3. GitOps 与 writer 验收
+
+- Action service: active
+- Repository identity: aios/aios
+- Authoritative repository: /data/aios-gitops/repo.git
+- Governed task-scoped write path: enabled for this branch task
+- legacy_global_writes_enabled: false
+- main write: false
+- production write: false
+- Baseline write: false
+- tag write: false
+- force push: false
+- Documentation branch creation/continuation: PASS
+- Non-main documentation commit: PASS
+- Remote branch OID equals local commit OID: PASS
+- Gitea branch and content readback: PASS
+
+旧的 aios-gitops-action-writer.service 不作为生产写开关；本验收以当前受治理 Action service 与受控 writer route 的实际响应为准。
+
+## 4. 文档完整性验收
+
+五个且仅五个路径属于 follow-up diff：
+
+- README.md
+- docs/requirements/V2R2C1-pre-push-review-fix.md
+- docs/architecture/release-provenance-and-registry-convergence.md
+- docs/acceptance/V2R2C1-pre-push-review-acceptance.md
+- docs/version-differences/V2R2C1-pre-push-mechanism-diff.md
+
+每个文件必须满足：
+
+- UTF-8 strict decode: PASS
+- CJK count > 0: PASS
+- U+FFFD count = 0: PASS
+- abnormal consecutive question marks = 0: PASS
+- source bytes SHA-256 equals Git blob SHA-256: PASS
+- Gitea contents readback equals Git blob bytes: PASS
+- git diff --check: PASS
+
+## 5. 提交方测试事实
+
+- Related tests: 760 passed
+- Skipped: 2 skipped
+- Full test result: 1 failed
+- Subtests: 154 passed
+- Failure: tests/test_signer_contract_compatibility.py::test_real_signer_help_and_self_test_are_available
+- /usr/local/libexec/aios-grant-signer-adapter self-test: return 42
+- Two independent reruns: both return 42
+
+Signer gate is FAIL / EXTERNAL. Full external gate is PARTIAL. Production ready remains false。不能把该失败解释为暂态成功，也不能把全量结果写成全通过。
+
+## 6. 分层状态
+
+| 层级 | 结果 |
+| --- | --- |
+| Mechanism code fix | COMPLETE |
+| Documentation branch write | COMPLETE |
+| Documentation remote readback | COMPLETE |
+| Full external gate | PARTIAL |
+| Signer gate | FAIL / EXTERNAL |
+| Production acceptance | NOT_RUN |
+| Owner review | REQUIRED |
+
+Documentation branch write/readback COMPLETE 只表示非 main 文档分支的受治理提交与远端内容一致，不表示 mechanism/release production write 已发生。
+
+## 7. 未执行生产动作
+
+- Mechanism/release production write: NOT_RUN
+- Merge to main: NOT_RUN
+- Production release: NOT_RUN
+- Production Baseline update: NOT_RUN
+- Tag creation/deletion/rewrite: NOT_RUN
+- Force push: NOT_RUN
+- systemd/nginx production reload or restart: NOT_RUN
+
+## 8. 治理结论
+
+force_push_used=false、tag_rewritten=false、direct_main_push=false、secrets_committed=false、production_mutation=false、legacy_global_writes_enabled=false。
+
+结论：文档语义一致性修复候选已准备为 READY_FOR_OWNER_REVIEW。Owner review 完成前不得 merge main，不得修改 Baseline、tag、历史版本身份或生产 runtime。

docs/architecture/release-provenance-and-registry-convergence.md

@@ -0,0 +1,156 @@
+# AIOS Release Provenance & Registry Convergence Architecture（最新机制语义版）
+
+## 1. 架构目标
+
+AIOS release path 把版本名提升为一组可验证且不可混淆的 Git、approval、artifact 与 runtime 身份。最新机制 HEAD 为 52b74ba61040246fffa8b10011c1199830e3d665；它位于非 main 的机制分支，不能被描述为新的 Production Baseline。
+
+架构必须同时解决：
+
+- mutable main 引起的 TOCTOU；
+- annotated tag object 与 peeled target 混淆；
+- source registry overlay 或 runtime projection 被误报为 Git/source convergence；
+- 未绑定 approval plan 驱动 registry 生成；
+- release 与 general artifact binding 混用；
+- builder 与 executor 各自验证不同身份；
+- legacy V2R2C1 receipt 兼容与 future binding fail closed。
+
+## 2. 身份模型
+
+| 符号 | 定义 | 当前或运行时绑定 |
+| --- | --- | --- |
+| R | authoritative repository identity | aios/aios；Git 目录为 /data/aios-gitops/repo.git |
+| A | annotated tag object | V2R2C1 tag object 07fc2746b688f5f243f29b39de39f32edc3dd9d2 |
+| T | A 的 peeled commit，即 release target | 998d30304787e31e5ae607c0bd86b5edb2b0e5ad |
+| E | evidence closure commit | a8b906d1bc24088c19647e98de6a679268f6d0a2 |
+| M | 本次 source-registry snapshot 固定的 main commit | 每次构建解析一次并写入 provenance |
+| C | 当前权威 main 或被 convergence probe 选定的 reference commit | 当前 authoritative main 为 3244f0f17f8845592ac94d585f5ebd65693c9f29 |
+| P | Owner-approved approval plan | sha256:c0df52008ae7f64cec5ee2d28947aa7b892066c670ab8295e4549140b1a7500e |
+
+R、A、T、E、M、C、P 不得用一个版本字符串互相替代。尤其是 mechanism branch HEAD、C、T 与 Production Baseline 是不同状态轴。
+
+## 3. Git Object Model 与 lineage
+
+对于 provenance-enabled release，代码必须建立以下关系：
+
+~~~text
+refs/tags/<version> -> A
+A^{commit}          = T
+T                   = registry peeled target
+T ancestor-of E
+T ancestor-of M     when the release lineage check requires it
+M ancestor-of C     when source-registry main convergence is checked
+SR                  = bytes(M:path)
+H                   = SHA-256(SR)
+~~~
+
+其中：
+
+- A 必须是 annotated tag object，不能由 lightweight tag 代替。
+- E 必须是 commit，且 release lineage 证明 T 到 E。
+- M 必须是 commit；source registry 的读取、校验、哈希和打包都使用 M:path 的同一字节。
+- C 是当前权威 main/reference commit，不是生产 target，也不是 mechanism branch HEAD。
+- M 与 C 的 convergence probe 还要比较相同 path 的精确字节；M ancestor-of C 仅在该 probe 的适用语义下成立。
+- 任何 same-name but different-OID tag、target、registry、approval 或 repository identity 冲突都停止自动流程。
+
+## 4. Pinned source registry 与 authority binding
+
+构建开始时由 authoritative repository 解析 M，并立即保存 M、path、SR 和 H。后续不得回到 mutable main 重新取 registry。
+
+registry record 必须绑定：
+
+- R、version、A、T、peeled target、E；
+- P 与其 authority/publication receipt；
+- M、source registry path、H；
+- generator 使用的 authoritative repository identity。
+
+authority binding 不是字符串格式检查。governed caller、generator、builder 与 executor 都必须能证明 approval authority 与 release identity 的关联。
+
+## 5. Artifact binding：release 与 general 分离
+
+release artifact 需要完整的 release binding：
+
+~~~text
+repository + version + tag_oid + target_oid
++ peeled_target_oid + evidence_closure_oid + approval_plan_id
++ source_registry_main_oid + source_registry_path + source_registry_sha256
+~~~
+
+general artifact 可以保持普通构建用途，但不得被当作 release artifact，也不得绕过 release binding。release mode 缺少 expected binding 或字段不匹配时返回 fail-closed 状态。
+
+artifact metadata、manifest 与 receipt 必须引用同一组对象。source registry payload 以 SR 的精确字节进入 artifact，不能只保存重新序列化后的等价 JSON。
+
+## 6. 组件职责与独立复核
+
+### version_registry.py
+
+解析 M，读取 M:path，验证 registry record、A/T/E lineage、approval authority，并分别报告 registry display、runtime bundle、attested projection 与 source convergence。
+
+### generate_source_registry.py
+
+在生成前获得 authoritative repository、repository identity 与 approval authority，校验 tag object、target、lineage、registry bytes 与 authority 关联。
+
+### governed_production_release.py
+
+执行 T、E、M 的适用 lineage 检查，构建由 T payload 与 M registry bytes 组成的 deterministic release artifact，并把复合 provenance 写入 metadata、manifest 和 receipt。
+
+### Autonomous release artifact builder
+
+release mode 强制 expected release identity；general mode 保持独立。builder 不得用缺失或局部 binding 伪造 release 成功。
+
+### approved_release_executor.py
+
+executor 以自己的 Git identity 重新读取并验证 artifact、canonical registry、source path、source SHA-256、BUILD-METADATA provenance、future tag object 与 authorization receipt。executor 不接受 builder 自述作为唯一证据。
+
+## 7. Runtime projection 与 Git convergence
+
+版本 API 必须区分：
+
+- legacy display list；
+- runtime bundle versions；
+- attested projection version；
+- runtime_bundle_contains_version；
+- source_registry_converged；
+- registry_consistency_status。
+
+projection 能看到版本不代表 runtime bundle 包含该版本，更不代表 M 与 C 的 source bytes 收敛。典型状态包括 PASS、RUNTIME_BUNDLE_MISSING、IDENTITY_MISMATCH、RUNTIME_ENTRY_INVALID、NO_ATTESTED_PROJECTION。
+
+当前生产 Baseline 仍为 V2R2C1 generation 4、PRODUCTION_DEPLOYED / ACTIVE。该运行时状态不能被机制分支 HEAD 或文档分支写入替代。
+
+## 8. Post-deploy source-registry recheck
+
+部署后的 executor 必须重新读取 authoritative Git 与 source registry path，重新计算 H，并核对：
+
+- repository identity；
+- release binding 的 A/T/E/P；
+- M、path、H；
+- artifact metadata 与实际 payload；
+- runtime projection 与 source convergence；
+- authorization receipt。
+
+任何 recheck 失败都必须保持 blocked，不得靠 runtime health、开放端口或 projection 可见性推断 release closure。
+
+## 9. Fail-closed 状态
+
+| 条件 | 结果 |
+| --- | --- |
+| source registry 缺失或精确字节不可读 | ReleaseBlocked |
+| repository、tag object、target、E、M、P 身份冲突 | fail closed |
+| lightweight tag | reject |
+| same-name different-OID tag | MANUAL_OWNER_REVIEW_REQUIRED |
+| T 不是要求的 E/M lineage ancestor | blocked / Owner review |
+| approval plan 未与 release identity 关联 | reject |
+| release mode 缺少 binding | ARTIFACT_RELEASE_BINDING_REQUIRED |
+| artifact registry digest 或 metadata 不匹配 | ReleaseBlocked |
+| projection 可见但 runtime bundle 缺版本 | RUNTIME_BUNDLE_MISSING |
+| runtime、projection、Git identity 不一致 | IDENTITY_MISMATCH |
+| post-deploy source registry recheck 失败 | production acceptance blocked |
+
+## 10. Legacy V2R2C1 兼容
+
+旧 V2R2C1 receipt 可以缺少 tag_oid、binding_schema、source_registry_main_oid、source_registry_path 和 source_registry_sha256。future binding 或 provenance-enabled binding 一旦提供 source provenance 字段，就必须完整提供并严格校验。
+
+兼容只允许旧 receipt 按旧语义读取，不允许新 release 回退到弱绑定模型，也不允许修改 versions/V2R2C1.json 的历史发布身份。
+
+## 11. 当前文档与生产边界
+
+文档 follow-up 分支的父提交为 869cc007649a39035f5b74513cecc55bb1ae673f，本次只允许五个 Markdown 路径。文档分支写入和远端回读是独立的 documentation closure；mechanism/release production write、main merge、Production release、Baseline update、tag rewrite、force push 与 production service mutation 均不在本任务中。

docs/requirements/V2R2C1-pre-push-review-fix.md

@@ -0,0 +1,169 @@
+# AIOS V2R2C1 Pre-Push Review Fix Requirements（最新机制语义版）
+
+## 1. 目的
+
+本需求记录 V2R2C1 Pre-Push Review 机制从初始 b660e2c 修复到最新 mechanism HEAD 的完整事实边界。它描述 release provenance、source registry、artifact、approval 与 runtime convergence 的可证明关系，不宣称新的生产发布。
+
+三个增量范围必须分开：
+
+- 初始 b660e2c 修复：14 个源码/测试文件。
+- b660e2c → 51b5f0e → 52b74ba 的完整审计：17 个路径，+1403/-79。
+- 最新 51b5f0e → 52b74ba tip 增量：12 个路径，+334/-39。
+
+当前机制参考为 52b74ba61040246fffa8b10011c1199830e3d665。文档 follow-up 以文档分支原 HEAD 869cc007649a39035f5b74513cecc55bb1ae673f 为父提交。
+
+## 2. 范围
+
+实现和审计覆盖：
+
+- pinned source registry snapshot 与 convergence；
+- governed release artifact build；
+- release executor 的独立 provenance 复核；
+- approval authority 绑定；
+- release/general artifact binding；
+- version API convergence 可观测性；
+- TOCTOU、identity mismatch 与 legacy receipt 回归测试；
+- 最新机制状态下的 README、requirements、architecture、acceptance、version differences 文档语义。
+
+不在本次范围内：
+
+- 修改 main；
+- merge；
+- Production release；
+- Production Baseline 或 versions/V2R2C1.json；
+- 创建、删除或重写正式 tag；
+- force push；
+- production runtime 修改；
+- systemd / nginx 生产服务切换、reload 或 restart；
+- 提交任何密钥。
+
+## 3. 功能需求
+
+### R1 — Source registry 固定 M
+
+一次构建只能解析一次 source registry 所在 main commit M。后续 registry 校验、字节读取、SHA-256 计算与 artifact 打包必须使用 M 下同一 Git blob，禁止在同一构建中重新读取 mutable refs/heads/main。
+
+验收要求：构建期间 main 前进时，artifact 仍绑定最初固定 M 的 registry 字节。
+
+### R2 — Release provenance 是复合身份
+
+release identity 至少包含：
+
+- repository R；
+- version；
+- annotated tag object A；
+- release target / peeled target T；
+- evidence closure E；
+- Owner-approved approval plan P。
+
+启用 source-registry provenance 时还必须包含：
+
+- pinned source-registry main commit M；
+- source registry path；
+- source registry 精确字节的 SHA-256。
+
+### R3 — Git object 与 lineage 可证明
+
+代码必须验证：
+
+- R 的仓库身份为 aios/aios；
+- refs/tags/<version> 指向 annotated tag object A；
+- A^commit 等于 T；
+- T 与 registry 中的 peeled target 一致；
+- E 是 commit；
+- T 是 E 的 ancestor；
+- M 是 commit；
+- 适用的 source-registry convergence 检查证明 T 到 M 的 lineage，并证明 M ancestor-of 当前权威 main/reference commit C；
+- M:path 的字节与 C:path 的字节在 convergence 检查中一致。
+
+关键身份冲突不得自动修复，必须进入 fail-closed 或 Owner review。
+
+### R4 — Approval plan 必须绑定 release authority
+
+仅有格式正确的 approval_plan_id 不足以授权。generator、governed caller 与 executor 必须读取权威 approval/publication receipt，并证明 authority 与 R、version、A、T、E、P 关联。缺少关联、来源不可信或字段不一致时必须拒绝。
+
+### R5 — Source registry generator 绑定权威仓库
+
+generate_source_registry.py 必须显式获得 authoritative repository、approval authority 与 repository identity。生成前校验 Git identity、tag object、target、lineage 与 approval authority，不得只凭本地参数生成表面合法的 registry。
+
+### R6 — Release artifact builder 强制 binding
+
+artifact builder 在 release mode 必须收到 expected release identity，并逐字段比较：
+
+- version；
+- repository；
+- tag_oid；
+- target_oid；
+- peeled_target_oid；
+- evidence_closure_oid；
+- approval_plan_id；
+- provenance-enabled 场景下的 M、path、sha256。
+
+release mode 缺少 binding 或字段不匹配必须 fail closed。general artifact 仍可使用其独立的非 release 语义，不能与 release binding 混淆。
+
+### R7 — Executor 独立复核 provenance
+
+生产 executor 不得只信任 builder 自述。它必须独立验证 source registry member、canonical JSON、registry identity、path、精确字节 SHA-256、BUILD-METADATA.json provenance、future binding 的 tag object 与 authorization receipt。executor 还必须以自己的 Git identity 重新核对关键对象。
+
+### R8 — V2R2C1 legacy receipt 兼容
+
+旧 V2R2C1 receipt 可以省略 future-only provenance 字段。新字段仅在 future release binding 或 provenance-enabled binding 中强制。兼容路径不得削弱新 binding 的严格校验，也不得修改历史 versions/V2R2C1.json。
+
+### R9 — 版本可见性不得冒充 convergence
+
+版本 API 必须分别暴露：
+
+- legacy display list；
+- runtime bundle versions；
+- attested projection version；
+- runtime bundle 是否包含该版本；
+- source registry 是否收敛；
+- registry consistency status。
+
+projection-only 可见时必须报告 RUNTIME_BUNDLE_MISSING 或等价非收敛状态，不能仅因 overlay 可见就报告 source convergence。
+
+### R10 — 回归测试覆盖 TOCTOU 与身份错误
+
+至少覆盖：
+
+- pinned source registry 后 main 前进；
+- lightweight tag 拒绝；
+- annotated tag 与 lineage 正常；
+- approval authority 缺失关联；
+- release binding 每个身份字段不匹配；
+- release mode 缺少 binding；
+- executor 对 source digest 与 build metadata 的独立复核；
+- 同名不同 tag object 进入 MANUAL_OWNER_REVIEW_REQUIRED；
+- projection-only 不等于 runtime convergence；
+- legacy V2R2C1 receipt 保持兼容。
+
+## 4. 当前验收状态
+
+| 项目 | 当前结果 |
+| --- | --- |
+| Code/mechanism fix | COMPLETE |
+| Documentation branch write | COMPLETE after this follow-up |
+| Documentation remote readback | COMPLETE after this follow-up |
+| Related test result | 760 passed |
+| Skipped | 2 skipped |
+| Full test result | 1 failed |
+| Subtests | 154 passed |
+| Full external gate | PARTIAL |
+| Signer gate | FAIL / EXTERNAL |
+| Production ready | false |
+| Owner review | REQUIRED |
+
+唯一失败为 tests/test_signer_contract_compatibility.py::test_real_signer_help_and_self_test_are_available。/usr/local/libexec/aios-grant-signer-adapter self-test 返回 42，两个独立重跑仍返回 42。该事实不能被解释为暂态成功，也不能将全量结果记为全通过。
+
+## 5. 安全与治理边界
+
+- force_push_used = false
+- tag_rewritten = false
+- direct_main_push = false
+- secrets_committed = false
+- production_mutation = false
+- legacy_global_writes_enabled = false
+- 未经 Owner review 不执行 merge 或 consequential production action。
+- 文档分支写入、Git/Gitea 远端回读与生产发布分别验收。
+- 任何 same-name / different-OID tag 冲突都进入 MANUAL_OWNER_REVIEW_REQUIRED。
+- 文档刷新不改写已投产 V2R2C1 registry identity。

docs/version-differences/V2R2C1-pre-push-mechanism-diff.md

@@ -0,0 +1,64 @@
+# V2R2C1 与 Pre-Push Mechanism Fix 版本差异（最新机制语义版）
+
+## 1. 状态轴
+
+| 状态 | OID / Version | 说明 |
+| --- | --- | --- |
+| Production Baseline | V2R2C1 / generation 4 | PRODUCTION_DEPLOYED / ACTIVE |
+| V2R2C1 tag object | 07fc2746b688f5f243f29b39de39f32edc3dd9d2 | annotated tag object |
+| V2R2C1 release target T | 998d30304787e31e5ae607c0bd86b5edb2b0e5ad | peeled target |
+| Production evidence closure E | a8b906d1bc24088c19647e98de6a679268f6d0a2 | Baseline 记录的证据闭包 |
+| Authoritative main C | 3244f0f17f8845592ac94d585f5ebd65693c9f29 | 当前权威工程主线 |
+| Latest mechanism HEAD | 52b74ba61040246fffa8b10011c1199830e3d665 | 非 main 待审机制分支 |
+| Documentation follow-up parent | 869cc007649a39035f5b74513cecc55bb1ae673f | 语义修复提交父提交 |
+
+以上状态不可互相替代。52b74ba 是最新 mechanism reference，不是新的 Production Baseline；文档 branch 也不是 main。
+
+## 2. 三层差异范围
+
+- Initial b660e2c review fix：14 个文件。
+- 完整 b660e2c → 51b5f0e → 52b74ba 审计：17 个路径，+1403/-79。
+- 最新 51b5f0e → 52b74ba 增量：12 个路径，+334/-39。
+
+## 3. 已发布 V2R2C1 机制
+
+V2R2C1 保留其 production baseline、annotated tag object、peeled target、artifact、evidence closure、approval、immutable staging 与 release executor 身份。历史 identity 保持不可变，versions/V2R2C1.json 不在本次修改范围。
+
+## 4. 最新机制增量
+
+52b74ba 所在机制链进一步明确：
+
+1. pinned source registry：一次解析 M，后续统一使用 M:path 的精确字节与 SHA-256。
+2. composite provenance：R、version、A、T、E、P 与 M、path、sha256 共同构成 release binding。
+3. Git object proof：区分 annotated tag object A、peeled target T、evidence closure E、pinned main M 与当前 main/reference C。
+4. authority proof：approval plan 必须与 release identity 真实关联。
+5. registry generator hardening：generator 接入权威 bare repository 与 approval authority。
+6. artifact binding：release mode 强制完整 binding，general artifact 保持独立语义。
+7. executor independent verification：executor 重新校验 registry bytes、metadata、tag object、authority 与 Git identity。
+8. convergence observability：区分 display list、runtime bundle、attested projection 与 source convergence。
+9. post-deploy recheck：部署后重新读取 authoritative Git、source registry 与 runtime projection。
+10. fail-closed：身份、digest、lineage、approval、binding 与 projection 不一致时停止自动流程。
+11. legacy compatibility：旧 V2R2C1 receipt 可继续读取，新 binding 不回退到弱校验。
+
+## 5. 兼容性差异
+
+| 项目 | Legacy V2R2C1 | Future / provenance-enabled |
+| --- | --- | --- |
+| tag_oid in executor binding | 可缺省 | 必须 |
+| binding schema | 可缺省 | aios.release-binding.* |
+| source registry main OID | 可缺省 | provenance 开启时必须 |
+| source registry SHA-256 | 可缺省 | provenance 开启时必须 |
+| source registry path | 默认/兼容 | 明确验证 |
+| artifact build release binding | 旧路径兼容 | release mode 强制 |
+| same-name different tag | 不允许重写 | MANUAL_OWNER_REVIEW_REQUIRED |
+| post-deploy registry check | 旧 receipt 可无 | 新 release 必须 |
+
+## 6. 当前验收状态
+
+相关测试为 760 passed、2 skipped、1 failed，154 subtests passed。唯一失败为 tests/test_signer_contract_compatibility.py::test_real_signer_help_and_self_test_are_available；self-test 返回 42，两个独立重跑仍返回 42。因此 Signer gate 为 FAIL / EXTERNAL，Full external gate 为 PARTIAL，Production ready 为 false。
+
+文档分支 write 与 remote readback 属于 COMPLETE 的独立 documentation closure。mechanism/release production write、main merge、Production release、Baseline update、tag rewrite 与 force push 均为 NOT_RUN。Owner review 仍为 REQUIRED。
+
+## 7. 处理边界
+
+不修改 versions/V2R2C1.json，不修改历史 tag 或 Production Baseline，不直接写 main，不执行 merge，不改变生产 runtime。若 Owner review 后需要正式交付机制，应建立新的受治理 release/version identity，并重新执行独立 authority 验收。

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_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_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"
+}