diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index c8d77d74..5508286c 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -322,3 +322,14 @@ x 必须为正。因阈值停下的调用(`stop_reason="max_budget"`)→ `Mo ((0, 1],缺省 0.85)、`budget_usd`(>0,缺省 30)、`judge_call_usd`(≥0,缺省 0.50)。`validate_seeds(init, knowledge_root)` 是运行时检查:种子必须是知识树中已存在的文件或目录,解析后不得逃出知识根目录,逐条返回问题。 解析只做静态校验,不读知识树。测试:`test_kb_init_config.py`。 + +## 2026-09-30 kb init:覆盖率度量 +`init_coverage`(纯函数,stdlib + PyYAML,不调模型):`load_owners(routes_yaml_text)` 读 `_routes.yaml` 的 `owners` +(schema_version 1;`models` 按名称匹配,不参与覆盖率;格式错误抛 `ValueError`);`routes_file(path, owners)` 按 +`scope_prefixes` 前缀匹配、保持路由顺序;`module_coverage(modules, owners)` —— 模块内**每个**源文件都被某个 owner 前缀 +覆盖才算覆盖;`make_include(source_roots, exclude, suffixes=())` 构造变更路径过滤器(测试/文档等一律由调用方的 +`exclude` glob 表达,不内置任何仓库字面量);`pr_weighted_coverage(prs, owners, include=, rule_pages=)` —— 每个 PR +改动的每个文件计一次,`routed` = 被某 owner 覆盖,`rule_bearing` = 覆盖它的 owner 页在 `rule_pages`(含有效规则的页) +中,`uncovered_hot` 按次数降序、路径升序;`churn_by_module(prs, modules)` 按最长模块前缀归属(已折叠目录归到折入的 +祖先),降序。空集合的比例记为 1.0(没有可漏的路由)。`ROOT_MODULE` 与 `profiles.establish.ROOT_MODULE` 相同, +由测试钉住(本包不依赖 `profiles`)。源根与变更路径按与 `profiles.establish.normalize_root` 相同的规则规范化(`.`、`./` 表示仓库根,接纳一切路径)。测试:`test_kb_init_coverage.py`。 diff --git a/doc/architecture/SPEC/profiles/establish.md b/doc/architecture/SPEC/profiles/establish.md index 9a4c0d62..c36ee30e 100644 --- a/doc/architecture/SPEC/profiles/establish.md +++ b/doc/architecture/SPEC/profiles/establish.md @@ -1,21 +1,24 @@ # profiles/establish.py —— 规范 - + -`LOC ~107 · profile(Stage 0–1.5 helper) · refactor-status: ok` +`LOC ~180 · profile(Stage 0–1.5 helper) · refactor-status: ok` ## 职责 确定性的建立期 helper。 ## 公开契约 `fact_id`、`build_doc_corpus`、`is_redundant`、`extract_directives`、 -`scan_modules`、`HUMAN_DOC_NAMES`。 +`scan_modules`、`scan_modules_at_depth`、`normalize_root`、`ROOT_MODULE`、`HUMAN_DOC_NAMES`。 ## 不变量 - `is_redundant`(对 README+docs 做 6 词 shingle)会丢弃任何**仓库自己的文档已经写过** 的 briefing 行(ETH 研究那条规则,**D5**)。 - `scan_modules` 是确定性的、按语言索引的,跳过非代码目录。 - `extract_directives` 限制行长(只收短祈使句)。 +- `scan_modules_at_depth`(kb init 的模块定义):源根以下至多 `depth` 层目录为模块,更深的文件归最深的那层祖先; + 非空行少于 `min_loc` 的模块由深到浅折入父目录模块,直到稳定;源根永不折叠(`normalize_root` 先把 `.`、`./pkg`、`/pkg/` 等写法规范化,仓库根为 `""` 且包含其他根,保证折叠必然终止);嵌套源根各自拥有自己的文件; + 每一层都跳过隐藏目录与非代码目录;`exclude` 按仓库相对路径做 fnmatch;未知语言返回 `{}`;结果确定、有序。 ## 边界 —— 不属于这里 纯确定性 helper —— 不含 LLM、不写 store、不含 step 逻辑。 @@ -24,7 +27,7 @@ 仅 stdlib。 ## 测试 -`test_profile_steps.py`(冗余过滤、模块扫描、指令抽取)。 +`test_profile_steps.py`(冗余过滤、模块扫描、指令抽取);`test_kb_init_coverage.py`(按深度的模块扫描)。 ## 重构备注 纯函数 —— 易测、易复用。冗余过滤器是承载 ETH 研究结论的关键防线;**保持它确定性**。 diff --git a/src/infermatrix_copilot/kb_service/init_coverage.py b/src/infermatrix_copilot/kb_service/init_coverage.py new file mode 100644 index 00000000..ea3b04db --- /dev/null +++ b/src/infermatrix_copilot/kb_service/init_coverage.py @@ -0,0 +1,168 @@ +"""Coverage metrics for ``kb init`` (design: kb-init v3 §7). + +Pure and deterministic. Two measures of how much of a repository its +knowledge routes reach: + +* **module coverage** — the share of modules (``profiles.establish. + scan_modules_at_depth``) whose every source file matches some owner's + ``scope_prefixes`` in ``_routes.yaml``; +* **PR-weighted coverage** — over a window of merged PRs, the share of changed + source files (counted once per PR that changed them) that route to an owner + (*routed*), and to an owner whose page carries rules (*rule-bearing*). + +An empty population counts as fully covered (ratio 1.0): there is nothing a +route could miss. +""" + +from __future__ import annotations + +import fnmatch +from collections import Counter +from dataclasses import dataclass +from pathlib import PurePosixPath +from typing import Callable, Mapping, Sequence + +import yaml + +# equals profiles.establish.ROOT_MODULE (kb_service stays free of profile +# imports; a test pins the two together) +ROOT_MODULE = "./" + + +@dataclass(frozen=True) +class Owner: + owner: str + path: str + prefixes: tuple[str, ...] + + +@dataclass(frozen=True) +class Coverage: + covered: list[str] + uncovered: list[str] + ratio: float + + +@dataclass(frozen=True) +class PrCoverage: + total: int + routed: int + rule_bearing: int + routed_ratio: float + rule_bearing_ratio: float + uncovered_hot: list[tuple[str, int]] + + +def load_owners(routes_yaml_text: str) -> list[Owner]: + """The ``owners`` of a ``_routes.yaml`` (schema_version 1); ``models`` are + name-matched, not path-matched, and play no part in coverage.""" + data = yaml.safe_load(routes_yaml_text) or {} + if not isinstance(data, dict): + raise ValueError("_routes.yaml must be a mapping") + if data.get("schema_version", 1) != 1: + raise ValueError(f"unsupported _routes.yaml schema_version {data.get('schema_version')!r}") + owners = data.get("owners") or [] + if not isinstance(owners, list): + raise ValueError("_routes.yaml owners must be a list") + out = [] + for item in owners: + if not isinstance(item, dict) or not item.get("owner") or not item.get("path"): + raise ValueError(f"malformed _routes.yaml owner: {item!r}") + prefixes = item.get("scope_prefixes") or [] + if not isinstance(prefixes, list) or any(not isinstance(p, str) or not p for p in prefixes): + raise ValueError(f"{item['owner']}: scope_prefixes must be a list of non-empty strings") + out.append(Owner(str(item["owner"]), str(item["path"]), tuple(prefixes))) + return out + + +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)] + + +def module_coverage(modules: Mapping[str, Mapping], owners: Sequence[Owner]) -> Coverage: + covered, uncovered = [], [] + for key in sorted(modules): + files = modules[key].get("files") or [] + (covered if all(routes_file(f, owners) for f in files) else uncovered).append(key) + total = len(covered) + len(uncovered) + return Coverage(covered, uncovered, len(covered) / total if total else 1.0) + + +def _normalize(path: str) -> str: + """Repo-relative POSIX form without leading ``./`` or slashes; the repository + root is ``""`` (same rule as ``profiles.establish.normalize_root``).""" + text = PurePosixPath(path.strip().strip("/") or ".").as_posix() + return "" if text == "." else text + + +def make_include(source_roots: Sequence[str], exclude: Sequence[str], + suffixes: Sequence[str] = ()) -> Callable[[str], bool]: + """A filter for changed paths: under a source root (any path when none are + given), with a source suffix (any when none are given), matching no + ``exclude`` fnmatch glob. Tests, docs and generated files are the caller's + ``exclude`` globs; nothing repository-specific is built in.""" + normalized = {_normalize(r) for r in source_roots} + # the repository root (".", "./", "") admits every path + roots = () if "" in normalized else tuple(sorted(r + "/" for r in normalized)) + sfx = tuple(suffixes) + + def include(path: str) -> bool: + path = _normalize(path) + if roots and not path.startswith(roots): + return False + if sfx and not path.endswith(sfx): + return False + return not any(fnmatch.fnmatch(path, glob) for glob in exclude) + + return include + + +def pr_weighted_coverage(prs: Sequence[Sequence[str]], owners: Sequence[Owner], *, + include: Callable[[str], bool], + rule_pages: set[str] | None = None) -> PrCoverage: + """Each file a PR changes counts once for that PR. ``rule_pages`` are the + knowledge paths (as ``_routes.yaml`` names them) holding at least one + active rule; without them nothing is rule-bearing.""" + pages = rule_pages or set() + total = routed = rule_bearing = 0 + missed: Counter[str] = Counter() + for files in prs: + for path in sorted({_normalize(f) for f in files}): + if not include(path): + continue + total += 1 + hits = routes_file(path, owners) + if hits: + routed += 1 + if any(o.path in pages for o in hits): + rule_bearing += 1 + else: + missed[path] += 1 + hot = sorted(missed.items(), key=lambda item: (-item[1], item[0])) + return PrCoverage(total, routed, rule_bearing, + routed / total if total else 1.0, + rule_bearing / total if total else 1.0, hot) + + +def module_of(path: str, modules: Mapping[str, Mapping]) -> str | None: + """The module owning ``path``: the longest module key that prefixes it (a + file in a folded directory belongs to the ancestor it folded into).""" + best = None + for key in modules: + prefix = "" if key == ROOT_MODULE else key + if path.startswith(prefix) and (best is None or len(prefix) > len(best[1])): + best = (key, prefix) + return best[0] if best else None + + +def churn_by_module(prs: Sequence[Sequence[str]], modules: Mapping[str, Mapping]) -> list[tuple[str, int]]: + """File changes per module over the PR window, highest first (ties by + module path). Paths no module owns are not counted.""" + counts: Counter[str] = Counter() + for files in prs: + for path in sorted({_normalize(f) for f in files}): + key = module_of(path, modules) + if key is not None: + counts[key] += 1 + return sorted(counts.items(), key=lambda item: (-item[1], item[0])) diff --git a/src/infermatrix_copilot/profiles/establish.py b/src/infermatrix_copilot/profiles/establish.py index 8984d391..6acb9edc 100644 --- a/src/infermatrix_copilot/profiles/establish.py +++ b/src/infermatrix_copilot/profiles/establish.py @@ -7,9 +7,11 @@ from __future__ import annotations +import fnmatch import hashlib import re -from pathlib import Path +from pathlib import Path, PurePosixPath +from typing import Sequence # human-authored agent instruction files: highest-trust briefing input, # ingested rather than re-derived @@ -100,3 +102,88 @@ def scan_modules(repo: Path, language: str, *, min_files: int = 3) -> dict: modules[entry.name] = {"local_paths": [f"{entry.name}/"], "wave": wave} return modules + + +ROOT_MODULE = "./" # the key of the module holding files directly in the repo root + + +def _loc(path: Path) -> int: + try: + text = path.read_text(encoding="utf-8", errors="replace") + except OSError: + return 0 + return sum(1 for line in text.splitlines() if line.strip()) + + +def normalize_root(root: str) -> str: + """A source root as a repo-relative POSIX directory without leading ``./`` + or slashes (``"src/pkg"``); the repository root itself is ``""``.""" + text = PurePosixPath(root.strip().strip("/") or ".").as_posix() + return "" if text == "." else text + + +def _module_parent(key: str) -> str: + parent = str(PurePosixPath(key.rstrip("/")).parent) + return ROOT_MODULE if parent == "." else parent + "/" + + +def scan_modules_at_depth(repo: Path, language: str, *, source_roots: Sequence[str], + depth: int, min_loc: int, + exclude: Sequence[str] = ()) -> dict[str, dict]: + """Deterministic directory-level modules for knowledge coverage. + + Every source file under a source root belongs to the module of its deepest + ancestor directory at most ``depth`` levels below that root (the root is + level 0). A module under ``min_loc`` non-blank lines folds into its parent + directory's module, deepest first and until stable; a source root never + folds. Returns ``{"pkg/sub/": {"files": [...], "loc": n}}`` (files directly + in the repo root: ``ROOT_MODULE``). Hidden and non-code directories are + skipped at every level, ``exclude`` fnmatch globs apply to repo-relative + file paths, and an unknown language scans nothing rather than guessing.""" + from .languages import suffixes as _suffixes + + sfx = _suffixes(language) + if not sfx or depth < 0: + return {} + roots = sorted({normalize_root(r) for r in source_roots}) or [""] + if "" in roots: + roots = [""] # the repository root already holds every other root + modules: dict[str, dict] = {} + seen: set[str] = set() + root_keys: set[str] = set() + # longest root first, so a nested root owns its own files + for root in sorted(roots, key=lambda r: (-len(r), r)): + base = repo / root if root else repo + if not base.is_dir(): + continue + root_key = f"{root}/" if root else ROOT_MODULE + root_keys.add(root_key) + for path in sorted(base.rglob("*")): + if not path.is_file() or path.suffix not in sfx: + continue + rel = path.relative_to(repo).as_posix() + if rel in seen or any(fnmatch.fnmatch(rel, glob) for glob in exclude): + continue + parts = path.relative_to(base).parts[:-1] + if any(p.startswith(".") or p in _NON_MODULE_DIRS for p in parts): + continue + seen.add(rel) + key = (f"{root}/" if root else "") + "".join(f"{p}/" for p in parts[:depth]) + entry = modules.setdefault(key or ROOT_MODULE, {"files": [], "loc": 0}) + entry["files"].append(rel) + entry["loc"] += _loc(path) + folding = True + while folding: # one fold per pass, deepest first: a folded child counts toward its parent + folding = False + for key in sorted(modules, key=lambda k: (-k.count("/"), k)): + parent = _module_parent(key) + if key in root_keys or parent == key or modules[key]["loc"] >= min_loc: + continue + entry = modules.pop(key) + target = modules.setdefault(parent, {"files": [], "loc": 0}) + target["files"].extend(entry["files"]) + target["loc"] += entry["loc"] + folding = True + break + return {key: {"files": sorted(entry["files"]), "loc": entry["loc"]} + for key, entry in sorted(modules.items()) if entry["files"]} diff --git a/test/test_kb_init_coverage.py b/test/test_kb_init_coverage.py new file mode 100644 index 00000000..84bdf8f6 --- /dev/null +++ b/test/test_kb_init_coverage.py @@ -0,0 +1,175 @@ +"""kb init coverage (design kb-init v3 §7): depth-bounded module scan, module +coverage, PR-weighted routed / rule-bearing coverage, churn ordering.""" + +import pytest + +from infermatrix_copilot.kb_service import init_coverage as cov +from infermatrix_copilot.profiles import establish +from infermatrix_copilot.profiles.establish import scan_modules_at_depth + + +def _write(root, rel, lines): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("".join(f"x{i} = {i}\n\n" for i in range(lines))) + + +def _tree(tmp_path): + _write(tmp_path, "pkg/__init__.py", 2) + _write(tmp_path, "pkg/big/a.py", 40) + _write(tmp_path, "pkg/big/deep/er/b.py", 30) # below depth 2: owned by pkg/big/deep/ + _write(tmp_path, "pkg/big/deep/c.py", 30) + _write(tmp_path, "pkg/small/d.py", 5) # folds into pkg/ + _write(tmp_path, "pkg/tests/test_x.py", 50) + _write(tmp_path, "pkg/docs/e.py", 50) # non-code dir: skipped + _write(tmp_path, "pkg/.hidden/f.py", 50) + _write(tmp_path, "pkg/big/README.md", 50) # not a source suffix + _write(tmp_path, "other/g.py", 50) # outside the source roots + + +def test_scan_depth_folding_and_exclude(tmp_path): + _tree(tmp_path) + modules = scan_modules_at_depth(tmp_path, "python", source_roots=["pkg/"], depth=2, + min_loc=20, exclude=["*/tests/*"]) + assert modules == { + "pkg/": {"files": ["pkg/__init__.py", "pkg/small/d.py"], "loc": 7}, + "pkg/big/": {"files": ["pkg/big/a.py"], "loc": 40}, + "pkg/big/deep/": {"files": ["pkg/big/deep/c.py", "pkg/big/deep/er/b.py"], "loc": 60}, + } + + +def test_scan_folds_until_stable_and_roots_never_fold(tmp_path): + _write(tmp_path, "pkg/a/b/c.py", 5) + _write(tmp_path, "pkg/a/d.py", 5) + modules = scan_modules_at_depth(tmp_path, "python", source_roots=["pkg"], depth=2, min_loc=8) + # pkg/a/b/ (5) -> pkg/a/ (10, stays); the root is kept whatever its size + assert modules == {"pkg/a/": {"files": ["pkg/a/b/c.py", "pkg/a/d.py"], "loc": 10}} + folded = scan_modules_at_depth(tmp_path, "python", source_roots=["pkg"], depth=2, min_loc=100) + assert folded == {"pkg/": {"files": ["pkg/a/b/c.py", "pkg/a/d.py"], "loc": 10}} + + +def test_scan_repo_root_nested_roots_and_unknown_language(tmp_path): + _write(tmp_path, "top.py", 3) + _write(tmp_path, "lib/x.py", 3) + _write(tmp_path, "lib/inner/y.py", 3) + whole = scan_modules_at_depth(tmp_path, "python", source_roots=[], depth=1, min_loc=0) + assert whole == {establish.ROOT_MODULE: {"files": ["top.py"], "loc": 3}, + "lib/": {"files": ["lib/inner/y.py", "lib/x.py"], "loc": 6}} + nested = scan_modules_at_depth(tmp_path, "python", source_roots=["lib", "lib/inner"], + depth=0, min_loc=0) + assert nested == {"lib/": {"files": ["lib/x.py"], "loc": 3}, + "lib/inner/": {"files": ["lib/inner/y.py"], "loc": 3}} + assert scan_modules_at_depth(tmp_path, "cobol", source_roots=[], depth=2, min_loc=0) == {} + + +def test_scan_is_deterministic(tmp_path): + _tree(tmp_path) + kwargs = dict(source_roots=["pkg/"], depth=2, min_loc=20) + assert scan_modules_at_depth(tmp_path, "python", **kwargs) == \ + scan_modules_at_depth(tmp_path, "python", **kwargs) + + +ROUTES = """\ +schema_version: 1 +owners: + - owner: core + path: repos/r/components/core/rules.md + signals: [core] + scope_prefixes: [pkg/big/] + - owner: misc + path: repos/r/components/misc/_index.md + signals: [misc] + scope_prefixes: [pkg/small/, pkg/big/deep/] +models: + dir: repos/r/models +""" + + +def test_load_owners_and_routing_order(): + owners = cov.load_owners(ROUTES) + assert [o.owner for o in owners] == ["core", "misc"] + assert [o.owner for o in cov.routes_file("pkg/big/deep/c.py", owners)] == ["core", "misc"] + assert cov.routes_file("pkg/none.py", owners) == [] + with pytest.raises(ValueError): + cov.load_owners("schema_version: 2\nowners: []\n") + with pytest.raises(ValueError): + cov.load_owners("owners:\n - owner: x\n") + with pytest.raises(ValueError): + cov.load_owners("owners:\n - {owner: x, path: p, scope_prefixes: ['']}\n") + + +def test_module_coverage_needs_every_file_routed(): + owners = cov.load_owners(ROUTES) + modules = {"pkg/": {"files": ["pkg/__init__.py", "pkg/small/d.py"]}, + "pkg/big/": {"files": ["pkg/big/a.py"]}} + result = cov.module_coverage(modules, owners) + assert result.covered == ["pkg/big/"] and result.uncovered == ["pkg/"] + assert result.ratio == 0.5 + assert cov.module_coverage({}, owners).ratio == 1.0 + + +def test_pr_weighted_counts_once_per_pr_and_rule_bearing(): + owners = cov.load_owners(ROUTES) + include = cov.make_include(["pkg/"], ["*/tests/*"], suffixes=[".py"]) + prs = [ + ["pkg/big/a.py", "pkg/big/a.py", "pkg/tests/test_x.py", "README.md"], + ["pkg/big/a.py", "pkg/small/d.py", "pkg/new.py"], + ["pkg/new.py", "pkg/other.py"], + ] + result = cov.pr_weighted_coverage(prs, owners, include=include, + rule_pages={"repos/r/components/core/rules.md"}) + assert (result.total, result.routed, result.rule_bearing) == (6, 3, 2) + assert result.routed_ratio == 0.5 and result.rule_bearing_ratio == pytest.approx(2 / 6) + assert result.uncovered_hot == [("pkg/new.py", 2), ("pkg/other.py", 1)] + bare = cov.pr_weighted_coverage(prs, owners, include=include) + assert bare.rule_bearing == 0 + assert cov.pr_weighted_coverage([], owners, include=include).routed_ratio == 1.0 + + +def test_make_include_without_roots_or_suffixes(): + include = cov.make_include([], ["docs/*"]) + assert include("anything/x.txt") and not include("docs/a.md") + + +def test_churn_by_module_uses_longest_prefix(): + modules = {cov.ROOT_MODULE: {}, "pkg/": {}, "pkg/big/": {}} + prs = [["pkg/big/a.py", "pkg/x.py"], ["pkg/big/gone.py", "setup.py"], ["pkg/big/a.py"]] + assert cov.churn_by_module(prs, modules) == [("pkg/big/", 3), ("./", 1), ("pkg/", 1)] + assert cov.module_of("elsewhere.py", {"pkg/": {}}) is None + + +def test_root_module_constant_matches_profiles(): + assert cov.ROOT_MODULE == establish.ROOT_MODULE + + +@pytest.mark.parametrize("root", ["pkg", "pkg/", "./pkg", "./pkg/", "/pkg"]) +def test_scan_normalizes_roots_and_folding_stops_at_the_root(tmp_path, root): + _write(tmp_path, "pkg/a/x.py", 3) + _write(tmp_path, "pkg/b/y.py", 3) + modules = scan_modules_at_depth(tmp_path, "python", source_roots=[root], depth=1, min_loc=10_000) + assert modules == {"pkg/": {"files": ["pkg/a/x.py", "pkg/b/y.py"], "loc": 6}} + assert establish.normalize_root(root) == "pkg" + + +@pytest.mark.parametrize("root", [".", "./", "", "/"]) +def test_scan_repo_root_spellings_terminate(tmp_path, root): + _write(tmp_path, "a/x.py", 3) + _write(tmp_path, "top.py", 3) + modules = scan_modules_at_depth(tmp_path, "python", source_roots=[root, "a"], depth=1, + min_loc=10_000) + assert modules == {establish.ROOT_MODULE: {"files": ["a/x.py", "top.py"], "loc": 6}} + + +@pytest.mark.parametrize("root", [".", "./", ""]) +def test_make_include_repo_root_spellings_admit_everything(root): + owners = cov.load_owners(ROUTES) + include = cov.make_include([root], [], suffixes=[".py"]) + assert include("pkg/big/a.py") and include("./pkg/new.py") and not include("README.md") + result = cov.pr_weighted_coverage([["./pkg/big/a.py", "pkg/new.py"]], owners, include=include) + assert (result.total, result.routed) == (2, 1) + assert result.uncovered_hot == [("pkg/new.py", 1)] + + +def test_make_include_normalizes_named_roots(): + include = cov.make_include(["./pkg/"], []) + assert include("pkg/x.py") and include("./pkg/x.py") and not include("pkgx/y.py")