fix(recovery): bind archive scans to their compressed byte witness

docs/architecture/rfcs/complete-state-recovery-v0.md

@@ -42,6 +42,13 @@ remains partial for compound worst cases and broader platform/provider profiles.
 This checkpoint does not qualify M2/M3, live adoption, remote providers or the G1
 team journey. No original identity or receipt is rewritten.
 
+Each gzip/tar scan witnesses the compressed bytes it actually consumes. Its
+digest and byte count must match preflight before manifest verification or audit
+publication; changing the input and restoring it before a later hash does not
+substitute for that witness. The copied scan retains the same check. Wrong input
+digests still refuse before tar decoding. Archive, expansion and metadata limits,
+inert status and execution-authority rules remain unchanged.
+
 ### Delivery after the writer cutoff (2026-10-16)
 
 The landed [R5/T4 cutoff](loopx-overall-roadmap-v0.md#post-6396-delivery-plan)

docs/architecture/rfcs/complete-state-recovery-v0.zh-CN.md

@@ -32,6 +32,11 @@ held，未知状态不改称可恢复，执行权限始终为 false。限制与
 最坏情形及更广的平台/provider 资格继续开放。
 本检查点不证明 M2/M3、live adoption、远端 provider 或 G1 团队验收，不重写原身份或回执。
 
+每次 gzip/tar 扫描都为实际消费的压缩字节生成见证。其 digest 和字节数必须在后续
+manifest 验证或审计发布前与 preflight 一致；修改输入后在末次 hash 前恢复，不能
+代替这份见证。复制扫描保留同样的检查。错误的输入 digest 仍在 tar 解码前拒绝。
+归档、展开、元数据容量、inert 状态及执行权限规则保留。
+
 
 ### writer 截止后的交付（2026-10-16）
 

loopx/control_plane/state_recovery_io.py

@@ -128,10 +128,33 @@ def _sha(handle: BinaryIO) -> tuple[str, int]:
     return digest.hexdigest(), count
 
 
+class _ArchiveInput(io.RawIOBase):
+    """Witness the compressed bytes actually consumed by the parser."""
+
+    def __init__(self, handle: BinaryIO):
+        super().__init__()
+        self.handle = handle
+        self.digest = hashlib.sha256()
+        self.total = 0
+
+    def readable(self) -> bool:
+        return True
+
+    def readinto(self, buffer: Any) -> int:
+        data = self.handle.read(min(len(buffer), MAX_ARCHIVE_BYTES - self.total + 1))
+        self.total += len(data)
+        _require(self.total <= MAX_ARCHIVE_BYTES, "recovery_capacity_exceeded")
+        self.digest.update(data)
+        buffer[:len(data)] = data
+        return len(data)
+
+
 class _BoundedGzip(io.RawIOBase):
     def __init__(self, handle: BinaryIO, archive_size: int):
         super().__init__()
-        self.source = gzip.GzipFile(fileobj=handle, mode="rb")
+        self.archive = _ArchiveInput(handle)
+        self.compressed = io.BufferedReader(self.archive, buffer_size=_CHUNK)
+        self.source = gzip.GzipFile(fileobj=self.compressed, mode="rb")
         self.limit = min(MAX_EXPANDED_BYTES, archive_size * MAX_EXPANSION_RATIO)
         self.total = 0
 
@@ -142,6 +165,13 @@ def read(self, size: int = -1) -> bytes:
         _require(self.total <= self.limit, "recovery_capacity_exceeded")
         return data
 
+    def close(self) -> None:
+        try:
+            self.source.close()
+        finally:
+            self.compressed.close()
+            super().close()
+
 
 class _RecoveryTarInfo(tarfile.TarInfo):
     def _reject_sparse(self, *_args: object) -> NoReturn:
@@ -350,7 +380,7 @@ def _scan(handle: BinaryIO, archive_size: int, output: Path | None = None) -> di
             ValueError, OverflowError, KeyError, RecursionError) as exc:
         raise StateRecoveryError("archive_unreadable") from exc
     finally:
