Pin the knowledge snapshot for Strict runs at reservation

doc/architecture/SPEC/app/run_service.md

@@ -1,6 +1,6 @@
 # app/run_service.py —— 规范
 
-<!-- verified-against: 2026-09-26 -->
+<!-- verified-against: 2026-09-28 -->
 
 `RunService` is the durable application boundary shared by the embedded
 Strict SDK and MCP transport. It owns policy-checked reserve/start, one
@@ -25,3 +25,17 @@ report pages remain available after terminal status.
 `mcp_server.CopilotMCP` remains a compatibility alias. The MCP server owns
 tool registration and protocol error projection, while the SDK owns typed
 request/result projection. Neither transport owns the run queue.
+
+## 2026-09-28 Knowledge snapshot pinning
+When `reserve_strict_review` / `reserve_quality_review` CREATE a reservation,
+they record the current `KnowledgeView` in `<run>/knowledge.json`: snapshot id,
+manifest tree hash, and resolved real paths (never the `active` symlink). An
+idempotent retry does not rewrite it. `_launch` sets the child's
+`KNOWLEDGE_DIR` and `KNOWLEDGE_ROOT` from that record, so a run that executes
+later (queued, or relaunched after a restart) still reviews with the knowledge
+active at reservation. If the pinned snapshot was pruned, the run is marked
+`failed`; it never silently switches knowledge. Without a snapshot root the
+run is pinned to this server's effective `knowledge_dir` (reported as `packaged`,
+or `unverified` for a custom directory), with `KNOWLEDGE_ROOT` removed from the
+child's environment. Runs reserved before pinning
+existed have no record and use the process default.

doc/architecture/SPEC/contract.md

@@ -1,6 +1,6 @@
 # contract.py —— 规范
 
-<!-- verified-against: 2026-09-22 -->
+<!-- verified-against: 2026-09-28 -->
 
 `旧版跨仓库契约兼容层 · refactor-status: compatibility-shim`
 
@@ -86,3 +86,7 @@ revision、supported repositories，以及 `supports_expected_head`、
 ## 重构备注
 新模块（PR2 mixed-mode contract 拆分）。保持它薄：任何"顺手在这里实现"
 的诱惑都在重造 thin_mcp_server 的巨石。
+
+## 2026-09-28 知识快照
+`build_review_result` 从 `<run>/knowledge.json` 读取钉住的知识，在 `diagnostics` 中给出 `knowledge_snapshot`
+与 `knowledge_tree_sha256`，供宿主（RB）与 review 一起记录。

src/infermatrix_copilot/app/run_service.py

@@ -29,6 +29,18 @@
 from .core import Copilot
 from .reservation import RunReservation
 
