From 4bdef4e112336819e0e04858724a3bbffd68013d Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 02:02:26 +0800 Subject: [PATCH] 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 (/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 --- doc/architecture/SPEC/kb_service.md | 7 ++ .../kb_service/activate.py | 64 ++++++++++++- test/conftest.py | 7 ++ test/test_kb_flow.py | 95 +++++++++++++++++++ 4 files changed, 171 insertions(+), 2 deletions(-) diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 0c95dc55c..2df97413c 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/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` 指向空目录，从不读取真实运行目录。 diff --git a/src/infermatrix_copilot/kb_service/activate.py b/src/infermatrix_copilot/kb_service/activate.py index 67f861301..48b1540a4 100644 --- a/src/infermatrix_copilot/kb_service/activate.py +++ b/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 (``/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) diff --git a/test/conftest.py b/test/conftest.py index a169c816e..157bcf00c 100644 --- a/test/conftest.py +++ b/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")) diff --git a/test/test_kb_flow.py b/test/test_kb_flow.py index c86deb665..695adcf6f 100644 --- a/test/test_kb_flow.py +++ b/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()}