Verify foundation native proofs before cache reuse and publication resume

Author: tzhouam

doc/architecture/SPEC/kb_service.md

@@ -903,3 +903,9 @@ worker 使用独立知识视图、阶段记录和订阅传输；唯一协调器
 预算限制；不可完整展示的行段显式保留为缺口。引用必须落在实际展示的单个区间
 中，不以最大行号冒充中间未读取的内容。旧功能仍使用原来源选择及覆盖判定，
 已覆盖页不重新提取。预算预留/结算采用原子操作；订阅费用未报告仍为未知。
+
+完成但尚未发布的订阅知识缓存及已准备发布的恢复也必须先做只读原生证明复核。
+复核绑定原输入身份、源码 pin、KB 基点、有效 owner、两个模型的完整归档及
+精确正文/证据。失败在任何记录或发布写入前拒绝，恢复原归档后可以继续同一
+已准备的发布；已发布的历史记录保持不变。已有订阅工作检查点的输入发生变化
+时须建立新批次，不能静默覆盖或重新抽取。

plugins/infermatrix-copilot/skills/imkbinit/SKILL.md

@@ -339,3 +339,9 @@ approved sections in deterministic order. Resume the same command/state to reuse
 completed native approvals; keep their external trace archives. Finite-budget
 runs retain the previous serial flow. New discovery features preferentially read
 the frozen report’s exact complete source spans; unread gaps cannot be cited.
+
+Completed unpublished caches and prepared-publication resumes also replay the
+original native proofs before any write or model call. Missing or changed
+archives reject reuse without changing the genuine record; restore the original
+archive to continue. Changed frozen inputs require a new batch. Published
+historical records remain unchanged.

src/infermatrix_copilot/kb_service/init_knowledge.py

@@ -126,10 +126,34 @@ def _input_options(self) -> dict:
                                                "zcode_start_interval_s": start_interval(self.rt)}
         return options
 
+    def _discovery_gate(self, chain: _Chain, pin: str) -> None:
+        self._knowledge_run_pin = pin
+        super()._discovery_gate(chain, pin)
+
     def _cache_reusable(self, previous: InitRecord) -> bool:
+        if previous.status == "published":
+            return True  # Preserve immutable historical publication records.
+        if self.rt.unlimited_subscription and previous.status != "published":
+            self._validate_unpublished_checkpoint(previous, previous.inputs_digest)
         return not (self.rt.unlimited_subscription and previous.unfinished)
 
+    def _resume_input_problems(self, previous: InitRecord, digest: str) -> list[str]:
+        if self.rt.unlimited_subscription and previous.status != "published":
+            # Raise before generic _blocked writes: restoring an archive must
+            # leave the same genuine prepared publication resumable.
+            self._validate_unpublished_checkpoint(previous, digest)
+        return super()._resume_input_problems(previous, digest)
+
+    def _validate_unpublished_checkpoint(self, previous: InitRecord, digest: str) -> None:
+        from .init_knowledge_parallel import validate_checkpoint
+        validate_checkpoint(self, previous, digest)
+
     def _restore_progress(self, previous: InitRecord | None) -> list[str]:
+        if (self.rt.unlimited_subscription and previous is not None
+                and previous.coverage.get("foundation_jobs")
+                and previous.inputs_digest != self.record.inputs_digest):
+            from .init_support import InitError
+            raise InitError("foundation inputs changed; create a new pinned batch")
         if (self.rt.unlimited_subscription and previous is not None
                 and previous.inputs_digest == self.record.inputs_digest):
             from copy import deepcopy

src/infermatrix_copilot/kb_service/init_knowledge_parallel.py

@@ -114,7 +114,7 @@ def call_json(self, role, **kwargs):
 
 
 def _worker(stage, job):
-    from .init_knowledge import SYSTEM_KNOWLEDGE, knowledge_prompt, knowledge_prompt, validate_sections
+    from .init_knowledge import SYSTEM_KNOWLEDGE, knowledge_prompt, validate_sections
     from .init_support import generate
     from .init_stages import _one_line
     worker = copy.copy(stage)
@@ -274,6 +274,49 @@ def _apply(stage, result):
     return accepted
 
 
