Keep knowledge snapshots that unfinished runs pinned

doc/architecture/SPEC/kb_service.md

@@ -174,3 +174,10 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝
 （提交的适配器基线落后于审计的上游）不阻塞巡检，而是起草一个配套 PR：插件以 `baseline_file`（相对适配器目录）指明基线
 文件，只替换其中的 `upstream:` 与 `inventories:` 块（其余内容与注释保留），经 `open_companion_pr` 以 draft 发布并转人工；
 每次巡检最多一个。巡检汇总列出 reconciliation 数量。
+
+## 2026-09-29 快照保留与运行钉住
+`activate.prune` 保留最新 10 个快照与当前激活快照之外，还保留被**未结束**运行钉住的快照（运行目录中的
+`knowledge.json` 指向该快照且 `run_status` 非终态）。运行目录位置取 `KB_RUN_ROOTS`（`os.pathsep` 分隔，可列多个服务）、
+否则运行服务实际生效的 `run_root`（环境变量或其支持的 `.env` 文件）。切换激活快照时记录旧快照的停用时间；
+停用不足 1 小时的快照不删除——在切换前刚解析到它的预留会在此窗口内写下钉住记录（运行服务在其他进程中，无法共享
+激活锁，因此以宽限期协调）。测试通过自动 fixture 把 `KB_RUN_ROOTS` 指向空目录，从不读取真实运行目录。

src/infermatrix_copilot/kb_service/activate.py

@@ -16,6 +16,7 @@
 import os
 import shutil
 import tempfile
+import time
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Iterator
@@ -25,6 +26,10 @@
 from .outbox import atomic_write_json
 
 KEEP = 10
+# a snapshot switched away from less than this ago is never pruned: a run that
+# resolved it just before the switch publishes its pin within this window
+PIN_GRACE = 3600
+DEACTIVATED = ".deactivated-{name}"
 
 
 class ActivationError(RuntimeError):
@@ -111,6 +116,11 @@ def verify_snapshot(snapshot: Path) -> KnowledgeView:
 
 def switch_active(state_dir: Path, snapshot: Path) -> None:
     link = active_link(state_dir)
+    if link.is_symlink():
+        previous = link.resolve()
+        if previous != snapshot.resolve() and previous.parent == snapshots_dir(state_dir).resolve():
+            # when it stopped being active: pruning honours a grace period after this
+            (snapshots_dir(state_dir) / DEACTIVATED.format(name=previous.name)).write_text("", encoding="utf-8")
     tmp = link.with_name(f".active.{os.getpid()}")
     if tmp.is_symlink() or tmp.exists():
         tmp.unlink()
@@ -149,13 +159,63 @@ def rollback(rt, sha: str) -> Path:
         return snapshot
 
 
