kb init: every routed owner page carries a Direct quick map

doc/architecture/SPEC/direct_routing.md

@@ -1,6 +1,6 @@
 # direct_routing.py —— 规范
 
-<!-- verified-against: 2026-09-29 -->
+<!-- verified-against: 2026-10-01 -->
 
 `LOC ~880 · Direct 模式完整策略包与仓库中立的知识路由 · refactor-status: stable`
 
@@ -21,7 +21,7 @@ Direct policy bundle）、`direct_knowledge_routes`、
 - **repo 守卫最先跑**：不支持的仓库在任何路由计算之前被拒 —— 修的是一个
   真实历史 bug（守卫曾排在空 intent 提前返回之后，向不支持的仓库泄漏
   owner 知识）。
-- **quick map fail-closed**：`_direct_quick_map` 返回内嵌代码地图与状态
+- **quick map fail-closed**：`_direct_quick_map(path)` 读文件后委托给 `_direct_quick_map_text(text)`（同一提取逻辑；kb init 用后者在落盘前检查它要路由的页面）返回内嵌代码地图与状态
   `{ok, truncated, unavailable}`，`truncated` 不是装饰 —— 把残图当全图
   与缺图同罪、且更难察觉；`_direct_route` 据此置
   `read_required = status != "ok"`（"自己去打开"是真回退，"什么都不给

doc/architecture/SPEC/kb_service.md

@@ -500,3 +500,30 @@ Direct 的路由优先级是：知识侧 `repos/<repo>/_routes.yaml` 存在则
 - 没有 routes 文件也没有 `review_routes` 的仓库：modules / deepen `blocked`（消息指明两种来源都没有）。
 测试：`test_kb_init_manifest_routed.py`（manifest 路由的 toy：三阶段 dry run 不写 `_routes.yaml`、规则落在 owner 页、
 briefing 文档不动、清单含建议项、覆盖率来源；知识侧路由的 toy 行为不变）。
+
+## 2026-10-01 路由到的页面必须带 Direct 快速入口（kb init）
+Direct 的产品就是内嵌地图：知识侧 `_routes.yaml` 路由到的每个 owner 页都要有 `## … Direct …` 段
+（`direct_routing._direct_quick_map_text`：第一个这样的标题到下一个 `## `，正文非空、≤3500 字符），
+`test_every_routed_page_yields_a_quick_map` 逐页把关；#265 路由到的页面没有。`init_quick_maps`：
+- `render_quick_map(page_text, signals=, prefixes=, key_files=)`：确定性渲染 `## 代码快速入口（Direct）`（标题以格式的 ID
+  扫描器 `ANY_RULE_HEADING` 读不成 ID 的词开头——`## Direct …` 会被读成 ID `Direct`、两页即 `duplicate_rule_id`；Direct 只要求标题行含 Direct）
+  （第二行是标记 `<!-- kb-init:quick-map -->`）：页面上每条 active 规则一行（触发行或标题 | 规则 ID |
+  规则里反引号命名的路径，否则 owner 前缀），没有规则的页面（地图卡、入口页）一行（触发词 | 入口 | 关键文件或前缀）；
+  ≤3000 字符，超出从末尾丢行并补一行 `| … | … | … |`，绝不超过 3500。
+- `with_quick_map(page_text, section)`：就地替换 init 自己写的段（带标记），没有则插到标题后第一个 `## ` 段之前；
+  没有标记的 Direct 段是人写的地图，原样保留。该标题不含 ` — `，永远不是规则标题
+  （`RULE_HEADING` / `_TOP_RULE`），L1 把它当 prose 块，页面上的规则不受影响。
+- `_Stage._refresh_quick_maps()`（`_conclude` 开头，三阶段共用）：head 里 routes 文件的每个 owner 页，本阶段改动过的
+  重新渲染（deepen 追加规则后行数随之更新；init 段之后页面新增的内容——如追加到入口页的索引行——保留在段尾），
+  还没有地图的补上，未改动且已有地图的不动（路由前缀漂移不制造改动）；`mapped()` 只在加上地图仍在页面容量内时
+  返回结果，否则该页不写并作为阻断问题上报（不能留下少一行的旧地图）；briefing 文档不写（记入清单，随后被检查挡下）；skeleton 从既有 `_index.md`
+  推导 owner 时跳过 briefing 文档（清单），`_owners_with_room` 把没有容量放地图的既有 owner 页改指到本阶段承接了
+  它溢出规则的页（`_place` 记录的 `_spilled_from`；同目录的其他新页——比如种子页——装的是别的知识，不算），没有这样的页就
+  丢弃该 owner 并记入清单。
+- 容量：`_apply` 在 `apply_operations` 之后把每个写入的规则页连同渲染出的地图（按 routes 文件里该页 owner 的真实触发词与前缀，`_map_inputs`）一起过 `page_over_capacity`（`_check_capacity_with_map`），超出即报 page full，由 `_place` 溢出到同级页——地图每条规则一行，只量规则会让页面
+  在加地图后越界；manifest 路由的仓库不写地图，也就不按地图量（否则装得下的规则会被溢出到 manifest 不路由的页面，
+  Direct 看不到它）。
+- 阻断检查：`validate_change(..., quick_map_pages=owner_pages(routes_text))` → `quick_map_problems`
+  用生产提取器逐页检查，缺页或 `unavailable` 即 blocked 并点名页面；本次写入或改动的页面被服务端截断（`truncated`，CI 拒绝新增的截断）也 blocked。manifest 路由的仓库没有 routes 文件、
+  不检查（Direct 的 manifest 回退对其页面报 `read_required`）。
+测试：`test_kb_init_quick_maps.py`。

