From a3aded4983113b9a2fc9ef0895260093daca3cdc Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 20:01:02 -0700
Subject: [PATCH 1/9] feat(archive): source activation runtime from out-of-tree
 signed handoff via --bundle-source
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Implements PR-2 of the local activation-release pipeline (workspace design
of record drafts/local-activation-release-cut-pipeline.md §3.2/§9.1c/
§9.12(2)/§10-A/D1, hardened by a 2-round distinct-family adversarial DESIGN
review before this implementation).

The signed+notarized runtime bundle ships as a release asset and is never
committed, so the builder can no longer read it from the git checkout:

- `--bundle-source <leaf .bundle dir>` supplies the signed 0o500 handoff.
  The committed manifest alone decides the mode (D1): activation manifest
  without --bundle-source = hard error (never a silent policy-only
  downgrade); policy-only manifest forbids the flag; an in-tree runtime/
  conflicts with an external source.
- classify_package() validates the manifest SHAPE only (split into
  _validate_activation_manifest + _validate_activation_bundle_tree); the
  physical leaf is validated as HOSTILE input: symlink argument rejected
  via lstat BEFORE resolve(), bounded non-following enumeration, member-set
  EQUALITY, per-member O_NOFOLLOW regular-file/uid/nlink-1/mode/size/sha256
  (strict 0o500 preserved — the external source is mode-preserving).
- Runtime archive metadata is synthesized ENTIRELY from a frozen manifest
  snapshot (0o755 traversal parents + sealed 0o500 leaf; file members get
  record mode/size); only bytes stream from the validated source.
- verify_archive stays self-contained (mode kwarg only — CI's
  build_release_evidence.py call is unchanged) and validates runtime
  members against a SINGLE frozen manifest byte-snapshot: exact member
  sequence + type + mode + size + sha256(archived) == record digest, plus
  archived runtime-manifest.json bytes == frozen bytes. A source-tree swap
  after build-time validation can never produce a passing archive (closes
  the validate→pack TOCTOU without fd-pinning).
- Build writes to an O_EXCL temp, verifies THAT, then os.replace()s to the
  destination; failure unlinks the temp — no publishable artifact on any
  failure path. --output must not alias a source tree.

12 new adversarial tests (fail-closed gating, symlink/hardlink/device
rejection, digest/size/mode drift, membership equality, tampered-archive +
manifest-swap detection via the frozen binding, no-output-on-failure,
determinism). 327 tests/ + 254 scripts/ green; CLI smoke-verified
(--print-mode, policy-only build, fail-closed bundle-source rejection).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../2026-07-17-build-archive-bundle-source.md |   2 +
 scripts/build_plugin_archive.py               | 430 ++++++++++++++----
 tests/test_plugin_archive.py                  | 268 ++++++++++-
 tests/test_release_evidence.py                |  14 +-
 4 files changed, 604 insertions(+), 110 deletions(-)
 create mode 100644 changelog.d/2026-07-17-build-archive-bundle-source.md

diff --git a/changelog.d/2026-07-17-build-archive-bundle-source.md b/changelog.d/2026-07-17-build-archive-bundle-source.md
new file mode 100644
index 0000000..4297c72
--- /dev/null
+++ b/changelog.d/2026-07-17-build-archive-bundle-source.md
@@ -0,0 +1,2 @@
+### Changed
+- `build_plugin_archive.py` now sources the activation runtime bundle from an out-of-tree signed handoff via `--bundle-source <leaf .bundle dir>` instead of the git checkout — the signed runtime ships as a release asset and is never committed. The committed `runtime-manifest.json` alone decides the mode (design decision D1): an activation manifest without `--bundle-source` is a hard error (never a silent policy-only downgrade), a policy-only manifest forbids the flag, and an in-tree `runtime/` conflicts with an external source. The bundle source is treated as hostile input (symlink argument rejected before resolution; bounded non-following enumeration; member-set equality; per-member O_NOFOLLOW regular-file/uid/nlink/mode/size/sha256 checks with the strict 0o500 install mode preserved). Runtime archive metadata is synthesized entirely from a frozen manifest snapshot (0o755 traversal parents + sealed 0o500 bundle leaf), `verify_archive` validates runtime members and the archived manifest bytes against that frozen snapshot only (self-contained — no bundle source needed, closing the validate→pack swap window), and the archive is built into an exclusive temp file that is verified before an atomic rename so a failed build never leaves a publishable artifact.
diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index a13c0b2..b960058 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -13,7 +13,6 @@
 import stat
 import tarfile
 from pathlib import Path, PurePosixPath
-from typing import Iterable
 
 try:
     from scripts.skill_structure import expected_skill_relpaths, skill_tree_differences
@@ -174,14 +173,23 @@ def _safe_source(path: Path) -> os.stat_result:
     return info
 
 
-def _load_manifest(plugin_path: Path) -> dict[str, object]:
+def _read_manifest_bytes(plugin_path: Path) -> bytes:
+    """Read the committed manifest exactly once for a frozen-snapshot check."""
+
     manifest_path = plugin_path / "runtime-manifest.json"
     info = _safe_source(manifest_path)
     if not stat.S_ISREG(info.st_mode):
         raise ValueError("runtime-manifest.json is not a regular file")
     try:
-        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    except (OSError, UnicodeError, ValueError, RecursionError) as exc:
+        return manifest_path.read_bytes()
+    except OSError as exc:
+        raise ValueError("runtime manifest is unreadable") from exc
+
+
+def _parse_manifest(data: bytes) -> dict[str, object]:
+    try:
+        manifest = json.loads(data.decode("utf-8"))
+    except (UnicodeError, ValueError, RecursionError) as exc:
         raise ValueError("runtime manifest is unreadable") from exc
     if (
         not isinstance(manifest, dict)
@@ -208,7 +216,20 @@ def _load_manifest(plugin_path: Path) -> dict[str, object]:
     return manifest
 
 
-def _validate_activation(plugin_path: Path, item: object) -> None:
+def _load_manifest(plugin_path: Path) -> dict[str, object]:
+    return _parse_manifest(_read_manifest_bytes(plugin_path))
+
+
+def _validate_activation_manifest(item: object) -> tuple[dict[str, object], ...]:
+    """Validate the activation artifact SHAPE from the committed manifest only.
+
+    Deliberately touches no filesystem bundle state: with the signed runtime
+    shipped as a release asset (never committed), the committed manifest is the
+    activation marker (design decision D1) and the physical bundle bytes are
+    validated separately against wherever they are sourced from
+    (`_validate_activation_bundle_tree`).
+    """
+
     fields = {
         "platform",
         "arch",
@@ -289,38 +310,86 @@ def _validate_activation(plugin_path: Path, item: object) -> None:
         or len(contracts_value) != len(contracts)
     ):
         raise ValueError("activation runtime manifest fields are invalid")
+    return records
 
-    runtime_root = plugin_path / "runtime"
-    allowed = {
-        Path("runtime"),
-        Path("runtime/darwin-arm64"),
-        RUNTIME_BUNDLE_REL,
-        *(RUNTIME_BUNDLE_REL / record["path"] for record in records),
-    }
-    observed = {path.relative_to(plugin_path) for path in runtime_root.rglob("*")}
-    observed.add(Path("runtime"))
-    if observed != allowed:
-        raise ValueError("activation package runtime bundle membership is not exact")
-    bundle = plugin_path / RUNTIME_BUNDLE_REL
-    bundle_info = _safe_source(bundle)
+
+def _validate_activation_bundle_tree(
+    bundle_leaf: Path, records: tuple[dict[str, object], ...]
+) -> None:
+    """Validate the physical runtime bundle leaf directory against the manifest.
+
+    The leaf is HOSTILE input (an operator-supplied out-of-tree handoff):
+    enumerate it bounded and non-following, require member-set EQUALITY with
+    the manifest records, and admit only regular files owned by the invoking
+    user with nlink 1, the exact install mode, exact size, and exact sha256.
+    Devices, FIFOs, sockets, and symlinks all fail the S_ISREG/O_NOFOLLOW
+    checks. `validate_file_records` has already rejected traversal, separator
+    confusables, and Unicode-normalization/case-fold collisions in the paths.
+    """
+
+    try:
+        leaf_info = bundle_leaf.lstat()
+    except OSError as exc:
+        raise ValueError("runtime bundle source is missing") from exc
     if (
-        not stat.S_ISDIR(bundle_info.st_mode)
-        or bundle_info.st_uid != os.getuid()
-        or stat.S_IMODE(bundle_info.st_mode) != runtime_bundle.INSTALL_MODE
+        stat.S_ISLNK(leaf_info.st_mode)
+        or not stat.S_ISDIR(leaf_info.st_mode)
+        or leaf_info.st_uid != os.getuid()
+        or stat.S_IMODE(leaf_info.st_mode) != runtime_bundle.INSTALL_MODE
     ):
-        raise ValueError("activation runtime bundle root identity is invalid")
+        raise ValueError("runtime bundle source root identity is invalid")
+    expected_names = {record["path"] for record in records}
+    observed_names: set[str] = set()
+    try:
+        with os.scandir(bundle_leaf) as entries:
+            for entry in entries:
+                observed_names.add(entry.name)
+                if len(observed_names) > len(expected_names):
+                    raise ValueError(
+                        "runtime bundle source membership is not exact"
+                    )
+    except OSError as exc:
+        raise ValueError("runtime bundle source could not be enumerated") from exc
+    if observed_names != expected_names:
+        raise ValueError("runtime bundle source membership is not exact")
+    member_flags = (
+        os.O_RDONLY
+        | getattr(os, "O_NOFOLLOW", 0)
+        | getattr(os, "O_CLOEXEC", 0)
+    )
     for record in records:
-        member = bundle / record["path"]
-        info = _safe_source(member)
-        if (
-            not stat.S_ISREG(info.st_mode)
-            or info.st_uid != os.getuid()
-            or info.st_nlink != 1
-            or stat.S_IMODE(info.st_mode) != record["install_mode"]
-            or info.st_size != record["size"]
-            or _sha256(member) != record["sha256"]
-        ):
-            raise ValueError("activation runtime bundle member identity is invalid")
+        member = bundle_leaf / record["path"]
+        try:
+            lexical = member.lstat()
+            descriptor = os.open(member, member_flags)
+        except OSError as exc:
+            raise ValueError("runtime bundle source member is unsafe") from exc
+        try:
+            info = os.fstat(descriptor)
+            if (
+                stat.S_ISLNK(lexical.st_mode)
+                or not stat.S_ISREG(info.st_mode)
+                or info.st_uid != os.getuid()
+                or info.st_nlink != 1
+                or stat.S_IMODE(info.st_mode) != record["install_mode"]
+                or info.st_size != record["size"]
+            ):
+                raise ValueError(
+                    "runtime bundle source member identity is invalid"
+                )
+            digest = hashlib.sha256()
+            while True:
+                chunk = os.read(descriptor, 1024 * 1024)
+                if not chunk:
+                    break
+                digest.update(chunk)
+            if digest.hexdigest() != record["sha256"]:
+                raise ValueError("runtime bundle source member digest is invalid")
+        finally:
+            try:
+                os.close(descriptor)
+            except OSError:
+                pass
 
 
 def classify_package(plugin_path: Path) -> str:
@@ -338,7 +407,11 @@ def classify_package(plugin_path: Path) -> str:
         return "policy-only"
     if len(artifacts) != 1:
         raise ValueError("activation package requires exactly one runtime artifact")
-    _validate_activation(plugin_path, artifacts[0])
+    # D1: the committed manifest is the activation marker. The physical bundle
+    # is NOT read here — it ships as a release asset (never committed) and is
+    # validated against its actual source by `_validate_activation_bundle_tree`
+    # (build) and against the frozen manifest records (verify_archive).
+    _validate_activation_manifest(artifacts[0])
     return "activation"
 
 
@@ -402,9 +475,37 @@ def _require_exact_third_party_notice_tree(plugin_path: Path) -> None:
             raise ValueError("activation third-party notice content digest is invalid")
 
 
-def _member_paths(plugin_path: Path, *, mode: str) -> list[Path]:
+# Runtime DIRECTORY members are packaging scaffolding, synthesized with fixed
+# canonical metadata — never statted from a mutable source tree. The two
+# traversal parents are plain 0o755 directories; the bundle leaf itself keeps
+# the sealed install mode.
+_RUNTIME_DIR_MODES: dict[str, int] = {
+    "runtime": 0o755,
+    "runtime/darwin-arm64": 0o755,
+    RUNTIME_BUNDLE_REL.as_posix(): runtime_bundle.INSTALL_MODE,
+}
+
+
+def _member_plan(
+    plugin_path: Path,
+    *,
+    mode: str,
+    records: tuple[dict[str, object], ...] = (),
+) -> list[tuple[str, Path | None]]:
+    """Return the canonical (archive name, source path) plan, sorted by name.
+
+    Runtime members carry a ``None`` source: their archive metadata derives
+    entirely from the frozen manifest ``records`` (and `_RUNTIME_DIR_MODES`),
+    never from a stat of a mutable tree. Every other member is sourced from
+    the git-tracked plugin tree.
+    """
+
     if mode not in {"policy-only", "activation"}:
         raise ValueError("unknown archive mode")
+    if mode == "activation" and not records:
+        raise ValueError("activation member plan requires frozen manifest records")
+    if mode == "policy-only" and records:
+        raise ValueError("policy-only member plan forbids runtime records")
     _require_exact_manifest_trees(plugin_path)
     differences = skill_tree_differences(plugin_path / "skills", SPECS_DIR)
     if differences:
@@ -415,39 +516,51 @@ def _member_paths(plugin_path: Path, *, mode: str) -> list[Path]:
         Path("skills"),
         *(Path("skills") / relative for relative in expected_skill_relpaths(SPECS_DIR)),
     ]
+    members: dict[str, Path | None] = {}
     if mode == "activation":
         _require_exact_third_party_notice_tree(plugin_path)
-        manifest = _load_manifest(plugin_path)
-        records = runtime_bundle.validate_file_records(
-            manifest["artifacts"][0]["files"]
-        )
-        relatives.extend(
-            (
-                Path("runtime"),
-                Path("runtime/darwin-arm64"),
-                RUNTIME_BUNDLE_REL,
-                *(RUNTIME_BUNDLE_REL / record["path"] for record in records),
-                *ACTIVATION_THIRD_PARTY_MEMBERS,
-            )
-        )
-    members: dict[str, Path] = {}
+        relatives.extend(ACTIVATION_THIRD_PARTY_MEMBERS)
+        for name in _RUNTIME_DIR_MODES:
+            members[name] = None
+        for record in records:
+            members[(RUNTIME_BUNDLE_REL / record["path"]).as_posix()] = None
     for relative in relatives:
         source = plugin_path / relative
         _safe_source(source)
         members[relative.as_posix()] = source
-    return [members[name] for name in sorted(members)]
+    return [(name, members[name]) for name in sorted(members)]
 
 
-def _canonical_names(paths: Iterable[Path], plugin_path: Path) -> list[str]:
-    return [path.relative_to(plugin_path).as_posix() for path in paths]
+def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
+    """Verify one built archive without trusting any mutable runtime source.
 
+    Non-runtime members are byte/mode-parity-checked against the git-tracked
+    plugin tree (which CI additionally binds to the signed-tag tree). Runtime
+    members are validated ONLY against a single frozen snapshot of the
+    committed manifest — exact member sequence, type, mode, size, and sha256,
+    plus the archived manifest bytes themselves — so a source-tree swap after
+    the build-time validation can never produce a passing archive, and no
+    runtime bundle source is needed to verify (CI runs this without one).
+    """
 
-def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
     plugin_path = plugin_path.resolve(strict=True)
     if classify_package(plugin_path) != mode:
         raise ValueError("archive mode no longer matches the source package")
