diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index c61adb6f..0d95b313 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -1,6 +1,6 @@ # kb_service/ —— 规范 - + `知识服务核心:仓库配置、账本、outbox、CLI · refactor-status: new` @@ -477,3 +477,26 @@ x 必须为正。因阈值停下的调用(`stop_reason="max_budget"`)→ `Mo ## 2026-09-30 新索引每页只登记一次(kb init) `init_stages.unlink_listed(text, listed)`:新建 `_index.md` 时,生成的导语里指向目录清单已列出页面的行内链接改成纯文本标签(支持尖括号目标与标题;锚点、`./` 忽略;图片与其他目标不动)。改写后再用 `l1.link_targets`(CommonMark 解析器)核验:若仍有指向清单页面的链接(改写不认识的写法,如引用式链接),整段可选导语丢弃并记入清单,绝不登记两次。目录清单是每个页面唯一的登记,知识树校验拒绝同一页面登记两次。回归来自 jiuwenswarm 试点。测试:`test_kb_init_skeleton.py`。 + +## 2026-10-01 kb init:由 adapter manifest 路由的仓库(回归:afd-plugin 首次真实运行,PR #265) +Direct 的路由优先级是:知识侧 `repos//_routes.yaml` 存在则用它,否则回退到 adapter manifest 的 +`review_routes`(`{prefix, owner, doc}`)。kb init 不得改变这一优先级下的路由结果,也不得把规则写进 briefing 文档: +- `init_coverage.owners_from_review_routes(manifest)`:把 `review_routes` 按 (owner, doc) 归组成 `Owner`(path = doc, + 前缀按声明顺序;同名 owner 的第二个 doc 以 doc 的 stem 作后缀);`owner_table(routes_text, manifest)` → + `("routes_file" | "manifest" | "none", owners)`,与 Direct 同一优先级:routes 文件有 owner 才算 `routes_file`;文件存在但 `owners: []` 时和 Direct 一样退回 manifest(且该文件保持不动——填入 owner 会覆盖 manifest 路由),只有 manifest 也没有路由时空文件才是 init 填写的知识侧表(损坏的 routes 文件抛错,绝不静默换成 manifest)。 + 每个阶段在 `_inputs` 里算出 `route_source` / `owners`,覆盖率、吸收、deepen 分组一律用它。 +- **manifest 路由的仓库**:任何阶段都不写 `_routes.yaml`(写了会覆盖 manifest,把 `ready` 变成 `scope_fallback`、 + 改变金样路由)。skeleton 把生成器的 owner 提案里 manifest 尚未路由的前缀、modules 阶段本要追加的前缀 / 新 owner, + 一律变成清单项 `review_routes (adapter PR): {prefix, owner, doc}`(`review_route_line`;adapter 是人工把关的, + 由 adapter PR 落地);modules 阶段仍写地图卡与分组页,覆盖率的 after 按"这些建议已合入"计算并在 PR 正文说明来源; + deepen 阶段把规则写到 manifest 所指的 owner 页(`_rule_page_for`:owner 页是规则页则就地追加,manifest 指向尚不存在的 + `rules*.md` 则创建它,否则旁边的 `rules.md` / `rules-code.md`)。文档不变量规则的落点:由 `_paths_named`(证据行 + + 规则正文里反引号命名的路径,`facts.claims_in`)经 `most_specific` 找到最具体覆盖最多路径的 owner(`_owner_for`), + 无 owner 覆盖的留在规则页。三个阶段都在记录里标 `coverage.routes_source`。 +- **briefing 文档只读**:`briefing_docs(manifest)` = `knowledge.briefing_docs` ∪ `briefing_docs_extra` ∪ + `performance_briefing_docs`;`adapters.base` 按硬上限渲染它们、超出部分静默截断(afd-plugin 的 16 条规则把索引挤出了 + briefing)。init 永不向这些页面追加规则或正文(`_writable_page` 改写到同目录的 `rules-init.md` 并记入清单;仓库规则页是 + briefing 文档时 skeleton 的规则页就是 `rules-init.md`);索引登记行不在此限(知识树校验要求新页面登记)。 +- 没有 routes 文件也没有 `review_routes` 的仓库:modules / deepen `blocked`(消息指明两种来源都没有)。 +测试:`test_kb_init_manifest_routed.py`(manifest 路由的 toy:三阶段 dry run 不写 `_routes.yaml`、规则落在 owner 页、 +briefing 文档不动、清单含建议项、覆盖率来源;知识侧路由的 toy 行为不变)。 diff --git a/src/infermatrix_copilot/kb_service/init_coverage.py b/src/infermatrix_copilot/kb_service/init_coverage.py index 5904104f..e64b9edd 100644 --- a/src/infermatrix_copilot/kb_service/init_coverage.py +++ b/src/infermatrix_copilot/kb_service/init_coverage.py @@ -19,6 +19,12 @@ An empty population counts as fully covered (ratio 1.0): there is nothing a route could miss. + +The owner table comes from ``owner_table``: a knowledge-side ``_routes.yaml`` +when the repository has one, else the adapter manifest's ``review_routes`` +(the precedence ``direct_routing`` applies), else nothing. A repository routed +by its manifest is measured exactly like one routed by a table; what differs +is that init never writes its routes (see ``init_stages``). """ from __future__ import annotations @@ -82,9 +88,71 @@ def load_owners(routes_yaml_text: str) -> list[Owner]: return out +def owners_from_review_routes(manifest: Mapping | None) -> list[Owner]: + """The owner table an adapter manifest's ``review_routes`` declares + (``{prefix, owner, doc}`` items, the form ``direct_routing`` falls back to + when a repository has no knowledge-side ``_routes.yaml``): one owner per + (owner, doc) pair in declaration order, its prefixes in declaration order. + An owner name that recurs with a second doc is suffixed with that doc's + stem so names stay unique. Items without a prefix or a doc are skipped.""" + routes = (manifest or {}).get("review_routes") or [] + if not isinstance(routes, list): + return [] + grouped: dict[tuple[str, str], list[str]] = {} + for item in routes: + if not isinstance(item, dict): + continue + prefix = str(item.get("prefix") or "").replace("\\", "/").strip() + doc = str(item.get("doc") or "").strip() + if not prefix or not doc: + continue + owner = str(item.get("owner") or prefix.rstrip("/")).strip() + grouped.setdefault((owner, doc), []) + if prefix not in grouped[(owner, doc)]: + grouped[(owner, doc)].append(prefix) + taken: set[str] = set() + out = [] + for (owner, doc), prefixes in grouped.items(): + name = owner if owner not in taken else f"{owner}/{PurePosixPath(doc).stem}" + n = 1 + while name in taken: # two docs with the same stem (components/a/rules.md, components/b/rules.md) + n += 1 + name = f"{owner}/{PurePosixPath(doc).stem}-{n}" + taken.add(name) + out.append(Owner(name, doc, tuple(prefixes))) + return out + + +def owner_table(routes_text: str | None, manifest: Mapping | None) -> tuple[str, list[Owner]]: + """Where a repository's routes come from and the owners they define, with + the precedence ``direct_routing`` applies: ``("routes_file", ...)`` for a + knowledge-side ``_routes.yaml`` that names owners, else ``("manifest", ...)`` + for adapter ``review_routes``, else ``("none", [])``. Direct falls back to + the manifest when the routes file has NO owners, so a valid but empty file + defers to manifest routes too (and must stay untouched: filling it would + override the manifest); only without manifest routes is an empty file the + knowledge-side table init fills. A malformed routes file raises + (``load_owners``); it is never silently replaced by the manifest.""" + file_owners = load_owners(routes_text) if routes_text is not None else None + if file_owners: + return "routes_file", file_owners + manifest_owners = owners_from_review_routes(manifest) + if manifest_owners: + return "manifest", manifest_owners + return ("routes_file", []) if file_owners is not None else ("none", []) + + +def reaches(path: str, prefix: str) -> bool: + """``prefix`` covers ``path``. A directory prefix ``pkg/`` also covers the + directory itself named without its slash (``facts.claims_in`` strips it + from a rule's backticked ``pkg/``), so a rule about a directory reaches + the owner of that directory.""" + return path.startswith(prefix) or (prefix.endswith("/") and path == prefix[:-1]) + + def routes_file(path: str, owners: Sequence[Owner]) -> list[Owner]: """Every owner whose scope prefixes reach ``path``, in routing order.""" - return [o for o in owners if any(path.startswith(p) for p in o.prefixes)] + return [o for o in owners if any(reaches(path, p) for p in o.prefixes)] def _page_depth(owner: Owner) -> int: @@ -100,7 +168,7 @@ def most_specific(path: str, owners: Sequence[Owner]) -> list[Owner]: best: tuple[int, int] | None = None out: list[Owner] = [] for owner in owners: - length = max((len(p) for p in owner.prefixes if path.startswith(p)), default=-1) + length = max((len(p) for p in owner.prefixes if reaches(path, p)), default=-1) if length < 0: continue key = (length, _page_depth(owner)) diff --git a/src/infermatrix_copilot/kb_service/init_deepen.py b/src/infermatrix_copilot/kb_service/init_deepen.py index ab9d3384..9f7fbe98 100644 --- a/src/infermatrix_copilot/kb_service/init_deepen.py +++ b/src/infermatrix_copilot/kb_service/init_deepen.py @@ -15,7 +15,11 @@ screening — the docs redundancy filter (D5), evidence inside the group's files, pinned claims, placement, the advisory judge — and are appended to the owner's rule page (never editing an existing rule; a full page spills to a -sibling). The pass stops when the rule-bearing PR-weighted coverage reaches +sibling; a briefing doc is never written to). Owners come from the +repository's ``_routes.yaml`` or, when it has none, from the adapter +manifest's ``review_routes`` (``init_coverage.owner_table``): a repository +routed by its manifest is deepened onto the pages the manifest routes to. +The pass stops when the rule-bearing PR-weighted coverage reaches ``init.coverage_target``, the budget runs out, or no hot group is left, and it always says which in the record's notes (and so the PR body): a pass never ends below the target without the reason and the hottest files still @@ -37,9 +41,8 @@ import yaml -from ..knowledge_service.lifecycle import LifecycleError, Page from .init_budget import BudgetExhausted -from .init_coverage import Owner, load_owners, make_include, most_specific, pr_weighted_coverage +from .init_coverage import Owner, make_include, most_specific, pr_weighted_coverage from .init_modules import scan_modules from .init_stages import ( ROUTES_NAME, _Candidate, _fence, _numbered, _one_line, _Stage, _title_of, adapter_missing, @@ -119,13 +122,10 @@ def _build(self, tree: Path) -> InitRecord: init = self.lifecycle.init routes_path = f"{self.repo_dir}/{ROUTES_NAME}" - text = self.base.get(routes_path) - if text is None: - return self._blocked([f"{routes_path} does not exist: run (and merge) the earlier stages first"]) - try: - owners = load_owners(text) - except (ValueError, yaml.YAMLError) as exc: - return self._blocked([f"{routes_path} is not a valid route table: {exc}"]) + if self.route_source == "none": + return self._blocked([f"{routes_path} does not exist and the adapter declares no review_routes: " + "run (and merge) the earlier stages first"]) + owners = self.owners # the routes file, or the adapter manifest's review_routes (same precedence as Direct) flip, problems = self._flip() if problems: return self._blocked(problems) @@ -183,6 +183,7 @@ def _build(self, tree: Path) -> InitRecord: "uncovered_hot": [list(item) for item in self._without_rules(prs, owners, include)[:MAX_HOT_LIST]], "unrouted_hot": [list(item) for item in current.uncovered_hot[:MAX_HOT_LIST]], "target": init.coverage_target, + "routes_source": self.route_source, } from ..knowledge_service.pinned_claims import Evidence @@ -234,15 +235,6 @@ def _without_rules(self, prs, owners: list[Owner], include) -> list[tuple[str, i return sorted(missed.items(), key=lambda item: (-item[1], item[0])) # -- rule pages ------------------------------------------------------------------ - def _active_rules(self, path: str) -> int: - text = self.head.get(path) - if text is None or not path.endswith(".md"): - return 0 - try: - return sum(1 for s in Page.parse(text).rules() if s.footer.status == "active") - except (LifecycleError, yaml.YAMLError): - return 0 - def _rule_pages(self, owners: list[Owner]) -> set[str]: """Owner paths that bear rules: the page itself holds an active rule, or (an entry or prose page) a page in the SAME directory does — never a @@ -253,6 +245,10 @@ def _rule_pages(self, owners: list[Owner]) -> set[str]: if self._active_rules(owner.path): bearing.add(owner.path) continue + redirected = self._redirect_of(owner.path) + if redirected != owner.path and self._active_rules(redirected): + bearing.add(owner.path) # a briefing doc's rules live beside it + continue if owner.path in self.head and self._is_rule_page(owner.path): continue # a rule page owns its rules itself: an empty one bears none directory = PurePosixPath(owner.path).parent @@ -260,26 +256,6 @@ def _rule_pages(self, owners: list[Owner]) -> set[str]: bearing.add(owner.path) return bearing - def _is_rule_page(self, path: str) -> bool: - try: - page = Page.parse(self.head[path]) - return page.frontmatter_data().get("type") == "rule" or bool(page.rules()) - except (KeyError, LifecycleError, yaml.YAMLError): - return False - - def _rule_page_for(self, owner: Owner) -> str: - """Where an owner's new rules go: its own page when that is a rule - page, else ``rules.md`` beside it (``rules-code.md`` when a prose - ``rules.md`` is in the way).""" - if owner.path in self.head and self._is_rule_page(owner.path): - return owner.path - directory = str(PurePosixPath(owner.path).parent) - for name in ("rules.md", "rules-code.md"): - path = f"{directory}/{name}" - if path not in self.head or self._is_rule_page(path): - return path - raise InitError(f"no rule page can be placed beside {owner.path}") - # -- one module -------------------------------------------------------------------- def _code_payload(self, tree: Path, files: list[str]) -> list[dict]: out, used = [], 0 diff --git a/src/infermatrix_copilot/kb_service/init_modules.py b/src/infermatrix_copilot/kb_service/init_modules.py index ca1b212d..c388ccfe 100644 --- a/src/infermatrix_copilot/kb_service/init_modules.py +++ b/src/infermatrix_copilot/kb_service/init_modules.py @@ -36,7 +36,7 @@ from .init_budget import BudgetExhausted from .init_coverage import ROOT_MODULE, Owner, load_owners, module_coverage, most_specific, routes_file from .init_stages import ( - _SLUG, ROUTES_NAME, _fence, _one_line, _page_frontmatter, _slug, _Stage, _title_of, + _SLUG, ROUTES_NAME, _fence, _one_line, _page_frontmatter, _slug, _Stage, _title_of, review_route_line, ) from .init_support import InitError, InitRecord, generate @@ -249,13 +249,23 @@ class _Modules(_Stage): def _build(self, tree: Path) -> InitRecord: routes_path = f"{self.repo_dir}/{ROUTES_NAME}" text = self.base.get(routes_path) - if text is None: - return self._blocked([f"{routes_path} does not exist: run (and merge) the skeleton stage first"]) - try: - doc = yaml.safe_load(text) or {} - load_owners(text) - except (ValueError, yaml.YAMLError) as exc: - return self._blocked([f"{routes_path} is not a valid route table: {exc}"]) + if self.route_source == "routes_file": + try: + doc = yaml.safe_load(text) or {} + load_owners(text) + except (ValueError, yaml.YAMLError) as exc: + return self._blocked([f"{routes_path} is not a valid route table: {exc}"]) + elif self.route_source == "manifest": + # the adapter manifest routes this repository: the same table, kept + # in memory only — what this stage would append becomes suggested + # review_routes entries, never a routes file (it would take + # precedence over the manifest and change every route) + doc = {"schema_version": 1, "owners": [ + {"owner": o.owner, "path": o.path, "signals": [], "scope_prefixes": list(o.prefixes)} + for o in self.owners]} + else: + return self._blocked([f"{routes_path} does not exist and the adapter declares no review_routes: " + "run (and merge) the skeleton stage first"]) self.routes_path, self.routes_before = routes_path, doc self.routes = copy.deepcopy(doc) if not isinstance(self.routes.get("owners"), list): @@ -289,6 +299,7 @@ def _build(self, tree: Path) -> InitRecord: "modules": {"before": round(before.ratio, 4), "after": round(after.ratio, 4), "total": len(modules)}, "unrouted": after.uncovered, "cards": {c.module: c.page for c in cards}, + "routes_source": self.route_source, } problems = routes_append_only(self.routes_before, self.routes) if problems: @@ -523,8 +534,28 @@ def _append_links(self, index: str, links: list[tuple[str, str]]) -> None: text += f"- [{title}]({link})\n" self.head[index] = text + def _suggest_routes(self) -> None: + """Every owner or prefix this stage would have appended to a routes + file, as the exact ``review_routes`` entries the adapter manifest + lacks (adapters are human-gated: an adapter PR, never a route file).""" + before = {str(o["owner"]): o for o in self.routes_before.get("owners") or []} + count = 0 + for owner in self.routes.get("owners") or []: + known = before.get(str(owner["owner"])) + had = list((known or {}).get("scope_prefixes") or []) + for prefix in owner.get("scope_prefixes") or []: + if prefix in had: + continue + self.record.checklist.append(review_route_line(prefix, str(owner["owner"]), str(owner["path"]))) + count += 1 + self.record.notes.append("routes come from the adapter manifest (review_routes): kb init writes no " + f"{ROUTES_NAME}; {count} review_routes entr{'y is' if count == 1 else 'ies are'} " + "suggested on the checklist") + def _conclude(self, rules, evidence, other=None, check_other=None) -> InitRecord: - if self.routes != self.routes_before: + if self.route_source == "manifest": + self._suggest_routes() + elif self.routes != self.routes_before: header = [] for line in self.base[self.routes_path].splitlines(keepends=True): if not line.startswith("#"): diff --git a/src/infermatrix_copilot/kb_service/init_stages.py b/src/infermatrix_copilot/kb_service/init_stages.py index db3e9e2d..0d87102c 100644 --- a/src/infermatrix_copilot/kb_service/init_stages.py +++ b/src/infermatrix_copilot/kb_service/init_stages.py @@ -41,6 +41,7 @@ ) from ..knowledge_service.pinned_claims import Evidence, check_rules, evidence_for from .init_budget import Budget, BudgetExhausted, PriceError +from .init_coverage import Owner, most_specific, owner_table, routes_file from .init_support import ( AUTHOR_ENV, KNOWLEDGE_PREFIX, STAGES, InitError, InitPublisher, InitRecord, InitRuntime, claim_problems, classify_verdict, collect_docs, generate, inputs_digest, judge, knowledge_changes, load_prepared, other_path_problems, @@ -49,6 +50,7 @@ from .models import ModelUnavailable ROUTES_NAME = "_routes.yaml" +RULES_INIT_NAME = "rules-init.md" # init's own rule page when the repository's rules page is off limits REPOS_INDEX = "repos/_index.md" MAX_OWNERS = 12 MAX_RULES_PER_CALL = 12 @@ -257,6 +259,31 @@ def _relative_link(from_dir: str, target: str) -> str: return os.path.relpath(target, from_dir).replace(os.sep, "/") +def briefing_docs(manifest: Mapping | None) -> set[str]: + """The knowledge pages an adapter injects into every run's prompt + (``knowledge.briefing_docs``, ``briefing_docs_extra`` and the deprecated + ``performance_briefing_docs``). ``adapters.base`` renders them under a hard + character cap that silently truncates: a rule appended there pushes the + page after it out of the briefing (afd-plugin, PR #265), so init never + writes rule content to them.""" + kn = (manifest or {}).get("knowledge") if isinstance(manifest, dict) else None + out: set[str] = set() + for key in ("briefing_docs", "briefing_docs_extra", "performance_briefing_docs"): + for item in (kn or {}).get(key) or []: + text = str(item or "").strip().lstrip("/") + if text: + out.add(text) + return out + + +def review_route_line(prefix: str, owner: str, doc: str) -> str: + """One ``review_routes`` item, in the form the adapter manifest takes + (a flow-style YAML mapping), for the checklist of a repository whose + routes live in its manifest: adapters are human-gated, so init suggests + the exact entry instead of writing a route.""" + return f"review_routes (adapter PR): {{prefix: {prefix}, owner: {owner}, doc: {doc}}}" + + _INLINE_LINK = re.compile(r"(?[^\[\]\n]*)\]\((?P<[^<>\n]*>|[^()\s]+)" r"(?:\s+(?:\"[^\"\n]*\"|'[^'\n]*'))?\)") @@ -326,6 +353,10 @@ class _Stage: _titles: dict[str, str] = field(default_factory=dict) _judge_stopped: bool = False + # the adapter's briefing docs, read from the manifest by run(); a stage built + # bare (tests) has none + briefing_docs: frozenset[str] = frozenset() + def run(self) -> InitRecord: rt, lc, stage = self.rt, self.lifecycle, self.STAGE init = lc.init @@ -384,6 +415,8 @@ def run(self) -> InitRecord: with tempfile.TemporaryDirectory(prefix="kb-init-") as scratch: tree = upstream.export(pin, Path(scratch) / "tree") self._inputs(tree) + if self.route_problem: + return self._blocked([self.route_problem]) return self._build(tree) except (ModelUnavailable, PriceError, FactsError) as exc: return self._blocked([f"{type(exc).__name__}: {exc}"]) @@ -464,6 +497,18 @@ def _inputs(self, tree: Path) -> None: self.existing = {p: t for p, t in self.base.items() if p.startswith(self.repo_dir + "/")} self.new_repo = not self.existing self.head: dict[str, str] = dict(self.base) + self.manifest = self._manifest() + self.briefing_docs = briefing_docs(self.manifest) + self.route_problem = "" + routes_path = f"{self.repo_dir}/{ROUTES_NAME}" + try: + self.route_source, self.owners = owner_table(self.base.get(routes_path), self.manifest) + except (ValueError, yaml.YAMLError) as exc: + # a malformed routes file is never replaced by the manifest, and no + # stage works on it: run() blocks before _build with this message + self.route_source, self.owners = "invalid", [] + self.route_problem = f"{routes_path} is not a valid route table: {exc}" + self._top_level: set[str] | None = None def _build(self, tree: Path) -> InitRecord: # pragma: no cover - every stage overrides it raise NotImplementedError @@ -482,15 +527,104 @@ def _manifest_text(self) -> str | None: return self.overlay[path] return self.rt.knowledge.show(self._base_sha, path) - def _language(self) -> str: - """The adapter's ``repo.language`` ("" when it declares none).""" + def _manifest(self) -> dict: + """The adapter manifest as a mapping ({} when absent or malformed).""" try: manifest = yaml.safe_load(self._manifest_text() or "") or {} except yaml.YAMLError: - return "" - repo = manifest.get("repo") if isinstance(manifest, dict) else None + return {} + return manifest if isinstance(manifest, dict) else {} + + def _language(self) -> str: + """The adapter's ``repo.language`` ("" when it declares none).""" + repo = self._manifest().get("repo") return str((repo or {}).get("language") or "") if isinstance(repo, dict) else "" + # -- rule pages ------------------------------------------------------------------ + def _active_rules(self, path: str) -> int: + text = self.head.get(path) + if text is None or not path.endswith(".md"): + return 0 + try: + return sum(1 for s in Page.parse(text).rules() if s.footer.status == "active") + except (LifecycleError, yaml.YAMLError): + return 0 + + def _is_rule_page(self, path: str) -> bool: + try: + page = Page.parse(self.head[path]) + return page.frontmatter_data().get("type") == "rule" or bool(page.rules()) + except (KeyError, LifecycleError, yaml.YAMLError): + return False + + def _rule_page_for(self, owner: Owner) -> str: + """Where an owner's new rules go: its own page when that is a rule + page (or a rule page the manifest names that does not exist yet), else + ``rules.md`` beside it (``rules-code.md`` when a prose ``rules.md`` is + in the way).""" + if owner.path in self.head and self._is_rule_page(owner.path): + return owner.path + if owner.path not in self.head and PurePosixPath(owner.path).name.startswith("rules"): + return owner.path # the manifest routes to a rule page init creates + directory = str(PurePosixPath(owner.path).parent) + for name in ("rules.md", "rules-code.md"): + path = f"{directory}/{name}" + if path not in self.head or self._is_rule_page(path): + return path + raise InitError(f"no rule page can be placed beside {owner.path}") + + def _redirect_of(self, page: str) -> str: + """Where init's rules for ``page`` live: the page itself, or, when it + is one of the adapter's briefing docs (never appended to), init's own + rule page beside it. Pure: coverage asks this too.""" + if page not in self.briefing_docs: + return page + return str(PurePosixPath(page).with_name(RULES_INIT_NAME)) + + def _writable_page(self, page: str) -> str: + """``_redirect_of(page)``, noting a redirect on the checklist once.""" + other = self._redirect_of(page) + if other != page: + note = (f"{page} is a briefing doc (knowledge.briefing_docs, rendered under a hard cap): " + f"init writes its rules to {other} instead") + if note not in self.record.checklist: + self.record.checklist.append(note) + return other + + def _target_page(self, candidate: "_Candidate", entries: list[Evidence]) -> str: + """The page a screened candidate is written to (stages refine it).""" + return self._writable_page(candidate.page) + + def _top_level_names(self) -> set[str]: + if self._top_level is None: + self._top_level = set(self.observer.top_level(self.record.pin)) + return self._top_level + + def _paths_named(self, candidate: "_Candidate", entries: list[Evidence]) -> list[str]: + """The repository paths a rule is about: its evidence lines and every + path or ``path::Symbol`` it names in backticks (``facts.claims_in``).""" + from ..knowledge_service.facts import claims_in + + paths = [e.path for e in entries] + for claim in claims_in(candidate.section, self._top_level_names(), active=True): + if claim.kind in ("path", "symbol") and claim.path and claim.path not in paths: + paths.append(claim.path) + return paths + + def _owner_for(self, paths: list[str]) -> Owner | None: + """The owner that most specifically reaches the most of ``paths`` + (routing order breaks ties); None when no owner reaches any.""" + counts: dict[str, int] = {} + by_name = {o.owner: o for o in self.owners} + for path in paths: + for owner in most_specific(path, self.owners): + counts[owner.owner] = counts.get(owner.owner, 0) + 1 + if not counts: + return None + order = [o.owner for o in self.owners] + best = max(counts, key=lambda name: (counts[name], -order.index(name))) + return by_name[best] + def _conclude(self, rules: Mapping[str, str], evidence: list[Evidence], other: Mapping[str, tuple[str | None, str | None]] | None = None, check_other=None) -> InitRecord: @@ -636,6 +770,7 @@ def _screen(self, candidates: list[_Candidate]) -> list[_Candidate]: if problems: self._drop(candidate, "; ".join(problems)) continue + candidate.page = self._target_page(candidate, entries) placed = self._place(candidate, running) if placed is None: continue @@ -839,10 +974,24 @@ def _default_page_title(self, page: str) -> str: return _one_line(getattr(self, "plan", {}).get("rules_title")) or f"{self.lifecycle.repo} rules" def _rules_page(self) -> tuple[str, bool]: - """(page for doc-invariant rules, whether it already exists).""" - path = f"{self.repo_dir}/rules.md" + """(page for doc-invariant rules, whether it already exists): the + repository's rule page, or init's own ``rules-init.md`` when that page + is one of the adapter's briefing docs (never appended to).""" + path = self._writable_page(f"{self.repo_dir}/rules.md") return path, path in self.base + def _target_page(self, candidate: _Candidate, entries: list[Evidence]) -> str: + """On a repository routed by its adapter manifest, a doc-invariant rule + goes to the page of the owner that most specifically reaches the code + it is about (its evidence and the paths it names), so Direct serves it + where the manifest routes; a rule no owner reaches keeps the rules + page. Briefing docs are never written to.""" + if self.route_source == "manifest" and candidate.origin == "docs": + owner = self._owner_for(self._paths_named(candidate, entries)) + if owner is not None: + return self._writable_page(self._rule_page_for(owner)) + return super()._target_page(candidate, entries) + def _pages_offered(self, rules_page: str) -> list[str]: pages = sorted(p for p in self.existing if p.endswith(".md")) for extra in (f"{self.repo_dir}/{INDEX_NAME}", f"{self.repo_dir}/architecture.md", rules_page): @@ -1085,16 +1234,10 @@ def _existing_owners(self) -> list[dict]: "scope_prefixes": prefixes}) return owners[:MAX_OWNERS] - def _routes(self, plan: dict) -> None: - path = f"{self.repo_dir}/{ROUTES_NAME}" - if path in self.base: - self.record.checklist.append(f"{path} exists: the skeleton stage leaves routes to the modules stage") - return - owners = self._existing_owners() - seen = {o["owner"] for o in owners} - owned_pages = {o["path"] for o in owners} - if owners: - self.record.notes.append("owners from existing pages: " + ", ".join(sorted(seen))) + def _plan_owners(self, plan: dict, seen: set[str], owned_pages: set[str]) -> list[dict]: + """The generator's owner proposals that hold up: a unique slug, a page + of this repository, prefixes that exist at the pin.""" + owners = [] for owner in list(plan.get("owners") or [])[:MAX_OWNERS]: if not isinstance(owner, dict): continue @@ -1122,6 +1265,38 @@ def _routes(self, plan: dict) -> None: signals = [_one_line(s, 60) for s in owner.get("signals") or [] if _one_line(s, 60)][:20] seen.add(slug) owners.append({"owner": slug, "path": page, "signals": signals, "scope_prefixes": prefixes}) + return owners + + def _suggest_routes(self, plan: dict) -> None: + """A repository routed by its adapter manifest: kb init writes no + ``_routes.yaml`` (it would take precedence over the manifest and change + every route); the generator's proposals become the exact + ``review_routes`` entries the manifest lacks, on the checklist.""" + self.record.notes.append("routes come from the adapter manifest (review_routes): kb init writes no " + f"{ROUTES_NAME}; entries it lacks are suggested on the checklist") + # a proposal for a page the manifest already routes is welcome here: only + # its prefixes the manifest lacks are suggested (none are written) + for owner in self._plan_owners(plan, set(), set()): + for prefix in owner["scope_prefixes"]: + if not routes_file(prefix, self.owners): + self.record.checklist.append(review_route_line(prefix, owner["owner"], owner["path"])) + self._note_shadowing([{"owner": o.owner, "path": o.path, "scope_prefixes": list(o.prefixes)} + for o in self.owners]) + + def _routes(self, plan: dict) -> None: + path = f"{self.repo_dir}/{ROUTES_NAME}" + if path in self.base: + self.record.checklist.append(f"{path} exists: the skeleton stage leaves routes to the modules stage") + return + if self.route_source == "manifest": + self._suggest_routes(plan) + return + owners = self._existing_owners() + seen = {o["owner"] for o in owners} + owned_pages = {o["path"] for o in owners} + if owners: + self.record.notes.append("owners from existing pages: " + ", ".join(sorted(seen))) + owners += self._plan_owners(plan, seen, owned_pages) 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()}) @@ -1177,18 +1352,24 @@ def _checklist(self) -> None: redundant += 1 self.record.checklist.append(f"existing rule {rule_id} restates the docs: {line.strip()[:120]}") routes = self.head.get(f"{self.repo_dir}/{ROUTES_NAME}") + owned: set[str] = set() + hint = "" if routes: try: - owned = {o.get("path") for o in (yaml.safe_load(routes) or {}).get("owners") or []} + owned = {str(o.get("path")) for o in (yaml.safe_load(routes) or {}).get("owners") or []} except yaml.YAMLError: owned = set() + elif self.route_source == "manifest": + owned = {o.path for o in self.owners} + hint = ": add a review_routes entry pointing at it (adapter PR)" + if routes or self.route_source == "manifest": owned_dirs = {str(PurePosixPath(p).parent) for p in owned if p and p.endswith("/" + INDEX_NAME)} for path in sorted(self.head): if not path.startswith(self.repo_dir + "/") or not path.endswith(".md") \ or PurePosixPath(path).name == INDEX_NAME or path in owned: continue if not any(path.startswith(d + "/") for d in owned_dirs): - self.record.checklist.append(f"no route reaches {path}") + self.record.checklist.append(f"no route reaches {path}{hint}") _TOKEN = re.compile(r"[a-z][a-z0-9_]{3,}") @@ -1251,7 +1432,12 @@ def _ratio(value) -> str: def _coverage_lines(coverage: dict) -> list[str]: """The coverage report of a stage (design §7) as PR-body lines.""" - lines = ["Coverage:", "", "| metric | before | after |", "|---|---|---|"] + lines = ["Coverage:", ""] + if coverage.get("routes_source") == "manifest": + lines += ["Routes come from the adapter manifest (`review_routes`); kb init writes no `_routes.yaml` " + "and suggests the entries it lacks on the checklist. \"after\" counts those suggestions as " + "if the adapter PR were merged.", ""] + lines += ["| metric | before | after |", "|---|---|---|"] for key, label in (("modules", "modules routed"), ("pr_routed", "PR-weighted, routed"), ("pr_rule_bearing", "PR-weighted, rule-bearing")): item = coverage.get(key) diff --git a/test/test_kb_init_manifest_routed.py b/test/test_kb_init_manifest_routed.py new file mode 100644 index 00000000..7a08ae9b --- /dev/null +++ b/test/test_kb_init_manifest_routed.py @@ -0,0 +1,338 @@ +"""kb init on a repository routed by its adapter manifest (``review_routes``). + +Regression for the first real run (afd-plugin skeleton, PR #265): a +knowledge-side ``_routes.yaml`` takes precedence over the manifest and changed +every route, and rules appended to a briefing doc pushed the index out of the +briefing cap. Offline, on the skeleton/modules/deepen suites' toy world. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import yaml + +from infermatrix_copilot.kb_service.init_coverage import ( + Owner, most_specific, owner_table, owners_from_review_routes, routes_file, +) +from infermatrix_copilot.kb_service.init_deepen import SYSTEM_CODE_RULES +from infermatrix_copilot.kb_service.init_modules import SYSTEM_CARD +from infermatrix_copilot.kb_service.init_stages import SYSTEM_MAP, review_route_line, run_stage +from infermatrix_copilot.kb_service.models import ModelReply +from infermatrix_copilot.knowledge_service.lifecycle import Page + +from test_kb_init_deepen import PKG_RULE, CodeGateway +from test_kb_init_modules import TOOLS +from test_kb_init_skeleton import ( # noqa: F401 - the fixture is used by name + DOC_RULE, DOC_RULE2, FakeGateway, _commit, _index, _lifecycle, _runtime, _tree, world, +) + +MANIFEST = """name: toy +repo: + full_name: o/toy + language: python +knowledge: + repo_subdir: repos/toy + briefing_docs: + - repos/toy/rules.md + - repos/toy/_index.md +review_routes: +- prefix: pkg/ + owner: core + doc: repos/toy/components/core/rules.md +knowledge_lifecycle: + enabled: false + mode: shadow +""" + +DOCS_ONLY_RULE = {"title": "Guides ship in both languages", + "body": "- 强制:every page under `docs/` has a translated twin before review.", + "evidence": [{"path": "docs/guide.md", "start": 1, "end": 2}]} + +CORE_RULES = ('---\ntitle: "Core rules"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\n' + "tags: [toy]\nsources: []\n---\n\n# Core rules\n\n" + "## Direct 代码快速入口\n\n| PR 描述在做什么 | 精确规则 | 第一批 live 源码 |\n|---|---|---|\n" + "| engine tick | (none yet) | `pkg/core.py` |\n") + + +def _manifest_routed_kb(world, *, core_rules: str = CORE_RULES) -> None: + """The toy knowledge base as afd-plugin's is shaped: a top-level rule + page that is a briefing doc, component pages the manifest routes to, and + no ``_routes.yaml``.""" + _commit(world["origin"], { + "adapters/toy/manifest.yaml": MANIFEST, + "knowledge/repos/toy/_index.md": _index("Toy", "toy", "- [Toy rules](rules.md)\n- [Components](components/_index.md)\n"), + "knowledge/repos/toy/rules.md": ( + '---\ntitle: "Toy rules"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\n' + "tags: [toy]\nsources: []\n---\n\n# Toy rules\n\n## TOY-1a — Ticks are counted\n\n" + "- 强制:every worker counts its ticks.\n"), + "knowledge/repos/toy/components/_index.md": _index("Components", "toy", "- [Core](core/_index.md)\n"), + "knowledge/repos/toy/components/core/_index.md": _index("Core", "toy", "The engine.\n\n- [Core rules](rules.md)\n"), + "knowledge/repos/toy/components/core/rules.md": core_rules, + }, "toy is routed by its manifest") + + +class ManifestGateway(FakeGateway): + """The map proposes the manifest's owner with one prefix it lacks.""" + + def call_json(self, role, *, system, prompt, validate=None, max_budget_usd=None): + reply = super().call_json(role, system=system, prompt=prompt, validate=validate, + max_budget_usd=max_budget_usd) + if system == SYSTEM_MAP: + reply.data["owners"] = [{"owner": "core", "title": "Core", "page": "repos/toy/components/core/rules.md", + "signals": ["engine"], "scope_prefixes": ["pkg/", "tests/"]}] + return reply + + +def _toy(**init): + return _lifecycle(**{"source_roots": ("pkg/", "tools/"), "min_module_loc": 0, **init}) + + +# -- the owner table ------------------------------------------------------------------ + +def test_owner_table_prefers_a_routes_file_then_the_manifest(): + manifest = yaml.safe_load(MANIFEST) + manifest["review_routes"] += [{"prefix": "pkg/util.py", "owner": "core", "doc": "repos/toy/components/core/rules.md"}, + {"prefix": "tools/", "owner": "core", "doc": "repos/toy/components/tools.md"}, + {"prefix": "nodoc/", "owner": "x"}] + owners = owners_from_review_routes(manifest) + assert owners == [Owner("core", "repos/toy/components/core/rules.md", ("pkg/", "pkg/util.py")), + Owner("core/tools", "repos/toy/components/tools.md", ("tools/",))] + assert owner_table(None, manifest) == ("manifest", owners) + # Direct falls back to the manifest when the routes file names no owners + # (direct_routing: ``routes_table is None or not routes_table["owners"]``), + # so an empty file defers too; it is the knowledge-side table only when + # the manifest routes nothing either + assert owner_table("schema_version: 1\nowners: []\n", manifest) == ("manifest", owners_from_review_routes(manifest)) + assert owner_table("schema_version: 1\nowners: []\n", {"review_routes": []}) == ("routes_file", []) + assert owner_table("schema_version: 1\nowners: []\n", None) == ("routes_file", []) + assert owner_table(None, {"name": "toy"}) == ("none", []) + # one owner name, three docs whose stems collide: three distinct owners, never a merged one + three = {"review_routes": [{"prefix": f"pkg/{n}.py", "owner": "core", "doc": f"repos/toy/components/{n}/rules.md"} + for n in ("a", "b", "c")]} + assert [o.owner for o in owners_from_review_routes(three)] == ["core", "core/rules", "core/rules-2"] + assert [o.path for o in owners_from_review_routes(three)] == [f"repos/toy/components/{n}/rules.md" for n in "abc"] + import pytest + + with pytest.raises(ValueError): + owner_table("schema_version: 2\nowners: []\n", manifest) # never silently replaced by the manifest + + +def test_a_malformed_routes_file_blocks_at_the_shared_input_boundary(world): + """Every stage reads its owner table in ``_Stage._inputs``; a routes file + this copilot cannot read blocks there, before any model call, instead of + escaping as a ValueError (the skeleton is the stage with no earlier chain + to satisfy; the later stages share the same path).""" + _commit(world["origin"], {"knowledge/repos/toy/_routes.yaml": "schema_version: 2\nowners: []\n"}, + "a routes file this copilot does not read") + gateway = FakeGateway() + record = run_stage(_runtime(world, gateway), _lifecycle(), "skeleton", dry_run=True) + assert record.status == "blocked" + assert record.problems == ["repos/toy/_routes.yaml is not a valid route table: unsupported _routes.yaml " + "schema_version 2"] + assert gateway.calls == [] + assert review_route_line("tests/", "core", "repos/toy/components/core/rules.md") == \ + "review_routes (adapter PR): {prefix: tests/, owner: core, doc: repos/toy/components/core/rules.md}" + + +# -- skeleton ---------------------------------------------------------------------------- + +def test_skeleton_writes_no_routes_file_and_places_rules_on_the_manifest_owner(world): + _manifest_routed_kb(world) + gateway = ManifestGateway(doc_rules=[DOC_RULE, DOC_RULE2, DOCS_ONLY_RULE]) + record = run_stage(_runtime(world, gateway), _toy(), "skeleton", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + assert "knowledge/repos/toy/_routes.yaml" not in tree + # the briefing doc is untouched (a dry-run snapshot holds changed files only); + # init's rules went to the owner page the manifest routes to + assert "knowledge/repos/toy/rules.md" not in tree + core = Page.parse(tree["knowledge/repos/toy/components/core/rules.md"]) + placed = {s.rule_id: s.text for s in core.rules()} + assert "pkg/util.py" in placed["TOY-I1"] and "pkg/core.py" in placed["TOY-I2"] + assert "## Direct 代码快速入口" in tree["knowledge/repos/toy/components/core/rules.md"] + # a rule about nothing the manifest routes keeps init's own page, never the briefing doc + own = Page.parse(tree["knowledge/repos/toy/rules-init.md"]) + assert [s.rule_id for s in own.rules()] == ["TOY-I3"] and "docs/" in own.rules()[0].text + assert "](rules-init.md)" in tree["knowledge/repos/toy/_index.md"] + assert any("repos/toy/rules.md is a briefing doc" in c for c in record.checklist) + # the generator's proposal became the exact review_routes entry the manifest lacks + assert review_route_line("tests/", "core", "repos/toy/components/core/rules.md") in record.checklist + assert not any("{prefix: pkg/" in c for c in record.checklist) + assert any(c.startswith("no route reaches repos/toy/rules-seed-other-rules.md: add a review_routes entry") + for c in record.checklist) + assert any(n.startswith("routes come from the adapter manifest") for n in record.notes) + body = (Path(record.pr["dry_run_dir"]) / "PR_BODY.md").read_text(encoding="utf-8") + assert "review_routes (adapter PR): {prefix: tests/" in body + + +def test_a_knowledge_routed_repository_is_unchanged(world): + """The toy of the other suites: no manifest routes, so the skeleton still + writes the routes file and its rules page.""" + record = run_stage(_runtime(world, FakeGateway()), _lifecycle(), "skeleton", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + assert "knowledge/repos/toy/_routes.yaml" in tree and "knowledge/repos/toy/rules.md" in tree + assert not any("review_routes (adapter PR)" in c for c in record.checklist) + + +# -- modules --------------------------------------------------------------------------------- + +class ManifestCardGateway(ManifestGateway): + def call_json(self, role, *, system, prompt, validate=None, max_budget_usd=None): + if system != SYSTEM_CARD: + return super().call_json(role, system=system, prompt=prompt, validate=validate, + max_budget_usd=max_budget_usd) + self.calls.append({"role": role.name, "system": system, "prompt": prompt, "max_budget_usd": max_budget_usd}) + payload = json.loads(prompt.split("\n", 1)[1].rsplit("", 1)[0]) + files = [f["path"] for f in payload["files"]] + data = {"title": f"Module {payload['module']}", "purpose": "Checks worker configs before a tick runs.", + "entry_points": [{"path": files[0], "what": "the entry"}], "key_files": [], "docs": [], + "signals": ["lint"], "group": "tooling", "group_title": "Tooling", "headings": {}} + if validate is not None: + validate(data) + return ModelReply(role, data, json.dumps(data), role.model, {}, 0.1, cost_usd=0.01) + + +def test_modules_suggests_review_routes_instead_of_writing_them(world): + _manifest_routed_kb(world) + _commit(world["upstream"], TOOLS, "tools") + gateway = ManifestCardGateway(doc_rules=[]) + skeleton = run_stage(_runtime(world, gateway), _toy(), "skeleton", dry_run=True) + assert skeleton.status == "dry_run", skeleton.problems + record = run_stage(_runtime(world, gateway), _toy(), "modules", dry_run=True) + assert record.status == "dry_run", record.problems + tree = _tree(record) + assert "knowledge/repos/toy/_routes.yaml" not in tree + # cards and their group are written; the routes they would have taken are suggested + assert "knowledge/repos/toy/components/tooling/tools-lint.md" in tree + assert "](tooling/_index.md)" in tree["knowledge/repos/toy/components/_index.md"] + suggested = [c for c in record.checklist if c.startswith("review_routes (adapter PR)")] + assert review_route_line("tools/lint/", "tooling", "repos/toy/components/tooling/_index.md") in suggested + assert review_route_line("tools/fmt/", "tooling", "repos/toy/components/tooling/_index.md") in suggested + assert review_route_line("tools/gen/", "tooling", "repos/toy/components/tooling/_index.md") in suggested + assert record.coverage["routes_source"] == "manifest" + assert record.coverage["modules"]["before"] < 1 and record.coverage["modules"]["after"] == 1 + assert any("3 review_routes entries are suggested" in n for n in record.notes) + body = (Path(record.pr["dry_run_dir"]) / "PR_BODY.md").read_text(encoding="utf-8") + assert "Routes come from the adapter manifest" in body + + +# -- deepen ---------------------------------------------------------------------------------- + +EMPTY_CORE_RULES = ('---\ntitle: "Core rules"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\n' + "tags: [toy]\nsources: []\n---\n\n# Core rules\n\n" + "## Direct 代码快速入口\n\n| PR 描述在做什么 | 精确规则 | 第一批 live 源码 |\n|---|---|---|\n" + "| engine tick | (none yet) | `pkg/core.py` |\n") + + +class ManifestCodeGateway(CodeGateway): + """Map proposals as ManifestGateway's; code rules as CodeGateway's.""" + + def call_json(self, role, *, system, prompt, validate=None, max_budget_usd=None): + reply = super().call_json(role, system=system, prompt=prompt, validate=validate, + max_budget_usd=max_budget_usd) + if system == SYSTEM_MAP: + reply.data["owners"] = [{"owner": "core", "title": "Core", "page": "repos/toy/components/core/rules.md", + "signals": ["engine"], "scope_prefixes": ["pkg/"]}] + return reply + + +def test_deepen_writes_onto_the_manifest_owner_page_and_still_flips(world): + _manifest_routed_kb(world, core_rules=EMPTY_CORE_RULES) + _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) + assert "knowledge/repos/toy/_routes.yaml" not in tree + core = Page.parse(tree["knowledge/repos/toy/components/core/rules.md"]) + assert [s.rule_id for s in core.rules()] and PKG_RULE["title"] in core.rules()[0].text + assert "knowledge/repos/toy/rules.md" not in tree # the briefing doc is never written to + assert "enabled: true" in tree["adapters/toy/manifest.yaml"] + assert record.coverage["routes_source"] == "manifest" + shown = [json.loads(c["prompt"].split("\n", 1)[1].rsplit("", 1)[0]) + for c in gateway.calls if c["system"] == SYSTEM_CODE_RULES] + assert [p["module"] for p in shown] == ["pkg/"] # the tools modules are unrouted: no group to deepen + assert any("no route reaches" in n and "(modules stage)" in n for n in record.notes) + + +def test_deepen_without_any_routes_blocks(world): + _commit(world["origin"], {"adapters/toy/manifest.yaml": "name: toy\nrepo:\n full_name: o/toy\n language: python\n" + "knowledge:\n repo_subdir: repos/toy\nknowledge_lifecycle:\n enabled: false\n mode: shadow\n"}, + "adapter without routes") + gateway = CodeGateway(doc_rules=[]) + 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) + # drop the routes file the skeleton wrote from its dry-run snapshot: the + # deepen stage then finds no routes file and no review_routes + routes = Path(world["tmp"]) / "state" / "init" / "toy" / "skeleton-dryrun" / "tree" / "knowledge/repos/toy/_routes.yaml" + routes.unlink() + record = run_stage(_runtime(world, gateway), _toy(seeds=()), "deepen", dry_run=True) + assert record.status == "blocked" + assert "declares no review_routes" in record.problems[0] + + +# -- gate findings on this change ----------------------------------------------------- + +DIR_RULE = {"title": "The package is reviewed as one unit", + "body": "- 强制:a change under `pkg/` is reviewed together with its guide.", + "evidence": [{"path": "docs/guide.md", "start": 1, "end": 2}]} + + +def test_a_rule_naming_a_directory_reaches_the_owner_of_that_directory(): + """``claims_in`` strips the trailing slash of ``pkg/``; the manifest + prefix keeps it. The directory still reaches its owner.""" + core = Owner("core", "repos/toy/components/core/rules.md", ("pkg/",)) + assert routes_file("pkg", [core]) == [core] and most_specific("pkg", [core]) == [core] + assert routes_file("pkg/core.py", [core]) == [core] + assert routes_file("pkgs", [core]) == [] and routes_file("pk", [core]) == [] + + +def test_a_directory_rule_with_docs_evidence_lands_on_the_manifest_owner(world): + _manifest_routed_kb(world) + gateway = ManifestGateway(doc_rules=[DIR_RULE], seed_rules=False) + record = run_stage(_runtime(world, gateway), _toy(seeds=()), "skeleton", 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 any(DIR_RULE["title"] in s.text for s in core.rules()) + assert "knowledge/repos/toy/rules.md" not in tree and "knowledge/repos/toy/rules-init.md" not in tree + + +BRIEFING_OWNER_MANIFEST = MANIFEST.replace("doc: repos/toy/components/core/rules.md", "doc: repos/toy/rules.md") +EMPTY_TOY_RULES = ('---\ntitle: "Toy rules"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\n' + "tags: [toy]\nsources: []\n---\n\n# Toy rules\n\nNothing yet.\n") + + +def test_rules_redirected_beside_a_briefing_doc_count_as_its_owners(world): + """The manifest routes ``pkg/`` to the top-level rule page, a briefing doc + init never appends to: deepen writes to ``rules-init.md`` beside it, and + coverage must see that page as the owner's, or deepen would regenerate + forever.""" + _commit(world["origin"], { + "adapters/toy/manifest.yaml": BRIEFING_OWNER_MANIFEST, + "knowledge/repos/toy/_index.md": _index("Toy", "toy", "- [Toy rules](rules.md)\n"), + "knowledge/repos/toy/rules.md": EMPTY_TOY_RULES, + }, "toy routes to its briefing doc") + _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) + assert "knowledge/repos/toy/rules.md" not in tree + beside = Page.parse(tree["knowledge/repos/toy/rules-init.md"]) + assert [s.rule_id for s in beside.rules()] + cov = record.coverage["pr_rule_bearing"] + assert cov["after"] > cov["before"], cov + assert not any("target not reached" in n and "0%" in n for n in record.notes)