-def prune(state_dir: Path, keep: int = KEEP) -> None:
+def run_roots() -> list[Path]:
+    """Where the review runs that may pin a snapshot keep their run
+    directories: ``KB_RUN_ROOTS`` (os.pathsep-separated), else the runs'
+    ``RUN_ROOT``, else its default."""
+    raw = os.environ.get("KB_RUN_ROOTS")
+    if raw:
+        return [Path(part).expanduser() for part in raw.split(os.pathsep) if part]
+    try:  # the runs' EFFECTIVE run_root (environment or the supported .env files)
+        from ..config import Settings
+
+        return [Path(Settings().run_root).expanduser()]
+    except Exception:  # noqa: BLE001 - unreadable config: the documented default
+        return [Path(os.environ.get("RUN_ROOT") or Path.home() / ".infermatrix-copilot" / "runs").expanduser()]
+
+
+def pinned_snapshots(roots: list[Path]) -> set[Path]:
+    """Snapshot directories pinned (``<run>/knowledge.json``) by runs that have
+    not finished: a queued or running Strict review must keep its knowledge."""
+    from .. import run_status as rs
+
+    pinned: set[Path] = set()
+    for root in roots:
+        if not root.is_dir():
+            continue
+        for run_dir in root.iterdir():
+            try:
+                pin = json.loads((run_dir / "knowledge.json").read_text(encoding="utf-8"))
+            except (OSError, ValueError):
+                continue
+            target = pin.get("knowledge_root") if isinstance(pin, dict) else None
+            if not target:
+                continue
+            status = rs.read_status(run_dir) or {}  # ONE read: a reclaim may change it meanwhile
+            state = status.get("state")
+            # an interrupted reservation that never launched (no child pid) can
+            # be reclaimed and will run with its original pin
+            if state not in rs.TERMINAL or (state == rs.INTERRUPTED and status.get("child_pid") is None):
+                pinned.add(Path(target).resolve())
+    return pinned
+
+
+def prune(state_dir: Path, keep: int = KEEP, *, roots: list[Path] | None = None) -> None:
+    """Keep the newest ``keep`` snapshots, the active one, and every snapshot an
+    unfinished run pinned; remove the rest."""
     root = snapshots_dir(state_dir)
     link = active_link(state_dir)
     active = link.resolve() if link.is_symlink() else None
+    pinned = pinned_snapshots(run_roots() if roots is None else roots)
     snapshots = sorted((p for p in root.iterdir() if p.is_dir() and not p.name.startswith(".")),
                        key=lambda p: p.stat().st_mtime, reverse=True)
+    now = time.time()
     for old in snapshots[keep:]:
-        if active is not None and old.resolve() == active:
+        resolved = old.resolve()
+        if (active is not None and resolved == active) or resolved in pinned:
             continue
+        marker = root / DEACTIVATED.format(name=old.name)
+        if marker.exists() and now - marker.stat().st_mtime < PIN_GRACE:
+            continue  # deactivated too recently: a reservation may be pinning it right now
         shutil.rmtree(old, ignore_errors=True)
+        marker.unlink(missing_ok=True)

test/conftest.py

@@ -89,3 +89,10 @@ def git(*args: str) -> None:
     git("add", ".")
     git("commit", "-q", "-m", "init")
     return repo
+
+
+@pytest.fixture(autouse=True)
+def _no_real_run_roots(tmp_path, monkeypatch):
+    """Snapshot pruning looks for runs that pin a snapshot; tests never read
+    the real ~/.infermatrix-copilot/runs."""
+    monkeypatch.setenv("KB_RUN_ROOTS", str(tmp_path / "no-runs"))

test/test_kb_flow.py

@@ -686,3 +686,98 @@ def apply_ack():
     acker.join(5)
     assert not errors, errors
     assert rt.ledger.changeset(changeset_id)["status"] == "pr_open"  # never stranded as paused