-    expected_paths = _member_paths(plugin_path, mode=mode)
-    expected_names = _canonical_names(expected_paths, plugin_path)
+    frozen_manifest = b""
+    records: tuple[dict[str, object], ...] = ()
+    if mode == "activation":
+        frozen_manifest = _read_manifest_bytes(plugin_path)
+        manifest = _parse_manifest(frozen_manifest)
+        artifacts = manifest["artifacts"]
+        if len(artifacts) != 1:
+            raise ValueError("activation package requires exactly one runtime artifact")
+        records = runtime_bundle.validate_file_records(artifacts[0]["files"])
+    plan = _member_plan(plugin_path, mode=mode, records=records)
+    expected_names = [name for name, _ in plan]
+    record_by_name = {
+        (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record
+        for record in records
+    }
     try:
         bundle = tarfile.open(archive_path, "r:gz")
     except (OSError, tarfile.TarError) as exc:
@@ -463,7 +576,33 @@ def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
                 raise ValueError("archive contains an unsafe member path")
         if names != expected_names:
             raise ValueError("archive member list is not canonical")
-        for source, member in zip(expected_paths, members, strict=True):
+        for (name, source), member in zip(plan, members, strict=True):
+            if source is None:
+                directory_mode = _RUNTIME_DIR_MODES.get(name)
+                if directory_mode is not None:
+                    if not member.isdir() or stat.S_IMODE(member.mode) != directory_mode:
+                        raise ValueError(
+                            f"archive runtime directory identity failed: {member.name}"
+                        )
+                    continue
+                record = record_by_name[name]
+                if (
+                    not member.isreg()
+                    or stat.S_IMODE(member.mode) != record["install_mode"]
+                    or member.size != record["size"]
+                ):
+                    raise ValueError(
+                        f"archive runtime member identity failed: {member.name}"
+                    )
+                archived = bundle.extractfile(member)
+                if (
+                    archived is None
+                    or hashlib.sha256(archived.read()).hexdigest() != record["sha256"]
+                ):
+                    raise ValueError(
+                        f"archive runtime member digest failed: {member.name}"
+                    )
+                continue
             source_info = _safe_source(source)
             source_is_file = stat.S_ISREG(source_info.st_mode)
             if source_is_file != member.isfile() or (
@@ -474,51 +613,153 @@ def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
                 raise ValueError(f"archive member mode parity failed: {member.name}")
             if source_is_file:
                 archived = bundle.extractfile(member)
-                if archived is None or archived.read() != source.read_bytes():
+                if archived is None:
+                    raise ValueError(f"archive member byte parity failed: {member.name}")
+                data = archived.read()
+                if name == "runtime-manifest.json" and mode == "activation":
+                    # Bind the archived manifest to the SAME frozen bytes the
+                    # runtime records above were parsed from (never a second
+                    # read) — closes the manifest-swap window.
+                    if data != frozen_manifest:
+                        raise ValueError(
+                            "archived runtime manifest does not match the frozen manifest"
+                        )
+                elif data != source.read_bytes():
                     raise ValueError(f"archive member byte parity failed: {member.name}")
         runtime_files = [
             member.name
             for member in members
             if member.isfile() and member.name.startswith("runtime/")
         ]
-        expected_runtime = (
-            [
-                (RUNTIME_BUNDLE_REL / record["path"]).as_posix()
-                for record in runtime_bundle.validate_file_records(
-                    _load_manifest(plugin_path)["artifacts"][0]["files"]
-                )
-            ]
-            if mode == "activation"
-            else []
-        )
+        expected_runtime = [
+            (RUNTIME_BUNDLE_REL / record["path"]).as_posix() for record in records
+        ]
         if runtime_files != expected_runtime:
             raise ValueError("archive runtime membership does not match release mode")
 
 
-def build_archive(root: Path, *, plugin: str, output: Path) -> str:
+def _synthesized_tarinfo(name: str, *, mode: int, size: int = 0, directory: bool = False) -> tarfile.TarInfo:
+    info = tarfile.TarInfo(name)
+    info.type = tarfile.DIRTYPE if directory else tarfile.REGTYPE
+    info.mode = mode
+    info.size = size
+    info.uid = 0
+    info.gid = 0
+    info.uname = ""
+    info.gname = ""
+    info.mtime = 0
+    return info
+
+
+def build_archive(
+    root: Path,
+    *,
+    plugin: str,
+    output: Path,
+    bundle_source: Path | None = None,
+) -> str:
     if plugin != PLUGIN_NAME:
         raise ValueError("agent-collab is the only releaseable plugin")
     plugin_path = (root / "plugins" / plugin).resolve(strict=True)
     mode = classify_package(plugin_path)
-    members = _member_paths(plugin_path, mode=mode)
+    records: tuple[dict[str, object], ...] = ()
+    bundle_leaf: Path | None = None
+    if mode == "policy-only":
+        # The committed manifest — never the flag — decides the mode.
+        if bundle_source is not None:
+            raise ValueError("policy-only manifest forbids --bundle-source")
+    else:
+        # Fail closed: an activation manifest without a matching signed bundle
+        # source is a hard error, never a silent policy-only downgrade.
+        if bundle_source is None:
+            raise ValueError("activation manifest requires --bundle-source")
+        runtime_root = plugin_path / "runtime"
+        if runtime_root.exists() or runtime_root.is_symlink():
+            raise ValueError(
+                "in-tree runtime conflicts with --bundle-source; remove one source"
+            )
+        try:
+            raw_source = bundle_source.lstat()
+        except OSError as exc:
+            raise ValueError("runtime bundle source is missing") from exc
+        # lstat the raw CLI argument BEFORE resolving — resolve() would erase
+        # the evidence that the argument itself was a symlink.
+        if stat.S_ISLNK(raw_source.st_mode) or not stat.S_ISDIR(raw_source.st_mode):
+            raise ValueError("runtime bundle source is unsafe")
+        bundle_leaf = bundle_source.resolve(strict=True)
+        manifest = _parse_manifest(_read_manifest_bytes(plugin_path))
+        artifacts = manifest["artifacts"]
+        if len(artifacts) != 1:
+            raise ValueError("activation package requires exactly one runtime artifact")
+        records = runtime_bundle.validate_file_records(artifacts[0]["files"])
+        _validate_activation_bundle_tree(bundle_leaf, records)
+    plan = _member_plan(plugin_path, mode=mode, records=records)
+    record_by_name = {
+        (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records
+    }
     output.parent.mkdir(parents=True, exist_ok=True)
-    with output.open("wb") as raw:
-        with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed:
-            with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as bundle:
-                for source in members:
-                    name = source.relative_to(plugin_path).as_posix()
-                    info = bundle.gettarinfo(str(source), arcname=name)
-                    info.uid = 0
-                    info.gid = 0
-                    info.uname = ""
-                    info.gname = ""
-                    info.mtime = 0
-                    if info.isfile():
-                        with source.open("rb") as stream:
-                            bundle.addfile(info, stream)
-                    else:
-                        bundle.addfile(info)
-    verify_archive(plugin_path, output, mode=mode)
+    output_resolved = output.parent.resolve(strict=True) / output.name
+    for forbidden in (plugin_path, bundle_leaf):
+        if forbidden is not None and output_resolved.is_relative_to(forbidden):
+            raise ValueError("archive output must not alias a source tree")
+    # Build into an exclusive temporary file, verify THAT, and only then move
+    # it to the requested destination — a failed build or failed verification
+    # never leaves a publishable artifact at the output path.
+    temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}"
+    member_flags = (
+        os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+    )
+    try:
+        descriptor = os.open(
+            temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644
+        )
+        with os.fdopen(descriptor, "wb") as raw:
+            with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed:
+                with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as bundle:
+                    for name, source in plan:
+                        if source is not None:
+                            info = bundle.gettarinfo(str(source), arcname=name)
+                            info.uid = 0
+                            info.gid = 0
+                            info.uname = ""
+                            info.gname = ""
+                            info.mtime = 0
+                            if info.isfile():
+                                with source.open("rb") as stream:
+                                    bundle.addfile(info, stream)
+                            else:
+                                bundle.addfile(info)
+                        elif name in _RUNTIME_DIR_MODES:
+                            bundle.addfile(
+                                _synthesized_tarinfo(
+                                    name, mode=_RUNTIME_DIR_MODES[name], directory=True
+                                )
+                            )
+                        else:
+                            # Runtime file member: metadata comes ENTIRELY from
+                            # the frozen manifest record; only the bytes stream
+                            # from the validated source (and verify_archive
+                            # re-binds those bytes to the record digest).
+                            record = record_by_name[name]
+                            info = _synthesized_tarinfo(
+                                name,
+                                mode=record["install_mode"],
+                                size=record["size"],
+                            )
+                            assert bundle_leaf is not None
+                            member_descriptor = os.open(
+                                bundle_leaf / record["path"], member_flags
+                            )
+                            with os.fdopen(member_descriptor, "rb") as stream:
+                                bundle.addfile(info, stream)
+        verify_archive(plugin_path, temp_path, mode=mode)
+        os.replace(temp_path, output_resolved)
+    except BaseException:
+        try:
+            os.unlink(temp_path)
+        except OSError:
+            pass
+        raise
     return mode
 
 
@@ -528,6 +769,16 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--plugin", default=PLUGIN_NAME)
     parser.add_argument("--output", type=Path)
     parser.add_argument("--print-mode", action="store_true")
+    parser.add_argument(
+        "--bundle-source",
+        type=Path,
+        default=None,
+        help=(
+            "path to the signed runtime bundle leaf directory "
+            "(…/agent-collab-runtime.bundle) from the local release handoff; "
+            "required for an activation manifest, forbidden for policy-only"
+        ),
+    )
     args = parser.parse_args(argv)
     try:
         if args.plugin != PLUGIN_NAME:
@@ -540,7 +791,12 @@ def main(argv: list[str] | None = None) -> int:
             return 0
         if args.output is None:
             raise ValueError("--output is required when building an archive")
-        mode = build_archive(args.repo_root, plugin=args.plugin, output=args.output)
+        mode = build_archive(
+            args.repo_root,
+            plugin=args.plugin,
+            output=args.output,
+            bundle_source=args.bundle_source,
+        )
     except (OSError, ValueError) as exc:
         print(f"FAIL: {exc}")
         return 1
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index d3674b0..6e42b91 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -8,10 +8,12 @@
 import shutil
 import stat
 import sys
+import os
 import tarfile
 import tempfile
 import unittest
 from pathlib import Path
+from unittest import mock
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -69,14 +71,19 @@ def _install_third_party_notices(self) -> None:
             self.assertTrue((licenses / name).is_file())
 
     def _activate(self) -> Path:
+        """Stage an activation manifest in-tree + a signed bundle OUT-of-tree.
+
+        The runtime bundle ships as a release asset (never committed), so the
+        fixture mirrors production: the committed tree carries only the
+        manifest, and the 0o500 bundle leaf lives in an external handoff
+        directory supplied to the builder via ``bundle_source``.
+        """
+
         self._install_third_party_notices()
-        runtime = (
-            self.plugin
-            / "runtime"
-            / "darwin-arm64"
-            / "agent-collab-runtime.bundle"
-            / "agent-collab-runtime"
+        self.bundle_leaf = (
+            self.root / "handoff" / "agent-collab-runtime.bundle"
         )
+        runtime = self.bundle_leaf / "agent-collab-runtime"
         runtime.parent.mkdir(parents=True)
         runtime.write_bytes(b"signed-runtime-fixture")
         library = runtime.parent / "libpython3.13.dylib"
@@ -206,10 +213,14 @@ def test_packaged_legal_files_match_repository_canonicals(self) -> None:
     def test_activation_archive_has_exactly_one_runtime_with_byte_mode_parity(self) -> None:
         runtime = self._activate()
         mode = self.builder.build_archive(
-            self.root, plugin="agent-collab", output=self.archive
+            self.root,
+            plugin="agent-collab",
+            output=self.archive,
+            bundle_source=self.bundle_leaf,
         )
         self.assertEqual(mode, "activation")
 
+        bundle_prefix = "runtime/darwin-arm64/agent-collab-runtime.bundle/"
         with tarfile.open(self.archive, "r:gz") as bundle:
             runtime_files = [
                 member
@@ -219,18 +230,37 @@ def test_activation_archive_has_exactly_one_runtime_with_byte_mode_parity(self)
             self.assertEqual(
                 [member.name for member in runtime_files],
                 [
-                    "runtime/darwin-arm64/agent-collab-runtime.bundle/agent-collab-runtime",
-                    "runtime/darwin-arm64/agent-collab-runtime.bundle/libpython3.13.dylib",
+                    bundle_prefix + "agent-collab-runtime",
+                    bundle_prefix + "libpython3.13.dylib",
                 ],
             )
             member = runtime_files[0]
             self.assertEqual(bundle.extractfile(member).read(), runtime.read_bytes())
             self.assertEqual(stat.S_IMODE(member.mode), 0o500)
 
+            # Synthesized runtime directory scaffolding carries fixed canonical
+            # modes: 0o755 traversal parents + the sealed 0o500 bundle leaf.
+            directory_modes = {
+                archived.name: stat.S_IMODE(archived.mode)
+                for archived in bundle.getmembers()
+                if archived.isdir() and archived.name.startswith("runtime")
+            }
+            self.assertEqual(
+                directory_modes,
+                {
+                    "runtime": 0o755,
+                    "runtime/darwin-arm64": 0o755,
+                    "runtime/darwin-arm64/agent-collab-runtime.bundle": 0o500,
+                },
+            )
+
             for archived in bundle.getmembers():
                 if not archived.isfile():
                     continue
-                source = self.plugin / archived.name
+                if archived.name.startswith(bundle_prefix):
+                    source = self.bundle_leaf / archived.name[len(bundle_prefix):]
+                else:
+                    source = self.plugin / archived.name
                 self.assertEqual(bundle.extractfile(archived).read(), source.read_bytes())
                 self.assertEqual(
                     stat.S_IMODE(archived.mode),
@@ -240,7 +270,10 @@ def test_activation_archive_has_exactly_one_runtime_with_byte_mode_parity(self)
     def test_activation_archive_includes_canonical_third_party_notice_tree(self) -> None:
         self._activate()
         mode = self.builder.build_archive(
-            self.root, plugin="agent-collab", output=self.archive
+            self.root,
+            plugin="agent-collab",
+            output=self.archive,
+            bundle_source=self.bundle_leaf,
         )
         self.assertEqual(mode, "activation")
 
@@ -278,7 +311,10 @@ def test_activation_rejects_missing_third_party_notice_member(self) -> None:
 
         with self.assertRaisesRegex(ValueError, "third-party notice tree"):
             self.builder.build_archive(
-                self.root, plugin="agent-collab", output=self.archive
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=self.bundle_leaf,
             )
 
     def test_activation_rejects_missing_top_level_third_party_notice(self) -> None:
@@ -287,7 +323,10 @@ def test_activation_rejects_missing_top_level_third_party_notice(self) -> None:
 
         with self.assertRaisesRegex(ValueError, "third-party notice tree"):
             self.builder.build_archive(
-                self.root, plugin="agent-collab", output=self.archive
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=self.bundle_leaf,
             )
 
     def test_activation_rejects_unexpected_third_party_notice_member(self) -> None:
@@ -298,7 +337,10 @@ def test_activation_rejects_unexpected_third_party_notice_member(self) -> None:
 
         with self.assertRaisesRegex(ValueError, "third-party notice tree"):
             self.builder.build_archive(
-                self.root, plugin="agent-collab", output=self.archive
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=self.bundle_leaf,
             )
 
     def test_activation_rejects_third_party_notice_content_drift(self) -> None:
@@ -308,7 +350,10 @@ def test_activation_rejects_third_party_notice_content_drift(self) -> None:
 
         with self.assertRaisesRegex(ValueError, "content digest"):
             self.builder.build_archive(
-                self.root, plugin="agent-collab", output=self.archive
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=self.bundle_leaf,
             )
 
     def test_activation_rejects_symlinked_third_party_notice_member(self) -> None:
@@ -319,7 +364,10 @@ def test_activation_rejects_symlinked_third_party_notice_member(self) -> None:
 
         with self.assertRaisesRegex(ValueError, "unsafe|third-party notice tree"):
             self.builder.build_archive(
-                self.root, plugin="agent-collab", output=self.archive
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=self.bundle_leaf,
             )
 
     def test_policy_only_rejects_unadvertised_runtime_and_missing_policy(self) -> None:
@@ -368,6 +416,194 @@ def test_archive_fails_closed_when_runtime_setup_entrypoint_is_missing(self) ->
                 self.root, plugin="agent-collab", output=self.archive
             )
 
+    def _build_activation(self) -> str:
+        return self.builder.build_archive(
+            self.root,
+            plugin="agent-collab",
+            output=self.archive,
+            bundle_source=self.bundle_leaf,
+        )
+
+    def test_activation_manifest_requires_bundle_source_fail_closed(self) -> None:
+        self._activate()
+        with self.assertRaisesRegex(ValueError, "requires --bundle-source"):
+            self.builder.build_archive(
+                self.root, plugin="agent-collab", output=self.archive
+            )
+        self.assertFalse(self.archive.exists())
+
+    def test_policy_only_manifest_forbids_bundle_source(self) -> None:
+        self._install_third_party_notices()
+        stray = self.root / "stray-bundle"
+        stray.mkdir()
+        with self.assertRaisesRegex(ValueError, "forbids --bundle-source"):
+            self.builder.build_archive(
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=stray,
+            )
+
+    def test_in_tree_runtime_conflicts_with_bundle_source(self) -> None:
+        self._activate()
+        (self.plugin / "runtime").mkdir()
+        with self.assertRaisesRegex(ValueError, "in-tree runtime conflicts"):
+            self._build_activation()
+
+    def test_bundle_source_rejects_symlink_argument_before_resolve(self) -> None:
+        self._activate()
+        alias = self.root / "leaf-alias"
+        alias.symlink_to(self.bundle_leaf)
+        with self.assertRaisesRegex(ValueError, "runtime bundle source is unsafe"):
+            self.builder.build_archive(
+                self.root,
+                plugin="agent-collab",
+                output=self.archive,
+                bundle_source=alias,
+            )
+
+    def test_bundle_source_rejects_content_and_size_drift(self) -> None:
+        self._activate()
+        library = self.bundle_leaf / "libpython3.13.dylib"
+        original = library.read_bytes()
+
+        library.chmod(0o700)
+        library.write_bytes(b"X" * len(original))  # same size, wrong bytes
+        library.chmod(0o500)
+        with self.assertRaisesRegex(ValueError, "digest is invalid"):
+            self._build_activation()
+
+        library.chmod(0o700)
+        library.write_bytes(original + b"tail")  # size drift
+        library.chmod(0o500)
+        with self.assertRaisesRegex(ValueError, "member identity is invalid"):
+            self._build_activation()
+        self.assertFalse(self.archive.exists())
+
+    def test_bundle_source_rejects_mode_drift(self) -> None:
+        self._activate()
+        (self.bundle_leaf / "libpython3.13.dylib").chmod(0o755)
+        with self.assertRaisesRegex(ValueError, "member identity is invalid"):
+            self._build_activation()
+
+        (self.bundle_leaf / "libpython3.13.dylib").chmod(0o500)
+        self.bundle_leaf.chmod(0o755)
+        with self.assertRaisesRegex(ValueError, "root identity is invalid"):
+            self._build_activation()
+
+    def test_bundle_source_requires_exact_membership(self) -> None:
+        self._activate()
+        self.bundle_leaf.chmod(0o700)
+        extra = self.bundle_leaf / "extra-member"
+        extra.write_bytes(b"unexpected")
+        self.bundle_leaf.chmod(0o500)
+        with self.assertRaisesRegex(ValueError, "membership is not exact"):
+            self._build_activation()
+
+        self.bundle_leaf.chmod(0o700)
+        extra.unlink()
+        (self.bundle_leaf / "libpython3.13.dylib").unlink()
+        self.bundle_leaf.chmod(0o500)
+        with self.assertRaisesRegex(ValueError, "membership is not exact"):
+            self._build_activation()
+
+    def test_bundle_source_rejects_symlinked_member(self) -> None:
+        self._activate()
+        self.bundle_leaf.chmod(0o700)
+        library = self.bundle_leaf / "libpython3.13.dylib"
+        library.unlink()
+        library.symlink_to("agent-collab-runtime")
+        self.bundle_leaf.chmod(0o500)
+        with self.assertRaisesRegex(
+            ValueError, "member is unsafe|member identity is invalid"
+        ):
+            self._build_activation()
+
+    def test_bundle_source_rejects_hardlinked_member(self) -> None:
+        self._activate()
+        os.link(
+            self.bundle_leaf / "agent-collab-runtime",
+            self.root / "hardlink-aside",
+        )
+        with self.assertRaisesRegex(ValueError, "member identity is invalid"):
+            self._build_activation()
+
+    def test_failed_verification_leaves_no_output_artifact(self) -> None:
+        self._activate()
+        with mock.patch.object(
+            self.builder, "verify_archive", side_effect=ValueError("forced failure")
+        ):
+            with self.assertRaisesRegex(ValueError, "forced failure"):
+                self._build_activation()
+        self.assertFalse(self.archive.exists())
+        leftovers = [
+            path.name
+            for path in self.archive.parent.iterdir()
+            if ".tmp" in path.name
+        ]
+        self.assertEqual(leftovers, [])
+
+    def test_verify_archive_binds_runtime_bytes_to_frozen_manifest(self) -> None:
+        self._activate()
+        mode = self._build_activation()
+        target = "runtime/darwin-arm64/agent-collab-runtime.bundle/agent-collab-runtime"
+
+        # Repack the archive swapping the runtime member's bytes for same-size
+        # garbage while keeping every member's metadata identical: source
+        # parity cannot notice (the source tree is untouched), only the frozen
+        # manifest digest binding can.
+        tampered = self.root / "tampered.plugin"
+        with tarfile.open(self.archive, "r:gz") as original:
+            members = [
+                (member, original.extractfile(member).read() if member.isfile() else None)
+                for member in original.getmembers()
+            ]
+        import gzip as gzip_module
+
+        with tampered.open("wb") as raw:
+            with gzip_module.GzipFile(
+                filename="", mode="wb", fileobj=raw, mtime=0
+            ) as compressed:
+                with tarfile.open(
+                    fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT
+                ) as bundle:
+                    for member, payload in members:
+                        if member.name == target:
+                            payload = b"Y" * member.size
+                        if payload is None:
+                            bundle.addfile(member)
+                        else:
+                            import io
+
+                            bundle.addfile(member, io.BytesIO(payload))
+
+        with self.assertRaisesRegex(ValueError, "runtime member digest failed"):
+            self.builder.verify_archive(self.plugin, tampered, mode=mode)
+
+    def test_verify_archive_binds_archived_manifest_to_frozen_bytes(self) -> None:
+        self._activate()
+        mode = self._build_activation()
+        manifest_path = self.plugin / "runtime-manifest.json"
+        # A parse-identical byte change (trailing whitespace) after the build:
+        # the archived manifest must equal the CURRENT frozen bytes exactly.
+        manifest_path.write_text(
+            manifest_path.read_text(encoding="utf-8") + " ", encoding="utf-8"
+        )
+        with self.assertRaisesRegex(ValueError, "frozen manifest"):
+            self.builder.verify_archive(self.plugin, self.archive, mode=mode)
+
+    def test_activation_archive_build_is_deterministic(self) -> None:
+        self._activate()
+        self._build_activation()
+        second = self.root / "agent-collab-second.plugin"
+        self.builder.build_archive(
+            self.root,
+            plugin="agent-collab",
+            output=second,
+            bundle_source=self.bundle_leaf,
+        )
+        self.assertEqual(self.archive.read_bytes(), second.read_bytes())
+
     def test_archive_rejects_unexpected_manifest_and_skill_members(self) -> None:
         extra_manifest = self.plugin / ".claude-plugin" / "extra.dat"
         extra_manifest.write_text("unexpected", encoding="utf-8")
diff --git a/tests/test_release_evidence.py b/tests/test_release_evidence.py
index be343a1..7e12618 100644
--- a/tests/test_release_evidence.py
+++ b/tests/test_release_evidence.py
@@ -102,12 +102,9 @@ def _build_activation(self):
         )
         for name in LEGAL_FILES:
             shutil.copy2(ROOT / name, repo / name)
-        bundle = (
-            plugin
-            / "runtime"
-            / "darwin-arm64"
-            / "agent-collab-runtime.bundle"
-        )
+        # The signed bundle ships as a release asset (never committed): the
+        # fixture stages it OUT-of-tree and supplies it via bundle_source.
+        bundle = self.root / "handoff" / "agent-collab-runtime.bundle"
         bundle.mkdir(parents=True)
         runtime_members = {
             "agent-collab-runtime": (
@@ -192,7 +189,10 @@ def _build_activation(self):
             json.dumps(manifest), encoding="utf-8"
         )
         mode = archive_builder.build_archive(
-            repo, plugin="agent-collab", output=self.archive
+            repo,
+            plugin="agent-collab",
+            output=self.archive,
+            bundle_source=bundle,
         )
         self.assertEqual(mode, "activation")
         evidence_builder.build_evidence(

From 1f684121d3cd94cbcfe802a6dc821190bd937c25 Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 20:14:58 -0700
Subject: [PATCH 2/9] fix(archive): true single-frozen-snapshot build,
 pack-time fd revalidation, canonical metadata verify
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex peer-review round 1 (REQUEST_CHANGES/H) on the --bundle-source
implementation; all 5 concerns integrated:

1. Single frozen snapshot now actually spans the build: the manifest is
   read ONCE through an O_NOFOLLOW fstat-checked descriptor; those exact
   bytes drive record validation, are embedded into the archive verbatim
   (io.BytesIO, never a pack-time re-read of the mutable checkout), and are
   handed to verify_archive via a new optional frozen_manifest parameter
   (standalone/CI callers omit it and read once from the checkout — the
   build_release_evidence.py call remains unchanged).
2. Pack-time revalidation of the OPEN runtime descriptor (fstat: regular
   file, uid, nlink 1, mode, size) with O_NONBLOCK so a post-validation
   FIFO swap cannot block the open; byte content stays re-bound to the
   manifest digest by verify_archive.
3. Temp cleanup only unlinks a temp file THIS invocation created
   (temp_created tracking — a pre-existing path makes O_EXCL fail without
   destroying the foreign file); output/source alias check runs BEFORE
   mkdir and re-checks the strictly resolved path after.
4. verify_archive enforces canonical metadata on EVERY member (regular
   file/dir only, uid/gid 0, empty uname/gname, mtime 0, no linkname, no
   PAX headers) and bounds every byte read with a size-parity check before
   extraction (hostile-archive decompression bound).
5. assert → explicit ValueError (survives python -O).

New tests: non-canonical mtime rejected; foreign temp file preserved on
O_EXCL collision; same-size manifest swap caught by the frozen-byte
binding + size-changing swap caught by size parity; same-size source drift
caught by byte parity + size drift caught bounded. 329 tests/ + 254
scripts/ green; CLI smoke-verified.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 158 ++++++++++++++++++++++++++++----
 tests/test_plugin_archive.py    |  77 ++++++++++++++--
 2 files changed, 212 insertions(+), 23 deletions(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index b960058..a2d7d64 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -7,6 +7,7 @@
 import gzip
 import hashlib
 import importlib.util
+import io
 import json
 import os
 import re
@@ -174,16 +175,43 @@ def _safe_source(path: Path) -> os.stat_result:
 
 
 def _read_manifest_bytes(plugin_path: Path) -> bytes:
-    """Read the committed manifest exactly once for a frozen-snapshot check."""
+    """Read the committed manifest exactly once for a frozen-snapshot check.
+
+    The read goes through a single O_NOFOLLOW descriptor whose fstat is
+    checked, so the snapshot is one coherent read of one regular file — a
+    concurrent swap yields either the old bytes or the new bytes, never a
+    mixture, and everything downstream derives from the returned snapshot.
+    """
 
     manifest_path = plugin_path / "runtime-manifest.json"
     info = _safe_source(manifest_path)
     if not stat.S_ISREG(info.st_mode):
         raise ValueError("runtime-manifest.json is not a regular file")
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+    try:
+        descriptor = os.open(manifest_path, flags)
+    except OSError as exc:
+        raise ValueError("runtime manifest is unreadable") from exc
     try:
-        return manifest_path.read_bytes()
+        described = os.fstat(descriptor)
+        if not stat.S_ISREG(described.st_mode):
+            raise ValueError("runtime-manifest.json is not a regular file")
+        chunks = []
+        while True:
+            chunk = os.read(descriptor, 1024 * 1024)
+            if not chunk:
+                break
+            chunks.append(chunk)
+            if sum(len(part) for part in chunks) > MAX_ARTIFACT_BYTES:
+                raise ValueError("runtime manifest is unreasonably large")
+        return b"".join(chunks)
     except OSError as exc:
         raise ValueError("runtime manifest is unreadable") from exc
+    finally:
+        try:
+            os.close(descriptor)
+        except OSError:
+            pass
 
 
 def _parse_manifest(data: bytes) -> dict[str, object]:
@@ -531,7 +559,13 @@ def _member_plan(
     return [(name, members[name]) for name in sorted(members)]
 
 
-def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
+def verify_archive(
+    plugin_path: Path,
+    archive_path: Path,
+    *,
+    mode: str,
+    frozen_manifest: bytes | None = None,
+) -> None:
     """Verify one built archive without trusting any mutable runtime source.
 
     Non-runtime members are byte/mode-parity-checked against the git-tracked
@@ -541,20 +575,27 @@ def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
     plus the archived manifest bytes themselves — so a source-tree swap after
     the build-time validation can never produce a passing archive, and no
     runtime bundle source is needed to verify (CI runs this without one).
+
+    ``frozen_manifest`` lets `build_archive` hand over the EXACT byte snapshot
+    it validated, packed, and embedded, so build+verify share one snapshot and
+    a mid-build manifest swap on disk is irrelevant. Standalone callers (CI)
+    omit it and the snapshot is read once from the checkout here.
     """
 
     plugin_path = plugin_path.resolve(strict=True)
     if classify_package(plugin_path) != mode:
         raise ValueError("archive mode no longer matches the source package")
-    frozen_manifest = b""
     records: tuple[dict[str, object], ...] = ()
     if mode == "activation":
-        frozen_manifest = _read_manifest_bytes(plugin_path)
+        if frozen_manifest is None:
+            frozen_manifest = _read_manifest_bytes(plugin_path)
         manifest = _parse_manifest(frozen_manifest)
         artifacts = manifest["artifacts"]
         if len(artifacts) != 1:
             raise ValueError("activation package requires exactly one runtime artifact")
         records = runtime_bundle.validate_file_records(artifacts[0]["files"])
+    else:
+        frozen_manifest = None
     plan = _member_plan(plugin_path, mode=mode, records=records)
     expected_names = [name for name, _ in plan]
     record_by_name = {
@@ -576,6 +617,26 @@ def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
                 raise ValueError("archive contains an unsafe member path")
         if names != expected_names:
             raise ValueError("archive member list is not canonical")
+        for member in members:
+            # Canonical archive metadata is part of the contract for EVERY
+            # member: only regular files and directories, normalized
+            # ownership, zero mtime, no link targets, and no PAX extension
+            # headers (canonical members never need any). This also rejects
+            # hostile archives that smuggle devices/links or oversized PAX
+            # payloads past the name checks.
+            if (
+                not (member.isfile() or member.isdir())
+                or member.uid != 0
+                or member.gid != 0
+                or member.uname != ""
+                or member.gname != ""
+                or member.mtime != 0
+                or member.linkname != ""
+                or member.pax_headers
+            ):
+                raise ValueError(
+                    f"archive member metadata is not canonical: {member.name}"
+                )
         for (name, source), member in zip(plan, members, strict=True):
             if source is None:
                 directory_mode = _RUNTIME_DIR_MODES.get(name)
@@ -612,6 +673,18 @@ def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None:
             if stat.S_IMODE(member.mode) != stat.S_IMODE(source_info.st_mode):
                 raise ValueError(f"archive member mode parity failed: {member.name}")
             if source_is_file:
+                # Bound the read BEFORE extracting: the declared member size
+                # must already match the expected payload exactly, so a
+                # hostile archive cannot force an unbounded decompression.
+                expected_size = (
+                    len(frozen_manifest)
+                    if name == "runtime-manifest.json" and frozen_manifest is not None
+                    else source_info.st_size
+                )
+                if member.size != expected_size:
+                    raise ValueError(
+                        f"archive member size parity failed: {member.name}"
+                    )
                 archived = bundle.extractfile(member)
                 if archived is None:
                     raise ValueError(f"archive member byte parity failed: {member.name}")
@@ -664,6 +737,7 @@ def build_archive(
     mode = classify_package(plugin_path)
     records: tuple[dict[str, object], ...] = ()
     bundle_leaf: Path | None = None
+    frozen_manifest: bytes | None = None
     if mode == "policy-only":
         # The committed manifest — never the flag — decides the mode.
         if bundle_source is not None:
@@ -687,7 +761,11 @@ def build_archive(
         if stat.S_ISLNK(raw_source.st_mode) or not stat.S_ISDIR(raw_source.st_mode):
             raise ValueError("runtime bundle source is unsafe")
         bundle_leaf = bundle_source.resolve(strict=True)
-        manifest = _parse_manifest(_read_manifest_bytes(plugin_path))
+        # Freeze the manifest ONCE: these exact bytes drive the bundle-tree
+        # validation, are embedded into the archive verbatim, and are handed
+        # to verify_archive — a mid-build manifest swap on disk is irrelevant.
+        frozen_manifest = _read_manifest_bytes(plugin_path)
+        manifest = _parse_manifest(frozen_manifest)
         artifacts = manifest["artifacts"]
         if len(artifacts) != 1:
             raise ValueError("activation package requires exactly one runtime artifact")
@@ -697,22 +775,34 @@ def build_archive(
     record_by_name = {
         (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records
     }
+    def _reject_source_alias(candidate: Path) -> None:
+        for forbidden in (plugin_path, bundle_leaf):
+            if forbidden is not None and candidate.is_relative_to(forbidden):
+                raise ValueError("archive output must not alias a source tree")
+
+    # Reject a source-tree destination BEFORE creating any directory (mkdir
+    # must never mutate a rejected destination), then re-check the strictly
+    # resolved path once the parent exists.
+    _reject_source_alias(output.resolve())
     output.parent.mkdir(parents=True, exist_ok=True)
     output_resolved = output.parent.resolve(strict=True) / output.name
-    for forbidden in (plugin_path, bundle_leaf):
-        if forbidden is not None and output_resolved.is_relative_to(forbidden):
-            raise ValueError("archive output must not alias a source tree")
+    _reject_source_alias(output_resolved)
     # Build into an exclusive temporary file, verify THAT, and only then move
     # it to the requested destination — a failed build or failed verification
     # never leaves a publishable artifact at the output path.
     temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}"
     member_flags = (
-        os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+        os.O_RDONLY
+        | getattr(os, "O_NOFOLLOW", 0)
+        | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NONBLOCK", 0)
     )
+    temp_created = False
     try:
         descriptor = os.open(
             temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644
         )
+        temp_created = True
         with os.fdopen(descriptor, "wb") as raw:
             with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed:
                 with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as bundle:
@@ -724,7 +814,16 @@ def build_archive(
                             info.uname = ""
                             info.gname = ""
                             info.mtime = 0
-                            if info.isfile():
+                            if (
+                                name == "runtime-manifest.json"
+                                and frozen_manifest is not None
+                            ):
+                                # Embed the EXACT frozen snapshot the runtime
+                                # records were validated against — never a
+                                # re-read of the mutable checkout.
+                                info.size = len(frozen_manifest)
+                                bundle.addfile(info, io.BytesIO(frozen_manifest))
+                            elif info.isfile():
                                 with source.open("rb") as stream:
                                     bundle.addfile(info, stream)
                             else:
@@ -746,19 +845,44 @@ def build_archive(
                                 mode=record["install_mode"],
                                 size=record["size"],
                             )
-                            assert bundle_leaf is not None
+                            if bundle_leaf is None:
+                                raise ValueError(
+                                    "runtime member requires a bundle source"
+                                )
                             member_descriptor = os.open(
                                 bundle_leaf / record["path"], member_flags
                             )
                             with os.fdopen(member_descriptor, "rb") as stream:
+                                # Revalidate the OPEN descriptor before
+                                # streaming: a post-validation swap to a
+                                # FIFO/device/hardlinked/resized entry is
+                                # rejected here (O_NONBLOCK keeps a swapped
+                                # FIFO from blocking the open), and the byte
+                                # content itself is re-bound to the manifest
+                                # digest by verify_archive below.
+                                streamed = os.fstat(stream.fileno())
+                                if (
+                                    not stat.S_ISREG(streamed.st_mode)
+                                    or streamed.st_uid != os.getuid()
+                                    or streamed.st_nlink != 1
+                                    or stat.S_IMODE(streamed.st_mode)
+                                    != record["install_mode"]
+                                    or streamed.st_size != record["size"]
+                                ):
+                                    raise ValueError(
+                                        "runtime bundle source member changed during packing"
+                                    )
                                 bundle.addfile(info, stream)
-        verify_archive(plugin_path, temp_path, mode=mode)
+        verify_archive(
+            plugin_path, temp_path, mode=mode, frozen_manifest=frozen_manifest
+        )
         os.replace(temp_path, output_resolved)
     except BaseException:
-        try:
-            os.unlink(temp_path)
-        except OSError:
-            pass
+        if temp_created:
+            try:
+                os.unlink(temp_path)
+            except OSError:
+                pass
         raise
     return mode
 
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index 6e42b91..be25d82 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -396,10 +396,19 @@ def test_archive_verifier_detects_source_byte_drift(self) -> None:
         mode = self.builder.build_archive(
             self.root, plugin="agent-collab", output=self.archive
         )
-        (self.plugin / "signing_policy.py").write_text("changed\n", encoding="utf-8")
+        policy = self.plugin / "signing_policy.py"
+        original = policy.read_bytes()
+
+        # Same-size drift is caught by byte parity …
+        policy.write_bytes(b"#" * len(original))
         with self.assertRaisesRegex(ValueError, "byte parity"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
+        # … and size drift is caught (bounded) before any byte read.
+        policy.write_bytes(original + b"\n# tail\n")
+        with self.assertRaisesRegex(ValueError, "size parity"):
+            self.builder.verify_archive(self.plugin, self.archive, mode=mode)
+
     def test_archive_size_limit_and_required_policy_member_are_canonical(self) -> None:
         self.assertEqual(self.builder.MAX_ARTIFACT_BYTES, 64 * 1024 * 1024)
         self.assertEqual(self.builder.RUNTIME_FILE_MODE, 0o500)
@@ -584,14 +593,70 @@ def test_verify_archive_binds_archived_manifest_to_frozen_bytes(self) -> None:
         self._activate()
         mode = self._build_activation()
         manifest_path = self.plugin / "runtime-manifest.json"
-        # A parse-identical byte change (trailing whitespace) after the build:
-        # the archived manifest must equal the CURRENT frozen bytes exactly.
-        manifest_path.write_text(
-            manifest_path.read_text(encoding="utf-8") + " ", encoding="utf-8"
-        )
+        # A SAME-SIZE, parse-valid, record-preserving byte change after the
+        # build (identity string tweak): size parity cannot notice, only the
+        # frozen-byte binding of the archived manifest can.
+        text = manifest_path.read_text(encoding="utf-8")
+        swapped = text.replace("Test Operator", "Test Operatox")
+        self.assertNotEqual(text, swapped)
+        self.assertEqual(len(text), len(swapped))
+        manifest_path.write_text(swapped, encoding="utf-8")
         with self.assertRaisesRegex(ValueError, "frozen manifest"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
+        # A size-changing swap fails closed too (bounded before any read).
+        manifest_path.write_text(text + " ", encoding="utf-8")
+        with self.assertRaisesRegex(ValueError, "size parity"):
+            self.builder.verify_archive(self.plugin, self.archive, mode=mode)
+
+    def test_verify_archive_rejects_non_canonical_member_metadata(self) -> None:
+        self._activate()
+        mode = self._build_activation()
+        target = "signing_policy.py"
+
+        # Repack with one member's mtime perturbed while bytes/modes stay
+        # identical: source parity cannot notice, only the canonical-metadata
+        # contract can.
+        tampered = self.root / "tampered-metadata.plugin"
+        import gzip as gzip_module
+        import io
+
+        with tarfile.open(self.archive, "r:gz") as original:
+            members = [
+                (member, original.extractfile(member).read() if member.isfile() else None)
+                for member in original.getmembers()
+            ]
+        with tampered.open("wb") as raw:
+            with gzip_module.GzipFile(
+                filename="", mode="wb", fileobj=raw, mtime=0
+            ) as compressed:
+                with tarfile.open(
+                    fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT
+                ) as bundle:
+                    for member, payload in members:
+                        if member.name == target:
+                            member.mtime = 1
+                        if payload is None:
+                            bundle.addfile(member)
+                        else:
+                            bundle.addfile(member, io.BytesIO(payload))
+
+        with self.assertRaisesRegex(ValueError, "metadata is not canonical"):
+            self.builder.verify_archive(self.plugin, tampered, mode=mode)
+
+    def test_failed_exclusive_temp_open_preserves_foreign_file(self) -> None:
+        self._activate()
+        # Pre-create the exact temp path this invocation would claim: the
+        # O_EXCL open must fail AND the foreign file must survive untouched.
+        foreign = (
+            self.archive.parent / f".{self.archive.name}.tmp.{os.getpid()}"
+        )
+        foreign.write_bytes(b"foreign artifact")
+        with self.assertRaises(OSError):
+            self._build_activation()
+        self.assertEqual(foreign.read_bytes(), b"foreign artifact")
+        self.assertFalse(self.archive.exists())
+
     def test_activation_archive_build_is_deterministic(self) -> None:
         self._activate()
         self._build_activation()

From 4a174bead4da3815ad038b2cabc98c268be2ca35 Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 20:34:48 -0700
Subject: [PATCH 3/9] fix(archive): bound hostile-archive decompression before
 tar parsing; GLM coverage tests

Codex peer-review round 2 closed concerns 1/2/3/5; the remaining #4 (H):
tarfile.open("r:gz") + getmembers() decompressed and parsed the ENTIRE
archive (PAX payloads included) before any canonical check ran, so a
crafted compressed asset could exhaust memory/time pre-validation. Fix:
_bounded_archive_stream() rejects a symlinked/oversized compressed file
(<= MAX_ARTIFACT_BYTES) and decompresses under a hard 2x MAX_ARTIFACT_BYTES
cap BEFORE tarfile ever parses a header; verify_archive then parses the
bounded in-memory bytes ("r:"). New test: a gzip bomb (tiny compressed,
>128 MiB decompressed) is rejected at the bound; a symlinked archive path
is refused outright.

Also integrates the GLM cross-check coverage concerns (PROCEED-WITH-
MODIFICATIONS/H on the round-1-fixed diff): tests for --output alias
rejection (in plugin tree, in bundle leaf, via symlinked parent),
pack-time-failure temp cleanup, and CLI --bundle-source wiring; comments
documenting the ustar <=100-char/ASCII name constraint and the residual
pid-suffix temp-path DoS note.

333 tests/ + 254 scripts/ green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 55 ++++++++++++++++++++--
 tests/test_plugin_archive.py    | 82 +++++++++++++++++++++++++++++++++
 2 files changed, 134 insertions(+), 3 deletions(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index a2d7d64..424e6c4 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -10,6 +10,7 @@
 import io
 import json
 import os
+from zlib import error as zlib_error
 import re
 import stat
 import tarfile
@@ -559,6 +560,45 @@ def _member_plan(
     return [(name, members[name]) for name in sorted(members)]
 
 
+# Hard resource bounds for reading a possibly-hostile archive: the compressed
+# file and its decompressed expansion are both capped BEFORE any tar/PAX
+# parsing happens, so a crafted asset cannot exhaust memory or time during
+# tarfile's own header processing.
+_MAX_COMPRESSED_ARCHIVE_BYTES = MAX_ARTIFACT_BYTES
+_MAX_DECOMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES
+
+
+def _bounded_archive_stream(archive_path: Path) -> io.BytesIO:
+    """Decompress an archive under hard byte caps before tar parsing."""
+
+    try:
+        info = archive_path.lstat()
+    except OSError as exc:
+        raise ValueError("plugin archive is unreadable") from exc
+    if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode):
+        raise ValueError("plugin archive is unreadable")
+    if info.st_size > _MAX_COMPRESSED_ARCHIVE_BYTES:
+        raise ValueError("plugin archive exceeds the canonical size bound")
+    chunks: list[bytes] = []
+    total = 0
+    try:
+        with archive_path.open("rb") as raw:
+            with gzip.GzipFile(fileobj=raw) as stream:
+                while True:
+                    chunk = stream.read(1024 * 1024)
+                    if not chunk:
+                        break
+                    total += len(chunk)
+                    if total > _MAX_DECOMPRESSED_ARCHIVE_BYTES:
+                        raise ValueError(
+                            "plugin archive decompression exceeds the canonical bound"
+                        )
+                    chunks.append(chunk)
+    except (OSError, EOFError, gzip.BadGzipFile, zlib_error) as exc:
+        raise ValueError("plugin archive is unreadable") from exc
+    return io.BytesIO(b"".join(chunks))
+
+
 def verify_archive(
     plugin_path: Path,
     archive_path: Path,
@@ -603,7 +643,9 @@ def verify_archive(
         for record in records
     }
     try:
-        bundle = tarfile.open(archive_path, "r:gz")
+        bundle = tarfile.open(
+            fileobj=_bounded_archive_stream(archive_path), mode="r:"
+        )
     except (OSError, tarfile.TarError) as exc:
         raise ValueError("plugin archive is unreadable") from exc
     with bundle:
@@ -623,7 +665,10 @@ def verify_archive(
             # ownership, zero mtime, no link targets, and no PAX extension
             # headers (canonical members never need any). This also rejects
             # hostile archives that smuggle devices/links or oversized PAX
-            # payloads past the name checks.
+            # payloads past the name checks. Deliberate constraint: member
+            # names must stay ASCII and <= 100 chars (the ustar name field) —
+            # a longer/non-ASCII path would make gettarinfo emit a PAX "path"
+            # header and the build would fail-closed against this check.
             if (
                 not (member.isfile() or member.isdir())
                 or member.uid != 0
@@ -789,7 +834,11 @@ def _reject_source_alias(candidate: Path) -> None:
     _reject_source_alias(output_resolved)
     # Build into an exclusive temporary file, verify THAT, and only then move
     # it to the requested destination — a failed build or failed verification
-    # never leaves a publishable artifact at the output path.
+    # never leaves a publishable artifact at the output path. The pid-suffixed
+    # temp name is predictable: in a shared/world-writable output directory an
+    # adversary could pre-create it as a DoS (O_EXCL fails, their file
+    # survives untouched — integrity is unaffected). Release builds write to
+    # operator-owned directories.
     temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}"
     member_flags = (
         os.O_RDONLY
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index be25d82..05a2d93 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -657,6 +657,88 @@ def test_failed_exclusive_temp_open_preserves_foreign_file(self) -> None:
         self.assertEqual(foreign.read_bytes(), b"foreign artifact")
         self.assertFalse(self.archive.exists())
 
+    def test_output_alias_of_source_trees_is_rejected(self) -> None:
+        self._activate()
+        for alias in (
+            self.plugin / "aliased-output.plugin",
+            self.bundle_leaf / "aliased-output.plugin",
+        ):
+            with self.subTest(alias=str(alias)):
+                with self.assertRaisesRegex(ValueError, "alias a source tree"):
+                    self.builder.build_archive(
+                        self.root,
+                        plugin="agent-collab",
+                        output=alias,
+                        bundle_source=self.bundle_leaf,
+                    )
+        # A symlinked parent that resolves into a source tree is caught too.
+        symdir = self.root / "sym-parent"
+        symdir.symlink_to(self.plugin)
+        with self.assertRaisesRegex(ValueError, "alias a source tree"):
+            self.builder.build_archive(
+                self.root,
+                plugin="agent-collab",
+                output=symdir / "aliased-output.plugin",
+                bundle_source=self.bundle_leaf,
+            )
+        self.assertFalse((self.plugin / "aliased-output.plugin").exists())
+
+    def test_pack_time_failure_leaves_no_output_or_temp(self) -> None:
+        self._activate()
+        with mock.patch.object(
+            self.builder,
+            "_synthesized_tarinfo",
+            side_effect=OSError("forced pack failure"),
+        ):
+            with self.assertRaises(OSError):
+                self._build_activation()
+        self.assertFalse(self.archive.exists())
+        leftovers = [
+            path.name
+            for path in self.archive.parent.iterdir()
+            if ".tmp" in path.name
+        ]
+        self.assertEqual(leftovers, [])
+
+    def test_cli_builds_activation_archive_with_bundle_source(self) -> None:
+        self._activate()
+        exit_code = self.builder.main(
+            [
+                "--repo-root",
+                str(self.root),
+                "--output",
+                str(self.archive),
+                "--bundle-source",
+                str(self.bundle_leaf),
+            ]
+        )
+        self.assertEqual(exit_code, 0)
+        self.assertTrue(self.archive.is_file())
+        self.builder.verify_archive(self.plugin, self.archive, mode="activation")
+
+    def test_verify_bounds_hostile_archive_decompression(self) -> None:
+        self._activate()
+        mode = self._build_activation()
+
+        # A gzip bomb (tiny compressed, huge decompressed) must be rejected
+        # by the hard decompression bound BEFORE any tar/PAX parsing.
+        import gzip as gzip_module
+
+        bomb = self.root / "bomb.plugin"
+        with bomb.open("wb") as raw:
+            with gzip_module.GzipFile(fileobj=raw, mode="wb") as stream:
+                zero_chunk = b"\0" * (1024 * 1024)
+                for _ in range(2 * 64 + 8):  # > 2 * MAX_ARTIFACT_BYTES
+                    stream.write(zero_chunk)
+        with self.assertRaisesRegex(ValueError, "decompression exceeds"):
+            self.builder.verify_archive(self.plugin, bomb, mode=mode)
+
+        # A symlinked archive path is refused outright.
+        alias = self.root / "archive-alias.plugin"
+        alias.symlink_to(self.archive)
+        with self.assertRaisesRegex(ValueError, "unreadable"):
+            self.builder.verify_archive(self.plugin, alias, mode=mode)
+
     def test_activation_archive_build_is_deterministic(self) -> None:
         self._activate()
         self._build_activation()

From e9f55ecfdb41e8cfd2f1a0b4e01dc18b2dabedce Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 20:41:55 -0700
Subject: [PATCH 4/9] =?UTF-8?q?fix(archive):=20fd-only=20compressed-size?=
 =?UTF-8?q?=20cap=20(O=5FNOFOLLOW=20+=20fstat=20+=20read-counting)=20?=
 =?UTF-8?q?=E2=80=94=20closes=20peer-review=20#4=20TOCTOU?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex round-3: the decompressed/PAX bound was confirmed fixed; the
compressed cap still had an lstat->open race. Now the archive is opened
O_NOFOLLOW, fstat'd on the descriptor, and the cap is enforced on the
bytes actually read from that same fd — no path-based window remains.
333 tests/ green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 62 +++++++++++++++++++++++----------
 1 file changed, 44 insertions(+), 18 deletions(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index 424e6c4..f3f1793 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -569,31 +569,57 @@ def _member_plan(
 
 
 def _bounded_archive_stream(archive_path: Path) -> io.BytesIO:
-    """Decompress an archive under hard byte caps before tar parsing."""
+    """Decompress an archive under hard byte caps before tar parsing.
 
+    All checks are descriptor-based (O_NOFOLLOW open + fstat + read-counting
+    on the SAME fd), so a concurrent path swap or file growth cannot bypass
+    the compressed-size cap — there is no lstat→open window to race.
+    """
+
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+    try:
+        descriptor = os.open(archive_path, flags)
+    except OSError as exc:
+        raise ValueError("plugin archive is unreadable") from exc
+    compressed_chunks: list[bytes] = []
+    compressed_total = 0
     try:
-        info = archive_path.lstat()
+        described = os.fstat(descriptor)
+        if not stat.S_ISREG(described.st_mode):
+            raise ValueError("plugin archive is unreadable")
+        if described.st_size > _MAX_COMPRESSED_ARCHIVE_BYTES:
+            raise ValueError("plugin archive exceeds the canonical size bound")
+        while True:
+            chunk = os.read(descriptor, 1024 * 1024)
+            if not chunk:
+                break
+            compressed_total += len(chunk)
+            if compressed_total > _MAX_COMPRESSED_ARCHIVE_BYTES:
+                # Enforced on the bytes actually read, so growth after the
+                # fstat cannot smuggle extra compressed input past the cap.
+                raise ValueError("plugin archive exceeds the canonical size bound")
+            compressed_chunks.append(chunk)
     except OSError as exc:
         raise ValueError("plugin archive is unreadable") from exc
-    if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode):
-        raise ValueError("plugin archive is unreadable")
-    if info.st_size > _MAX_COMPRESSED_ARCHIVE_BYTES:
-        raise ValueError("plugin archive exceeds the canonical size bound")
+    finally:
+        try:
+            os.close(descriptor)
+        except OSError:
+            pass
     chunks: list[bytes] = []
     total = 0
     try:
-        with archive_path.open("rb") as raw:
-            with gzip.GzipFile(fileobj=raw) as stream:
-                while True:
-                    chunk = stream.read(1024 * 1024)
-                    if not chunk:
-                        break
-                    total += len(chunk)
-                    if total > _MAX_DECOMPRESSED_ARCHIVE_BYTES:
-                        raise ValueError(
-                            "plugin archive decompression exceeds the canonical bound"
-                        )
-                    chunks.append(chunk)
+        with gzip.GzipFile(fileobj=io.BytesIO(b"".join(compressed_chunks))) as stream:
+            while True:
+                chunk = stream.read(1024 * 1024)
+                if not chunk:
+                    break
+                total += len(chunk)
+                if total > _MAX_DECOMPRESSED_ARCHIVE_BYTES:
+                    raise ValueError(
+                        "plugin archive decompression exceeds the canonical bound"
+                    )
+                chunks.append(chunk)
     except (OSError, EOFError, gzip.BadGzipFile, zlib_error) as exc:
         raise ValueError("plugin archive is unreadable") from exc
     return io.BytesIO(b"".join(chunks))

From f68a4f7c2713f30c20c0467b1a33b53e03e45c01 Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 21:45:22 -0700
Subject: [PATCH 5/9] refactor(archive): regenerate-and-compare verifier
 (adversarial design-review pivot)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Per the adversarial-architecture-review tripwire: the post-hoc peer-review
loop kept surfacing NEW bypasses in ONE surface (how verify_archive reads a
hostile archive) across rounds — the signal the surface was under-architected.
Reverted it to the design phase, wrote a threat-class enumeration, and ran a
2-round adversarial DESIGN review. Round 1 (RECONSIDER/H) surfaced a whole
layer of structural bypasses the code loop hadn't reached (hidden PAX/GNU/
sparse physical headers, gzip-header malleability, sparse/contiguous
typeflags, dir-payload smuggling, parser differentials) AND the architectural
pivot: don't validate hostile tar bytes with tarfile — regenerate the
canonical tar from trusted inputs and byte-compare. Round 2 PROCEED-WITH-
MODIFICATIONS/H confirmed the pivot closes all 8 classes; 4 refinements folded
in.

verify_archive now NEVER hands candidate tar bytes to tarfile:
- A shared deterministic USTAR serializer (_emit_canonical_tar, used by BOTH
  build and verify) reproduces the canonical tar; runtime payloads zero-filled
  at verify. USTAR_FORMAT + a _finalize_tarinfo guard make an extension record
  impossible → byte-reproducible across the operator (3.13.14) and CI
  (3.10/3.12/3.14); a golden-bytes test binds the serializer output.
- The candidate is fd-only bounded-read (O_NOFOLLOW|O_NONBLOCK + fstat, no
  path-swap window) with a canonical gzip-header check, single-stream/no-
  trailing validation, and hard compressed/decompressed caps enforced on bytes
  actually read; then sliced ONLY at offsets the verifier computed.
- Structural byte-compare OUTSIDE the runtime ranges binds every header,
  typeflag, ordering, padding, EOF, dialect, and the archived manifest bytes;
  each runtime payload is bound to the frozen-manifest digest; the range map
  is asserted complete/one-to-one/disjoint/in-bounds (design item 6).
- Build emits the shared canonical tar, gzips it (canonical header) into an
  O_EXCL temp, verifies THAT, then os.replace()s; runtime payloads are
  re-read+re-digested through fresh fds at pack time.

New tests: golden USTAR serializer bytes, cross-call reproducibility + range-
map integrity, raw structural tamper caught without parsing, canonical vs
non-canonical gzip header. 336 tests/ + 254 scripts/ green; CLI smoke-verified
(canonical gzip header 1f8b080000000000, round-trip verify).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../2026-07-17-build-archive-bundle-source.md |   3 +-
 scripts/build_plugin_archive.py               | 563 ++++++++++--------
 tests/test_plugin_archive.py                  | 137 ++++-
 3 files changed, 431 insertions(+), 272 deletions(-)

diff --git a/changelog.d/2026-07-17-build-archive-bundle-source.md b/changelog.d/2026-07-17-build-archive-bundle-source.md
index 4297c72..a750b4d 100644
--- a/changelog.d/2026-07-17-build-archive-bundle-source.md
+++ b/changelog.d/2026-07-17-build-archive-bundle-source.md
@@ -1,2 +1,3 @@
 ### Changed
-- `build_plugin_archive.py` now sources the activation runtime bundle from an out-of-tree signed handoff via `--bundle-source <leaf .bundle dir>` instead of the git checkout — the signed runtime ships as a release asset and is never committed. The committed `runtime-manifest.json` alone decides the mode (design decision D1): an activation manifest without `--bundle-source` is a hard error (never a silent policy-only downgrade), a policy-only manifest forbids the flag, and an in-tree `runtime/` conflicts with an external source. The bundle source is treated as hostile input (symlink argument rejected before resolution; bounded non-following enumeration; member-set equality; per-member O_NOFOLLOW regular-file/uid/nlink/mode/size/sha256 checks with the strict 0o500 install mode preserved). Runtime archive metadata is synthesized entirely from a frozen manifest snapshot (0o755 traversal parents + sealed 0o500 bundle leaf), `verify_archive` validates runtime members and the archived manifest bytes against that frozen snapshot only (self-contained — no bundle source needed, closing the validate→pack swap window), and the archive is built into an exclusive temp file that is verified before an atomic rename so a failed build never leaves a publishable artifact.
+- `build_plugin_archive.py` now sources the activation runtime bundle from an out-of-tree signed handoff via `--bundle-source <leaf .bundle dir>` instead of the git checkout — the signed runtime ships as a release asset and is never committed. The committed `runtime-manifest.json` alone decides the mode (design decision D1): an activation manifest without `--bundle-source` is a hard error (never a silent policy-only downgrade), a policy-only manifest forbids the flag, and an in-tree `runtime/` conflicts with an external source. The bundle source is treated as hostile input (symlink argument rejected before resolution; bounded non-following enumeration; member-set equality; per-member O_NOFOLLOW regular-file/uid/nlink/mode/size/sha256 checks with the strict 0o500 install mode preserved), and its payloads are re-read + re-digested through fresh descriptors at pack time.
+- `verify_archive` is rearchitected to **regenerate the canonical tar from trusted inputs and byte-compare** rather than parse a possibly-hostile archive: it never hands candidate tar bytes to `tarfile`. A shared deterministic `USTAR` serializer (used by both build and verify) reproduces the canonical tar with runtime payloads zero-filled; the candidate is only bounded-inflated and then sliced at offsets the verifier computed. Every structural property — headers, exact typeflags, ordering, padding, EOF, dialect, and the absence of hidden PAX/GNU/sparse records — is bound by the byte-compare outside the runtime ranges; each runtime payload is bound to the frozen-manifest digest; the archived `runtime-manifest.json` bytes are bound to the frozen snapshot. Reading the candidate is fd-only (`O_NOFOLLOW`/`O_NONBLOCK` + `fstat`, no path-swap window) under hard compressed- and decompressed-size caps, with a canonical gzip-header check and single-stream (no trailing/concatenated data) validation. The archive is built into an exclusive temp that is verified before an atomic rename, so a failed build never leaves a publishable artifact. Hardened through a 2-round up-front adversarial design review of the hostile-archive-reading threat model (the regenerate-and-compare pivot) before implementation.
diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index f3f1793..d7f7fed 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -10,10 +10,10 @@
 import io
 import json
 import os
-from zlib import error as zlib_error
 import re
 import stat
 import tarfile
+import zlib
 from pathlib import Path, PurePosixPath
 
 try:
@@ -560,29 +560,134 @@ def _member_plan(
     return [(name, members[name]) for name in sorted(members)]
 
 
-# Hard resource bounds for reading a possibly-hostile archive: the compressed
-# file and its decompressed expansion are both capped BEFORE any tar/PAX
-# parsing happens, so a crafted asset cannot exhaust memory or time during
-# tarfile's own header processing.
+# Hard resource bounds for reading a possibly-hostile archive. The verifier
+# NEVER parses hostile tar/PAX bytes: it regenerates the ONE canonical tar from
+# trusted inputs and byte-compares, so a candidate archive is only ever sliced
+# by offsets WE computed. Only the bounded gzip inflater touches candidate
+# bytes, under hard compressed- and decompressed-size caps.
 _MAX_COMPRESSED_ARCHIVE_BYTES = MAX_ARTIFACT_BYTES
 _MAX_DECOMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES
+_USTAR_BLOCK = 512
+# Fixed canonical gzip header: magic (1f 8b), DEFLATE method (08), flags 0
+# (rejects FTEXT/FHCRC/FEXTRA/FNAME/FCOMMENT — the malleability/amplification
+# vectors), and mtime 0. Bytes 8-9 (XFL, OS) are compressor/platform-derived,
+# informational, and not attacker-amplifiable, so they are deliberately not
+# bound — the inflated-tar byte-compare binds all CONTENT regardless.
+_CANONICAL_GZIP_PREFIX = b"\x1f\x8b\x08\x00\x00\x00\x00\x00"
+
+
+def _synthesized_tarinfo(
+    name: str, *, mode: int, size: int = 0, directory: bool = False
+) -> tarfile.TarInfo:
+    info = tarfile.TarInfo(name)
+    info.type = tarfile.DIRTYPE if directory else tarfile.REGTYPE
+    info.mode = mode
+    info.size = size
+    info.uid = 0
+    info.gid = 0
+    info.uname = ""
+    info.gname = ""
+    info.mtime = 0
+    return info
 
 
-def _bounded_archive_stream(archive_path: Path) -> io.BytesIO:
-    """Decompress an archive under hard byte caps before tar parsing.
+def _finalize_tarinfo(info: tarfile.TarInfo) -> tarfile.TarInfo:
+    info.uid = 0
+    info.gid = 0
+    info.uname = ""
+    info.gname = ""
+    info.mtime = 0
+    if info.pax_headers:
+        # USTAR_FORMAT would already have raised for an unrepresentable field;
+        # this belt-and-suspenders makes it impossible to emit an extension
+        # record so build and verify are byte-reproducible across environments.
+        raise ValueError(f"archive member is not USTAR-representable: {info.name}")
+    return info
 
-    All checks are descriptor-based (O_NOFOLLOW open + fstat + read-counting
-    on the SAME fd), so a concurrent path swap or file growth cannot bypass
-    the compressed-size cap — there is no lstat→open window to race.
+
+def _emit_canonical_tar(
+    plan: list[tuple[str, Path | None]],
+    *,
+    plugin_path: Path,
+    frozen_manifest: bytes | None,
+    record_by_name: dict[str, dict[str, object]],
+    runtime_payloads: dict[str, bytes],
+) -> tuple[bytes, dict[str, tuple[int, int]]]:
+    """Serialize the ONE canonical uncompressed tar deterministically.
+
+    USTAR_FORMAT (tarfile raises on anything needing a PAX/GNU extension
+    record), zeroed ownership/mtime, exact ``_member_plan`` order. Non-runtime
+    payloads come from the trusted ``plugin_path`` tree (the manifest member
+    from ``frozen_manifest``); runtime FILE payloads come from
+    ``runtime_payloads`` (real bytes at build, zero-fill at verify). Both build
+    and verify call this, so they produce byte-identical structure. Returns
+    ``(tar_bytes, runtime_ranges)`` mapping each runtime file member name to its
+    ``(data_offset, length)`` within the tar.
     """
 
-    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+    buffer = io.BytesIO()
+    ranges: dict[str, tuple[int, int]] = {}
+    with tarfile.open(
+        fileobj=buffer, mode="w", format=tarfile.USTAR_FORMAT
+    ) as tar:
+        for name, source in plan:
+            if source is None and name in _RUNTIME_DIR_MODES:
+                tar.addfile(
+                    _finalize_tarinfo(
+                        _synthesized_tarinfo(
+                            name, mode=_RUNTIME_DIR_MODES[name], directory=True
+                        )
+                    )
+                )
+                continue
+            if source is None:
+                record = record_by_name[name]
+                payload = runtime_payloads[name]
+                if len(payload) != record["size"]:
+                    raise ValueError("runtime payload size mismatch during emit")
+                info = _finalize_tarinfo(
+                    _synthesized_tarinfo(
+                        name, mode=record["install_mode"], size=record["size"]
+                    )
+                )
+                header_offset = buffer.tell()
+                tar.addfile(info, io.BytesIO(payload))
+                ranges[name] = (header_offset + _USTAR_BLOCK, record["size"])
+                continue
+            info = _finalize_tarinfo(tar.gettarinfo(str(source), arcname=name))
+            if name == "runtime-manifest.json" and frozen_manifest is not None:
+                info.size = len(frozen_manifest)
+                tar.addfile(info, io.BytesIO(frozen_manifest))
+            elif info.isfile():
+                with source.open("rb") as stream:
+                    tar.addfile(info, stream)
+            else:
+                tar.addfile(info)
+    return buffer.getvalue(), ranges
+
+
+def _read_bounded_compressed(archive_path: Path) -> bytes:
+    """Read the compressed archive fd-only under a hard size cap.
+
+    ``O_NOFOLLOW`` refuses a symlinked final component; ``O_NONBLOCK`` keeps a
+    substituted FIFO from blocking the open before the ``fstat`` regular-file
+    check; the cap is enforced on the bytes ACTUALLY read from the descriptor,
+    so a path swap or post-fstat growth cannot smuggle input past it. There is
+    no ``lstat``→``open`` window.
+    """
+
+    flags = (
+        os.O_RDONLY
+        | getattr(os, "O_NOFOLLOW", 0)
+        | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NONBLOCK", 0)
+    )
     try:
         descriptor = os.open(archive_path, flags)
     except OSError as exc:
         raise ValueError("plugin archive is unreadable") from exc
-    compressed_chunks: list[bytes] = []
-    compressed_total = 0
+    chunks: list[bytes] = []
+    total = 0
     try:
         described = os.fstat(descriptor)
         if not stat.S_ISREG(described.st_mode):
@@ -593,12 +698,10 @@ def _bounded_archive_stream(archive_path: Path) -> io.BytesIO:
             chunk = os.read(descriptor, 1024 * 1024)
             if not chunk:
                 break
-            compressed_total += len(chunk)
-            if compressed_total > _MAX_COMPRESSED_ARCHIVE_BYTES:
-                # Enforced on the bytes actually read, so growth after the
-                # fstat cannot smuggle extra compressed input past the cap.
+            total += len(chunk)
+            if total > _MAX_COMPRESSED_ARCHIVE_BYTES:
                 raise ValueError("plugin archive exceeds the canonical size bound")
-            compressed_chunks.append(chunk)
+            chunks.append(chunk)
     except OSError as exc:
         raise ValueError("plugin archive is unreadable") from exc
     finally:
@@ -606,23 +709,87 @@ def _bounded_archive_stream(archive_path: Path) -> io.BytesIO:
             os.close(descriptor)
         except OSError:
             pass
+    return b"".join(chunks)
+
+
+def _inflate_bounded(compressed: bytes) -> bytes:
+    """Inflate a single gzip stream under a hard output cap.
+
+    Binds the fixed canonical gzip header, caps decompressed OUTPUT per call
+    (so a bomb trips before materializing the expansion), validates CRC/ISIZE
+    (zlib raises on mismatch at end-of-stream), and requires the stream to be
+    complete with NO trailing/concatenated data.
+    """
+
+    if len(compressed) < 10 or compressed[:8] != _CANONICAL_GZIP_PREFIX:
+        raise ValueError("plugin archive gzip header is not canonical")
+    decompressor = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS)
+    chunk_size = 1024 * 1024
     chunks: list[bytes] = []
     total = 0
+    pending = compressed
     try:
-        with gzip.GzipFile(fileobj=io.BytesIO(b"".join(compressed_chunks))) as stream:
-            while True:
-                chunk = stream.read(1024 * 1024)
-                if not chunk:
-                    break
-                total += len(chunk)
+        while not decompressor.eof:
+            before = len(pending)
+            piece = decompressor.decompress(pending, chunk_size)
+            pending = decompressor.unconsumed_tail
+            if piece:
+                total += len(piece)
                 if total > _MAX_DECOMPRESSED_ARCHIVE_BYTES:
                     raise ValueError(
                         "plugin archive decompression exceeds the canonical bound"
                     )
-                chunks.append(chunk)
-    except (OSError, EOFError, gzip.BadGzipFile, zlib_error) as exc:
+                chunks.append(piece)
+            elif len(pending) >= before:
+                # No output produced and no input consumed: the stream cannot
+                # make progress (truncated/malformed). Stop; the eof check below
+                # rejects it.
+                break
+    except zlib.error as exc:
         raise ValueError("plugin archive is unreadable") from exc
-    return io.BytesIO(b"".join(chunks))
+    if not decompressor.eof:
+        raise ValueError("plugin archive gzip stream is incomplete")
+    if decompressor.unused_data:
+        raise ValueError("plugin archive has trailing data after the gzip stream")
+    return b"".join(chunks)
+
+
+def _assert_runtime_range_map(
+    ranges: dict[str, tuple[int, int]],
+    record_by_name: dict[str, dict[str, object]],
+    total_length: int,
+) -> None:
+    """Assert the regenerated runtime range map is well-formed (design item 6).
+
+    Complete + one-to-one with the manifest runtime entries, each range
+    exact-length, in-bounds, and pairwise disjoint. The archived
+    ``runtime-manifest.json`` member is non-runtime by construction and thus
+    lies OUTSIDE every range (bound by the structural byte-compare).
+    """
+
+    if set(ranges) != set(record_by_name):
+        raise ValueError("archive runtime range map is not one-to-one with the manifest")
+    for name, (offset, length) in ranges.items():
+        if length != record_by_name[name]["size"]:
+            raise ValueError("archive runtime range length does not match the manifest")
+    previous_end = 0
+    for offset, length in sorted(ranges.values()):
+        if offset < previous_end or offset < 0 or offset + length > total_length:
+            raise ValueError("archive runtime range map is out of bounds or overlapping")
+        previous_end = offset + length
+
+
+def _bytes_equal_outside_ranges(
+    candidate: bytes, canonical: bytes, ranges: list[tuple[int, int]]
+) -> bool:
+    """True iff candidate == canonical at every byte NOT inside a runtime range."""
+
+    position = 0
+    for offset, length in sorted(ranges):
+        if candidate[position:offset] != canonical[position:offset]:
+            return False
+        position = offset + length
+    return candidate[position:] == canonical[position:]
 
 
 def verify_archive(
@@ -632,20 +799,17 @@ def verify_archive(
     mode: str,
     frozen_manifest: bytes | None = None,
 ) -> None:
-    """Verify one built archive without trusting any mutable runtime source.
-
-    Non-runtime members are byte/mode-parity-checked against the git-tracked
-    plugin tree (which CI additionally binds to the signed-tag tree). Runtime
-    members are validated ONLY against a single frozen snapshot of the
-    committed manifest — exact member sequence, type, mode, size, and sha256,
-    plus the archived manifest bytes themselves — so a source-tree swap after
-    the build-time validation can never produce a passing archive, and no
-    runtime bundle source is needed to verify (CI runs this without one).
-
-    ``frozen_manifest`` lets `build_archive` hand over the EXACT byte snapshot
-    it validated, packed, and embedded, so build+verify share one snapshot and
-    a mid-build manifest swap on disk is irrelevant. Standalone callers (CI)
-    omit it and the snapshot is read once from the checkout here.
+    """Verify a built archive by REGENERATING the canonical tar and comparing.
+
+    The candidate archive's tar bytes are never handed to ``tarfile``. Instead
+    the ONE canonical tar is regenerated from trusted inputs (plugin tree +
+    frozen manifest) with runtime payloads zero-filled; the candidate is only
+    bounded-inflated and then sliced by offsets WE computed. Every structural
+    property (headers, typeflags, ordering, padding, EOF, dialect, no hidden
+    records) is bound by the byte-compare OUTSIDE the runtime ranges; each
+    runtime payload is bound to the frozen-manifest digest. ``frozen_manifest``
+    lets ``build_archive`` share the exact snapshot it packed; CI omits it and
+    reads the snapshot once from the signed-tag checkout.
     """
 
     plugin_path = plugin_path.resolve(strict=True)
@@ -663,136 +827,88 @@ def verify_archive(
     else:
         frozen_manifest = None
     plan = _member_plan(plugin_path, mode=mode, records=records)
-    expected_names = [name for name, _ in plan]
     record_by_name = {
-        (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record
-        for record in records
+        (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records
     }
-    try:
-        bundle = tarfile.open(
-            fileobj=_bounded_archive_stream(archive_path), mode="r:"
-        )
-    except (OSError, tarfile.TarError) as exc:
-        raise ValueError("plugin archive is unreadable") from exc
-    with bundle:
-        members = bundle.getmembers()
-        names = [member.name for member in members]
-        if len(names) != len(set(names)):
-            raise ValueError("archive contains duplicate members")
-        for name in names:
-            pure = PurePosixPath(name)
-            if pure.is_absolute() or ".." in pure.parts or "\\" in name:
-                raise ValueError("archive contains an unsafe member path")
-        if names != expected_names:
-            raise ValueError("archive member list is not canonical")
-        for member in members:
-            # Canonical archive metadata is part of the contract for EVERY
-            # member: only regular files and directories, normalized
-            # ownership, zero mtime, no link targets, and no PAX extension
-            # headers (canonical members never need any). This also rejects
-            # hostile archives that smuggle devices/links or oversized PAX
-            # payloads past the name checks. Deliberate constraint: member
-            # names must stay ASCII and <= 100 chars (the ustar name field) —
-            # a longer/non-ASCII path would make gettarinfo emit a PAX "path"
-            # header and the build would fail-closed against this check.
+    zero_payloads = {
+        name: b"\x00" * record["size"] for name, record in record_by_name.items()
+    }
+    canonical, ranges = _emit_canonical_tar(
+        plan,
+        plugin_path=plugin_path,
+        frozen_manifest=frozen_manifest,
+        record_by_name=record_by_name,
+        runtime_payloads=zero_payloads,
+    )
+    candidate = _inflate_bounded(_read_bounded_compressed(archive_path))
+    if len(candidate) != len(canonical):
+        raise ValueError("archive does not match the canonical layout")
+    _assert_runtime_range_map(ranges, record_by_name, len(canonical))
+    for name, (offset, length) in ranges.items():
+        record = record_by_name[name]
+        if (
+            length != record["size"]
+            or hashlib.sha256(candidate[offset : offset + length]).hexdigest()
+            != record["sha256"]
+        ):
+            raise ValueError(f"archive runtime member digest failed: {name}")
+    if not _bytes_equal_outside_ranges(candidate, canonical, list(ranges.values())):
+        raise ValueError("archive structure does not match the canonical layout")
+
+
+def _read_runtime_payloads(
+    bundle_leaf: Path, records: tuple[dict[str, object], ...]
+) -> dict[str, bytes]:
+    """Read + re-validate + re-digest each runtime payload into memory.
+
+    Reading through a fresh ``O_NOFOLLOW``/``O_NONBLOCK`` descriptor and
+    re-checking identity here binds the EXACT bytes that will be emitted into
+    the archive (closing any window between the earlier tree validation and the
+    pack) — and the bytes are additionally re-bound to the manifest digest by
+    ``verify_archive`` on the temp before publish.
+    """
+
+    flags = (
+        os.O_RDONLY
+        | getattr(os, "O_NOFOLLOW", 0)
+        | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NONBLOCK", 0)
+    )
+    payloads: dict[str, bytes] = {}
+    for record in records:
+        name = (RUNTIME_BUNDLE_REL / record["path"]).as_posix()
+        try:
+            descriptor = os.open(bundle_leaf / record["path"], flags)
+        except OSError as exc:
+            raise ValueError("runtime bundle source member is unsafe") from exc
+        try:
+            info = os.fstat(descriptor)
             if (
-                not (member.isfile() or member.isdir())
-                or member.uid != 0
-                or member.gid != 0
-                or member.uname != ""
-                or member.gname != ""
-                or member.mtime != 0
-                or member.linkname != ""
-                or member.pax_headers
+                not stat.S_ISREG(info.st_mode)
+                or info.st_uid != os.getuid()
+                or info.st_nlink != 1
+                or stat.S_IMODE(info.st_mode) != record["install_mode"]
+                or info.st_size != record["size"]
             ):
-                raise ValueError(
-                    f"archive member metadata is not canonical: {member.name}"
-                )
-        for (name, source), member in zip(plan, members, strict=True):
-            if source is None:
-                directory_mode = _RUNTIME_DIR_MODES.get(name)
-                if directory_mode is not None:
-                    if not member.isdir() or stat.S_IMODE(member.mode) != directory_mode:
-                        raise ValueError(
-                            f"archive runtime directory identity failed: {member.name}"
-                        )
-                    continue
-                record = record_by_name[name]
-                if (
-                    not member.isreg()
-                    or stat.S_IMODE(member.mode) != record["install_mode"]
-                    or member.size != record["size"]
-                ):
-                    raise ValueError(
-                        f"archive runtime member identity failed: {member.name}"
-                    )
-                archived = bundle.extractfile(member)
-                if (
-                    archived is None
-                    or hashlib.sha256(archived.read()).hexdigest() != record["sha256"]
-                ):
-                    raise ValueError(
-                        f"archive runtime member digest failed: {member.name}"
-                    )
-                continue
-            source_info = _safe_source(source)
-            source_is_file = stat.S_ISREG(source_info.st_mode)
-            if source_is_file != member.isfile() or (
-                not source_is_file and not member.isdir()
+                raise ValueError("runtime bundle source member changed during packing")
+            data = b""
+            while len(data) <= record["size"]:
+                chunk = os.read(descriptor, 1024 * 1024)
+                if not chunk:
+                    break
+                data += chunk
+            if (
+                len(data) != record["size"]
+                or hashlib.sha256(data).hexdigest() != record["sha256"]
             ):
-                raise ValueError(f"archive member type parity failed: {member.name}")
-            if stat.S_IMODE(member.mode) != stat.S_IMODE(source_info.st_mode):
-                raise ValueError(f"archive member mode parity failed: {member.name}")
-            if source_is_file:
-                # Bound the read BEFORE extracting: the declared member size
-                # must already match the expected payload exactly, so a
-                # hostile archive cannot force an unbounded decompression.
-                expected_size = (
-                    len(frozen_manifest)
-                    if name == "runtime-manifest.json" and frozen_manifest is not None
-                    else source_info.st_size
-                )
-                if member.size != expected_size:
-                    raise ValueError(
-                        f"archive member size parity failed: {member.name}"
-                    )
-                archived = bundle.extractfile(member)
-                if archived is None:
-                    raise ValueError(f"archive member byte parity failed: {member.name}")
-                data = archived.read()
-                if name == "runtime-manifest.json" and mode == "activation":
-                    # Bind the archived manifest to the SAME frozen bytes the
-                    # runtime records above were parsed from (never a second
-                    # read) — closes the manifest-swap window.
-                    if data != frozen_manifest:
-                        raise ValueError(
-                            "archived runtime manifest does not match the frozen manifest"
-                        )
-                elif data != source.read_bytes():
-                    raise ValueError(f"archive member byte parity failed: {member.name}")
-        runtime_files = [
-            member.name
-            for member in members
-            if member.isfile() and member.name.startswith("runtime/")
-        ]
-        expected_runtime = [
-            (RUNTIME_BUNDLE_REL / record["path"]).as_posix() for record in records
-        ]
-        if runtime_files != expected_runtime:
-            raise ValueError("archive runtime membership does not match release mode")
-
-
-def _synthesized_tarinfo(name: str, *, mode: int, size: int = 0, directory: bool = False) -> tarfile.TarInfo:
-    info = tarfile.TarInfo(name)
-    info.type = tarfile.DIRTYPE if directory else tarfile.REGTYPE
-    info.mode = mode
-    info.size = size
-    info.uid = 0
-    info.gid = 0
-    info.uname = ""
-    info.gname = ""
-    info.mtime = 0
-    return info
+                raise ValueError("runtime bundle source member changed during packing")
+            payloads[name] = data
+        finally:
+            try:
+                os.close(descriptor)
+            except OSError:
+                pass
+    return payloads
 
 
 def build_archive(
@@ -809,6 +925,7 @@ def build_archive(
     records: tuple[dict[str, object], ...] = ()
     bundle_leaf: Path | None = None
     frozen_manifest: bytes | None = None
+    runtime_payloads: dict[str, bytes] = {}
     if mode == "policy-only":
         # The committed manifest — never the flag — decides the mode.
         if bundle_source is not None:
@@ -833,8 +950,8 @@ def build_archive(
             raise ValueError("runtime bundle source is unsafe")
         bundle_leaf = bundle_source.resolve(strict=True)
         # Freeze the manifest ONCE: these exact bytes drive the bundle-tree
-        # validation, are embedded into the archive verbatim, and are handed
-        # to verify_archive — a mid-build manifest swap on disk is irrelevant.
+        # validation, are embedded into the archive verbatim, and are handed to
+        # verify_archive — a mid-build manifest swap on disk is irrelevant.
         frozen_manifest = _read_manifest_bytes(plugin_path)
         manifest = _parse_manifest(frozen_manifest)
         artifacts = manifest["artifacts"]
@@ -842,112 +959,46 @@ def build_archive(
             raise ValueError("activation package requires exactly one runtime artifact")
         records = runtime_bundle.validate_file_records(artifacts[0]["files"])
         _validate_activation_bundle_tree(bundle_leaf, records)
+        runtime_payloads = _read_runtime_payloads(bundle_leaf, records)
     plan = _member_plan(plugin_path, mode=mode, records=records)
     record_by_name = {
         (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records
     }
+
     def _reject_source_alias(candidate: Path) -> None:
         for forbidden in (plugin_path, bundle_leaf):
             if forbidden is not None and candidate.is_relative_to(forbidden):
                 raise ValueError("archive output must not alias a source tree")
 
-    # Reject a source-tree destination BEFORE creating any directory (mkdir
-    # must never mutate a rejected destination), then re-check the strictly
-    # resolved path once the parent exists.
+    # Reject a source-tree destination BEFORE creating any directory (mkdir must
+    # never mutate a rejected destination), then re-check the strictly resolved
+    # path once the parent exists.
     _reject_source_alias(output.resolve())
     output.parent.mkdir(parents=True, exist_ok=True)
     output_resolved = output.parent.resolve(strict=True) / output.name
     _reject_source_alias(output_resolved)
-    # Build into an exclusive temporary file, verify THAT, and only then move
-    # it to the requested destination — a failed build or failed verification
-    # never leaves a publishable artifact at the output path. The pid-suffixed
-    # temp name is predictable: in a shared/world-writable output directory an
-    # adversary could pre-create it as a DoS (O_EXCL fails, their file
-    # survives untouched — integrity is unaffected). Release builds write to
-    # operator-owned directories.
-    temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}"
-    member_flags = (
-        os.O_RDONLY
-        | getattr(os, "O_NOFOLLOW", 0)
-        | getattr(os, "O_CLOEXEC", 0)
-        | getattr(os, "O_NONBLOCK", 0)
+    # Regenerate the ONE canonical tar (shared with verify_archive), then gzip
+    # it into an exclusive temp, verify THAT, and only then os.replace() to the
+    # destination — a failed build or verification never leaves a publishable
+    # artifact. The pid-suffixed temp name is predictable: in a shared/
+    # world-writable output dir an adversary could pre-create it as a DoS
+    # (O_EXCL fails, their file survives untouched — integrity unaffected).
+    # Release builds write to operator-owned directories.
+    canonical_tar, _ranges = _emit_canonical_tar(
+        plan,
+        plugin_path=plugin_path,
+        frozen_manifest=frozen_manifest,
+        record_by_name=record_by_name,
+        runtime_payloads=runtime_payloads,
     )
+    temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}"
     temp_created = False
     try:
-        descriptor = os.open(
-            temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644
-        )
+        descriptor = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
         temp_created = True
         with os.fdopen(descriptor, "wb") as raw:
-            with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed:
-                with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as bundle:
-                    for name, source in plan:
-                        if source is not None:
-                            info = bundle.gettarinfo(str(source), arcname=name)
-                            info.uid = 0
-                            info.gid = 0
-                            info.uname = ""
-                            info.gname = ""
-                            info.mtime = 0
-                            if (
-                                name == "runtime-manifest.json"
-                                and frozen_manifest is not None
-                            ):
-                                # Embed the EXACT frozen snapshot the runtime
-                                # records were validated against — never a
-                                # re-read of the mutable checkout.
-                                info.size = len(frozen_manifest)
-                                bundle.addfile(info, io.BytesIO(frozen_manifest))
-                            elif info.isfile():
-                                with source.open("rb") as stream:
-                                    bundle.addfile(info, stream)
-                            else:
-                                bundle.addfile(info)
-                        elif name in _RUNTIME_DIR_MODES:
-                            bundle.addfile(
-                                _synthesized_tarinfo(
-                                    name, mode=_RUNTIME_DIR_MODES[name], directory=True
-                                )
-                            )
-                        else:
-                            # Runtime file member: metadata comes ENTIRELY from
-                            # the frozen manifest record; only the bytes stream
-                            # from the validated source (and verify_archive
-                            # re-binds those bytes to the record digest).
-                            record = record_by_name[name]
-                            info = _synthesized_tarinfo(
-                                name,
-                                mode=record["install_mode"],
-                                size=record["size"],
-                            )
-                            if bundle_leaf is None:
-                                raise ValueError(
-                                    "runtime member requires a bundle source"
-                                )
-                            member_descriptor = os.open(
-                                bundle_leaf / record["path"], member_flags
-                            )
-                            with os.fdopen(member_descriptor, "rb") as stream:
-                                # Revalidate the OPEN descriptor before
-                                # streaming: a post-validation swap to a
-                                # FIFO/device/hardlinked/resized entry is
-                                # rejected here (O_NONBLOCK keeps a swapped
-                                # FIFO from blocking the open), and the byte
-                                # content itself is re-bound to the manifest
-                                # digest by verify_archive below.
-                                streamed = os.fstat(stream.fileno())
-                                if (
-                                    not stat.S_ISREG(streamed.st_mode)
-                                    or streamed.st_uid != os.getuid()
-                                    or streamed.st_nlink != 1
-                                    or stat.S_IMODE(streamed.st_mode)
-                                    != record["install_mode"]
-                                    or streamed.st_size != record["size"]
-                                ):
-                                    raise ValueError(
-                                        "runtime bundle source member changed during packing"
-                                    )
-                                bundle.addfile(info, stream)
+            with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed:
+                compressed.write(canonical_tar)
         verify_archive(
             plugin_path, temp_path, mode=mode, frozen_manifest=frozen_manifest
         )
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index 05a2d93..de63376 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -8,6 +8,7 @@
 import shutil
 import stat
 import sys
+import io
 import os
 import tarfile
 import tempfile
@@ -399,14 +400,15 @@ def test_archive_verifier_detects_source_byte_drift(self) -> None:
         policy = self.plugin / "signing_policy.py"
         original = policy.read_bytes()
 
-        # Same-size drift is caught by byte parity …
+        # Same-size drift: the regenerated canonical tar differs from the
+        # candidate outside any runtime range → structural byte-compare fails.
         policy.write_bytes(b"#" * len(original))
-        with self.assertRaisesRegex(ValueError, "byte parity"):
+        with self.assertRaisesRegex(ValueError, "structure does not match"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
-        # … and size drift is caught (bounded) before any byte read.
+        # Size drift changes the total canonical length → rejected up front.
         policy.write_bytes(original + b"\n# tail\n")
-        with self.assertRaisesRegex(ValueError, "size parity"):
+        with self.assertRaisesRegex(ValueError, "does not match the canonical layout"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
     def test_archive_size_limit_and_required_policy_member_are_canonical(self) -> None:
@@ -574,7 +576,7 @@ def test_verify_archive_binds_runtime_bytes_to_frozen_manifest(self) -> None:
                 filename="", mode="wb", fileobj=raw, mtime=0
             ) as compressed:
                 with tarfile.open(
-                    fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT
+                    fileobj=compressed, mode="w", format=tarfile.USTAR_FORMAT
                 ) as bundle:
                     for member, payload in members:
                         if member.name == target:
@@ -594,19 +596,20 @@ def test_verify_archive_binds_archived_manifest_to_frozen_bytes(self) -> None:
         mode = self._build_activation()
         manifest_path = self.plugin / "runtime-manifest.json"
         # A SAME-SIZE, parse-valid, record-preserving byte change after the
-        # build (identity string tweak): size parity cannot notice, only the
-        # frozen-byte binding of the archived manifest can.
+        # build (identity string tweak): the archived manifest member is a
+        # non-runtime member, so the regenerated canonical (embedding the new
+        # on-disk bytes) differs from the candidate → structural compare fails.
         text = manifest_path.read_text(encoding="utf-8")
         swapped = text.replace("Test Operator", "Test Operatox")
         self.assertNotEqual(text, swapped)
         self.assertEqual(len(text), len(swapped))
         manifest_path.write_text(swapped, encoding="utf-8")
-        with self.assertRaisesRegex(ValueError, "frozen manifest"):
+        with self.assertRaisesRegex(ValueError, "structure does not match"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
-        # A size-changing swap fails closed too (bounded before any read).
+        # A size-changing swap changes the canonical total length → rejected.
         manifest_path.write_text(text + " ", encoding="utf-8")
-        with self.assertRaisesRegex(ValueError, "size parity"):
+        with self.assertRaisesRegex(ValueError, "does not match the canonical layout"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
     def test_verify_archive_rejects_non_canonical_member_metadata(self) -> None:
@@ -631,7 +634,7 @@ def test_verify_archive_rejects_non_canonical_member_metadata(self) -> None:
                 filename="", mode="wb", fileobj=raw, mtime=0
             ) as compressed:
                 with tarfile.open(
-                    fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT
+                    fileobj=compressed, mode="w", format=tarfile.USTAR_FORMAT
                 ) as bundle:
                     for member, payload in members:
                         if member.name == target:
@@ -641,7 +644,7 @@ def test_verify_archive_rejects_non_canonical_member_metadata(self) -> None:
                         else:
                             bundle.addfile(member, io.BytesIO(payload))
 
-        with self.assertRaisesRegex(ValueError, "metadata is not canonical"):
+        with self.assertRaisesRegex(ValueError, "structure does not match|canonical layout"):
             self.builder.verify_archive(self.plugin, tampered, mode=mode)
 
     def test_failed_exclusive_temp_open_preserves_foreign_file(self) -> None:
@@ -720,25 +723,129 @@ def test_verify_bounds_hostile_archive_decompression(self) -> None:
         self._activate()
         mode = self._build_activation()
 
-        # A gzip bomb (tiny compressed, huge decompressed) must be rejected
-        # by the hard decompression bound BEFORE any tar/PAX parsing.
+        # A gzip bomb (tiny compressed, huge decompressed) with a CANONICAL
+        # header (mtime=0) must be rejected by the hard decompression bound,
+        # not materialized.
         import gzip as gzip_module
 
         bomb = self.root / "bomb.plugin"
         with bomb.open("wb") as raw:
-            with gzip_module.GzipFile(fileobj=raw, mode="wb") as stream:
+            # filename="" so no FNAME flag is set (canonical header), matching
+            # what the builder's os.fdopen path emits.
+            with gzip_module.GzipFile(
+                filename="", fileobj=raw, mode="wb", mtime=0
+            ) as stream:
                 zero_chunk = b"\0" * (1024 * 1024)
                 for _ in range(2 * 64 + 8):  # > 2 * MAX_ARTIFACT_BYTES
                     stream.write(zero_chunk)
         with self.assertRaisesRegex(ValueError, "decompression exceeds"):
             self.builder.verify_archive(self.plugin, bomb, mode=mode)
 
+        # A non-canonical gzip header (non-zero mtime) is rejected up front.
+        stamped = self.root / "stamped.plugin"
+        with stamped.open("wb") as raw:
+            with gzip_module.GzipFile(
+                fileobj=raw, mode="wb", mtime=1_700_000_000
+            ) as stream:
+                stream.write(b"whatever")
+        with self.assertRaisesRegex(ValueError, "gzip header is not canonical"):
+            self.builder.verify_archive(self.plugin, stamped, mode=mode)
+
         # A symlinked archive path is refused outright.
         alias = self.root / "archive-alias.plugin"
         alias.symlink_to(self.archive)
         with self.assertRaisesRegex(ValueError, "unreadable"):
             self.builder.verify_archive(self.plugin, alias, mode=mode)
 
+    def test_ustar_serializer_is_golden_across_python_versions(self) -> None:
+        # Design item 5: the regenerate-and-compare architecture requires the
+        # USTAR serializer to be byte-identical in the operator's build env
+        # (3.13.14) and CI (3.10/3.12/3.14). This golden binds the exact bytes;
+        # CI's version matrix fails here if any Python serializes USTAR
+        # differently, so a legitimate cross-env archive can never fail the
+        # verify byte-compare undetected.
+        buffer = io.BytesIO()
+        with tarfile.open(
+            fileobj=buffer, mode="w", format=tarfile.USTAR_FORMAT
+        ) as tar:
+            tar.addfile(
+                self.builder._synthesized_tarinfo(
+                    "runtime/darwin-arm64/agent-collab-runtime.bundle",
+                    mode=0o500,
+                    directory=True,
+                )
+            )
+            tar.addfile(
+                self.builder._synthesized_tarinfo(
+                    "runtime/darwin-arm64/agent-collab-runtime.bundle/agent-collab-runtime",
+                    mode=0o500,
+                    size=5,
+                ),
+                io.BytesIO(b"hello"),
+            )
+        data = buffer.getvalue()
+        self.assertEqual(
+            hashlib.sha256(data).hexdigest(),
+            "366c3a87cf260b77a3c937b6c44ad6f91d91f1073d59d195a9ae761ec66b6bd0",
+        )
+
+    def test_canonical_tar_is_byte_reproducible_across_calls(self) -> None:
+        # Design item 5: the canonical inflated tar must be byte-identical on
+        # regeneration (the cross-environment reproducibility contract the
+        # regenerate-and-compare architecture depends on). CI runs this on
+        # 3.10/3.12/3.14, so a serializer-version divergence would fail here.
+        self._activate()
+        plugin = self.plugin.resolve(strict=True)
+        frozen = self.builder._read_manifest_bytes(plugin)
+        records = self.builder.runtime_bundle.validate_file_records(
+            self.builder._parse_manifest(frozen)["artifacts"][0]["files"]
+        )
+        plan = self.builder._member_plan(plugin, mode="activation", records=records)
+        rbn = {
+            (self.builder.RUNTIME_BUNDLE_REL / r["path"]).as_posix(): r
+            for r in records
+        }
+        zero = {name: b"\x00" * r["size"] for name, r in rbn.items()}
+        first, ranges_a = self.builder._emit_canonical_tar(
+            plan, plugin_path=plugin, frozen_manifest=frozen,
+            record_by_name=rbn, runtime_payloads=zero,
+        )
+        second, ranges_b = self.builder._emit_canonical_tar(
+            plan, plugin_path=plugin, frozen_manifest=frozen,
+            record_by_name=rbn, runtime_payloads=zero,
+        )
+        self.assertEqual(first, second)
+        self.assertEqual(ranges_a, ranges_b)
+        # The reported ranges are one-to-one with the manifest, disjoint, and
+        # in-bounds (item 6).
+        self.builder._assert_runtime_range_map(ranges_a, rbn, len(first))
+
+    def test_verify_rejects_raw_structural_tamper_without_parsing(self) -> None:
+        # A hostile archive whose runtime payload digest is preserved but whose
+        # STRUCTURE differs (here: a directory member's mode flipped) must be
+        # rejected by the byte-compare, with no reliance on tarfile parsing the
+        # candidate. We tamper the inflated tar bytes directly.
+        self._activate()
+        mode = self._build_activation()
+        raw = self.archive.read_bytes()
+        import gzip as gzip_module
+
+        inflated = gzip_module.decompress(raw)
+        # Flip one byte inside the first 512-byte header block (the manifest or
+        # first member's metadata region) — a structural mutation.
+        tampered_tar = bytearray(inflated)
+        tampered_tar[100] ^= 0x01
+        tampered = self.root / "raw-tamper.plugin"
+        with tampered.open("wb") as fh:
+            with gzip_module.GzipFile(
+                filename="", fileobj=fh, mode="wb", mtime=0
+            ) as gz:
+                gz.write(bytes(tampered_tar))
+        with self.assertRaisesRegex(
+            ValueError, "structure does not match|canonical layout"
+        ):
+            self.builder.verify_archive(self.plugin, tampered, mode=mode)
+
     def test_activation_archive_build_is_deterministic(self) -> None:
         self._activate()
         self._build_activation()

From 2b3414aff0460f3fd6ccf9320e78afc9ac8cd82f Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 22:00:08 -0700
Subject: [PATCH 6/9] fix(archive): umask-normalize non-runtime modes; O(n)
 payload read (GLM cross-check)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

GLM cross-check on the regenerate-and-compare implementation: PROCEED-WITH-
MODIFICATIONS (12-point checklist all correct, no fail-open). Integrated both:

1. Cross-environment mode reproducibility (concern 1): non-runtime member
   modes came from gettarinfo (the working-tree checkout mode, which a
   checkout applies the local umask to). With build (operator host) and
   verify (CI host) now on different machines, a umask difference could make
   the regenerated canonical's mode fields differ → false-reject a legitimate
   archive. git tracks ONLY the exec bit, so normalize non-runtime modes to
   0o755/0o644 by the exec bit — umask-independent. Runtime members keep their
   sealed 0o500 from the manifest.
2. O(n^2) payload assembly (concern 2): _read_runtime_payloads now
   list-appends + b"".join instead of repeated b"" concatenation.

New test: non-runtime modes normalized to 0o644/0o755 even when a source file
is 0o600, and the archive still verifies. 337 tests/ + 254 scripts/ green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 19 +++++++++++++++----
 tests/test_plugin_archive.py    | 22 ++++++++++++++++++++++
 2 files changed, 37 insertions(+), 4 deletions(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index d7f7fed..67839c6 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -654,7 +654,15 @@ def _emit_canonical_tar(
                 tar.addfile(info, io.BytesIO(payload))
                 ranges[name] = (header_offset + _USTAR_BLOCK, record["size"])
                 continue
-            info = _finalize_tarinfo(tar.gettarinfo(str(source), arcname=name))
+            info = tar.gettarinfo(str(source), arcname=name)
+            # git tracks ONLY the exec bit, but a working-tree checkout applies
+            # the local umask to the rest — so normalize non-runtime modes to
+            # 0o755/0o644 by the exec bit alone. Without this the operator's
+            # build host and CI (different umask, different checkouts) could
+            # emit different mode fields and the verify byte-compare would
+            # false-reject a legitimate archive.
+            info.mode = 0o755 if (info.isdir() or (info.mode & 0o111)) else 0o644
+            info = _finalize_tarinfo(info)
             if name == "runtime-manifest.json" and frozen_manifest is not None:
                 info.size = len(frozen_manifest)
                 tar.addfile(info, io.BytesIO(frozen_manifest))
@@ -891,12 +899,15 @@ def _read_runtime_payloads(
                 or info.st_size != record["size"]
             ):
                 raise ValueError("runtime bundle source member changed during packing")
-            data = b""
-            while len(data) <= record["size"]:
+            parts: list[bytes] = []
+            read_total = 0
+            while read_total <= record["size"]:
                 chunk = os.read(descriptor, 1024 * 1024)
                 if not chunk:
                     break
-                data += chunk
+                parts.append(chunk)
+                read_total += len(chunk)
+            data = b"".join(parts)
             if (
                 len(data) != record["size"]
                 or hashlib.sha256(data).hexdigest() != record["sha256"]
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index de63376..39004b6 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -789,6 +789,28 @@ def test_ustar_serializer_is_golden_across_python_versions(self) -> None:
             "366c3a87cf260b77a3c937b6c44ad6f91d91f1073d59d195a9ae761ec66b6bd0",
         )
 
+    def test_non_runtime_modes_are_umask_normalized(self) -> None:
+        # Non-runtime member modes must be exactly 0o644 (files) / 0o755 (dirs)
+        # regardless of the checkout host's umask, so the operator's build and
+        # CI's regeneration byte-match. Perturb a source file to 0o600 (as a
+        # umask-077 checkout would) and confirm the archive normalizes it.
+        self._activate()
+        (self.plugin / "signing_policy.py").chmod(0o600)
+        mode = self._build_activation()
+        self.assertEqual(mode, "activation")
+        with tarfile.open(self.archive, "r:gz") as bundle:
+            for member in bundle.getmembers():
+                if member.name.startswith("runtime/"):
+                    continue  # runtime members keep the sealed 0o500
+                expected = 0o755 if member.isdir() else 0o644
+                self.assertEqual(
+                    stat.S_IMODE(member.mode),
+                    expected,
+                    f"{member.name} mode not normalized",
+                )
+        # And it still verifies (build and verify both normalize identically).
+        self.builder.verify_archive(self.plugin, self.archive, mode="activation")
+
     def test_canonical_tar_is_byte_reproducible_across_calls(self) -> None:
         # Design item 5: the canonical inflated tar must be byte-identical on
         # regeneration (the cross-environment reproducibility contract the

From bd977f843703a249120a41a48c27d17ae245a1b9 Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 22:09:44 -0700
Subject: [PATCH 7/9] fix(archive): single frozen-manifest snapshot;
 O_NONBLOCK+read-cap in bundle validator; full 10-byte gzip header
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex peer-review round 1 on the regenerate-and-compare implementation
(REQUEST_CHANGES/H); concerns 1-3 integrated (concern 4 was already
satisfied — see below):

1. Single-snapshot classification: classify_package() re-read the manifest
   separately from build/verify's freeze, so an A->B->A swap could classify A
   but package B. New _classify_from_manifest(plugin_path, bytes) derives the
   mode, the FULL activation-shape validation, AND the runtime records from
   ONE byte snapshot; build_archive and verify_archive freeze once and
   classify from that snapshot (policy-only included). classify_package() now
   reads once and delegates.
2. _validate_activation_bundle_tree hardened to match _read_runtime_payloads:
   O_NONBLOCK on the member open (a swapped FIFO cannot block) + a read cap at
   the fstat'd size (a file that grows after fstat cannot cause unbounded
   reading; read_total != size now fails closed).
3. The gzip-header check now binds the FULL canonical 10 bytes
   (1f8b 08 00 00000000 02 ff) — CPython writes XFL (level-9 best) and OS (ff)
   deterministically, so a re-gzip at a different level (XFL change) is
   rejected; the build->verify round-trip tests fail closed on any Python that
   diverges. New test: an identical-content archive re-gzipped at level 1 is
   rejected on the XFL byte.

Concern 4 (cross-env reproducibility proof) was already met by
test_ustar_serializer_is_golden_across_python_versions (golden sha256 of the
USTAR serializer, run on CI's 3.10/3.12/3.14 matrix) — the reviewer saw only
the script diff, not the test file. 337 tests/ + 254 scripts/ green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 93 +++++++++++++++++++--------------
 tests/test_plugin_archive.py    | 16 ++++++
 2 files changed, 70 insertions(+), 39 deletions(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index 67839c6..d2dc382 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -385,6 +385,7 @@ def _validate_activation_bundle_tree(
         os.O_RDONLY
         | getattr(os, "O_NOFOLLOW", 0)
         | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NONBLOCK", 0)  # a swapped FIFO must not block the open
     )
     for record in records:
         member = bundle_leaf / record["path"]
@@ -407,12 +408,16 @@ def _validate_activation_bundle_tree(
                     "runtime bundle source member identity is invalid"
                 )
             digest = hashlib.sha256()
-            while True:
+            read_total = 0
+            while read_total <= record["size"]:
+                # Cap the read at the fstat'd size: a file that grows after the
+                # fstat cannot cause unbounded reading.
                 chunk = os.read(descriptor, 1024 * 1024)
                 if not chunk:
                     break
+                read_total += len(chunk)
                 digest.update(chunk)
-            if digest.hexdigest() != record["sha256"]:
+            if read_total != record["size"] or digest.hexdigest() != record["sha256"]:
                 raise ValueError("runtime bundle source member digest is invalid")
         finally:
             try:
@@ -421,27 +426,45 @@ def _validate_activation_bundle_tree(
                 pass
 
 
-def classify_package(plugin_path: Path) -> str:
-    plugin_path = plugin_path.resolve(strict=True)
+def _classify_from_manifest(
+    plugin_path: Path, manifest_bytes: bytes
+) -> tuple[str, tuple[dict[str, object], ...]]:
+    """Classify + fully validate the package from ONE manifest byte snapshot.
+
+    The mode decision, the full activation-manifest shape validation, and the
+    runtime records ALL derive from the same ``manifest_bytes`` — so a caller
+    that freezes the manifest once and passes those exact bytes here cannot be
+    raced (an A→B→A manifest swap between classification and packing is
+    impossible; the frozen bytes are the single source).
+    """
+
     policy = plugin_path / "signing_policy.py"
     policy_info = _safe_source(policy)
     if not stat.S_ISREG(policy_info.st_mode):
         raise ValueError("signing_policy.py must be a regular file")
-    manifest = _load_manifest(plugin_path)
+    manifest = _parse_manifest(manifest_bytes)
     artifacts = manifest["artifacts"]
     runtime_root = plugin_path / "runtime"
     if not artifacts:
         if runtime_root.exists() or runtime_root.is_symlink():
             raise ValueError("policy-only package contains an unadvertised runtime")
-        return "policy-only"
+        return "policy-only", ()
     if len(artifacts) != 1:
         raise ValueError("activation package requires exactly one runtime artifact")
     # D1: the committed manifest is the activation marker. The physical bundle
     # is NOT read here — it ships as a release asset (never committed) and is
     # validated against its actual source by `_validate_activation_bundle_tree`
     # (build) and against the frozen manifest records (verify_archive).
-    _validate_activation_manifest(artifacts[0])
-    return "activation"
+    records = _validate_activation_manifest(artifacts[0])
+    return "activation", records
+
+
+def classify_package(plugin_path: Path) -> str:
+    plugin_path = plugin_path.resolve(strict=True)
+    mode, _records = _classify_from_manifest(
+        plugin_path, _read_manifest_bytes(plugin_path)
+    )
+    return mode
 
 
 def _require_exact_manifest_trees(plugin_path: Path) -> None:
@@ -568,12 +591,15 @@ def _member_plan(
 _MAX_COMPRESSED_ARCHIVE_BYTES = MAX_ARTIFACT_BYTES
 _MAX_DECOMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES
 _USTAR_BLOCK = 512
-# Fixed canonical gzip header: magic (1f 8b), DEFLATE method (08), flags 0
-# (rejects FTEXT/FHCRC/FEXTRA/FNAME/FCOMMENT — the malleability/amplification
-# vectors), and mtime 0. Bytes 8-9 (XFL, OS) are compressor/platform-derived,
-# informational, and not attacker-amplifiable, so they are deliberately not
-# bound — the inflated-tar byte-compare binds all CONTENT regardless.
-_CANONICAL_GZIP_PREFIX = b"\x1f\x8b\x08\x00\x00\x00\x00\x00"
+# The complete fixed canonical 10-byte gzip header the builder emits: magic
+# (1f 8b), DEFLATE method (08), flags 0 (rejects FTEXT/FHCRC/FEXTRA/FNAME/
+# FCOMMENT — the malleability/amplification vectors), mtime 0, XFL 02
+# (best-compression, GzipFile default level 9), OS ff (unknown). CPython's gzip
+# writes XFL and OS deterministically (hardcoded ff for OS; 02 for level 9), so
+# binding all ten is byte-reproducible across the operator's build Python and
+# CI's 3.10/3.12/3.14 — and the build→verify round-trip tests fail closed on
+# any version that diverges rather than silently accepting a variant.
+_CANONICAL_GZIP_HEADER = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff"
 
 
 def _synthesized_tarinfo(
@@ -729,7 +755,7 @@ def _inflate_bounded(compressed: bytes) -> bytes:
     complete with NO trailing/concatenated data.
     """
 
-    if len(compressed) < 10 or compressed[:8] != _CANONICAL_GZIP_PREFIX:
+    if compressed[:10] != _CANONICAL_GZIP_HEADER:
         raise ValueError("plugin archive gzip header is not canonical")
     decompressor = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS)
     chunk_size = 1024 * 1024
@@ -821,19 +847,14 @@ def verify_archive(
     """
 
     plugin_path = plugin_path.resolve(strict=True)
-    if classify_package(plugin_path) != mode:
+    # Freeze ONE manifest snapshot: mode classification, full shape validation,
+    # runtime records, and the embedded manifest member all derive from it (a
+    # caller-supplied snapshot from build_archive, or a single read here in CI).
+    if frozen_manifest is None:
+        frozen_manifest = _read_manifest_bytes(plugin_path)
+    resolved_mode, records = _classify_from_manifest(plugin_path, frozen_manifest)
+    if resolved_mode != mode:
         raise ValueError("archive mode no longer matches the source package")
-    records: tuple[dict[str, object], ...] = ()
-    if mode == "activation":
-        if frozen_manifest is None:
-            frozen_manifest = _read_manifest_bytes(plugin_path)
-        manifest = _parse_manifest(frozen_manifest)
-        artifacts = manifest["artifacts"]
-        if len(artifacts) != 1:
-            raise ValueError("activation package requires exactly one runtime artifact")
-        records = runtime_bundle.validate_file_records(artifacts[0]["files"])
-    else:
-        frozen_manifest = None
     plan = _member_plan(plugin_path, mode=mode, records=records)
     record_by_name = {
         (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records
@@ -932,10 +953,13 @@ def build_archive(
     if plugin != PLUGIN_NAME:
         raise ValueError("agent-collab is the only releaseable plugin")
     plugin_path = (root / "plugins" / plugin).resolve(strict=True)
-    mode = classify_package(plugin_path)
-    records: tuple[dict[str, object], ...] = ()
+    # Freeze the manifest ONCE: mode classification, full shape validation,
+    # runtime records, the bundle-tree validation, the embedded manifest member,
+    # and verify_archive ALL derive from this single snapshot, so a mid-build
+    # A→B→A manifest swap on disk cannot change what gets packaged.
+    frozen_manifest: bytes | None = _read_manifest_bytes(plugin_path)
+    mode, records = _classify_from_manifest(plugin_path, frozen_manifest)
     bundle_leaf: Path | None = None
-    frozen_manifest: bytes | None = None
     runtime_payloads: dict[str, bytes] = {}
     if mode == "policy-only":
         # The committed manifest — never the flag — decides the mode.
@@ -960,15 +984,6 @@ def build_archive(
         if stat.S_ISLNK(raw_source.st_mode) or not stat.S_ISDIR(raw_source.st_mode):
             raise ValueError("runtime bundle source is unsafe")
         bundle_leaf = bundle_source.resolve(strict=True)
-        # Freeze the manifest ONCE: these exact bytes drive the bundle-tree
-        # validation, are embedded into the archive verbatim, and are handed to
-        # verify_archive — a mid-build manifest swap on disk is irrelevant.
-        frozen_manifest = _read_manifest_bytes(plugin_path)
-        manifest = _parse_manifest(frozen_manifest)
-        artifacts = manifest["artifacts"]
-        if len(artifacts) != 1:
-            raise ValueError("activation package requires exactly one runtime artifact")
-        records = runtime_bundle.validate_file_records(artifacts[0]["files"])
         _validate_activation_bundle_tree(bundle_leaf, records)
         runtime_payloads = _read_runtime_payloads(bundle_leaf, records)
     plan = _member_plan(plugin_path, mode=mode, records=records)
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index 39004b6..ad82809 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -751,6 +751,22 @@ def test_verify_bounds_hostile_archive_decompression(self) -> None:
         with self.assertRaisesRegex(ValueError, "gzip header is not canonical"):
             self.builder.verify_archive(self.plugin, stamped, mode=mode)
 
+        # The FULL 10-byte header is bound: re-gzipping identical content at a
+        # different compression level (changes the XFL byte) is rejected even
+        # though the inflated content is identical.
+        inflated_ok = gzip_module.decompress(self.archive.read_bytes())
+        xfl_variant = self.root / "xfl.plugin"
+        with xfl_variant.open("wb") as raw:
+            with gzip_module.GzipFile(
+                filename="", fileobj=raw, mode="wb", mtime=0, compresslevel=1
+            ) as gz:
+                gz.write(inflated_ok)
+        self.assertNotEqual(
+            xfl_variant.read_bytes()[:10], self.archive.read_bytes()[:10]
+        )
+        with self.assertRaisesRegex(ValueError, "gzip header is not canonical"):
+            self.builder.verify_archive(self.plugin, xfl_variant, mode=mode)
+
         # A symlinked archive path is refused outright.
         alias = self.root / "archive-alias.plugin"
         alias.symlink_to(self.archive)

From 7dafc001573a438f239e054854e2eb7de47654ec Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 22:32:21 -0700
Subject: [PATCH 8/9] fix(archive): create build temp 0o600 not 0o644 (CodeQL
 CWE-732 py/overly-permissive-file)

The build temp (which becomes the release archive via os.replace) was
created world-readable. It is read + uploaded by the building user; no
other local user needs read access. Owner-only 0o600 clears the high-
severity CodeQL alert and is strictly safer for the release artifact.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index d2dc382..8c13748 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -1020,7 +1020,9 @@ def _reject_source_alias(candidate: Path) -> None:
     temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}"
     temp_created = False
     try:
-        descriptor = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
+        # Owner-only (0o600): the archive is read + uploaded by the building
+        # user; no other local user needs read access to the release artifact.
+        descriptor = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
         temp_created = True
         with os.fdopen(descriptor, "wb") as raw:
             with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed:

From f054668689dae848d9817c097ecd99372e34d0a6 Mon Sep 17 00:00:00 2001
From: John Osumi <931193+sumitake@users.noreply.github.com>
Date: Thu, 16 Jul 2026 22:40:48 -0700
Subject: [PATCH 9/9] fix(archive): compressed-archive cap headroom over
 runtime payload (Codex PR-review P2)

The compressed cap equaled MAX_ARTIFACT_BYTES (the runtime-payload cap), but
the archive also holds the manifest, skills, licenses, plugin metadata, tar
headers/padding, and gzip framing. A poorly-compressible runtime near the cap
would make the compressed .plugin exceed 64 MiB and verify_archive would
reject the builder's OWN legitimate output ("exceeds the canonical size
bound"). Both caps now carry 2x (128 MiB) headroom over the payload cap; the
decompressed cap still bounds a gzip bomb. New test asserts the cap
relationship.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 scripts/build_plugin_archive.py | 11 ++++++++++-
 tests/test_plugin_archive.py    | 15 +++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py
index 8c13748..7201c9d 100644
--- a/scripts/build_plugin_archive.py
+++ b/scripts/build_plugin_archive.py
@@ -588,7 +588,16 @@ def _member_plan(
 # trusted inputs and byte-compares, so a candidate archive is only ever sliced
 # by offsets WE computed. Only the bounded gzip inflater touches candidate
 # bytes, under hard compressed- and decompressed-size caps.
-_MAX_COMPRESSED_ARCHIVE_BYTES = MAX_ARTIFACT_BYTES
+#
+# The runtime payload alone may be as large as MAX_ARTIFACT_BYTES (64 MiB); the
+# archive additionally holds the manifest, skills, licenses, plugin metadata,
+# tar headers/padding, and gzip framing, and a poorly-compressible (already
+# compressed/encrypted) runtime near the limit barely shrinks. So BOTH the
+# compressed and decompressed caps carry headroom above the payload cap
+# (2x = 128 MiB) — otherwise the builder could reject its OWN legitimate output
+# — while the decompressed cap still bounds a gzip bomb well below memory
+# exhaustion.
+_MAX_COMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES
 _MAX_DECOMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES
 _USTAR_BLOCK = 512
 # The complete fixed canonical 10-byte gzip header the builder emits: magic
diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py
index ad82809..09bd33d 100644
--- a/tests/test_plugin_archive.py
+++ b/tests/test_plugin_archive.py
@@ -411,6 +411,21 @@ def test_archive_verifier_detects_source_byte_drift(self) -> None:
         with self.assertRaisesRegex(ValueError, "does not match the canonical layout"):
             self.builder.verify_archive(self.plugin, self.archive, mode=mode)
 
+    def test_compressed_archive_cap_has_headroom_over_runtime_payload(self) -> None:
+        # The runtime payload alone may be up to MAX_ARTIFACT_BYTES; the archive
+        # also holds the manifest/skills/licenses/headers/gzip framing, and a
+        # poorly-compressible runtime near the cap barely shrinks. The compressed
+        # cap must exceed the payload cap or the builder would reject its own
+        # legitimate output; the decompressed cap must still bound a bomb.
+        self.assertGreater(
+            self.builder._MAX_COMPRESSED_ARCHIVE_BYTES,
+            self.builder.MAX_ARTIFACT_BYTES,
+        )
+        self.assertGreaterEqual(
+            self.builder._MAX_DECOMPRESSED_ARCHIVE_BYTES,
+            self.builder._MAX_COMPRESSED_ARCHIVE_BYTES,
+        )
+
     def test_archive_size_limit_and_required_policy_member_are_canonical(self) -> None:
         self.assertEqual(self.builder.MAX_ARTIFACT_BYTES, 64 * 1024 * 1024)
         self.assertEqual(self.builder.RUNTIME_FILE_MODE, 0o500)