Skip to content

Commit 4bdef4e

Browse files
tzhouamclaude
andcommitted
Keep knowledge snapshots that unfinished runs pinned
Strict and quality runs pin the knowledge snapshot active at reservation (#213), and the design requires that a snapshot referenced by any unfinished run is never deleted. Pruning kept only the newest ten plus the active one, so a long queue could lose a pinned snapshot and fail those runs. - activate.prune also keeps every snapshot that a run pinned (<run>/knowledge.json) while that run is not terminal, or is an interrupted reservation that never launched (it can be reclaimed and will run with its original pin). - Switching the active snapshot records when the previous one stopped being active, and pruning never removes a snapshot deactivated less than an hour ago. A reservation that resolved it just before the switch publishes its pin within that window. The review runs live in other processes (and possibly other hosts) that cannot share the service's activation lock, so the coordination is this grace period. - The run directories come from KB_RUN_ROOTS (os.pathsep-separated, for several servers), else the runs' EFFECTIVE run_root (environment or the .env files Settings reads). - An autouse test fixture points KB_RUN_ROOTS at an empty directory, so tests never read the real ~/.infermatrix-copilot/runs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent fb61e8f commit 4bdef4e

4 files changed

Lines changed: 171 additions & 2 deletions

File tree

‎doc/architecture/SPEC/kb_service.md‎

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

‎src/infermatrix_copilot/kb_service/activate.py‎

Lines changed: 62 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import os
1717
import shutil
1818
import tempfile
19+
import time
1920
from contextlib import contextmanager
2021
from pathlib import Path
2122
from typing import Iterator
@@ -25,6 +26,10 @@
2526
from .outbox import atomic_write_json
2627

2728
KEEP = 10
29+
# a snapshot switched away from less than this ago is never pruned: a run that
30+
# resolved it just before the switch publishes its pin within this window
31+
PIN_GRACE = 3600
32+
DEACTIVATED = ".deactivated-{name}"
2833

2934

3035
class ActivationError(RuntimeError):
@@ -111,6 +116,11 @@ def verify_snapshot(snapshot: Path) -> KnowledgeView:
111116

112117
def switch_active(state_dir: Path, snapshot: Path) -> None:
113118
link = active_link(state_dir)
119+
if link.is_symlink():
120+
previous = link.resolve()
121+
if previous != snapshot.resolve() and previous.parent == snapshots_dir(state_dir).resolve():
122+
# when it stopped being active: pruning honours a grace period after this
123+
(snapshots_dir(state_dir) / DEACTIVATED.format(name=previous.name)).write_text("", encoding="utf-8")
114124
tmp = link.with_name(f".active.{os.getpid()}")
115125
if tmp.is_symlink() or tmp.exists():
116126
tmp.unlink()
@@ -149,13 +159,63 @@ def rollback(rt, sha: str) -> Path:
149159
return snapshot
150160

151161

152-
def prune(state_dir: Path, keep: int = KEEP) -> None:
162+
def run_roots() -> list[Path]:
163+
"""Where the review runs that may pin a snapshot keep their run
164+
directories: ``KB_RUN_ROOTS`` (os.pathsep-separated), else the runs'
165+
``RUN_ROOT``, else its default."""
166+
raw = os.environ.get("KB_RUN_ROOTS")
167+
if raw:
168+
return [Path(part).expanduser() for part in raw.split(os.pathsep) if part]
169+
try: # the runs' EFFECTIVE run_root (environment or the supported .env files)
170+
from ..config import Settings
171+
172+
return [Path(Settings().run_root).expanduser()]
173+
except Exception: # noqa: BLE001 - unreadable config: the documented default
174+
return [Path(os.environ.get("RUN_ROOT") or Path.home() / ".infermatrix-copilot" / "runs").expanduser()]
175+
176+
177+
def pinned_snapshots(roots: list[Path]) -> set[Path]:
178+
"""Snapshot directories pinned (``<run>/knowledge.json``) by runs that have
179+
not finished: a queued or running Strict review must keep its knowledge."""
180+
from .. import run_status as rs
181+
182+
pinned: set[Path] = set()
183+
for root in roots:
184+
if not root.is_dir():
185+
continue
186+
for run_dir in root.iterdir():
187+
try:
188+
pin = json.loads((run_dir / "knowledge.json").read_text(encoding="utf-8"))
189+
except (OSError, ValueError):
190+
continue
191+
target = pin.get("knowledge_root") if isinstance(pin, dict) else None
192+
if not target:
193+
continue
194+
status = rs.read_status(run_dir) or {} # ONE read: a reclaim may change it meanwhile
195+
state = status.get("state")
196+
# an interrupted reservation that never launched (no child pid) can
197+
# be reclaimed and will run with its original pin
198+
if state not in rs.TERMINAL or (state == rs.INTERRUPTED and status.get("child_pid") is None):
199+
pinned.add(Path(target).resolve())
200+
return pinned
201+
202+
203+
def prune(state_dir: Path, keep: int = KEEP, *, roots: list[Path] | None = None) -> None:
204+
"""Keep the newest ``keep`` snapshots, the active one, and every snapshot an
205+
unfinished run pinned; remove the rest."""
153206
root = snapshots_dir(state_dir)
154207
link = active_link(state_dir)
155208
active = link.resolve() if link.is_symlink() else None
209+
pinned = pinned_snapshots(run_roots() if roots is None else roots)
156210
snapshots = sorted((p for p in root.iterdir() if p.is_dir() and not p.name.startswith(".")),
157211
key=lambda p: p.stat().st_mtime, reverse=True)
212+
now = time.time()
158213
for old in snapshots[keep:]:
159-
if active is not None and old.resolve() == active:
214+
resolved = old.resolve()
215+
if (active is not None and resolved == active) or resolved in pinned:
160216
continue
217+
marker = root / DEACTIVATED.format(name=old.name)
218+
if marker.exists() and now - marker.stat().st_mtime < PIN_GRACE:
219+
continue # deactivated too recently: a reservation may be pinning it right now
161220
shutil.rmtree(old, ignore_errors=True)
221+
marker.unlink(missing_ok=True)

‎test/conftest.py‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,3 +89,10 @@ def git(*args: str) -> None:
8989
git("add", ".")
9090
git("commit", "-q", "-m", "init")
9191
return repo
92+
93+
94+
@pytest.fixture(autouse=True)
95+
def _no_real_run_roots(tmp_path, monkeypatch):
96+
"""Snapshot pruning looks for runs that pin a snapshot; tests never read
97+
the real ~/.infermatrix-copilot/runs."""
98+
monkeypatch.setenv("KB_RUN_ROOTS", str(tmp_path / "no-runs"))

‎test/test_kb_flow.py‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -686,3 +686,98 @@ def apply_ack():
686686
acker.join(5)
687687
assert not errors, errors
688688
assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" # never stranded as paused
689+
690+
691+
def test_prune_keeps_snapshots_pinned_by_unfinished_runs(tmp_path):
692+
import json
693+
import os
694+
695+
from infermatrix_copilot import run_status as rs
696+
697+
state = tmp_path / "state"
698+
for n in range(4):
699+
path = snapshots_dir(state) / f"s{n}"
700+
path.mkdir(parents=True)
701+
os.utime(path, (1000 + n, 1000 + n))
702+
active_link(state).symlink_to((snapshots_dir(state) / "s3").resolve())
703+
runs = tmp_path / "runs"
704+
for n in range(4, 6):
705+
path = snapshots_dir(state) / f"s{n}"
706+
path.mkdir(parents=True)
707+
os.utime(path, (900 + n, 900 + n))
708+
for name, snapshot, status in (("running", "s1", {"state": rs.RUNNING}), ("queued", "s0", {"state": rs.QUEUED}),
709+
("done", "s2", {"state": rs.DONE}),
710+
("never-launched", "s4", {"state": rs.INTERRUPTED, "child_pid": None}),
711+
("partly-ran", "s5", {"state": rs.INTERRUPTED, "child_pid": 4242})):
712+
run_dir = runs / name
713+
run_dir.mkdir(parents=True)
714+
(run_dir / "knowledge.json").write_text(json.dumps(
715+
{"snapshot": snapshot, "knowledge_root": str(snapshots_dir(state) / snapshot)}))
716+
(run_dir / rs.STATUS_NAME).write_text(json.dumps(status))
717+
prune(state, keep=1, roots=[runs])
718+
remaining = sorted(p.name for p in snapshots_dir(state).iterdir())
719+
# live runs and a reclaimable (never-launched) reservation keep their pins;
720+
# a finished run's, or one that ran and was interrupted, does not
721+
assert remaining == ["s0", "s1", "s3", "s4"]
722+
723+
724+
def test_run_roots_come_from_the_environment(monkeypatch, tmp_path):
725+
import os
726+
727+
from infermatrix_copilot.kb_service.activate import run_roots
728+
729+
monkeypatch.setenv("KB_RUN_ROOTS", os.pathsep.join([str(tmp_path / "a"), str(tmp_path / "b")]))
730+
assert run_roots() == [tmp_path / "a", tmp_path / "b"]
731+
monkeypatch.delenv("KB_RUN_ROOTS")
732+
monkeypatch.setenv("RUN_ROOT", str(tmp_path / "c"))
733+
assert run_roots() == [tmp_path / "c"] # Settings reads RUN_ROOT too
734+
735+
736+
def test_a_snapshot_switched_away_from_is_kept_through_the_pin_grace_period(tmp_path):
737+
import os
738+
import time
739+
740+
from infermatrix_copilot.kb_service.activate import DEACTIVATED, PIN_GRACE, switch_active
741+
742+
state = tmp_path / "state"
743+
for n in range(3):
744+
path = snapshots_dir(state) / f"s{n}"
745+
path.mkdir(parents=True)
746+
os.utime(path, (1000 + n, 1000 + n))
747+
switch_active(state, snapshots_dir(state) / "s0")
748+
switch_active(state, snapshots_dir(state) / "s2") # s0 just stopped being active
749+
prune(state, keep=1, roots=[])
750+
assert sorted(p.name for p in snapshots_dir(state).iterdir() if not p.name.startswith(".")) == ["s0", "s2"]
751+
marker = snapshots_dir(state) / DEACTIVATED.format(name="s0")
752+
old = time.time() - PIN_GRACE - 1
753+
os.utime(marker, (old, old)) # the grace period has passed
754+
prune(state, keep=1, roots=[])
755+
assert sorted(p.name for p in snapshots_dir(state).iterdir() if not p.name.startswith(".")) == ["s2"]
756+
assert not marker.exists()
757+
758+
759+
def test_run_roots_fall_back_to_the_effective_settings(monkeypatch, tmp_path):
760+
from infermatrix_copilot import config
761+
from infermatrix_copilot.kb_service.activate import run_roots
762+
763+
monkeypatch.delenv("KB_RUN_ROOTS")
764+
765+
class FakeSettings:
766+
run_root = tmp_path / "from-dotenv"
767+
768+
monkeypatch.setattr(config, "Settings", FakeSettings)
769+
assert run_roots() == [tmp_path / "from-dotenv"]
770+
771+
772+
def test_pin_discovery_reads_each_run_status_once(tmp_path, monkeypatch):
773+
import json
774+
775+
from infermatrix_copilot import run_status as rs
776+
from infermatrix_copilot.kb_service.activate import pinned_snapshots
777+
778+
run_dir = tmp_path / "runs" / "r"
779+
run_dir.mkdir(parents=True)
780+
(run_dir / "knowledge.json").write_text(json.dumps({"snapshot": "s", "knowledge_root": str(tmp_path / "s")}))
781+
answers = iter([{"state": rs.INTERRUPTED, "child_pid": None}, {"state": rs.QUEUED}])
782+
monkeypatch.setattr(rs, "read_status", lambda _d: next(answers)) # a reclaim lands between reads
783+
assert pinned_snapshots([tmp_path / "runs"]) == {(tmp_path / "s").resolve()}

0 commit comments

Comments
 (0)