From 7d5d5828d5cfaeca57d52d462660b2befbf962c5 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Thu, 1 Oct 2026 01:55:05 +0800 Subject: [PATCH] kb init: every routed owner page carries a Direct quick map MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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: first such heading to the next "## ", non-empty, at most 3500 chars) and test_every_routed_page_yields_a_quick_map fails CI otherwise. The first real run (afd-plugin skeleton, PR #265) routed pages that had none. init_quick_maps renders that section deterministically on every owner page init writes or routes to: one row per active rule on the page (its 触发 line or title, its ID, the paths it names, else the owner's prefixes) or one 入口 row for a map card or entry page; at most 3000 chars, rows dropped from the end behind a "…" row, never over Direct's 3500. The heading is "## 代码快速入口（Direct）": the knowledge format's loose ID scanner reads "## Direct …" as the ID "Direct", and L1 refuses one ID on two pages, so the heading opens with a word it cannot take for an ID. The section is the first after the title, marked as init's own, and L1 sees its changes as prose: rules on the page are untouched. A map init did not write (no marker) is hand-written and never rewritten. _Stage._conclude refreshes the maps of the owner pages the stage changed (deepen's appended rules become rows; what the page gained after the generated table — an index line, prose, another table — stays after it) and adds one to a routed page that has none; a regenerated map that no longer fits the page is a blocking problem, never a stale map. Then validate_change blocks — with the page named — when any owner page yields no map, or when a page this change wrote would be truncated as served (CI refuses new truncation), using the same extractor the server runs (_direct_quick_map now delegates to a text-taking _direct_quick_map_text). A manifest-routed repository has no routes file and is exempt. Capacity counts the map: _apply measures each written rule page with its map as it will really be rendered (the owner's signals and prefixes from the routes file), so a page that would overflow reports "page full" and the next rule spills to a sibling as before; a pre-existing owner page with no room for even its map is routed to the rule page init wrote beside it, or dropped with a checklist line; a briefing doc is never made an owner. The two skeleton capacity tests now count the map on the routed page. Co-Authored-By: Claude Fable 5.1 --- doc/architecture/SPEC/direct_routing.md | 4 +- doc/architecture/SPEC/kb_service.md | 27 ++ src/infermatrix_copilot/direct_routing.py | 12 +- .../kb_service/init_quick_maps.py | 264 ++++++++++++++ .../kb_service/init_stages.py | 115 +++++- test/test_kb_init_quick_maps.py | 344 ++++++++++++++++++ test/test_kb_init_skeleton.py | 18 +- 7 files changed, 770 insertions(+), 14 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/init_quick_maps.py create mode 100644 test/test_kb_init_quick_maps.py diff --git a/doc/architecture/SPEC/direct_routing.md b/doc/architecture/SPEC/direct_routing.md index 920c3e9c..3105f4ae 100644 --- a/doc/architecture/SPEC/direct_routing.md +++ b/doc/architecture/SPEC/direct_routing.md @@ -1,6 +1,6 @@ # direct_routing.py —— 规范 - + `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"`（"自己去打开"是真回退，"什么都不给 diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 0d95b313..139cdc2c 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -500,3 +500,30 @@ Direct 的路由优先级是：知识侧 `repos//_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） + （第二行是标记 ``）：页面上每条 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`。 diff --git a/src/infermatrix_copilot/direct_routing.py b/src/infermatrix_copilot/direct_routing.py index ef860dc9..7aa9bfa0 100644 --- a/src/infermatrix_copilot/direct_routing.py +++ b/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) diff --git a/src/infermatrix_copilot/kb_service/init_quick_maps.py b/src/infermatrix_copilot/kb_service/init_quick_maps.py new file mode 100644 index 00000000..13548052 --- /dev/null +++ b/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 = "" +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.+?)\s*$") +_TITLE = re.compile(r"^#{2,3}\s+[A-Za-z0-9][A-Za-z0-9-]{1,40}\s+[—-]\s+(?P.+?)\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 diff --git a/src/infermatrix_copilot/kb_service/init_stages.py b/src/infermatrix_copilot/kb_service/init_stages.py index 0d87102c..9833e6c0 100644 --- a/src/infermatrix_copilot/kb_service/init_stages.py +++ b/src/infermatrix_copilot/kb_service/init_stages.py @@ -29,7 +29,7 @@ from collections import Counter from dataclasses import dataclass, field from pathlib import Path, PurePosixPath -from typing import Any, Mapping +from typing import Any, Mapping, Sequence import yaml @@ -38,6 +38,7 @@ from ..knowledge_service.lifecycle import LifecycleError, Page from ..knowledge_service.ops import ( INDEX_NAME, KnowledgeOperation, all_rule_ids, all_tombstoned_ids, apply_operations, index_line, + page_over_capacity, ) from ..knowledge_service.pinned_claims import Evidence, check_rules, evidence_for from .init_budget import Budget, BudgetExhausted, PriceError @@ -109,12 +110,15 @@ def validate_change(base: Mapping[str, str], head: Mapping[str, str], *, observer, rules: Mapping[str, str], evidence: list[Evidence], other: Mapping[str, tuple[str | None, str | None]] | None = None, - check_other=None) -> list[str]: + check_other=None, quick_map_pages: Sequence[str] = ()) -> list[str]: """Blocking problems of one init change. ``base``/``head`` are knowledge-relative trees, ``rules`` the rules the stage wrote (rule ID -> text), ``other`` the repository-relative paths outside ``knowledge/`` - (base, head) that ``check_other(path, before, after)`` must accept.""" + (base, head) that ``check_other(path, before, after)`` must accept, + ``quick_map_pages`` the owner pages of a knowledge-side routes file, each + of which must yield a Direct quick map (``init_quick_maps``).""" from ..knowledge_service.l1 import check_changeset, check_index_links, check_tree + from .init_quick_maps import quick_map_problems result = check_changeset(base, head, knowledge_changes(base, head), bootstrap=True) problems = [f"L1 {i.code} {i.path}: {i.detail}" for i in result.issues] @@ -122,6 +126,7 @@ def validate_change(base: Mapping[str, str], head: Mapping[str, str], *, observe problems += [f"index {i.code} {i.path}: {i.detail}" for i in check_index_links(base, head)] problems += claim_problems(observer, rules, evidence) problems += other_path_problems(other or {}, check_other) + problems += quick_map_problems(head, quick_map_pages, base) return problems @@ -630,9 +635,19 @@ def _conclude(self, rules: Mapping[str, str], evidence: list[Evidence], check_other=None) -> InitRecord: """The deterministic checks of the change (design §9.3), then the publication. ``other`` are repository paths outside ``knowledge/`` - (before, after) that ``check_other`` must accept.""" - problems = validate_change(self.base, self.head, observer=self.observer, rules=rules, - evidence=evidence, other=other, check_other=check_other) + (before, after) that ``check_other`` must accept. Every owner page of + a knowledge-side routes file first gets init's Direct quick map and + must then yield one (``init_quick_maps``).""" + from .init_quick_maps import owner_pages + + problems = self._refresh_quick_maps() + try: + routed = owner_pages(self.head.get(f"{self.repo_dir}/{ROUTES_NAME}")) + except (ValueError, yaml.YAMLError): + routed = [] # the routes file itself is reported by the validators + problems += validate_change(self.base, self.head, observer=self.observer, rules=rules, + evidence=evidence, other=other, check_other=check_other, + quick_map_pages=routed) changed: dict[str, str] = {KNOWLEDGE_PREFIX + p: t for p, t in self.head.items() if self.base.get(p) != t} for path, (before, after) in (other or {}).items(): if after is not None and after != before: @@ -650,6 +665,31 @@ def _conclude(self, rules: Mapping[str, str], evidence: list[Evidence], return self._blocked(problems) return self._publish(changed) + def _refresh_quick_maps(self) -> list[str]: + """Init's Direct quick map on every owner page of the knowledge-side + routes file in ``head`` (the file it wrote or extends), rendered from + the page as it now stands (``init_quick_maps.refresh_quick_maps``): + the checklist lines are recorded, the blocking problems returned.""" + from .init_quick_maps import refresh_quick_maps + + text = self.head.get(f"{self.repo_dir}/{ROUTES_NAME}") + if text is None: + return [] + self.head, notes, problems = refresh_quick_maps(self.head, self.base, text, briefing_docs=self.briefing_docs) + for note in notes: + if note not in self.record.checklist: + self.record.checklist.append(note) + return problems + + def _map_inputs(self, page: str) -> tuple[list[str], list[str]]: + """What ``page``'s map is rendered with: its owner's signals and + prefixes in the routes file this stage will conclude with (head, + else base), or nothing when no routes name it yet.""" + from .init_quick_maps import map_inputs + + routes = f"{self.repo_dir}/{ROUTES_NAME}" + return map_inputs(self.head.get(routes, self.base.get(routes)), page) + # -- model inputs and calls ----------------------------------------------------- def _doc_payload(self) -> list[dict]: out, used = [], 0 @@ -845,6 +885,9 @@ def _place(self, candidate: _Candidate, tree: dict[str, str]) -> dict[str, str] if candidate.page == previous: self._drop(candidate, f"refused by the knowledge format: {exc}") return None + spilled = self.__dict__.setdefault("_spilled_from", {}).setdefault(previous, []) + if candidate.page not in spilled: + spilled.append(candidate.page) # the pages that took this page's overflow, in order self.record.notes.append(f"{previous} is full: {candidate.rule_id} goes to {candidate.page}") self._drop(candidate, "no page could take it") return None @@ -869,8 +912,29 @@ def _apply(self, candidates: list[_Candidate], files: dict[str, str]) -> dict[st for c in candidates] result = apply_operations(work, ops, release=self.release, today=self.today) work.update(result.files) + self._check_capacity_with_map(work, {c.page for c in candidates}) return work + def _check_capacity_with_map(self, work: Mapping[str, str], pages: set[str]) -> None: + """A rule page must stay under the format's capacity WITH the Direct + quick map init renders on it (one row per rule): ``apply_operations`` + only measures the rules, so the map is counted here, and a page that + would overflow with it is reported as full — the same signal that + moves the next rule to a sibling page (``_place``).""" + from .init_quick_maps import has_hand_written_map, render_quick_map, with_quick_map + + if self.route_source == "manifest": + return # no routes file, no map written: the format's own measure is the whole truth + for page in sorted(pages): + text = work.get(page) + if text is None or has_hand_written_map(text): + continue + signals, prefixes = self._map_inputs(page) # the map as it will really be rendered + with_map = with_quick_map(text, render_quick_map(text, signals=signals, prefixes=prefixes)) + over = page_over_capacity(with_map) + if over: + raise LifecycleError(f"{page}: page full once its Direct quick map is counted ({over})") + def _write_rules(self, kept: list[_Candidate]) -> None: """The kept rules on their placed pages. Dropping failed rules only shrinks pages, so the placement found while screening still fits.""" @@ -1229,6 +1293,13 @@ def _existing_owners(self) -> list[dict]: if prefix not in prefixes: prefixes.append(prefix) name = PurePosixPath(path).parent.name + if prefixes and path in self.briefing_docs: + # a routed page must carry a Direct quick map, and init never + # writes to a briefing doc: it cannot be an owner + self.record.checklist.append( + f"{path} names {', '.join(prefixes[:4])} but is a briefing doc: add a Direct quick map by " + f"hand and route it in {ROUTES_NAME}, or route another page") + continue if prefixes and _slug(name) not in {o["owner"] for o in owners}: owners.append({"owner": _slug(name), "path": path, "signals": [name.replace("-", " ")], "scope_prefixes": prefixes}) @@ -1300,12 +1371,44 @@ def _routes(self, plan: dict) -> None: if not owners: owners.append({"owner": _slug(self.lifecycle.repo), "path": f"{self.repo_dir}/{INDEX_NAME}", "signals": [self.lifecycle.repo], "scope_prefixes": self._root_prefixes()}) + owners = self._owners_with_room(owners) self._note_shadowing(owners) header = (f"# Direct-mode owner routing for {self.lifecycle.repo}, written by kb init " f"(pin {self.record.pin[:12]}).\n") self.head[path] = header + yaml.safe_dump({"schema_version": 1, "owners": owners}, allow_unicode=True, sort_keys=False) + def _owners_with_room(self, owners: list[dict]) -> list[dict]: + """Owners whose page can carry its Direct quick map. A pre-existing + page at the format's capacity is swapped for the page that took the + rules overflowing from it this stage (``_place`` records those; any + other new page beside it — a seed page, say — holds knowledge about + something else); with no such page the owner is dropped and the + checklist says why.""" + from .init_quick_maps import has_hand_written_map, mapped + + out = [] + for owner in owners: + page = str(owner["path"]) + text = self.head.get(page) + if text is None or page not in self.base or has_hand_written_map(text): + out.append(owner) + continue + if mapped(text, signals=owner.get("signals") or [], prefixes=owner.get("scope_prefixes") or []): + out.append(owner) + continue + sibling = next((p for p in self.__dict__.get("_spilled_from", {}).get(page, []) + if p in self.head and p not in self.base and self._is_rule_page(p) + and mapped(self.head[p], signals=[], prefixes=[])), None) + if sibling: + self.record.notes.append(f"owner {owner['owner']}: {page} has no room for a Direct quick map; " + f"routed to {sibling} instead") + out.append({**owner, "path": sibling}) + else: + self.record.checklist.append(f"owner {owner['owner']} dropped: {page} has no room for a Direct " + "quick map (page capacity); split the page by hand and route it") + return out + def _note_shadowing(self, owners: list[dict]) -> None: """A prefix that is an ancestor of another owner's prefix (e.g. a bare package directory next to its components) stays as the least specific diff --git a/test/test_kb_init_quick_maps.py b/test/test_kb_init_quick_maps.py new file mode 100644 index 00000000..2603f2ec --- /dev/null +++ b/test/test_kb_init_quick_maps.py @@ -0,0 +1,344 @@ +"""Every page kb init routes to through a knowledge-side ``_routes.yaml`` +carries a Direct quick map, checked with the production extractor. + +Regression for the first real run (afd-plugin skeleton, PR #265): routed pages +without a ``## … Direct …`` section fail ``test_every_routed_page_yields_a_quick_map``. +Offline, on the skeleton/modules/deepen suites' toy world. +""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from infermatrix_copilot.direct_routing import _direct_quick_map, _direct_quick_map_text +from infermatrix_copilot.kb_service.init_quick_maps import ( + HARD_CAP, MARKER, MAX_CHARS, QUICK_MAP_HEADING, has_hand_written_map, map_inputs, owner_pages, + quick_map_problems, refresh_quick_maps, render_quick_map, with_quick_map, +) +from infermatrix_copilot.kb_service.init_stages import run_stage, validate_change +from infermatrix_copilot.kb_service.init_support import knowledge_changes +from infermatrix_copilot.knowledge_service.facts import FactsError +from infermatrix_copilot.knowledge_service.l1 import check_changeset, check_tree +from infermatrix_copilot.knowledge_service.lifecycle import ANY_RULE_HEADING, RULE_HEADING, Page + +from test_kb_init_deepen import CodeGateway, _chain, _churn +from test_kb_init_modules import CardGateway, _modules_lifecycle, _world_with_tools +from test_kb_init_skeleton import ( # noqa: F401 - the fixture is used by name + FakeGateway, _commit, _lifecycle, _runtime, _tree, world, +) + +RULE_PAGE = ( + '---\ntitle: "Core rules"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\n' + "tags: [toy]\nsources: []\n---\n\n# Core rules\n\nHow the engine is reviewed.\n\n" + "## TOY-1a — Ticks are idempotent\n\n- 触发：a PR edits `pkg/core.py` or `pkg/util.py::helper`.\n" + "- 强制：a second `Engine.step` call returns the same value.\n\n" + "<!-- kb:rule status=active since=2026-09-01 -->\n\n" + "## TOY-1b — Helpers stay pure\n\n- 强制：`pkg/util.py` never imports the engine.\n\n" + "<!-- kb:rule status=active since=2026-09-01 -->\n" +) + + +def _routed_pages(tree: dict[str, str]) -> list[str]: + routes = yaml.safe_load(tree["knowledge/repos/toy/_routes.yaml"]) + return list(dict.fromkeys(str(o["path"]) for o in routes["owners"])) + + +def _production_check(tree: dict[str, str], pages: list[str], tmp_path: Path) -> dict[str, str]: + """What ``test_every_routed_page_yields_a_quick_map`` would see: the file + on disk through ``_direct_quick_map``.""" + out = {} + for page in pages: + target = tmp_path / page + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(tree[f"knowledge/{page}"], encoding="utf-8") + text, status = _direct_quick_map(str(target)) + assert text.startswith(QUICK_MAP_HEADING) or status == "unavailable" + out[page] = status + return out + + +# -- rendering --------------------------------------------------------------------- + +def test_the_heading_is_never_a_rule_heading(): + assert RULE_HEADING.match(QUICK_MAP_HEADING) is None + section = render_quick_map(RULE_PAGE, signals=["engine"], prefixes=["pkg/"]) + page = with_quick_map(RULE_PAGE, section) + parsed = Page.parse(page) + assert [s.rule_id for s in parsed.rules()] == ["TOY-1a", "TOY-1b"] + assert parsed.sections[0].heading == QUICK_MAP_HEADING and not parsed.sections[0].is_rule + # not even the loose ID scanner (which L1's duplicate-ID check uses across + # pages) reads the heading as an ID: "Direct" is not the first word + assert [m.group("rule") for m in ANY_RULE_HEADING.finditer(page)] == ["TOY-1a", "TOY-1b"] + assert not check_tree({"repos/toy/rules.md": page, "repos/toy/rules-2.md": page.replace("TOY-1", "TOY-2"), + "repos/toy/_index.md": "---\ntitle: t\ntype: index\n---\n# t\n\n- [a](rules.md)\n- [b](rules-2.md)\n"}) + + +def test_rules_become_rows_with_their_trigger_and_paths(): + section = render_quick_map(RULE_PAGE, signals=["engine", "tick"], prefixes=["pkg/"]) + lines = section.splitlines() + assert lines[0] == QUICK_MAP_HEADING and lines[1] == MARKER + assert "触发词：engine、tick。" in section + assert "| a PR edits `pkg/core.py` or `pkg/util.py::helper`. | TOY-1a | `pkg/core.py`、`pkg/util.py::helper` |" in lines + assert "| Helpers stay pure | TOY-1b | `pkg/util.py` |" in lines + text, status = _direct_quick_map_text(with_quick_map(RULE_PAGE, section)) + assert status == "ok" and "TOY-1b" in text + + +def test_a_page_without_rules_gets_one_entry_row(): + card = "---\ntitle: c\ntype: architecture\n---\n\n# Card\n\nThe lint worker.\n\n**Key files**\n\n- `tools/lint/y.py`\n" + section = render_quick_map(card, signals=["lint"], prefixes=["tools/lint/"]) + assert "| lint | 入口 | `tools/lint/y.py` |" in section + page = with_quick_map(card, section) + assert page.endswith("\n") and _direct_quick_map_text(page)[1] == "ok" + assert Page.parse(page).sections[0].heading == QUICK_MAP_HEADING + bare = "---\ntitle: c\ntype: architecture\n---\n\n# Card\n\nNothing named here.\n" + assert render_quick_map(bare, signals=[], prefixes=["tools/lint/"]).count("| （无触发词） | 入口 | `tools/lint/` |") == 1 + + +def test_the_map_never_exceeds_the_caps(): + rules = "".join( + f"## TOY-{n}x — Rule number {n} with a long title that repeats itself to fill the row\n\n" + f"- 触发：a PR edits `pkg/module_{n}/handler.py` or `pkg/module_{n}/state.py` and mentions {n}.\n" + f"- 强制：keep {n}.\n\n<!-- kb:rule status=active since=2026-09-01 -->\n\n" for n in range(1, 120)) + page = RULE_PAGE.split("## TOY-1a")[0] + rules + section = render_quick_map(page, signals=[f"signal {i}" for i in range(12)], prefixes=["pkg/"]) + assert len(section) <= MAX_CHARS + 1 and len(section) < HARD_CAP + assert section.rstrip("\n").endswith("| … | … | … |") + assert _direct_quick_map_text(with_quick_map(page, section))[1] == "ok" + assert "| " in section and "\\|" not in section + + +def test_pipes_in_cells_are_escaped(): + page = RULE_PAGE.replace("Helpers stay pure", "Helpers | stay pure") + section = render_quick_map(page, signals=["a|b"], prefixes=[]) + assert "| Helpers \\| stay pure | TOY-1b |" in section and "触发词：a\\|b。" in section + + +def test_regeneration_replaces_only_the_init_map_and_keeps_a_hand_written_one(): + first = with_quick_map(RULE_PAGE, render_quick_map(RULE_PAGE, signals=["engine"], prefixes=["pkg/"])) + grown = first.replace("## TOY-1b", "## TOY-1c — A third rule\n\n- 强制：x.\n\n" + "<!-- kb:rule status=active since=2026-09-01 -->\n\n## TOY-1b") + second = with_quick_map(grown, render_quick_map(grown, signals=["engine"], prefixes=["pkg/"])) + assert second.count(QUICK_MAP_HEADING) == 1 and second.count(MARKER) == 1 + parsed = Page.parse(second) + assert parsed.sections[0].heading == QUICK_MAP_HEADING + assert [s.rule_id for s in parsed.rules()] == ["TOY-1a", "TOY-1c", "TOY-1b"] + assert "| A third rule | TOY-1c |" in second + hand = RULE_PAGE.replace("# Core rules\n", "# Core rules\n\n## Direct code map\n\n| a | b | c |\n|---|---|---|\n| x | y | z |\n") + assert has_hand_written_map(hand) and not has_hand_written_map(first) + assert with_quick_map(hand, render_quick_map(hand, signals=["engine"], prefixes=["pkg/"])) == hand + + +def test_regeneration_keeps_what_the_page_gained_after_the_map(): + """The map is the run of table rows after its separator; an index line, + prose and even a second table appended after it belong to the page.""" + index = ("---\ntitle: \"Tooling\"\ntype: index\ntags: [toy]\nsources: []\n---\n\n# Tooling\n\n" + "- [Card](card.md)\n") + first = with_quick_map(index, render_quick_map(index, signals=["lint"], prefixes=["tools/"])) + gained = first.rstrip("\n") + "\n- [Tooling rules](rules.md)\n\nSee also:\n\n| a | b |\n|---|---|\n| 1 | 2 |\n" + second = with_quick_map(gained, render_quick_map(gained, signals=["lint", "fmt"], prefixes=["tools/"])) + assert second.count(QUICK_MAP_HEADING) == 1 and "触发词：lint、fmt。" in second + assert "- [Tooling rules](rules.md)" in second and "See also:" in second and "| 1 | 2 |" in second + assert second.index("| lint、fmt | 入口 |") < second.index("- [Tooling rules](rules.md)") < second.index("| 1 | 2 |") + lines = second.splitlines() + assert lines.count("|---|---|") == 1 and lines.count("|---|---|---|") == 1 + + +def test_a_map_that_no_longer_fits_blocks_instead_of_going_stale(): + filler = "".join(f"- note {i}: keep the tick log readable for operators.\n" for i in range(470)) + page = RULE_PAGE.replace("How the engine is reviewed.\n", "How the engine is reviewed.\n\n" + filler) + routes = ("schema_version: 1\nowners:\n- {owner: core, path: repos/toy/rules.md, signals: [engine]," + " scope_prefixes: [pkg/]}\n") + head, notes, problems = refresh_quick_maps({"repos/toy/rules.md": page}, {}, routes, briefing_docs=()) + assert not problems and MARKER in head["repos/toy/rules.md"] + grown = head["repos/toy/rules.md"] + "".join( + f"## TOY-{n}z — More\n\n- 强制：{n}.\n\n<!-- kb:rule status=active since=2026-09-01 -->\n\n" for n in range(4)) + head2, _, problems = refresh_quick_maps({"repos/toy/rules.md": grown}, head, routes, briefing_docs=()) + assert problems == ["quick map: repos/toy/rules.md has no room for its Direct quick map (page capacity): " + "split the page by hand, or route another page"] + assert head2["repos/toy/rules.md"] == grown # nothing half-written + assert map_inputs(routes, "repos/toy/rules.md") == (["engine"], ["pkg/"]) + assert map_inputs(routes, "repos/toy/other.md") == ([], []) + + +def test_a_newly_truncated_map_blocks_but_an_old_one_is_tolerated(): + # level-three rules right after the map are part of the served section + rules = "".join(f"### TOY-{n}q — Rule {n}\n\n- 强制：{'x' * 60} {n}.\n\n" + f"<!-- kb:rule status=active since=2026-09-01 -->\n\n" for n in range(60)) + page = RULE_PAGE.split("## TOY-1a")[0] + rules + mapped_page = with_quick_map(page, render_quick_map(page, signals=[], prefixes=["pkg/"])) + assert _direct_quick_map_text(mapped_page)[1] == "truncated" + files = {"repos/toy/rules.md": mapped_page} + assert quick_map_problems(files, ["repos/toy/rules.md"], base={}) == [ + "quick map: repos/toy/rules.md: the Direct section as served exceeds 3500 characters (truncated); keep " + "the map first and short, and start the rules with a `## ` heading so level-three rules are not part of it"] + assert quick_map_problems(files, ["repos/toy/rules.md"], base=files) == [] # unchanged: known debt + assert len(quick_map_problems(files, ["repos/toy/rules.md"])) == 1 # no base given: strict + + +def test_a_regenerated_map_is_a_prose_change_for_l1(): + base = with_quick_map(RULE_PAGE, render_quick_map(RULE_PAGE, signals=["engine"], prefixes=["pkg/"])) + head = with_quick_map(base, render_quick_map(base, signals=["engine", "tick"], prefixes=["pkg/"])) + assert base != head + index = "---\ntitle: t\ntype: index\n---\n# t\n\n- [a](rules.md)\n" + before = {"repos/toy/rules.md": base, "repos/toy/_index.md": index} + after = {"repos/toy/rules.md": head, "repos/toy/_index.md": index} + result = check_changeset(before, after, knowledge_changes(before, after)) + assert not result.issues, result.issues + assert {b.kind for b in result.blocks} == {"prose"} + + +def test_quick_map_problems_name_the_page(): + good = with_quick_map(RULE_PAGE, render_quick_map(RULE_PAGE, signals=[], prefixes=[])) + files = {"repos/toy/rules.md": good, "repos/toy/bare.md": RULE_PAGE, + "repos/toy/empty.md": RULE_PAGE.replace("## TOY-1a", "## Direct map\n\n## TOY-1a")} + problems = quick_map_problems(files, ["repos/toy/rules.md", "repos/toy/bare.md", "repos/toy/empty.md", + "repos/toy/missing.md"]) + assert len(problems) == 3 + assert problems[0].startswith("quick map: repos/toy/bare.md yields no Direct quick map") + assert problems[1].startswith("quick map: repos/toy/empty.md yields no Direct quick map") + assert problems[2] == "quick map: routed page repos/toy/missing.md does not exist" + assert owner_pages("schema_version: 1\nowners:\n- {owner: a, path: p.md, scope_prefixes: [x/]}\n" + "- {owner: b, path: p.md, scope_prefixes: [y/]}\n") == ["p.md"] + assert owner_pages(None) == [] + + +def test_validate_change_blocks_a_routed_page_without_a_map(): + index = "---\ntitle: t\ntype: index\n---\n# t\n\n- [a](rules.md)\n" + files = {"repos/toy/rules.md": RULE_PAGE, "repos/toy/_index.md": index} + + class Unreachable: # the claim checks are not under test: nothing to check + def head(self): + return "a" * 40 + + def top_level(self, sha): + return set() + + problems = validate_change(files, files, observer=Unreachable(), rules={}, evidence=[], + quick_map_pages=["repos/toy/rules.md"]) + assert problems == [p for p in problems if p.startswith("quick map: repos/toy/rules.md")] + assert len(problems) == 1 + + +# -- the stages -------------------------------------------------------------------- + +def test_skeleton_routes_only_pages_that_yield_a_map(world, tmp_path): + record = run_stage(_runtime(world), _lifecycle(), "skeleton", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + pages = _routed_pages(tree) + assert pages == ["repos/toy/rules.md"] + assert set(_production_check(tree, pages, tmp_path).values()) == {"ok"} + page = tree["knowledge/repos/toy/rules.md"] + parsed = Page.parse(page) + assert parsed.sections[0].heading == QUICK_MAP_HEADING and MARKER in parsed.sections[0].text + ids = [s.rule_id for s in parsed.rules()] + assert ids and all(f"| {i} |" in parsed.sections[0].text for i in ids) + + +def test_modules_maps_every_new_group_page(world, tmp_path): + _world_with_tools(world) + gateway = CardGateway() + tree: dict[str, str] = {} # a dry-run tree holds the stage's changes: overlay the chain + for stage in ("skeleton", "modules"): + record = run_stage(_runtime(world, gateway), _modules_lifecycle(), stage, dry_run=True) + assert record.status == "dry_run", (stage, record.problems) + tree.update(_tree(record)) + pages = _routed_pages(tree) + assert len(pages) >= 2 and any(p.endswith("/_index.md") for p in pages) + assert set(_production_check(tree, pages, tmp_path).values()) == {"ok"} + for page in pages: + assert Page.parse(tree[f"knowledge/{page}"]).sections[0].heading == QUICK_MAP_HEADING + + +def test_deepen_regenerates_the_map_with_the_new_rules(world, tmp_path): + _world_with_tools(world) + _churn(world) + gateway = CodeGateway() + tree: dict[str, str] = {} + for stage in ("skeleton", "modules", "deepen"): + record = run_stage(_runtime(world, gateway), _modules_lifecycle(), stage, dry_run=True) + assert record.status == "dry_run", (stage, record.problems) + tree.update(_tree(record)) + pages = _routed_pages(tree) + assert set(_production_check(tree, pages, tmp_path).values()) == {"ok"} + written = set(record.verdicts) + assert written + for page in pages: + text = tree[f"knowledge/{page}"] + assert text.count(QUICK_MAP_HEADING) == 1 and text.count(MARKER) == 1 + parsed = Page.parse(text) + assert parsed.sections[0].heading == QUICK_MAP_HEADING + for rule in parsed.rules(): + assert f"| {rule.rule_id} |" in parsed.sections[0].text + + +def test_a_briefing_doc_is_never_made_an_owner(world): + """An existing repository whose only entry page names code: it cannot be + routed when the adapter reads it into every prompt (init does not write + to it, so it could never carry init's map).""" + _commit(world["origin"], { + "adapters/toy/manifest.yaml": "name: toy\nknowledge:\n repo_subdir: repos/toy\n briefing_docs:\n" + " - repos/toy/components/core/_index.md\n", + "knowledge/repos/toy/_index.md": "---\ntitle: \"Toy\"\ncreated: 2026-09-01\nupdated: 2026-09-01\n" + "type: index\ntags: [toy]\nsources: []\n---\n\n# Toy\n\n" + "- [core](components/_index.md)\n", + "knowledge/repos/toy/components/_index.md": "---\ntitle: \"Components\"\ncreated: 2026-09-01\n" + "updated: 2026-09-01\ntype: index\ntags: [toy]\nsources: []\n" + "---\n\n# Components\n\n- [core](core/_index.md)\n", + "knowledge/repos/toy/components/core/_index.md": "---\ntitle: \"Core\"\ncreated: 2026-09-01\n" + "updated: 2026-09-01\ntype: index\ntags: [toy]\n" + "sources: []\n---\n\n# Core\n\n主源码：`pkg/`\n", + }, "existing toy pages") + record = run_stage(_runtime(world), _lifecycle(seeds=()), "skeleton", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + assert "repos/toy/components/core/_index.md" not in _routed_pages(tree) + assert any("components/core/_index.md" in c and "briefing doc" in c for c in record.checklist) + assert "knowledge/repos/toy/components/core/_index.md" not in tree # untouched: not in the change + + +def test_a_full_owner_page_is_only_rerouted_to_the_page_that_took_its_overflow(world): + """A seed page beside a full rules page is not that page's overflow: + routing the engine's code to it would serve unrelated knowledge.""" + from test_kb_init_skeleton import _full_rules_page + + _commit(world["origin"], _full_rules_page(487), "full rules page") + gateway = FakeGateway(doc_rules=[], seed_rules=True) # nothing spills; a seed page appears + record = run_stage(_runtime(world, gateway), _lifecycle(), "skeleton", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + assert "knowledge/repos/toy/rules-seed-other-rules.md" in tree + routes = yaml.safe_load(tree["knowledge/repos/toy/_routes.yaml"]) + assert [o["path"] for o in routes["owners"]] == [] + assert any(c.startswith("owner core dropped: repos/toy/rules.md has no room") for c in record.checklist) + + +def test_a_manifest_routed_page_is_measured_without_a_map(world): + """No routes file, no map: counting a hypothetical map would spill a rule + that fits to a sibling the manifest never routes to, and Direct would + miss it.""" + from test_kb_init_manifest_routed import ManifestCodeGateway, _manifest_routed_kb, _toy + from test_kb_init_modules import TOOLS + + # a rule page with no rule yet (deepen skips owners that already bear + # rules), 493 non-empty lines: the code rule fits, a map on top would not + filler = "".join(f"- note {i}: keep the tick log readable for operators.\n" for i in range(483)) + near_full = ('---\ntitle: "Core rules"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\n' + "tags: [toy]\nsources: []\n---\n\n# Core rules\n\n" + filler) + _manifest_routed_kb(world, core_rules=near_full) + _commit(world["upstream"], TOOLS, "tools") + _commit(world["upstream"], {"pkg/util.py": "def helper():\n return 2\n\n# tick helper\n"}, "util") + gateway = ManifestCodeGateway(doc_rules=[], seed_rules=False) + for stage in ("skeleton", "modules"): + record = run_stage(_runtime(world, gateway), _toy(seeds=()), stage, dry_run=True) + assert record.status in ("dry_run", "empty"), (stage, record.problems) + record = run_stage(_runtime(world, gateway), _toy(seeds=()), "deepen", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + core = Page.parse(tree["knowledge/repos/toy/components/core/rules.md"]) + assert core.rules() and "knowledge/repos/toy/components/core/rules-2.md" not in tree + assert not any("is full" in n for n in record.notes) + assert QUICK_MAP_HEADING not in tree["knowledge/repos/toy/components/core/rules.md"] diff --git a/test/test_kb_init_skeleton.py b/test/test_kb_init_skeleton.py index 37849ce0..3001a413 100644 --- a/test/test_kb_init_skeleton.py +++ b/test/test_kb_init_skeleton.py @@ -314,24 +314,30 @@ def _judged_texts(gateway: FakeGateway) -> dict[str, str]: def test_rules_that_fill_the_page_together_spill_to_a_sibling(world): - # 493 lines: the first new rule (4 lines with its footer) fits, the second does not + # 493 lines: the page is the routed owner, so its Direct quick map (one row + # per rule) counts toward its capacity — the map for its one rule fits + # (499 lines), the first new rule with its own row would not _commit(world["origin"], _full_rules_page(483), "near-full rules page") gateway = FakeGateway(doc_rules=[DOC_RULE, DOC_RULE2], seed_rules=False) record = run_stage(_runtime(world, gateway), _lifecycle(seeds=()), "skeleton", dry_run=True) assert record.status == "dry_run", record.problems tree = _tree(record) - assert [s.rule_id for s in Page.parse(tree["knowledge/repos/toy/rules.md"]).rules()] == ["TOY-1a", "TOY-I1"] + page = Page.parse(tree["knowledge/repos/toy/rules.md"]) + assert [s.rule_id for s in page.rules()] == ["TOY-1a"] + assert page.sections[0].heading.endswith("（Direct）") and "| TOY-1a |" in page.sections[0].text sibling = Page.parse(tree["knowledge/repos/toy/rules-doc-invariants.md"]) - assert [s.rule_id for s in sibling.rules()] == ["TOY-I2"] + assert [s.rule_id for s in sibling.rules()] == ["TOY-I1", "TOY-I2"] assert "](rules-doc-invariants.md)" in tree["knowledge/repos/toy/_index.md"] # the judge saw both rules where they are written judged = _judged_texts(gateway) assert "TOY-I1" in judged["TOY-I1"] and "TOY-I2" in judged["TOY-I2"] + assert record.verdicts["TOY-I1"]["page"] == "repos/toy/rules-doc-invariants.md" assert record.verdicts["TOY-I2"]["page"] == "repos/toy/rules-doc-invariants.md" def test_a_rule_too_big_for_the_page_on_its_own_goes_to_a_sibling(world): - # 497 lines: not even the first new rule fits on rules.md + # 497 lines: not even the first new rule fits on rules.md, nor its Direct + # quick map — so the owner is routed to the sibling page init writes _commit(world["origin"], _full_rules_page(487), "full rules page") gateway = FakeGateway(doc_rules=[DOC_RULE, DOC_RULE2], seed_rules=False) record = run_stage(_runtime(world, gateway), _lifecycle(seeds=()), "skeleton", dry_run=True) @@ -340,6 +346,10 @@ def test_a_rule_too_big_for_the_page_on_its_own_goes_to_a_sibling(world): assert "knowledge/repos/toy/rules.md" not in tree # the full page is untouched sibling = Page.parse(tree["knowledge/repos/toy/rules-doc-invariants.md"]) assert [s.rule_id for s in sibling.rules()] == ["TOY-I1", "TOY-I2"] + assert sibling.sections[0].heading.endswith("（Direct）") + routes = yaml.safe_load(tree["knowledge/repos/toy/_routes.yaml"]) + assert [o["path"] for o in routes["owners"]] == ["repos/toy/rules-doc-invariants.md"] + assert any("no room for a Direct quick map" in n for n in record.notes) assert all(_judged_texts(gateway)[rid] for rid in ("TOY-I1", "TOY-I2")) assert not record.dropped