+KNOWLEDGE_PIN = "knowledge.json"
+
+
+def read_knowledge_pin(run_dir: str | Path) -> dict:
+    """The knowledge a reserved run was pinned to ({} for runs reserved
+    before pinning existed, which use the process default)."""
+    try:
+        data = json.loads((Path(run_dir) / KNOWLEDGE_PIN).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+    return data if isinstance(data, dict) else {}
+
 
 class RunService:
     """The server core: a serialized run queue over isolated subprocesses, plus
@@ -130,6 +142,21 @@ def _launch(self, run_id: str, *, strict_compat: bool = False) -> None:
             self.settings.allowed_repo_roots)
         if self.settings.strict_backend:
             env["STRICT_BACKEND"] = self.settings.strict_backend
+        pin = read_knowledge_pin(run_dir)
+        if pin.get("knowledge_dir"):
+            # the snapshot pinned at reservation, even if another is active now;
+            # a snapshot pruned since then fails the run instead of silently
+            # reviewing with different knowledge
+            if not (Path(pin["knowledge_dir"]) / "AGENTS.md").is_file():
+                rs.mark(run_dir, rs.FAILED, note=(
+                    f"pinned knowledge snapshot {pin.get('snapshot')} is no longer available"))
+                self._reap()
+                return
+            env["KNOWLEDGE_DIR"] = pin["knowledge_dir"]
+            if pin.get("knowledge_root"):
+                env["KNOWLEDGE_ROOT"] = pin["knowledge_root"]
+            else:
+                env.pop("KNOWLEDGE_ROOT", None)  # the packaged tree, whatever is configured now
         popen_kwargs: dict[str, Any] = {}
         if os.name == "nt":
             # Codex/Claude launch the MCP server over stdio.  Without a new
@@ -198,7 +225,7 @@ def reserve_strict_review(self, spec_dict: dict) -> tuple[str, bool]:
         run_id, created = self.reservations.reserve(
             spec, owner_server_id=self.server_id, owner_server_pid=self.pid,
             idempotency_key=str(spec_dict.get("idempotency_key") or ""))
-        if created:
+        if created and self._pin_or_fail(run_id):
             self._q.put((run_id, True))
         return run_id, created
 
@@ -215,10 +242,63 @@ def reserve_quality_review(self, spec_dict: dict) -> tuple[str, bool]:
         run_id, created = self.reservations.reserve(
             spec, owner_server_id=self.server_id, owner_server_pid=self.pid,
             idempotency_key=str(spec_dict.get("idempotency_key") or ""))
-        if created:
+        if created and self._pin_or_fail(run_id):
             self._q.put((run_id, False))
         return run_id, created
 
+    # -- knowledge pinning ---------------------------------------------------------
+    def _pin_or_fail(self, run_id: str) -> bool:
+        """Pin, or make the reservation durably terminal: a reservation that
+        exists but is never enqueued would stay queued forever (retries see
+        created=False), so a pinning failure fails the run with its reason."""
+        try:
+            self._pin_knowledge(run_id)
+            return True
+        except Exception as exc:  # noqa: BLE001 - any failure must end the run visibly
+            rs.mark(self.run_root / run_id, rs.FAILED,
+                    note=f"could not pin the knowledge snapshot: {type(exc).__name__}: {exc}")
+            return False
+
+    def _pin_knowledge(self, run_id: str) -> dict:
+        """Record, at reservation, the knowledge this run must review with.
+
+        The knowledge service swaps the active snapshot at runtime; a Strict
+        run that starts later (queued, or re-launched after a restart) must
+        still read the snapshot that was active when it was reserved, and its
+        result must say which one that was. The resolved real paths are
+        recorded, never the ``active`` symlink."""
+        from ..knowledge_view import KnowledgeView
+
+        existing = read_knowledge_pin(self.run_root / run_id)
+        if existing:
+            # reclaiming an interrupted, never-started reservation also reports
+            # created=True: its reservation-time pin stands
+            return existing
+        view = KnowledgeView.current()
+        if view.snapshot == "packaged":
+            env_root = ""
+        elif view.verified:
+            env_root = str(view.root.parent)  # the snapshot dir (MANIFEST.json + knowledge/)
+        else:
+            env_root = str(view.root)
+        snapshot, knowledge_dir = view.public_snapshot, str(view.root)
+        if view.snapshot == "packaged":
+            # no snapshot root selected: the run reviews with this server's
+            # EFFECTIVE knowledge_dir (a deployment may configure its own), and
+            # KNOWLEDGE_ROOT is cleared at launch, so a relaunch under a newly
+            # configured root cannot review with a snapshot while reporting this
+            configured = Path(self.settings.knowledge_dir).resolve()
+            if configured != view.root.resolve():
+                snapshot = "unverified"
+            knowledge_dir = str(configured)
+        pin = {"snapshot": snapshot, "tree_sha256": view.tree_sha256,
+               "knowledge_dir": knowledge_dir, "knowledge_root": env_root}
+        path = self.run_root / run_id / KNOWLEDGE_PIN
+        tmp = path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(pin, sort_keys=True), encoding="utf-8")
+        os.replace(tmp, path)
+        return pin
+
     def strict_readiness(self, repo: str, repo_path: str = "") -> list[str]:
         """Return actionable setup gaps before reserving a Strict run.
 

src/infermatrix_copilot/contract.py

@@ -240,6 +240,12 @@ def build_review_result(run_dir: Path | str) -> dict[str, Any]:
         if events:
             diagnostics[name] = events
 
+    # which knowledge the run reviewed with (pinned at reservation)
+    pin = _read_json(run_dir / "knowledge.json")
+    if pin.get("snapshot"):
+        diagnostics["knowledge_snapshot"] = str(pin["snapshot"])
+        diagnostics["knowledge_tree_sha256"] = str(pin.get("tree_sha256") or "")
+
     metrics = _read_json(run_dir / "metrics.json")
     cost = metrics.get("cost") or {}
     if cost:

test/test_mcp.py

@@ -610,3 +610,163 @@ def test_subprocess_tamper_defense(settings):
     assert st["owner_server_id"] == core.server_id
     assert (rd / "console.log").exists()
     core.close()
+
+
+# -- Strict runs pin the knowledge snapshot at reservation ---------------------------
+
+def _reserve_created(core, monkeypatch, run_id="run-20260928-120000-aaaaaa"):
+    (core.run_root / run_id).mkdir(parents=True, exist_ok=True)
+    rs.init_queued(core.run_root / run_id, run_id=run_id, owner_server_id=core.server_id,
+                   owner_server_pid=core.pid)
+    monkeypatch.setattr(core._q, "put", lambda item: None)
+    monkeypatch.setattr(core.reservations, "reserve", lambda *_a, **_k: (run_id, True))
+    return core.reserve_strict_review({
+        "kind": "pr_review", "repo": "vllm-omni", "pr": 7, "post": False,
+        "params": {"review_depth": "standard"}, "expected_head_sha": "a" * 40,
+        "idempotency_key": "attempt-pin"})[0]
+
+
+def _fake_launch(core, monkeypatch, run_id):
+    from infermatrix_copilot import mcp_server
+
+    captured = {}
+
+    class FakeProcess:
+        pid = 424243
+        returncode = 0
+
+        def wait(self):
+            return 0
+
+    def fake_popen(argv, **kwargs):
+        captured["env"] = kwargs["env"]
+        return FakeProcess()
+
+    monkeypatch.setattr(mcp_server.subprocess, "Popen", fake_popen)
+    core._launch(run_id)
+    return captured
+
+
+def test_a_strict_run_reviews_with_the_snapshot_active_when_it_was_reserved(settings, monkeypatch, tmp_path):
+    from infermatrix_copilot.contract import build_review_result
+    from infermatrix_copilot.knowledge_view import KNOWLEDGE_ROOT_ENV, _load_view
+    from test_knowledge_view_routing import _snapshot
+
+    old = _snapshot(tmp_path, "c" * 40, retire=False)
+    new = _snapshot(tmp_path, "d" * 40, retire=True)
+    active = tmp_path / "active"
+    active.symlink_to(old)
+    monkeypatch.setenv(KNOWLEDGE_ROOT_ENV, str(active))
+    _load_view.cache_clear()
+    core = _core(settings)
+    run_id = _reserve_created(core, monkeypatch)
+    swap = tmp_path / "active.tmp"
+    swap.symlink_to(new)
+    swap.replace(active)                       # a newer snapshot is activated meanwhile
+
+    env = _fake_launch(core, monkeypatch, run_id)["env"]
+    assert env["KNOWLEDGE_DIR"] == str((old / "knowledge").resolve())
+    assert env["KNOWLEDGE_ROOT"] == str(old.resolve())
+    diagnostics = build_review_result(core.run_root / run_id)["diagnostics"]
+    manifest = json.loads((old / "MANIFEST.json").read_text())
+    assert diagnostics["knowledge_snapshot"] == "c" * 40
+    assert diagnostics["knowledge_tree_sha256"] == manifest["tree_sha256"]
+    core.close()
+
+
+def test_a_pruned_pinned_snapshot_fails_the_run_instead_of_switching_knowledge(settings, monkeypatch, tmp_path):
+    import shutil
+
+    from infermatrix_copilot.knowledge_view import KNOWLEDGE_ROOT_ENV, _load_view
+    from test_knowledge_view_routing import _snapshot
+
+    old = _snapshot(tmp_path, "e" * 40, retire=False)
+    monkeypatch.setenv(KNOWLEDGE_ROOT_ENV, str(old))
+    _load_view.cache_clear()
+    core = _core(settings)
+    run_id = _reserve_created(core, monkeypatch)
+    shutil.rmtree(old)
+    assert _fake_launch(core, monkeypatch, run_id) == {}          # never launched
+    status = rs.read_status(core.run_root / run_id)
+    assert status["state"] == rs.FAILED and "e" * 40 in status["note"]
+    core.close()
+
+
+def test_packaged_knowledge_is_pinned_even_if_a_root_is_configured_later(settings, monkeypatch, tmp_path):
+    from infermatrix_copilot.contract import build_review_result
+    from infermatrix_copilot.knowledge_view import KNOWLEDGE_ROOT_ENV, KnowledgeView, _load_view
+
+    monkeypatch.delenv(KNOWLEDGE_ROOT_ENV, raising=False)
+    monkeypatch.delenv("KNOWLEDGE_DIR", raising=False)
+    _load_view.cache_clear()
+    core = _core(settings)
+    run_id = _reserve_created(core, monkeypatch)
+    packaged = str(KnowledgeView.current().root)
+    monkeypatch.setenv(KNOWLEDGE_ROOT_ENV, str(tmp_path / "later-activated"))   # config changes before launch
+    env = _fake_launch(core, monkeypatch, run_id)["env"]
+    assert env["KNOWLEDGE_DIR"] == packaged and KNOWLEDGE_ROOT_ENV not in env
+    assert build_review_result(core.run_root / run_id)["diagnostics"]["knowledge_snapshot"] == "packaged"
+    core.close()
+
+
+def test_a_reclaimed_reservation_keeps_its_original_pin(settings, monkeypatch, tmp_path):
+    from infermatrix_copilot.knowledge_view import KNOWLEDGE_ROOT_ENV, _load_view
+    from test_knowledge_view_routing import _snapshot
+
+    old = _snapshot(tmp_path, "7" * 40, retire=False)
+    new = _snapshot(tmp_path, "8" * 40, retire=True)
+    active = tmp_path / "active"
+    active.symlink_to(old)
+    monkeypatch.setenv(KNOWLEDGE_ROOT_ENV, str(active))
+    _load_view.cache_clear()
+    core = _core(settings)
+    run_id = _reserve_created(core, monkeypatch)
+    swap = tmp_path / "active.tmp"
+    swap.symlink_to(new)
+    swap.replace(active)
+    assert _reserve_created(core, monkeypatch, run_id) == run_id     # reclaimed: created=True again
+    pin = json.loads((core.run_root / run_id / "knowledge.json").read_text())
+    assert pin["snapshot"] == "7" * 40
+    core.close()
+
+
+def test_a_pinning_failure_fails_the_run_instead_of_stranding_it(settings, monkeypatch):
+    from infermatrix_copilot.knowledge_view import KnowledgeView, KnowledgeViewError
+
+    def broken():
+        raise KnowledgeViewError("KNOWLEDGE_ROOT does not resolve")
+
+    monkeypatch.setattr(KnowledgeView, "current", classmethod(lambda cls: broken()))
+    core = _core(settings)
+    queued = []
+    run_id = "run-20260928-120000-bbbbbb"
+    (core.run_root / run_id).mkdir(parents=True)
+    rs.init_queued(core.run_root / run_id, run_id=run_id, owner_server_id=core.server_id,
+                   owner_server_pid=core.pid)
+    monkeypatch.setattr(core._q, "put", queued.append)
+    monkeypatch.setattr(core.reservations, "reserve", lambda *_a, **_k: (run_id, True))
+    core.reserve_strict_review({
+        "kind": "pr_review", "repo": "vllm-omni", "pr": 7, "post": False,
+        "params": {"review_depth": "standard"}, "expected_head_sha": "a" * 40,
+        "idempotency_key": "attempt-broken"})
+    status = rs.read_status(core.run_root / run_id)
+    assert queued == [] and status["state"] == rs.FAILED
+    assert "could not pin the knowledge snapshot" in status["note"]
+    core.close()
+
+
+def test_a_configured_knowledge_dir_is_pinned_not_replaced_by_the_packaged_tree(settings, monkeypatch, tmp_path):
+    from infermatrix_copilot.contract import build_review_result
+    from infermatrix_copilot.knowledge_view import KNOWLEDGE_ROOT_ENV, KnowledgeView, _load_view
+
+    monkeypatch.delenv(KNOWLEDGE_ROOT_ENV, raising=False)
+    _load_view.cache_clear()
+    custom = tmp_path / "custom-knowledge"
+    shutil.copytree(KnowledgeView.current().root, custom, ignore=shutil.ignore_patterns("__pycache__"))
+    settings.knowledge_dir = custom
+    core = _core(settings)
+    run_id = _reserve_created(core, monkeypatch)
+    env = _fake_launch(core, monkeypatch, run_id)["env"]
+    assert env["KNOWLEDGE_DIR"] == str(custom.resolve()) and KNOWLEDGE_ROOT_ENV not in env
+    assert build_review_result(core.run_root / run_id)["diagnostics"]["knowledge_snapshot"] == "unverified"
+    core.close()