diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 0c95dc55..2df97413 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 67f86130..48b1540a 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 a169c816..157bcf00 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 c86deb66..695adcf6 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()}