Add the knowledge merge flow, snapshot activation, release sweep and scheduler

doc/architecture/SPEC/engine/steps/knowledge.md

@@ -16,3 +16,6 @@ auto_merge 仓库写 outbox 的 `open_pr` 项，shadow 只记录）。逻辑全
 
 ## 测试
 `test_kb_intake_gate.py`。
+
+## 2026-09-28 合并、巡检、激活步骤
+新增 `knowledge.advance_merges`、`knowledge.sweep`、`knowledge.activate`；`kb run` 路径自行持有租约，调度器路径复用其租约。

doc/architecture/SPEC/kb_service.md

@@ -41,3 +41,23 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv
 - `runtime`：collect → intake → gate → publish；shadow 只记录；变更集文件存 `changesets/<id>.json`。
 - `calibration`：按仓库的校准集评分（坏样例须全部拦下、好样例误拒 ≤ 20%）。
 - `runner`：经标准 executor 运行 `kb-*` playbook。CLI 新增 `kb run`、`kb calibrate`。
+
+## 2026-09-28 合并流程、激活、巡检与调度
+- `merge`：publisher 回执推进状态（pr_requested → pr_open → verdict_posted → queued → merged）；
+  对 PR 的精确 head 签发 kb-gate 判定（清单、逐块、一致性）并经 outbox `post_verdict` 发布；
+  PR 阶段 `kb-gate` 成功后才发 `enqueue`，每仓库同时至多 1 个排队；head 变化 → human；
+  关闭未合并 → closed；合并后记录退役以便下个发版 purge。`pause_open_prs` 发出出队 + 转 draft 项。
+- `activate`：每个 main SHA 生成不可变快照（`snapshots/<sha>/knowledge` + MANIFEST），
+  加载校验、树级检查、各仓库 `_routes.yaml` 可加载后原子切换 `active`；支持回滚；保留最近 10 个及当前激活。
+- `sweep`：发版触发（首次只记录基线）/ 兜底周期；T1 结构报告（唯一自动修复是补索引链接）；
+  T2/T3 按规则页让生成器对照发版 diff 逐条给出 keep/edit/replace/retire；purge；全局熔断强制 human。
+- `scheduler`（`kb serve`）：终身持有租约；每 tick 重签控制记录与暂停清单、处理回执、按仓库
+  advance/intake/sweep（仓库间隔离）、main 变化即激活；24 小时内 2 个知识 PR 被人关闭 → 暂停该仓库并出队。
+- CLI：`kb serve [--once]`、`kb activate`、`kb rollback --to SHA`；`kb pause` 同时为已开 PR 发暂停项。
+- 每个变更集记录其待回执的 outbox 项（`pending_item`）：未过期时不重复签发；回执只作用于与之匹配的项和
+  对应的前置状态，迟到的重复项回执不改变状态。
+- 回滚后写入 `rollback_pin`：调度器不会重新激活被回滚的那个 main，只有更新的 main 才会激活。
+- 巡检按页记录进度（`sweep_progress`）：页面评估失败与“保持”区分，失败页在下次重试，3 次后交给人；
+  所有页面落定后基线才推进；兜底巡检只做 T1 与 purge。
+- 熔断暂停与状态变化在同一加锁事务中发布控制记录与暂停清单。
+

doc/architecture/SPEC/knowledge_service.md

@@ -44,3 +44,6 @@ passing per-block judgements, the consistency judgement with the page hashes it
 saw, upstream fact attestations and the decision source (`auto` or
 `human-approved` with its reviews). `check_binding` rejects replay on another
 PR, head or window. Standard library only; vendored by kb-gate.
+
+## 2026-09-28 T1 索引修复
+L1 的索引检查允许为目录中已存在但未列入索引的页面补链接（巡检 T1 修复），新页面仍必须被链接。

doc/architecture/SPEC/playbooks/PLAYBOOKS.md