+def validate_checkpoint(stage, previous, digest):
+    """Read-only replay before cache reuse or any prepared publication write."""
+    import tempfile
+    from pathlib import Path
+    from .feature_discovery_index import build_discovery_index, discovery_scope
+    from .init_knowledge_inputs import source_owners
+
+    saved = previous.coverage.get("foundation_jobs", {})
+    if (previous.stage != "knowledge" or previous.repo != stage.lifecycle.repo
+            or previous.pin != stage._knowledge_run_pin
+            or previous.kb_base_sha != stage._base_sha or previous.inputs_digest != digest
+            or (saved and saved.get("binding") != digest)):
+        raise InitError("foundation checkpoint frozen input identity differs")
+    if not saved.get("tasks"):
+        return  # A genuine empty stage may reuse an already-covered merged KB.
+    context = copy.copy(stage)
+    context.record = copy.deepcopy(previous)
+    context.base = {**stage.rt.knowledge.knowledge_files(previous.kb_base_sha),
+                    **{p[len("knowledge/"):]: text for p, text in stage.overlay.items()
+                       if p.startswith("knowledge/")}}
+    context.tags = [stage.lifecycle.repo]
+    context.today = stage.rt.today()
+    problems = context._precheck()
+    if problems:
+        raise InitError("foundation checkpoint policy invalid: " + "; ".join(problems))
+    upstream = stage.rt.upstream(stage.lifecycle.repo, stage.lifecycle.full_name)
+    context.observer = upstream.observer(previous.pin, pull=stage.rt.pull)
+    # Export and index live only in a temporary directory. Do not repair or
+    # rewrite stage records, previews, pacing journals or shared index caches.
+    with tempfile.TemporaryDirectory(prefix="kb-foundation-proof-") as scratch:
+        tree = upstream.export(previous.pin, Path(scratch) / "tree")
+        context._inputs(tree)
+        if context.route_problem:
+            raise InitError(context.route_problem)
+        policy = getattr(context, "coverage_policy", None)
+        docs = list(stage.lifecycle.init.doc_globs) + list(policy.catalog_sources if policy else ())
+        index = build_discovery_index(tree, pin=previous.pin,
+                    scope=discovery_scope(stage.lifecycle.init, policy), doc_globs=sorted(set(docs)))
+        context.owners = list(source_owners(list(index.production), context.owners, policy).values())
+        for result in saved["tasks"].values():
+            _validate_result(context, result)
+
+
 def restore_jobs(stage):
     saved = stage.record.coverage.get("foundation_jobs", {})
     if saved and saved.get("binding") != stage.record.inputs_digest:

test/test_kb_foundation_parallel.py

@@ -111,11 +111,15 @@ def test_late_discovery_ranges_are_complete_and_gaps_are_not_offered(tmp_path):
         parallel.preferred_sources(stage, tmp_path, "new-feature", ["giant.ts"], 100000)
 
 
-@pytest.mark.parametrize("damage", ["judge", "generator", "owner", "facet", "offered", "section"])
-def test_native_checkpoint_replay_exact_approval_and_tamper_rejected(world, damage):
+def _complete_native(world, *, publish=False):
     seed_gateway = KnowledgeGateway()
     _chain(world, seed_gateway)
-    state = _runtime(world).state_dir
+    if publish:
+        from test_kb_init_skeleton import _commit
+        skeleton = InitRecord.load(_runtime(world).state_dir, "toy", "skeleton")
+        modules = InitRecord.load(_runtime(world).state_dir, "toy", "modules")
+        _commit(world["origin"], {**_tree(skeleton), **_tree(modules)}, "merge prerequisite pages")
+    state = _runtime(world).state_dir if not publish else world["tmp"] / "publish-state"
     traces = TraceStore(state / "init" / "traces")
     offline = KnowledgeGateway()
 
@@ -135,10 +139,24 @@ def complete(self, *, system, messages, model, effort, role, **kwargs):
             return Reply(blocks=[Block(type="text", text=text)], model=model)
 
     gateway = ModelGateway(None, transport_factory=lambda _: NativeTransport(), recorder=trace_recorder(traces))
-    rt = _runtime(world, gateway, generator=ModelRole("generator", "zcode", "GLM-5.3"))
+    rt = _runtime(world, gateway, state_dir=state, generator=ModelRole("generator", "zcode", "GLM-5.3"))
+    if publish:
+        from test_kb_init_skeleton import FakeGh
+        rt.gh_run = FakeGh(fail_create=1)
     rt.environ = {**rt.environ, "KB_KNOWLEDGE_CONCURRENCY": "2", "KB_KNOWLEDGE_START_INTERVAL_S": "0.001"}
-    record = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True)
-    assert record.status == "dry_run", record.problems
+    if publish:
+        rt.environ.update(ALLOW_PUSH="1", ALLOW_POST="1", KB_INIT_GIT_AUTHOR="Tester <tester@example.com>")
+    record = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=not publish,
+                       from_existing=publish, unlimited_subscription=True)
+    assert record.status == ("blocked" if publish else "dry_run"), record.problems
+    assert not record.unfinished
+    return rt, record, offline, traces
+
+
+
+@pytest.mark.parametrize("damage", ["judge", "generator", "owner", "facet", "offered", "section"])
+def test_native_checkpoint_replay_exact_approval_and_tamper_rejected(world, damage):
+    rt, record, offline, traces = _complete_native(world)
     artifacts = [a for task in record.coverage["foundation_jobs"]["tasks"].values() for a in task["artifacts"]]
     assert len(artifacts) == 12 and all(a["judge_receipt"]["native_trace_id"] for a in artifacts)
     first_tree = _tree(record)
