ci(mirror): fence five exact forwarded owner blobs

.github/workflows/paired-root.yml

@@ -53,6 +53,11 @@ jobs:
       - name: Test paired-root admission rules
         working-directory: mirror
         run: PYTHONDONTWRITEBYTECODE=1 python3 -m unittest scripts.ci.test_paired_root -v
+      - name: Check exact forwarded owner blobs
+        run: |
+          PYTHONDONTWRITEBYTECODE=1 python3 mirror/scripts/ci/paired_root.py check-forwarded \
+            --libraries Libraries --mirror mirror \
+            --libraries-sha "$LIBRARIES_SHA" --mirror-sha "$MIRROR_SHA"
       - name: Assemble pinned disposable source pair
         id: assemble
         shell: bash

scripts/ci/paired_root.py

@@ -14,6 +14,16 @@
 
 
 REQUIRED_POLY_KV_DROP = 'fib-quant 0.1.0-alpha.1'
+# Only whole files already reviewed and forwarded from this owner revision.
+# Partial src/db.rs is deliberately absent: its FK fix does not equal the
+# owner file. This is byte/mode drift detection, not semantic equivalence.
+FORWARDED_EXACT_PATHS = (
+    'examples/governed_append_canary.rs',
+    'src/types.rs',
+    'tests/foreign_key_integrity.rs',
+    'tests/search_tests.rs',
+    'tests/storage_lifecycle.rs',
+)
 
 
 def require_exact_sha(observed: str, expected: str, owner: str) -> None:
@@ -91,6 +101,18 @@ def tracked_files(repo: Path, sha: str, *, strict: bool) -> dict[str, tuple[str,
         raise ValueError(f'no tracked files in {repo}')
     return records
 
+def check_forwarded_paths(libraries: Path, mirror: Path, libraries_sha: str, mirror_sha: str) -> None:
+    if not re.fullmatch(r'[0-9a-f]{40}', libraries_sha) or not re.fullmatch(r'[0-9a-f]{40}', mirror_sha):
+        raise ValueError('exact 40-hex source revisions required')
+    owner = tracked_files(libraries, libraries_sha, strict=False)
+    mirrored = tracked_files(mirror, mirror_sha, strict=True)
+    for path in FORWARDED_EXACT_PATHS:
+        owner_blob = owner.get('semantic-memory/' + path)
+        mirror_blob = mirrored.get(path)
+        if owner_blob is None or mirror_blob is None or owner_blob != mirror_blob:
+            raise ValueError(f'forwarded path differs from pinned Libraries owner: {path}')
+    print(f'{len(FORWARDED_EXACT_PATHS)} forwarded paths match exact owner Git blobs and modes')
+
 
 def extract_archive(repo: Path, sha: str, destination: Path, *, owner: bool) -> None:
     proc = subprocess.Popen(
@@ -182,13 +204,18 @@ def main() -> None:
     lock = sub.add_parser('check-lock')
     lock.add_argument('--before', required=True)
     lock.add_argument('--after', required=True)
+    forwarded = sub.add_parser('check-forwarded')
+    for flag in ('libraries', 'mirror', 'libraries-sha', 'mirror-sha'):
+        forwarded.add_argument('--' + flag, required=True)
     args = parser.parse_args()
     if args.action == 'assemble':
         assemble(Path(args.libraries), Path(args.mirror), args.libraries_sha, args.mirror_sha,
                  Path(args.scratch), args.pr_head_sha)
-    else:
+    elif args.action == 'check-lock':
         validate_lock_delta(tomllib.loads(Path(args.before).read_text()), tomllib.loads(Path(args.after).read_text()))
         print('scratch lock delta is exactly the declared PolyKV FibQuant dependency drop')
+    else:
+        check_forwarded_paths(Path(args.libraries), Path(args.mirror), args.libraries_sha, args.mirror_sha)
 
 
 if __name__ == '__main__':

scripts/ci/test_paired_root.py

@@ -63,6 +63,39 @@ def test_wrong_source_sha_is_rejected_before_assembly(self):
             paired_root.require_exact_sha('short', 'short', 'mirror')
         paired_root.require_exact_sha('a' * 40, 'a' * 40, 'Libraries')
 
+    def test_forwarded_blobs_bind_exact_paths_and_modes_not_unrelated_files(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            owner, mirror = root / 'owner', root / 'mirror'
+            forwarded = {path: 'owner:' + path for path in paired_root.FORWARDED_EXACT_PATHS}
+            owner_sha = self.fixture_repo(owner, {
+                **{'semantic-memory/' + path: value for path, value in forwarded.items()},
+                'semantic-memory/src/db.rs': 'partial owner-only change',
+            })
+            mirror_sha = self.fixture_repo(mirror, {**forwarded, 'src/db.rs': 'intentional partial difference'})
+            paired_root.check_forwarded_paths(owner, mirror, owner_sha, mirror_sha)
+            with self.assertRaisesRegex(ValueError, 'exact 40-hex'):
+                paired_root.check_forwarded_paths(owner, mirror, 'bad', mirror_sha)
+            path = next(iter(forwarded))
+            (mirror / path).write_text('different')
+            subprocess.run(['git', '-C', str(mirror), 'commit', '-qam', 'changed bytes'], check=True)
+            changed = subprocess.check_output(['git', '-C', str(mirror), 'rev-parse', 'HEAD'], text=True).strip()
+            with self.assertRaisesRegex(ValueError, path):
+                paired_root.check_forwarded_paths(owner, mirror, owner_sha, changed)
+            (mirror / path).write_text(forwarded[path])
+            (mirror / path).chmod(0o755)
+            subprocess.run(['git', '-C', str(mirror), 'add', '--', path], check=True)
+            subprocess.run(['git', '-C', str(mirror), 'commit', '-qm', 'changed mode'], check=True)
+            mode_changed = subprocess.check_output(['git', '-C', str(mirror), 'rev-parse', 'HEAD'], text=True).strip()
+            with self.assertRaisesRegex(ValueError, path):
+                paired_root.check_forwarded_paths(owner, mirror, owner_sha, mode_changed)
+            (mirror / path).chmod(0o644)
+            subprocess.run(['git', '-C', str(mirror), 'rm', '-qf', '--', path], check=True)
+            subprocess.run(['git', '-C', str(mirror), 'commit', '-qm', 'missing path'], check=True)
+            missing = subprocess.check_output(['git', '-C', str(mirror), 'rev-parse', 'HEAD'], text=True).strip()
+            with self.assertRaisesRegex(ValueError, path):
+                paired_root.check_forwarded_paths(owner, mirror, owner_sha, missing)
+
     def test_archive_member_rejects_duplicates_and_file_directory_collisions(self):
         seen = {}
         paired_root.claim_archive_member('src', 'directory', seen)