From a1249d5ed4c6003ba109a96159bef8a343581ef6 Mon Sep 17 00:00:00 2001
From: tzhouam <tzhouam@connect.ust.hk>
Date: Tue, 29 Sep 2026 01:01:00 +0800
Subject: [PATCH] Archive the trace store weekly and pull it to the GPU box

Traces are kept forever, and the design requires a weekly off-machine copy
(archive plus hash manifest) so a single bot host is never their only home.

- archive.make_archive (scheduler, every 7 days) packs everything not yet
  archived into <state_dir>/archive/traces-<seq>-<time>.tar.gz. What was
  archived is tracked explicitly in archive/archived.json (path -> content
  hash), never by timestamps. Blobs are immutable and archived once; a record
  file that grew is archived again; a blob renamed into place mid-archive is
  in the next archive. Names lead with a monotonic sequence, so name order
  is creation (and restore) order even if the clock moves backwards, and
  they are never reused or overwritten. index.db is rebuilt from the records and never archived. A .sha256 manifest (the archive's hash, then
  one line per member) is written last and marks the archive complete.
- The publisher, which already reaches the bot host over SSH, pulls every
  complete archive it does not have (kb publish, each round) and keeps it in
  $KB_PUBLISHER_STATE/archive only if the hash verifies. A mismatch is traced
  and rejected; transport failures are traced, never fatal.
- Restoring means extracting the archives in name order into an empty trace
  root, then running TraceStore.rebuild_index; the test does exactly that.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 doc/architecture/SPEC/kb_service.md           |   7 ++
 doc/features/knowledge-service.md             |   3 +
 src/infermatrix_copilot/kb_service/archive.py |  87 +++++++++++++++
 src/infermatrix_copilot/kb_service/cli.py     |   2 +-
 .../kb_service/publisher.py                   |  48 +++++++-
 .../kb_service/scheduler.py                   |   9 ++
 test/test_kb_archive.py                       | 105 ++++++++++++++++++
 7 files changed, 259 insertions(+), 2 deletions(-)
 create mode 100644 src/infermatrix_copilot/kb_service/archive.py
 create mode 100644 test/test_kb_archive.py

diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md
index d23760ea..8417248d 100644
--- a/doc/architecture/SPEC/kb_service.md
+++ b/doc/architecture/SPEC/kb_service.md
@@ -152,3 +152,10 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv
 各类操作数、待人工处理数、模型调用次数/token/耗时）写入 `<state_dir>/reports/sweep-<repo>-<tag>.md`；仅当仓库为
 `auto_merge` 且上游公开时，另通过新的 outbox 项 `open_issue`（7 天有效）在知识仓库发布一个 issue。发布器按标题复用已有
 issue（同一报告只开一个），标签不存在时不带标签重试，绝不丢报告。