@@ -54,3 +54,6 @@ success, steps[]`。
 
 ## kb-intake（candidate）
 知识服务按仓库运行：`knowledge.collect_events` → `knowledge.intake` → `knowledge.publish`。candidate 状态，planner 不可见，由知识服务调度。
+
+## kb-merge / kb-sweep / kb-activate（candidate）
+知识服务的合并推进、发版巡检与快照激活；candidate 状态，由 `kb serve` 调度或 `kb run` 手动运行。

doc/features/knowledge-service.md

@@ -3,7 +3,7 @@
 > **一览**
 > | | |
 > |---|---|
-> | **状态** | 🚧 分阶段落地。已合并：Direct 路由多仓库化与按请求知识视图（#203）、Knowledge Ops API 2.0 与 L1（#204）、generated-baseline 发版审计（#205）、服务核心（#206）；本 PR 加入 intake、质量门与校准；本页随后续 PR 更新 |
+> | **状态** | 🚧 分阶段落地。已合并：Direct 路由多仓库化与按请求知识视图（#203）、Knowledge Ops API 2.0 与 L1（#204）、generated-baseline 发版审计（#205）、服务核心（#206）、intake 与质量门（#207）；本 PR 加入合并流程、快照激活、发版巡检与调度器；本页随后续 PR 更新 |
 > | **做什么** | 知识的创建、维护、退役、删除与发版巡检全部在 Copilot；每次变更经质量门，通过即经合并队列自动合并；ReviewBot 只读消费 |
 > | **怎么开关** | 每个仓库在 `adapters/<repo>/manifest.yaml` 的 `knowledge_lifecycle`（人工审阅的高风险段）中设置 `enabled` 与 `mode: shadow\|auto_merge` |
 > | **设计** | 经 GPT-6 sol 评审批准的设计文档（知识库管理重构设计，多仓库，v7） |
@@ -50,3 +50,17 @@ infermatrix-copilot kb calibrate --repo vllm-omni                  # 评审模
 模型：`KB_GENERATOR`（默认 `claude-code:claude-opus-5-5`）、`KB_JUDGE`（默认
 `codex:gpt-6-sol:medium`）。每次模型调用写入 `$KB_STATE_DIR/traces/model_calls.jsonl`。
 
+## 调度、合并、激活与巡检
+
+```bash
+infermatrix-copilot kb serve                 # 常驻调度器（systemd），持有单写入租约
+infermatrix-copilot kb activate              # 立即激活知识仓库 main 的快照
+infermatrix-copilot kb rollback --to <sha>   # 回指到较早的快照（下一请求生效）
+```
+
+合并流程：`open_pr` 回执 → 为该 PR 的精确 head 签发判定并发布评论 → PR 阶段 `kb-gate`
+通过后入合并队列（每仓库至多 1 个）→ 合并后激活新快照。发版巡检在上游新 release 后运行
+T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。
+
+评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。
+

playbooks/kb-activate.yaml

@@ -0,0 +1,16 @@
+# Build, verify and atomically activate the knowledge snapshot of the knowledge
+# repository's main (what Direct/Strict reviews read). Candidate.
+name: kb-activate
+version: 1
+status: candidate
+provenance:
+  created_by: human
+  source: knowledge-base redesign (Copilot owns the knowledge base), v7
+task_kinds: [knowledge_maintenance]
+repos: []
+requires: []
+params:
+  repo: "any enabled knowledge repository (activation is tree-wide)"
+steps:
+  - {id: activate, step: knowledge.activate}
+success: state/kb/active points at a verified snapshot of main

playbooks/kb-merge.yaml

@@ -0,0 +1,17 @@
+# Knowledge service merge flow for ONE repository: apply publisher acks, sign
+# verdicts for opened PRs, enqueue once the PR-stage kb-gate passes (one queued
+# knowledge PR per repository), record merges. Candidate: scheduler-driven.
+name: kb-merge
+version: 1
+status: candidate
+provenance:
+  created_by: human
+  source: knowledge-base redesign (Copilot owns the knowledge base), v7
+task_kinds: [knowledge_maintenance]
+repos: []
+requires: []
+params:
+  repo: "knowledge repository name"
+steps:
+  - {id: advance, step: knowledge.advance_merges}
+success: every in-flight knowledge PR advanced one state or routed to people

playbooks/kb-sweep.yaml

@@ -0,0 +1,18 @@
+# Knowledge service release sweep for ONE repository: T1 structure, T2/T3
+# re-check of every active rule against the release diff, purge of rules
+# retired at least one release ago. Every change set goes through the gate;
+# a sweep-wide circuit breaker sends them all to people. Candidate.
+name: kb-sweep
+version: 1
+status: candidate
+provenance:
+  created_by: human
+  source: knowledge-base redesign (Copilot owns the knowledge base), v7
+task_kinds: [knowledge_maintenance]
+repos: []
+requires: []
+params:
+  repo: "knowledge repository name"
+steps:
+  - {id: sweep, step: knowledge.sweep}
+success: the release is swept (or none was due); change sets gated and staged

src/infermatrix_copilot/engine/steps/knowledge.py

@@ -83,3 +83,69 @@ async def publish(ctx: StepContext) -> StepResult:
     status = publish_changeset(rt, lifecycle, changeset_id)
     return StepResult(True, summary=f"change set {changeset_id}: {status}",
                       outputs={"state_updates": {"kb_changeset_status": status}})
+
+
+def _with_lease(rt, work):
+    """Run ``work(owner)`` under the single-writer lease (``kb run`` path); the
+    scheduler already holds it and passes its own owner."""
+    if rt.lease_owner:
+        return work(rt.lease_owner)
+    with rt.ledger.lease() as owner:
+        return work(owner)
+
+
+@step("knowledge.advance_merges", kind="deterministic", risk="knowledge",
+      description="Advance in-flight knowledge PRs: sign verdicts, enqueue, record merges")
+async def advance_merges(ctx: StepContext) -> StepResult:
+    rt, lifecycle = _lifecycle(ctx)
+    if lifecycle is None:
+        return StepResult(False, FailureKind.BLOCKED, "repository is unknown or disabled")
+    from ...kb_service import merge
+
+    if rt.outbox is not None and rt.publisher_public_key is not None:
+        merge.apply_acks(rt, rt.outbox.collect_acks(rt.publisher_public_key))
+    events = _with_lease(rt, lambda owner: merge.advance(rt, lifecycle))
+    return StepResult(True, summary="; ".join(events) or "nothing to advance",
+                      outputs={"state_updates": {"kb_merge_events": events}})
+
+
+@step("knowledge.sweep", kind="agent", risk="knowledge",
+      description="Release sweep: T1 structure, T2/T3 rule re-checks, purge; each page through the gate")
+async def sweep(ctx: StepContext) -> StepResult:
+    rt, lifecycle = _lifecycle(ctx)
+    if lifecycle is None or not lifecycle.full_name:
+        return StepResult(False, FailureKind.BLOCKED, "repository is unknown, disabled or has no upstream")
+    from ...kb_service.runtime import publish as publish_changeset
+    from ...kb_service.sweep import UpstreamRepo, detect_release, run_sweep
+
+    upstream = UpstreamRepo(rt.state_dir / "upstream" / f"{lifecycle.repo}.git", lifecycle.full_name)
+
+    def work(owner):
+        found = detect_release(rt, lifecycle, upstream)
+        if found is None:
+            return None
+        report = run_sweep(rt, lifecycle, owner, found, upstream)
+        for changeset_id in report["changesets"]:
+            publish_changeset(rt, lifecycle, changeset_id)
+        return report
+
+    report = _with_lease(rt, work)
+    if report is None:
+        return StepResult(True, summary="no release to sweep")
+    return StepResult(True, summary=f"sweep {report['sweep']['tag']}: {len(report['changesets'])} change set(s)",
+                      outputs={"state_updates": {"kb_sweep_changesets": report["changesets"]}})
+
+
+@step("knowledge.activate", kind="deterministic", risk="knowledge",
+      description="Build, verify and activate the knowledge snapshot of the knowledge repository's main")
+async def activate_snapshot(ctx: StepContext) -> StepResult:
+    rt = _runtime(ctx)
+    from ...kb_service.activate import ActivationError, activate
+
+    try:
+        sha = rt.knowledge.fetch()
+        snapshot = activate(rt, sha)
+    except ActivationError as exc:
+        return StepResult(False, FailureKind.ESCALATE, f"activation refused: {exc}")
+    return StepResult(True, summary=f"active snapshot {sha}",
+                      outputs={"state_updates": {"kb_active_snapshot": sha, "kb_snapshot_path": str(snapshot)}})

src/infermatrix_copilot/kb_service/activate.py

@@ -0,0 +1,161 @@
+"""Knowledge snapshot activation: what reviews read, switched atomically.
+
+Every merged knowledge change (from the service or from people) becomes an
+immutable snapshot ``snapshots/<merge sha>/`` holding ``knowledge/`` and a
+``MANIFEST.json``. Before ``active`` is switched to it, the snapshot must load
+as a verified ``KnowledgeView``, have no tree-level issues, and every
+repository's ``_routes.yaml`` must load. The switch is an atomic symlink
+replace; rollback points ``active`` back at an earlier snapshot. The latest
+``KEEP`` snapshots are kept, and so is anything ``active`` points at.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+import shutil
+import tempfile
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator
+
+from ..knowledge_service.l1 import check_tree
+from ..knowledge_view import KnowledgeView, _load_view, build_manifest
+from .outbox import atomic_write_json
+
+KEEP = 10
+
+
+class ActivationError(RuntimeError):
+    """A snapshot failed its checks; ``active`` was not switched."""
+
+
+@contextmanager
+def activation_lock(state_dir: Path) -> Iterator[None]:
+    """Serialises activation, rollback and the scheduler's "read the rollback
+    pin, then maybe activate" decision across processes, so the symlink, the
+    activation record and the pin never disagree and a tick cannot undo a
+    rollback that finished while it was deciding."""
+    state_dir.mkdir(parents=True, exist_ok=True)
+    fd = os.open(state_dir / ".activation.lock", os.O_RDWR | os.O_CREAT, 0o600)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX)
+        yield
+    finally:
+        fcntl.flock(fd, fcntl.LOCK_UN)
+        os.close(fd)
+
+
+def snapshots_dir(state_dir: Path) -> Path:
+    return state_dir / "snapshots"
+
+
+def active_link(state_dir: Path) -> Path:
+    return state_dir / "active"
+
+
+def build_snapshot(state_dir: Path, sha: str, files: dict[str, str], *, extra: dict[str, str]) -> Path:
+    """Write the snapshot directory (idempotent: an existing one is reused).
+    ``files`` are the governed knowledge texts; ``extra`` any other files the
+    tree needs to be served (e.g. AGENTS.md, README.md)."""
+    target = snapshots_dir(state_dir) / sha
+    if (target / "MANIFEST.json").is_file():
+        return target
+    staging = Path(tempfile.mkdtemp(prefix=f".{sha}.", dir=snapshots_dir(state_dir).mkdir(
+        parents=True, exist_ok=True) or snapshots_dir(state_dir)))
+    try:
+        root = staging / "knowledge"
+        for rel, text in {**extra, **files}.items():
+            path = root / rel
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(text, encoding="utf-8")
+        atomic_write_json(staging / "MANIFEST.json", build_manifest(root, sha))
+        os.replace(staging, target)
+    except BaseException:
+        shutil.rmtree(staging, ignore_errors=True)
+        raise
+    return target
+
+
+def verify_snapshot(snapshot: Path) -> KnowledgeView:
+    from ..knowledge_view import KnowledgeViewError
+
+    _load_view.cache_clear()
+    try:
+        view = _load_view(str(snapshot.resolve()))
+    except KnowledgeViewError as exc:
+        raise ActivationError(f"snapshot does not load: {exc}") from exc
+    # _load_view checks names and structure; hash EVERY file now, since not
+    # every consumer reads through KnowledgeView.path
+    for rel in view.files or {}:
+        try:
+            view.path(rel)
+        except (KnowledgeViewError, FileNotFoundError, ValueError) as exc:
+            raise ActivationError(f"snapshot content does not match its manifest: {exc}") from exc
+    files = {p.relative_to(view.root).as_posix(): p.read_text(encoding="utf-8")
+             for p in view.root.rglob("*") if p.is_file() and p.suffix in (".md", ".yaml")}
+    issues = check_tree({k: v for k, v in files.items() if k.startswith(("repos/", "general/"))})
+    if issues:
+        raise ActivationError(f"snapshot has tree issues: {[i.to_dict() for i in issues[:5]]}")
+    from ..direct_routing import _load_routes
+
+    for repo_dir in sorted((view.root / "repos").iterdir()):
+        if (repo_dir / "_routes.yaml").is_file():
+            try:
+                _load_routes(view, repo_dir.name)
+            except Exception as exc:
+                raise ActivationError(f"{repo_dir.name}/_routes.yaml does not load: {exc}") from exc
+    return view
+
+
+def switch_active(state_dir: Path, snapshot: Path) -> None:
+    link = active_link(state_dir)
+    tmp = link.with_name(f".active.{os.getpid()}")
+    if tmp.is_symlink() or tmp.exists():
+        tmp.unlink()
+    tmp.symlink_to(snapshot.resolve())
+    os.replace(tmp, link)
+
+
+def activate(rt, sha: str, *, locked: bool = False) -> Path:
+    """Build, verify and switch to the snapshot of knowledge at ``sha``."""
+    if not locked:
+        with activation_lock(rt.state_dir):
+            return activate(rt, sha, locked=True)
+    files = rt.knowledge.knowledge_files(sha)
+    extra = rt.knowledge.top_level_knowledge(sha)
+    snapshot = build_snapshot(rt.state_dir, sha, files, extra=extra)
+    verify_snapshot(snapshot)
+    switch_active(rt.state_dir, snapshot)
+    rt.ledger.record_activation(sha, {"files": len(files)})
+    rt.ledger.set_cursor("*", "rollback_pin", "")
+    prune(rt.state_dir)
+    return snapshot
+
+
+def rollback(rt, sha: str) -> Path:
+    """Point ``active`` at an earlier snapshot and pin it: the scheduler will not
+    re-activate the main it was rolled back from, only a LATER main."""
+    with activation_lock(rt.state_dir):
+        snapshot = snapshots_dir(rt.state_dir) / sha
+        if not (snapshot / "MANIFEST.json").is_file():
+            raise ActivationError(f"no snapshot for {sha}")
+        verify_snapshot(snapshot)
+        switch_active(rt.state_dir, snapshot)
+        rt.ledger.record_activation(sha, {"rollback": True})
+        rt.ledger.set_cursor("*", "rollback_pin",
+                             json.dumps({"snapshot": sha, "main": rt.knowledge.main_sha()}))
+        return snapshot
+
+
+def prune(state_dir: Path, keep: int = KEEP) -> None:
+    root = snapshots_dir(state_dir)
+    link = active_link(state_dir)
+    active = link.resolve() if link.is_symlink() else None
+    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)
+    for old in snapshots[keep:]:
+        if active is not None and old.resolve() == active:
+            continue
+        shutil.rmtree(old, ignore_errors=True)