+
+
+def test_prune_keeps_snapshots_pinned_by_unfinished_runs(tmp_path):
+    import json
+    import os
+
+    from infermatrix_copilot import run_status as rs
+
+    state = tmp_path / "state"
+    for n in range(4):
+        path = snapshots_dir(state) / f"s{n}"
+        path.mkdir(parents=True)
+        os.utime(path, (1000 + n, 1000 + n))
+    active_link(state).symlink_to((snapshots_dir(state) / "s3").resolve())
+    runs = tmp_path / "runs"
+    for n in range(4, 6):
+        path = snapshots_dir(state) / f"s{n}"
+        path.mkdir(parents=True)
+        os.utime(path, (900 + n, 900 + n))
+    for name, snapshot, status in (("running", "s1", {"state": rs.RUNNING}), ("queued", "s0", {"state": rs.QUEUED}),
+                                   ("done", "s2", {"state": rs.DONE}),
+                                   ("never-launched", "s4", {"state": rs.INTERRUPTED, "child_pid": None}),
+                                   ("partly-ran", "s5", {"state": rs.INTERRUPTED, "child_pid": 4242})):
+        run_dir = runs / name
+        run_dir.mkdir(parents=True)
+        (run_dir / "knowledge.json").write_text(json.dumps(
+            {"snapshot": snapshot, "knowledge_root": str(snapshots_dir(state) / snapshot)}))
+        (run_dir / rs.STATUS_NAME).write_text(json.dumps(status))
+    prune(state, keep=1, roots=[runs])
+    remaining = sorted(p.name for p in snapshots_dir(state).iterdir())
+    # live runs and a reclaimable (never-launched) reservation keep their pins;
+    # a finished run's, or one that ran and was interrupted, does not
+    assert remaining == ["s0", "s1", "s3", "s4"]
+
+
+def test_run_roots_come_from_the_environment(monkeypatch, tmp_path):
+    import os
+
+    from infermatrix_copilot.kb_service.activate import run_roots
+
+    monkeypatch.setenv("KB_RUN_ROOTS", os.pathsep.join([str(tmp_path / "a"), str(tmp_path / "b")]))
+    assert run_roots() == [tmp_path / "a", tmp_path / "b"]
+    monkeypatch.delenv("KB_RUN_ROOTS")
+    monkeypatch.setenv("RUN_ROOT", str(tmp_path / "c"))
+    assert run_roots() == [tmp_path / "c"]                  # Settings reads RUN_ROOT too
+
+
+def test_a_snapshot_switched_away_from_is_kept_through_the_pin_grace_period(tmp_path):
+    import os
+    import time
+
+    from infermatrix_copilot.kb_service.activate import DEACTIVATED, PIN_GRACE, switch_active
+
+    state = tmp_path / "state"
+    for n in range(3):
+        path = snapshots_dir(state) / f"s{n}"
+        path.mkdir(parents=True)
+        os.utime(path, (1000 + n, 1000 + n))
+    switch_active(state, snapshots_dir(state) / "s0")
+    switch_active(state, snapshots_dir(state) / "s2")     # s0 just stopped being active
+    prune(state, keep=1, roots=[])
+    assert sorted(p.name for p in snapshots_dir(state).iterdir() if not p.name.startswith(".")) == ["s0", "s2"]
+    marker = snapshots_dir(state) / DEACTIVATED.format(name="s0")
+    old = time.time() - PIN_GRACE - 1
+    os.utime(marker, (old, old))                          # the grace period has passed
+    prune(state, keep=1, roots=[])
+    assert sorted(p.name for p in snapshots_dir(state).iterdir() if not p.name.startswith(".")) == ["s2"]
+    assert not marker.exists()
+
+
+def test_run_roots_fall_back_to_the_effective_settings(monkeypatch, tmp_path):
+    from infermatrix_copilot import config
+    from infermatrix_copilot.kb_service.activate import run_roots
+
+    monkeypatch.delenv("KB_RUN_ROOTS")
+
+    class FakeSettings:
+        run_root = tmp_path / "from-dotenv"
+
+    monkeypatch.setattr(config, "Settings", FakeSettings)
+    assert run_roots() == [tmp_path / "from-dotenv"]
+
+
+def test_pin_discovery_reads_each_run_status_once(tmp_path, monkeypatch):
+    import json
+
+    from infermatrix_copilot import run_status as rs
+    from infermatrix_copilot.kb_service.activate import pinned_snapshots
+
+    run_dir = tmp_path / "runs" / "r"
+    run_dir.mkdir(parents=True)
+    (run_dir / "knowledge.json").write_text(json.dumps({"snapshot": "s", "knowledge_root": str(tmp_path / "s")}))
+    answers = iter([{"state": rs.INTERRUPTED, "child_pid": None}, {"state": rs.QUEUED}])
+    monkeypatch.setattr(rs, "read_status", lambda _d: next(answers))   # a reclaim lands between reads
+    assert pinned_snapshots([tmp_path / "runs"]) == {(tmp_path / "s").resolve()}