+
+## 2026-09-29 留痕离机归档
+`archive.make_archive`（调度器每 7 天）把 `<state_dir>/traces` 中尚未归档的内容（不含可重建的 `index.db`）打包为
+`archive/traces-<序号>-<时间>.tar.gz`（序号在前且单调递增，名称顺序即恢复顺序，不受时钟回拨影响；从不复用或覆盖）。已归档内容以 `archive/archived.json`（路径 → 内容
+哈希）显式记录而非依赖时间戳：blob 不可变只归档一次，增长的记录文件再次归档，归档期间才落盘的 blob 进入下一个归档，并写 `.sha256` 清单（首行为包的哈希，其后每个成员一行；清单最后写入，
+标志归档完整）。发布器（GPU 盒，经已有的 SSH 通道）每轮拉取本地没有的归档，只有哈希校验通过才保存到
+`$KB_PUBLISHER_STATE/archive/`，否则记入 trace 并拒绝。恢复：按名称顺序解开到空目录后运行 `TraceStore.rebuild_index`。
diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md
index 9ffa2e1b..63da9853 100644
--- a/doc/features/knowledge-service.md
+++ b/doc/features/knowledge-service.md
@@ -111,6 +111,9 @@ infermatrix-copilot kb publish --remote bot-host:/path/to/kb-state --once   # 
 ALLOW_POST=1 ALLOW_PUSH=1 infermatrix-copilot kb publish --remote bot-host:/path/to/kb-state   # 常驻
 ```
 
+每轮还会拉取 bot 主机上每周生成的留痕归档（`$KB_STATE_DIR/archive/`），哈希校验通过后保存到
+`$KB_PUBLISHER_STATE/archive/`，作为离机副本。
+
 没有 `ALLOW_POST=1` 时只把将要执行的动作写入 `$KB_PUBLISHER_STATE/traces/publisher.jsonl`；推送分支还需要
 `ALLOW_PUSH=1`。控制记录超过 10 分钟未更新（服务可能宕机）时整轮不执行任何动作。
 
diff --git a/src/infermatrix_copilot/kb_service/archive.py b/src/infermatrix_copilot/kb_service/archive.py
new file mode 100644
index 00000000..60bb17f7
--- /dev/null
+++ b/src/infermatrix_copilot/kb_service/archive.py
@@ -0,0 +1,87 @@
+"""Weekly off-machine archive of the trace store.
+
+Traces are kept forever; a single bot host must not be their only copy. Once
+a week the scheduler packs everything not yet archived in
+``<state_dir>/traces`` into ``<state_dir>/archive/traces-<seq>-<time>.tar.gz``
+with a ``.sha256`` manifest (the archive's hash, then one line per member),
+written last to mark it complete. What was archived is tracked explicitly in
+``archive/archived.json`` (path -> content hash), never by timestamps: a
+blob renamed into place while an archive was being made is simply picked up by
+the next one, and a record file that grew since is archived again. Names carry
+a monotonic sequence FIRST, so name order is creation order whatever the
+wall clock does, and are never reused or overwritten. ``index.db`` is
+rebuilt from the records and never archived.
+
+The publisher on the GPU box, which already reaches the bot host over SSH,
+pulls every archive it does not have and keeps it only if the hash verifies.
+Restoring is extracting the archives in name order into an empty trace root
+(later copies of a record file replace earlier ones) and running
+``TraceStore.rebuild_index``.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import re
+import tarfile
+import time
+from pathlib import Path
+
+SEQUENCE = "trace_archive_seq"
+NAME = re.compile(r"traces-\d{6}-\d{8}-\d{6}\.tar\.gz")  # sequence first: name order = restore order
+INDEX = "archived.json"
+
+
+def _sha(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def make_archive(rt) -> Path | None:
+    """Pack everything not yet archived; None when nothing is new."""
+    root = rt.state_dir / "traces"
+    out_dir = rt.state_dir / "archive"
+    index_path = out_dir / INDEX
+    archived: dict[str, str] = json.loads(index_path.read_text(encoding="utf-8")) if index_path.exists() else {}
+    members: list[tuple[str, bytes]] = []
+    for path in sorted(root.rglob("*")) if root.is_dir() else []:
+        if not path.is_file() or path.name == "index.db" or path.name.endswith(".tmp"):
+            continue
+        rel = path.relative_to(root).as_posix()
+        if rel.startswith("blobs/") and rel in archived:
+            continue  # content-addressed: never changes once written
+        data = path.read_bytes()
+        digest = _sha(data)
+        if archived.get(rel) != digest:
+            members.append((rel, data))
+            archived[rel] = digest
+    if not members:
+        return None
+    sequence = int(rt.ledger.get_cursor("*", SEQUENCE) or 0) + 1
+    rt.ledger.set_cursor("*", SEQUENCE, str(sequence))  # consumed even if packing fails: never reused
+    name = f"traces-{sequence:06d}-" + time.strftime("%Y%m%d-%H%M%S", time.gmtime(rt.clock())) + ".tar.gz"
+    out_dir.mkdir(parents=True, exist_ok=True)
+    if (out_dir / name).exists():
+        raise FileExistsError(f"refusing to overwrite archive {name}")
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w:gz") as tar:
+        for rel, data in members:
+            info = tarfile.TarInfo(rel)
+            info.size, info.mode, info.mtime = len(data), 0o644, int(rt.clock())
+            tar.addfile(info, io.BytesIO(data))
+    blob = buffer.getvalue()
+    manifest = f"{_sha(blob)}  {name}\n" + "".join(f"{_sha(data)}  {rel}\n" for rel, data in members)
+    tmp = out_dir / f".{name}.tmp"
+    tmp.write_bytes(blob)
+    tmp.replace(out_dir / name)
+    (out_dir / f"{name}.sha256").write_text(manifest, encoding="utf-8")  # written last: marks it complete
+    tmp_index = out_dir / f".{INDEX}.tmp"
+    tmp_index.write_text(json.dumps(archived, sort_keys=True), encoding="utf-8")
+    tmp_index.replace(index_path)  # only after the archive is complete
+    return out_dir / name
+
+
+def verify_archive(data: bytes, manifest: str, name: str) -> bool:
+    first = (manifest.splitlines() or [""])[0].split()
+    return len(first) == 2 and first[1] == name and _sha(data) == first[0]
diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py
index d069d640..ab46b9e1 100644
--- a/src/infermatrix_copilot/kb_service/cli.py
+++ b/src/infermatrix_copilot/kb_service/cli.py
@@ -300,7 +300,7 @@ def _publish(args) -> int:
         allow_push=os.environ.get("ALLOW_PUSH") == "1",
     )
     if args.once:
-        json.dump(publisher.run_once(), sys.stdout, sort_keys=True)
+        json.dump({**publisher.run_once(), **publisher.sync_archives()}, sys.stdout, sort_keys=True)
         print()
         return 0
     publisher.serve(interval=args.interval)
diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py
index b9b88f93..0d3a368c 100644
--- a/src/infermatrix_copilot/kb_service/publisher.py
+++ b/src/infermatrix_copilot/kb_service/publisher.py
@@ -44,6 +44,7 @@
 from .outbox import CONTROL_MAX_AGE, OutboxError, OutboxItem, check_item
 
 ITEM_ID = re.compile(r"[0-9]+-[0-9a-f]{12}")
+ARCHIVE_NAME = re.compile(r"traces-\d{6}-\d{8}-\d{6}\.tar\.gz")
 GOVERNED_PATH = re.compile(r"knowledge/(?:repos/[A-Za-z0-9._-]+|general)/(?:[A-Za-z0-9._-]+/)*[A-Za-z0-9._-]+\.(?:md|yaml)")
 BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+")
 HOLD_LABEL = "kb:hold"
@@ -63,6 +64,7 @@ class Transport(Protocol):
     def list_items(self) -> list[str]: ...
     def read(self, rel: str) -> bytes: ...
     def write_ack(self, item_id: str, data: bytes) -> None: ...
+    def list_archives(self) -> list[str]: ...
 
 
 @dataclass
@@ -79,6 +81,12 @@ def list_items(self) -> list[str]:
     def read(self, rel: str) -> bytes:
         return (self.root / rel).read_bytes()
 
+    def list_archives(self) -> list[str]:
+        folder = self.root / "archive"
+        return sorted(p.name for p in folder.glob("traces-*.tar.gz")
+                      if ARCHIVE_NAME.fullmatch(p.name) and (folder / f"{p.name}.sha256").exists()) \
+            if folder.is_dir() else []
+
     def write_ack(self, item_id: str, data: bytes) -> None:
         folder = self.root / "inbox" / "acks"
         folder.mkdir(parents=True, exist_ok=True)
@@ -116,6 +124,11 @@ def list_items(self) -> list[str]:
     def read(self, rel: str) -> bytes:
         return self._ssh(f"cat {shlex.quote(self.root + '/' + rel)}")
 
+    def list_archives(self) -> list[str]:
+        out = self._ssh(f"ls -1 {shlex.quote(self.root + '/archive')} 2>/dev/null || true").decode().split()
+        names = set(out)
+        return sorted(n for n in names if ARCHIVE_NAME.fullmatch(n) and f"{n}.sha256" in names)
+
     def write_ack(self, item_id: str, data: bytes) -> None:
         folder = shlex.quote(self.root + "/inbox/acks")
         tmp, final = shlex.quote(f"{self.root}/inbox/acks/.{item_id}.tmp"), \
@@ -187,6 +200,39 @@ def _send_ack(self, ack: dict) -> bool:
             return False
 
     # one round -----------------------------------------------------------------
+    def sync_archives(self) -> dict:
+        """Pull every weekly trace archive not yet here, keeping only those whose
+        hash verifies (the GPU box is the off-machine copy)."""
+        from .archive import verify_archive
+
+        dest = self.state_dir / "archive"
+        counts = {"pulled": 0, "rejected": 0}
+        try:
+            names = self.transport.list_archives()
+        except (OutboxError, OSError) as exc:
+            self._trace(event="archive_unreachable", error=str(exc))
+            return counts
+        for name in names:
+            if (dest / f"{name}.sha256").exists():
+                continue
+            try:
+                data = self.transport.read(f"archive/{name}")
+                manifest = self.transport.read(f"archive/{name}.sha256").decode("utf-8", "replace")
+            except (OutboxError, OSError) as exc:
+                self._trace(event="archive_unreachable", archive=name, error=str(exc))
+                continue
+            if not verify_archive(data, manifest, name):
+                counts["rejected"] += 1
+                self._trace(event="archive_rejected", archive=name)
+                continue
+            dest.mkdir(parents=True, exist_ok=True)
+            (dest / f".{name}.tmp").write_bytes(data)
+            (dest / f".{name}.tmp").replace(dest / name)
+            (dest / f"{name}.sha256").write_text(manifest, encoding="utf-8")  # last: marks it complete
+            counts["pulled"] += 1
+            self._trace(event="archive_pulled", archive=name, bytes=len(data))
+        return counts
+
     def run_once(self) -> dict:
         summary = {"performed": 0, "failed": 0, "resent": 0, "undelivered": 0, "dry_run": 0, "skipped": 0}
         try:
@@ -444,5 +490,5 @@ def serve(self, *, interval: float, stop=None) -> None:
 
         stop = stop or threading.Event()
         while not stop.is_set():
-            self._trace(event="round", **self.run_once())
+            self._trace(event="round", **self.run_once(), **self.sync_archives())
             stop.wait(interval)
diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py
index 3b49f8f6..5979bb22 100644
--- a/src/infermatrix_copilot/kb_service/scheduler.py
+++ b/src/infermatrix_copilot/kb_service/scheduler.py
@@ -26,6 +26,7 @@
 
 from . import merge
 from .activate import ActivationError, activate, activation_lock
+from .archive import make_archive
 from .audit import audit_main
 from .report import flush_reports
 from .external import poll_external
@@ -43,6 +44,7 @@ class Scheduler:
     intake_every: float = 900.0
     release_every: float = 3600.0
     audit_every: float = 24 * 3600.0
+    archive_every: float = 7 * 24 * 3600.0
     log: list[dict] = field(default_factory=list)
     _last: dict[str, float] = field(default_factory=dict)
 
@@ -77,6 +79,13 @@ def tick(self) -> None:
                 self._repo_tick(lifecycle)
             except Exception as exc:  # isolate repositories from each other
                 self._record(lifecycle.repo, "error", error=repr(exc), trace=traceback.format_exc()[-2000:])
+        if self._due("archive", self.archive_every):
+            try:  # the weekly off-machine copy of the traces (pulled by the publisher)
+                archive = make_archive(rt)
+                if archive is not None:
+                    self._record("*", "archive", archive=archive.name)
+            except Exception as exc:
+                self._record("*", "error", error=repr(exc), trace=traceback.format_exc()[-2000:])
         if self._due("audit", self.audit_every):
             try:  # every knowledge change on main must have been recorded
                 for finding in audit_main(rt, self._pause):
diff --git a/test/test_kb_archive.py b/test/test_kb_archive.py
new file mode 100644
index 00000000..f294abe2
--- /dev/null
+++ b/test/test_kb_archive.py
@@ -0,0 +1,105 @@
+"""Weekly trace archives: made on the bot host, pulled and verified by the publisher."""
+
+from __future__ import annotations
+
+import io
+import tarfile
+import time
+
+from infermatrix_copilot.kb_service.archive import make_archive, verify_archive
+from infermatrix_copilot.trace_store import TraceStore
+from test_kb_flow import _flow_runtime
+
+
+def _runtime(tmp_path):
+    rt, lifecycle = _flow_runtime(tmp_path)
+    rt.clock = lambda: time.time() + 5                    # files written "now" are before the archive time
+    rt.traces = TraceStore(rt.state_dir / "traces", environ={})
+    return rt
+
+
+def test_an_archive_holds_new_records_and_blobs_and_restores_the_store(tmp_path):
+    rt = _runtime(tmp_path)
+    record = rt.traces.append("decision", inputs={"prompt": "why"}, result={"status": "pass"})
+    path = make_archive(rt)
+    assert path is not None and (path.parent / f"{path.name}.sha256").exists()
+    manifest = (path.parent / f"{path.name}.sha256").read_text()
+    assert verify_archive(path.read_bytes(), manifest, path.name)
+    with tarfile.open(fileobj=io.BytesIO(path.read_bytes())) as tar:
+        names = tar.getnames()
+        assert "index.db" not in names and any(n.startswith("records/") for n in names)
+        restored = tmp_path / "restored"
+        tar.extractall(restored, filter="data")
+    store = TraceStore(restored)
+    assert store.rebuild_index() == 1 and store.get(record["id"]) == record
+    assert store.blob(record["inputs"]["prompt"]) == "why"
+
+
+def test_archives_are_incremental_and_never_reuse_a_name(tmp_path):
+    rt = _runtime(tmp_path)
+    rt.clock = lambda: 1790000000.0                       # every archive in the same second
+    rt.traces.append("decision", inputs={"prompt": "first"}, result={"status": "pass"})
+    first = make_archive(rt)
+    assert make_archive(rt) is None                       # nothing new since
+    rt.traces.append("decision", inputs={"prompt": "second"}, result={"status": "fail"})
+    second = make_archive(rt)
+    assert second is not None and second.name != first.name and first.exists()
+    with tarfile.open(fileobj=io.BytesIO(second.read_bytes())) as tar:
+        names = tar.getnames()
+    assert len([n for n in names if n.startswith("blobs/")]) == 1   # only the new blob
+    assert any(n.startswith("records/") for n in names)              # the grown record file again
+
+
+def test_a_blob_that_lands_during_archiving_is_in_the_next_archive(tmp_path):
+    import os
+
+    rt = _runtime(tmp_path)
+    rt.traces.append("decision", inputs={"prompt": "early"}, result={"status": "pass"})
+    blobs = rt.state_dir / "traces" / "blobs" / "ab"
+    blobs.mkdir(parents=True, exist_ok=True)
+    pending = blobs / ".late.tmp"                          # put_blob's temp file, mid-write
+    pending.write_bytes(b"late")
+    os.utime(pending, (1, 1))                              # an old mtime survives the rename
+    first = make_archive(rt)
+    pending.replace(blobs / ("ab" + "0" * 62 + ".gz"))     # renamed after the scan
+    second = make_archive(rt)
+    assert second is not None
+    with tarfile.open(fileobj=io.BytesIO(second.read_bytes())) as tar:
+        assert tar.getnames() == ["blobs/ab/ab" + "0" * 62 + ".gz"]
+    assert first.name != second.name
+
+
+def test_the_publisher_pulls_verified_archives_once_and_rejects_tampered_ones(tmp_path):
+    from test_kb_publisher import _setup
+
+    rt, lifecycle, _changeset_id, pub, _gh = _setup(tmp_path)
+    rt.clock = lambda: time.time() + 5
+    rt.traces = TraceStore(rt.state_dir / "traces", environ={})
+    rt.traces.append("decision", result={"status": "pass"})
+    path = make_archive(rt)
+    assert pub.sync_archives() == {"pulled": 1, "rejected": 0}
+    assert (tmp_path / "publisher" / "archive" / path.name).read_bytes() == path.read_bytes()
+    assert pub.sync_archives() == {"pulled": 0, "rejected": 0}          # once
+    other = path.parent / "traces-999999-20000101-000000.tar.gz"
+    other.write_bytes(b"not what the manifest says")
+    (path.parent / f"{other.name}.sha256").write_text(f"{'0' * 64}  {other.name}\n")
+    assert pub.sync_archives() == {"pulled": 0, "rejected": 1}
+    assert not (tmp_path / "publisher" / "archive" / other.name).exists()
+
+
+def test_restoring_in_name_order_survives_a_clock_moving_backwards(tmp_path):
+    rt = _runtime(tmp_path)
+    rt.clock = lambda: 1790000000.0
+    first_record = rt.traces.append("decision", result={"status": "pass"})
+    first = make_archive(rt)
+    rt.clock = lambda: 1780000000.0                       # the clock jumped back
+    second_record = rt.traces.append("decision", result={"status": "fail"})
+    second = make_archive(rt)
+    assert sorted([first.name, second.name]) == [first.name, second.name]
+    restored = tmp_path / "restored"
+    for archive in sorted((rt.state_dir / "archive").glob("traces-*.tar.gz")):
+        with tarfile.open(archive) as tar:
+            tar.extractall(restored, filter="data")
+    store = TraceStore(restored)
+    store.rebuild_index()
+    assert store.get(first_record["id"]) and store.get(second_record["id"])