src/infermatrix_copilot/direct_routing.py

@@ -287,7 +287,15 @@ def _signal_matches(text: str, signal: str) -> bool:
 
 
 def _direct_quick_map(path: str, max_chars: int = 3500) -> tuple[str, str]:
-    """Return the embedded Direct code map and its status, never the whole page.
+    """Return the embedded Direct code map of the page at ``path`` and its
+    status (``_direct_quick_map_text`` on the file's text)."""
+    return _direct_quick_map_text(Path(path).read_text(encoding="utf-8"), max_chars)
+
+
+def _direct_quick_map_text(text: str, max_chars: int = 3500) -> tuple[str, str]:
+    """Return the embedded Direct code map in ``text`` and its status, never
+    the whole page. Text-taking so a producer (kb init) can check a page it
+    has not written to disk yet against the same extraction the server runs.
 
     Status is ``ok`` / ``truncated`` / ``unavailable``. The last two both mean the
     host cannot rely on the excerpt alone.
@@ -298,7 +306,7 @@ def _direct_quick_map(path: str, max_chars: int = 3500) -> tuple[str, str]:
     A partial map presented as whole is the same lie as a missing one, just harder
     to notice.
     """
-    lines = Path(path).read_text(encoding="utf-8").splitlines()
+    lines = text.splitlines()
     start = next(
         (
             index for index, line in enumerate(lines)

src/infermatrix_copilot/kb_service/init_quick_maps.py

@@ -0,0 +1,264 @@
+"""Direct quick maps on the pages kb init routes to.
+
+Direct's product is the embedded map: for a repository routed by a
+knowledge-side ``_routes.yaml``, every owner page must carry a ``## … Direct …``
+section (``direct_routing._direct_quick_map_text``: the first such heading up to
+the next ``## ``, non-empty, at most 3500 characters), and
+``test_every_routed_page_yields_a_quick_map`` fails CI on any page that does
+not. The first real run (afd-plugin skeleton, PR #265) routed pages that had
+none.
+
+kb init therefore renders that section deterministically on every owner page
+of the routes file it writes or extends — the rules on the page as rows, or one
+row for a map card or an entry page — keeps it the first section after the
+title, regenerates it whenever it appends rules to the page (deepen), and
+blocks a change whose routed pages would not yield a map. A map init did not
+write (no ``MARKER``) is somebody's hand-written map and is never rewritten. A
+repository routed by its adapter manifest carries no routes file and is exempt:
+Direct's manifest fallback reports ``read_required`` for its pages.
+
+Standard library plus the knowledge format; nothing here names a repository.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import replace
+from typing import Iterable, Mapping, Sequence
+
+from ..knowledge_service.facts import CODE_SPAN, PATH
+from ..knowledge_service.lifecycle import LifecycleError, Page, Section
+
+# The ID scanner of the knowledge format (``lifecycle.ANY_RULE_HEADING``) reads
+# ``## Word ...`` as the ID ``Word`` and L1 refuses one ID on two pages, so the
+# heading opens with a word that scanner cannot take for an ID; Direct only
+# needs "Direct" somewhere on the heading line.
+QUICK_MAP_HEADING = "## 代码快速入口（Direct）"
+MARKER = "<!-- kb-init:quick-map -->"
+MAX_CHARS = 3000          # what init renders (rows are dropped past it)
+HARD_CAP = 3500           # what Direct serves whole (``_direct_quick_map`` cap)
+MAX_CELL = 80
+MAX_SOURCES = 3
+_DIRECT = re.compile(r"^##\s+.*Direct", re.IGNORECASE)
+_TRIGGER = re.compile(r"^\s*[-*]\s*(?:\*\*)?触发(?:\*\*)?\s*[:：]\s*(?P<text>.+?)\s*$")
+_TITLE = re.compile(r"^#{2,3}\s+[A-Za-z0-9][A-Za-z0-9-]{1,40}\s+[—-]\s+(?P<title>.+?)\s*$")
+_HEADER = ("| PR 描述在做什么 | 精确规则 | 第一批 live 源码 |", "|---|---|---|")
+_MORE = "| … | … | … |"
+
+
+def _cell(text: str, limit: int = MAX_CELL) -> str:
+    text = " ".join(str(text).split()).replace("|", "\\|")
+    return text if len(text) <= limit else text[:limit - 1].rstrip() + "…"
+
+
+def _paths_in(text: str, limit: int = MAX_SOURCES) -> list[str]:
+    """The repository paths (``path`` or ``path::Symbol``) ``text`` names in
+    backticks, in order of first appearance."""
+    out: list[str] = []
+    for match in CODE_SPAN.finditer(text):
+        token = match.group("tok").strip()
+        if PATH.fullmatch(token.split("::", 1)[0]) and token not in out:
+            out.append(token)
+            if len(out) >= limit:
+                break
+    return out
+
+
+def _trigger(section: Section) -> str:
+    for line in section.text.splitlines()[1:]:
+        match = _TRIGGER.match(line)
+        if match:
+            return match.group("text")
+    match = _TITLE.match(section.heading)
+    return match.group("title") if match else section.rule_id
+
+
+def _sources(paths: Sequence[str], fallback: Sequence[str]) -> str:
+    chosen = list(paths) or list(fallback)[:MAX_SOURCES]
+    return "、".join(f"`{p}`" for p in chosen) if chosen else "—"
+
+
+def render_quick_map(page_text: str, *, signals: Sequence[str], prefixes: Sequence[str],
+                     key_files: Sequence[str] = ()) -> str:
+    """The Direct section for ``page_text``: one row per active rule on the
+    page (its 触发 line or title, its ID, the paths it names), or one row for
+    a page without rules (a map card or an entry page: the owner's signals,
+    its key files or scope prefixes). Deterministic; at most ``MAX_CHARS``
+    (rows are dropped from the end behind a ``…`` row), never over
+    ``HARD_CAP``."""
+    try:
+        rules = [s for s in Page.parse(page_text).rules() if s.footer.status == "active"]
+    except LifecycleError:
+        rules = []
+    rows: list[str] = []
+    for section in rules:
+        rows.append(f"| {_cell(_trigger(section))} | {section.rule_id} | "
+                    f"{_sources(_paths_in(section.text), prefixes)} |")
+    if not rows:
+        files = list(key_files) or _paths_in(page_text, MAX_SOURCES)
+        rows.append(f"| {_cell('、'.join(signals) or '（无触发词）')} | 入口 | {_sources(files, prefixes)} |")
+    lead = (("触发词：" + "、".join(_cell(s, 40) for s in list(signals)[:12]) + "。") if signals else "")
+    lead += "PR 描述或改动命中下表一行时，先读该行的规则，再读页面其余部分。"
+
+    def render(kept: list[str], more: bool) -> str:
+        return "\n".join([QUICK_MAP_HEADING, MARKER, "", lead, "", *_HEADER, *kept, *([_MORE] if more else [])])
+
+    kept, more = list(rows), False
+    text = render(kept, more)
+    while len(text) > MAX_CHARS and kept:
+        kept.pop()
+        more = True
+        text = render(kept, more)
+    if len(text) > HARD_CAP:      # pragma: no cover - the lead alone is far under the cap
+        raise ValueError(f"quick map cannot fit under {HARD_CAP} characters")
+    return text + "\n"
+
+
+def _after_table(section_text: str) -> str:
+    """The lines of init's map section that follow ITS table (content the
+    page gained after the map was written: an index line, prose, even a
+    second table); "" when there are none. The generated table is the run of
+    ``|`` lines right after the ``|---|---|---|`` separator, so anything past
+    that run — including another table behind prose — is the page's."""
+    lines = section_text.splitlines()
+    separator = next((i for i, line in enumerate(lines) if line.strip() == _HEADER[1]), None)
+    if separator is None:
+        return ""
+    end = separator + 1
+    while end < len(lines) and lines[end].startswith("|"):
+        end += 1
+    return "\n".join(lines[end:])
+
+
+def has_map(page_text: str) -> bool:
+    """The page carries a Direct section (init's or hand-written)."""
+    return any(_DIRECT.match(s.heading) for s in Page.parse(page_text).sections)
+
+
+def has_hand_written_map(page_text: str) -> bool:
+    """A Direct section that init did not write (no ``MARKER``)."""
+    for section in Page.parse(page_text).sections:
+        if _DIRECT.match(section.heading):
+            return MARKER not in section.text
+    return False
+
+
+def with_quick_map(page_text: str, section_text: str) -> str:
+    """``page_text`` with ``section_text`` as its Direct section: init's own
+    map replaced in place, a page without one gets it as the first section
+    after the title, a hand-written map is left exactly as it is."""
+    page = Page.parse(page_text)
+    body = section_text.rstrip("\n") + "\n\n"
+    sections = list(page.sections)
+    for index, section in enumerate(sections):
+        if _DIRECT.match(section.heading):
+            if MARKER not in section.text:
+                return page_text
+            # what the page gained after the map (an index line appended to an
+            # entry page, say) belongs to the page, not to the map: keep it
+            tail = _after_table(section.text)
+            last = index == len(sections) - 1
+            trailing = section.text[len(section.text.rstrip("\n")):] if last else "\n\n"
+            rendered = section_text.rstrip("\n") + ("\n\n" + tail.rstrip("\n") if tail.strip() else "")
+            sections[index] = Section(rendered + (trailing or "\n"))
+            return replace(page, sections=tuple(sections)).render()
+    head = page.head if page.head.endswith("\n\n") else page.head.rstrip("\n") + "\n\n"
+    if not sections:
+        body = section_text.rstrip("\n") + "\n"
+    return replace(page, head=head, sections=(Section(body), *sections)).render()
+
+
+def mapped(page_text: str, *, signals: Sequence[str], prefixes: Sequence[str]) -> str | None:
+    """``page_text`` with its Direct quick map, or None when the map would
+    push the page over the knowledge format's capacity (``page_over_capacity``:
+    the caller keeps the page as it is and says so)."""
+    from ..knowledge_service.ops import page_over_capacity
+
+    text = with_quick_map(page_text, render_quick_map(page_text, signals=signals, prefixes=prefixes))
+    return None if page_over_capacity(text) else text
+
+
+def map_inputs(routes_text: str | None, page: str) -> tuple[list[str], list[str]]:
+    """The (signals, scope_prefixes) the routes file gives ``page``'s first
+    owner — what its map is rendered with — or empty lists."""
+    import yaml
+
+    try:
+        doc = yaml.safe_load(routes_text or "") or {}
+    except yaml.YAMLError:
+        return [], []
+    for owner in (doc.get("owners") or []) if isinstance(doc, dict) else []:
+        if isinstance(owner, dict) and str(owner.get("path") or "") == page:
+            return ([str(x) for x in owner.get("signals") or []],
+                    [str(x) for x in owner.get("scope_prefixes") or []])
+    return [], []
+
+
+def refresh_quick_maps(head: Mapping[str, str], base: Mapping[str, str], routes_text: str | None, *,
+                       briefing_docs: Iterable[str]) -> tuple[dict[str, str], list[str], list[str]]:
+    """``head`` with init's Direct quick map on every owner page of the
+    routes file that this change touched or that has no map yet, plus the
+    checklist lines and the BLOCKING problems that produced. An owner page
+    unchanged since ``base`` that already has a map is left alone (routing
+    drift makes no churn); a briefing doc is never written (checklist; the
+    presence check then blocks); a page whose regenerated map no longer
+    fits the format's capacity is a problem, never a stale map."""
+    out, checklist, problems = dict(head), [], []
+    off_limits = set(briefing_docs)
+    try:
+        pages = owner_pages(routes_text)
+    except ValueError:
+        return out, checklist, problems       # the routes file is reported by the validators
+    for page in pages:
+        text = out.get(page)
+        if text is None or not page.endswith(".md"):
+            continue
+        if page in off_limits:
+            checklist.append(f"{page} is a briefing doc (knowledge.briefing_docs): init does not write its "
+                             "Direct quick map; add one by hand or route another page")
+            continue
+        if text == base.get(page) and has_map(text):
+            continue
+        signals, prefixes = map_inputs(routes_text, page)
+        with_map = mapped(text, signals=signals, prefixes=prefixes)
+        if with_map is None:
+            problems.append(f"quick map: {page} has no room for its Direct quick map (page capacity): "
+                            "split the page by hand, or route another page")
+            continue
+        out[page] = with_map
+    return out, checklist, problems
+
+
+def owner_pages(routes_text: str | None) -> list[str]:
+    """The owner pages a routes file names, in routing order, each once."""
+    from .init_coverage import load_owners
+
+    if routes_text is None:
+        return []
+    return list(dict.fromkeys(o.path for o in load_owners(routes_text)))
+
+
+def quick_map_problems(files: Mapping[str, str], pages: Iterable[str],
+                       base: Mapping[str, str] | None = None) -> list[str]:
+    """Why Direct could not serve the routed ``pages`` of ``files``: the same
+    extraction the server runs (``_direct_quick_map_text``), page by page.
+    Blocking: a route without a map hands the host nothing, and a map the
+    server would truncate is not whole — CI refuses newly truncated maps,
+    so a page this change wrote or changed (``base``) may not truncate."""
+    from ..direct_routing import _direct_quick_map_text
+
+    problems = []
+    for page in pages:
+        text = files.get(page)
+        if text is None:
+            problems.append(f"quick map: routed page {page} does not exist")
+            continue
+        _, status = _direct_quick_map_text(text)
+        if status == "unavailable":
+            problems.append(f"quick map: {page} yields no Direct quick map (a routed page needs a "
+                            f"`{QUICK_MAP_HEADING[3:]}` section with a non-empty body)")
+        elif status == "truncated" and (base is None or base.get(page) != text):
+            problems.append(f"quick map: {page}: the Direct section as served exceeds {HARD_CAP} characters "
+                            "(truncated); keep the map first and short, and start the rules with a `## ` "
+                            "heading so level-three rules are not part of it")
+    return problems