@@ -261,3 +279,70 @@ def test_shared_page_assembly_preserves_existing_sections_and_both_owners():
     assert historical in stage.head[page]
     assert stage.head[page].index("owner=left") < stage.head[page].index("owner=right")
     assert len(stage.record.evidence) == 2
+
+
+@pytest.mark.parametrize("damage", ["missing_generator", "missing_judge", "tampered_generator", "tampered_judge", "binding", "config", "role"])
+def test_completed_cache_cannot_skip_proof_replay_or_overwrite_on_failure(world, damage):
+    rt, record, offline, traces = _complete_native(world)
+    assert record.status == "dry_run" and record.unfinished == []
+    artifact = next(iter(record.coverage["foundation_jobs"]["tasks"].values()))["artifacts"][0]
+    if damage.startswith("missing_") or damage.startswith("tampered_"):
+        role = damage.rsplit("_", 1)[1]
+        native = traces.get(artifact[role + "_receipt"]["native_trace_id"])
+        attempt = traces.root / "attempts" / native["result"]["native_attempt_id"] / "attempt.json"
+        if damage.startswith("missing_"):
+            attempt.unlink()
+        else:
+            saved = json.loads(attempt.read_text())
+            saved["status"] = "interrupted"
+            attempt.write_text(json.dumps(saved))
+    elif damage == "binding":
+        record.coverage["foundation_jobs"]["binding"] = "different-inputs"
+        record.save(rt.state_dir)
+    elif damage == "config":
+        rt.gateway._settings = SimpleNamespace(zcode_reasoning_level="low")
+    else:
+        rt.judge = ModelRole("judge", "codex", "different-model", "medium")
+    path = InitRecord.path(rt.state_dir, "toy", "knowledge")
+    before, calls = path.read_bytes(), len(offline.calls)
+    with pytest.raises(InitError, match="foundation"):
+        run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True)
+    assert path.read_bytes() == before and len(offline.calls) == calls
+
+
+@pytest.mark.parametrize("damage", ["missing_generator", "missing_judge", "digest", "role", "config"])
+def test_prepared_publication_checks_original_proofs_before_resume(world, damage):
+    rt, record, offline, traces = _complete_native(world, publish=True)
+    assert record.pr["prepared"] and record.status == "blocked" and not record.unfinished
+    artifact = next(iter(record.coverage["foundation_jobs"]["tasks"].values()))["artifacts"][0]
+    restore = None
+    if damage.startswith("missing_"):
+        role = damage.rsplit("_", 1)[1]
+        native = traces.get(artifact[role + "_receipt"]["native_trace_id"])
+        attempt = traces.root / "attempts" / native["result"]["native_attempt_id"] / "attempt.json"
+        restore = attempt, attempt.read_bytes()
+        attempt.unlink()
+    elif damage == "digest":
+        record.inputs_digest = "wrong-inputs"
+        record.save(rt.state_dir)
+    elif damage == "role":
+        rt.judge = ModelRole("judge", "codex", "different-model", "medium")
+    else:
+        rt.gateway._settings = SimpleNamespace(zcode_reasoning_level="low")
+    path = InitRecord.path(rt.state_dir, "toy", "knowledge")
+    before, calls, pushes = path.read_bytes(), len(offline.calls), len(rt.gh_run.pushed)
+    with pytest.raises(InitError, match="foundation"):
+        run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, from_existing=True, unlimited_subscription=True)
+    assert path.read_bytes() == before and len(offline.calls) == calls and len(rt.gh_run.pushed) == pushes
+    if restore:
+        restore[0].write_bytes(restore[1])
+        resumed = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, from_existing=True, unlimited_subscription=True)
+        assert resumed.status == "published" and len(offline.calls) == calls
+        # A previously published record remains immutable historical output,
+        # including retained limitations; archive loss cannot trigger a rerun.
+        resumed.unfinished.append("retained historical scan limitation")
+        resumed.save(rt.state_dir)
+        restore[0].unlink()
+        published_bytes = path.read_bytes()
+        cached = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, from_existing=True, unlimited_subscription=True)
+        assert cached.status == "published" and path.read_bytes() == published_bytes and len(offline.calls) == calls