-        stream.source.close()
+        stream.close()
     _require(embedded is not None, "embedded_manifest_missing")
     _check_links(links)
     # Links are materialized last and no member may live beneath a link.
@@ -361,7 +391,8 @@ def _scan(handle: BinaryIO, archive_size: int, output: Path | None = None) -> di
         for name, target in links:
             _directories(output, PurePosixPath(name).parts[:-1])
             os.symlink(target, output / name)
-    return {"embedded_manifest": embedded, "files": files, "logical_bytes": logical,
+    return {"archive_sha256": stream.archive.digest.hexdigest(), "archive_size_bytes": stream.archive.total,
+            "embedded_manifest": embedded, "files": files, "logical_bytes": logical,
             "sqlite_members": sqlite_members, "standalone_sqlite_members": standalone,
             "entry_count": len(names)}
 
@@ -383,6 +414,8 @@ def verify_state_backup(
         _runtime("state.recovery.archive.verify", {"external_manifest": external,
             "archive_sha256": archive_sha, "archive_size_bytes": archive_size})
         observation = _scan(archive_handle, archive_size)
+        _require(observation["archive_sha256"] == archive_sha
+                 and observation["archive_size_bytes"] == archive_size, "recovery_input_changed")
         verified = _runtime("state.recovery.manifest.verify", {
             "external_manifest": external, "embedded_manifest": observation["embedded_manifest"],
             "archive_sha256": archive_sha, "archive_size_bytes": archive_size,

skills/loopx-self-repair/references/repair-patterns.md

@@ -5,6 +5,7 @@ teaches a reusable control-plane lesson.
 
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
+| `archive_verification_byte_epoch_gap` | Backup preview or execution accepts compressed input rewritten after preflight; restoring it before the final hash conceals the change. | External manifest and preflight archive identity, the digest/count of bytes actually consumed by gzip/tar, copy observations and destination publication. | Separate mutable-file reads supply the hash and parser facts; a final hash cannot prove which bytes were parsed. | Keep early wrong-digest refusal. Generate a bounded compressed-byte witness in the parser's actual input stream, compare it with preflight before audit, and compare copy witnesses before publication. Preserve capacity checks, inert status and no execution authority. Cover preview, execute and ABA restoration with real producer archives and filesystem fault injection. |
 | `legacy_goal_command_current_work_gap` | A configured Goal command remains executable heartbeat guidance after current Todo selection, a quiet/replan route, or settlement has superseded it. | The native full decision and original capture, exact selected/adaptive work, final interaction channels, and command provenance. | Legacy recommendation precedence or copied refinements retain a command without binding it to current work; a later reentry can supersede an earlier projection again. | Keep legacy fields at their existing typed quota projection boundary and derive admission, action and spend policy from the canonical interaction owner on each reentry. Preserve source configuration, admitted unscoped command compatibility and exact governed-intent commands. Exercise real File/SQLite CLI selection, default/detail/Envelope views, capture, settlement and replay; never execute a stale command or create a human gate to resolve a projection contradiction. |
 | `outcome_review_source_capacity` | Repeated milestone reviews fail with `goal_acceptance_invalid_request` and `outcome sources exceeds its supported capacity` or `outcome source capacity reached`. | The exact failed operation/receipt, inspected original outcome sources and requirements, and its review evidence; verify whether refinement committed before a later writeback failure. | Requirement provenance is used as an append-only progress log; older native/CLI contracts required a source even for a review. | Keep the original outcome and all origins. Existing-outcome reviews/work links omit source (native: source_summary); put observations in review evidence/notes and owning materials. Creation/new requirements still need real provenance. On an older build, use an inspected existing source with exactly the same ref/summary for unchanged requirements only. Never prune origins, inflate limits, invent a new outcome to escape capacity, or silently relabel a new request. Preserve failed receipts and exact-operation recovery; qualify repeated reviews at the source limit and ownership/CAS/replay negative cases through real backends. |
 | `outcome_state_byte_capacity` | Outcome refinement fails with declaration or review capacity exceeded, even with one source or a review-only patch. | Preserve the exact response and operation identity; inspect the failing budget and its candidate/retained/limit UTF-8 bytes, then the affected current outcome. A source-count failure is a different constraint. | Declarations and current reviews have independent aggregate budgets across all supporting outcomes. An expanded review can exceed its budget without adding any source. Packet bytes and model tokens do not measure either budget. | For review overflow, compact only proposed review notes/evidence references while retaining every conclusion and its provenance; keep full evidence in its owning material. Shortening reviews cannot repair declaration overflow. Do not change original requirements/origins, delete history, inflate limits or create a replacement outcome to evade either budget. If no lossless update fits, report the capacity blocker. Validation rejection writes no outcome state. After readback, CLI correction can reuse the rejected operation; native conversation envelopes require a new operation id for changed intent. For uncertain writes, inspect the original receipt first and retain unchanged retry intent. Qualify both budget boundaries, rejection/no mutation, correction and replay through real CLI/backends. |

tests/test_state_recovery.py

@@ -645,6 +645,74 @@ def test_digest_refusal_precedes_tar_decoding_and_writes_nothing(tmp_path):
     assert not list(tmp_path.glob(".loopx-recovery-*"))
 
 
+@pytest.mark.parametrize("execute", [False, True])
+def test_archive_bytes_changed_between_reads_are_rejected_even_after_restore(tmp_path, monkeypatch, execute):
+    archive, manifest = _backup(tmp_path)
+    original_bytes = archive.read_bytes()
+    changed = bytearray(original_bytes)
+    changed[4] ^= 1  # gzip mtime changes identity without changing tar members.
+    changed_bytes = bytes(changed)
+    archive_inode = archive.stat().st_ino
+    original_fdopen = os.fdopen
+
+    class ChangingArchive(io.BufferedReader):
+        passes = 0
+        eof = False
+
+        def note_eof(self):
+            if not self.eof:
+                self.passes += 1
+                self.eof = True
+
+        def read(self, size=-1):
+            value = super().read(size)
+            if not value:
+                self.note_eof()
+            return value
+
+        def readinto(self, buffer):
+            count = super().readinto(buffer)
+            if not count:
+                self.note_eof()
+            return count
+
+        def seek(self, offset, whence=0):
+            if offset == 0 and whence == 0:
+                if self.passes == 1:
+                    archive.write_bytes(changed_bytes)
+                elif self.passes >= 3:
+                    archive.write_bytes(original_bytes)
+                # Force real file reads rather than replaying a buffered prefix.
+                super().seek(0, os.SEEK_END)
+                self.eof = False
+            return super().seek(offset, whence)
+
+    def input_file(fd, *args, **kwargs):
+        handle = original_fdopen(fd, *args, **kwargs)
+        return ChangingArchive(handle.detach()) if os.fstat(fd).st_ino == archive_inode else handle
+
+    monkeypatch.setattr(os, "fdopen", input_file)
+    destination = tmp_path / "unpublished"
+    with pytest.raises(StateRecoveryError, match="recovery_input_changed"):
+        verify_state_backup(archive=archive, manifest=manifest, destination=destination, execute=execute)
+    assert not destination.exists()
+    assert not list(tmp_path.glob(".loopx-recovery-*"))
+
+
+def test_archive_stream_enforces_the_compressed_byte_budget(monkeypatch):
+    from loopx.control_plane import state_recovery_io as recovery
+
+    data = io.BytesIO()
+    with tarfile.open(fileobj=data, mode="w:gz") as archive:
+        member = tarfile.TarInfo("manifest.json")
+        member.size = 2
+        archive.addfile(member, io.BytesIO(b"{}"))
+    compressed = data.getvalue()
+    monkeypatch.setattr(recovery, "MAX_ARCHIVE_BYTES", len(compressed) - 1)
+    with pytest.raises(StateRecoveryError, match="recovery_capacity_exceeded"):
+        recovery._scan(io.BytesIO(compressed), len(compressed))
+
+
 def test_external_manifest_cannot_relabel_the_embedded_capture(tmp_path):
     archive, manifest = _backup(tmp_path)
     data = json.loads(manifest.read_text())