diff --git a/doc/architecture/SPEC/adapters/base.md b/doc/architecture/SPEC/adapters/base.md index 98d7c80e..b50ec8ef 100644 --- a/doc/architecture/SPEC/adapters/base.md +++ b/doc/architecture/SPEC/adapters/base.md @@ -1,6 +1,6 @@ # adapters/base.py —— 规范 - + `LOC ~423 · 边缘（仓库知识） · refactor-status: ok` @@ -58,3 +58,6 @@ **把 `capabilities` 推导保持为 planner 信任的唯一来源** —— `app.repository_context` 只为实际 checkout **补上** `repo.path`， 并将同一 adapter policy 交给执行层。 + +## 2026-09-28 退役规则不再提供 +`render_briefing_docs` 经 `visible_text` 去掉退役规则。 diff --git a/doc/architecture/SPEC/knowledge_docs.md b/doc/architecture/SPEC/knowledge_docs.md index 509e5007..2fdc7af5 100644 --- a/doc/architecture/SPEC/knowledge_docs.md +++ b/doc/architecture/SPEC/knowledge_docs.md @@ -33,3 +33,6 @@ ## 2026-09-28 快照校验 可选 `verify(rel)` 回调在 `read`/`search` 读取每个文件前调用；MCP 传入 `KnowledgeView.path`，快照中缺失或被改动的文件使读取/搜索 fail-closed。 + +## 2026-09-28 退役规则不再提供 +`read`/`search` 经 `visible_text` 去掉退役规则后再返回。 diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index 9df7c42c..f2723dd0 100644 --- a/doc/architecture/SPEC/knowledge_service.md +++ b/doc/architecture/SPEC/knowledge_service.md @@ -1,6 +1,6 @@ # knowledge_service/ — provider curation components - + The provider owns knowledge curation beneath the public SDK v1 facade. `KnowledgeCurator` composes four domain components over one explicit work @@ -17,3 +17,11 @@ local commits and proposal export. The public `sdk.v1.knowledge` module only re-exports the curator and validator error. Existing proposal IDs, errors, wire projections, validator order and rollback behavior are compatibility contracts across the move. + +Knowledge Ops API 2.0 sits beside the v1 curator and does not change it: +`lifecycle` parses rule pages (byte-exact) and their `kb:rule` footers, +`ops.apply_operations` applies typed add / edit_same_meaning / replace / retire +/ purge changes with their mechanical consequences, and `l1` is the +deterministic half of the quality gate (`check_tree`, `check_changeset`). The +three modules need only the standard library and PyYAML and import nothing else +from the package, so the kb-gate verifier bundle can vendor them. diff --git a/doc/architecture/SPEC/knowledge_service/l1.md b/doc/architecture/SPEC/knowledge_service/l1.md new file mode 100644 index 00000000..630001db --- /dev/null +++ b/doc/architecture/SPEC/knowledge_service/l1.md @@ -0,0 +1,34 @@ +# knowledge_service/l1.py —— 规范 + + + +`LOC ~420 · 知识质量门的确定性一层（L1） · refactor-status: stable` + +## 职责 +- `check_tree(files)`：尾注合法、supersede 双向一致、tombstone ID 未被复用、 + `_routes.yaml` 合法且引用页面存在。 +- `check_changeset(base, head, changes, external_texts=)`： + - 路径白名单（`knowledge/{repos/,general}/**/*.md`、`_routes.yaml`、`_tombstones.yaml`） + 与文件模式 `100644`；拒绝 `knowledge/{skills,tools,.claude}/`、顶层文件、非 UTF-8。 + - 无新增重复 ID、不复用退役或 tombstone ID、无指向本次退役/删除规则的悬空引用、 + 无新增树级问题、页面不超拆分线。 + - 机械项按规则重算：frontmatter 只允许 `updated`（不倒退）与 `sources`（等于 + `expected_sources`）变化；`_index.md` 只允许为新页追加索引行；tombstones 等于本次 + purge 的 ID 且为规范格式。不符即 issue，不能靠判定覆盖。 + - 其余变更分块：`rule`（add / edit / retire / supersede / purge，按段内容哈希）与 + `prose`（页头、非规则段、非规则页、`_routes.yaml`）。每块有稳定 `block_id`， + 供 L2 判定与签名逐块绑定。 + - `external_refs`：知识目录之外引用了本次退役/删除 ID 的文件（配套 PR 流程）。 + - 嵌套 `###` 规则随父规则退役、删除并记入 tombstones；编辑不得增删嵌套 ID；保护状态以 + 旧尾注为准且不得被移除；页面类型不得借改 `type:` 退出生命周期；每个新页面都必须在 + 同目录 `_index.md` 出现。 + - purge 时序：tombstone 的 `purged_at` 不得等于该规则的 `retired_at`；传入可信的 + `release` 时 `purged_at` 必须等于它。 + +## 不变量 +纯函数；标准库 + PyYAML + `lifecycle`/`ops`，验证包原样携带。段比较忽略尾部空行 +（追加段落时前一段多出的分隔空行不算改动）。 + +## 测试 +`test_knowledge_ops_l1.py`（伪造的 sources/frontmatter/删除 active 规则/重复 ID、白名单与 +模式、prose 分块、外部引用、真实知识树上执行各类操作后两个既有校验器均通过）。 diff --git a/doc/architecture/SPEC/knowledge_service/lifecycle.md b/doc/architecture/SPEC/knowledge_service/lifecycle.md new file mode 100644 index 00000000..fcc9996c --- /dev/null +++ b/doc/architecture/SPEC/knowledge_service/lifecycle.md @@ -0,0 +1,24 @@ +# knowledge_service/lifecycle.py —— 规范 + + + +`LOC ~330 · 规则页数据模型与生命周期尾注 · refactor-status: stable` + +## 职责 +把规则页解析为 frontmatter、head 与 `## ` 段（跳过代码围栏内的标题），未改动的页 +解析再渲染逐字节相同。规则段可带 `` 尾注（status、since、retired_at、 +reason、evidence、supersedes、superseded_by、protected）；无尾注即 active。 + +## 公开契约 +`Page.parse/render`、`Page.rules/rule/replace_section/append_section`、 +`Page.sources/with_sources`（按页面原风格写 flow 或 block 列表）、`Section.footer/ +with_footer/body_without_footer/content_sha256/citations`、`Footer.parse/render/check`、 +`expected_sources`（旧列表 + active 规则新引用，只追加）、`visible_text`（去掉退役规则， +无退役规则时逐字节原样返回）。 + +## 不变量 +- 仅依赖标准库与 PyYAML、不 import 包内其他模块：kb-gate 验证包原样携带。 +- 尾注键未知、重复、格式错误或状态约束不满足时抛 `LifecycleError`。 + +## 测试 +`test_knowledge_ops_l1.py`（全树逐字节往返、服务端隐藏退役规则）。 diff --git a/doc/architecture/SPEC/knowledge_service/ops.md b/doc/architecture/SPEC/knowledge_service/ops.md new file mode 100644 index 00000000..78ed3dcc --- /dev/null +++ b/doc/architecture/SPEC/knowledge_service/ops.md @@ -0,0 +1,27 @@ +# knowledge_service/ops.py —— 规范 + + + +`LOC ~360 · Knowledge Ops API 2.0（类型化知识变更） · refactor-status: stable` + +## 职责 +`apply_operations(files, ops, release, today)` 是纯函数：输入知识相对路径到文本的映射与 +类型化操作（`add`、`edit_same_meaning`、`replace`、`retire`、`purge`），输出变更后的文件。 +随后写入机械结果：`updated:`、只追加的 `sources:`、新页面在同目录 `_index.md` 的一行、 +purge 的 `_tombstones.yaml` 条目。 + +## 不变量 +- 新 ID（含嵌套 `###`）在全树唯一，且不在 tombstones 中。 +- `replace` 在一次调用内同时退役旧规则（`reason=superseded`、`superseded_by`）并添加带 + `supersedes` 的后继规则。 +- `purge` 只针对已退役且 `retired_at` 不是本次发版的规则。 +- `protected` 规则需 `allow_protected`（人工路径）。 +- 超过拆分线（32 KiB / 500 非空行）的页面被拒绝，提示改用新的 `rules-.md`。 +- 任何非法操作抛 `LifecycleError`，输入映射从不被修改（无部分应用）。 +- 版本常量 `KNOWLEDGE_OPS_API_VERSION = 2.0.0`，与 v1 `KnowledgeCurator`（1.1.0）并存。 + +## 依赖（允许） +标准库 + PyYAML + `lifecycle`（验证包原样携带）。 + +## 测试 +`test_knowledge_ops_l1.py`。 diff --git a/doc/architecture/SPEC/sdk.md b/doc/architecture/SPEC/sdk.md index 15feb6c8..66c6802b 100644 --- a/doc/architecture/SPEC/sdk.md +++ b/doc/architecture/SPEC/sdk.md @@ -121,3 +121,6 @@ Direct 1.1 / Strict 1.3 accept typed `CarriedFinding` inputs (unique IDs, source ## 2026-09-28 知识视图 `DirectClient` 每次 `plan` 解析一个 `KnowledgeView`，文档引用与 `resource_revision` 均取自该视图；`validate` 按计划时钉住的知识根校验资源未变，激活新快照不会使进行中的 review 失效；`diagnostics.knowledge_snapshot` 报告所用快照。`read_document(..., review_context_id=...)` 从该 review 计划时钉住的知识树读取（含分页），未签发的 context id 被拒绝；不带 context id 时读当前激活树。 + +## 2026-09-28 退役规则不再提供 +`DirectClient` 返回的文档内容与 `sha256` 均基于去掉退役规则后的服务文本（无退役规则时与磁盘字节一致）。 diff --git a/doc/knowledge/SCHEMA.md b/doc/knowledge/SCHEMA.md index c1253875..50758fa3 100644 --- a/doc/knowledge/SCHEMA.md +++ b/doc/knowledge/SCHEMA.md @@ -64,6 +64,33 @@ contradictions: [相对路径] # 可选：与本页冲突的页面 先把仍然有效的独有结论合入最近 owner 的幸存页面，再删除被取代或重复页面并修复 入链。Git 历史负责追溯旧版本；`knowledge/` 内不再维护 `_archive/` 副本。 +## 规则生命周期尾注 + +规则段（`## — <标题>`）可以以一行机器可读尾注结束，渲染时不可见： + +``` + + +``` + +- 没有尾注 = `status=active`（兼容全部既有规则）。 +- `status=retired` 必须带 `retired_at`、`reason`（`upstream-removed` / `superseded` / + `incorrect` / `duplicate`）与 `evidence`；`superseded` 还必须带 `superseded_by`，且 + 后继规则带对应的 `supersedes`。 +- 退役规则的正文在磁盘上保留一个发版周期供复核与撤销，但**立即停止对外提供**： + Direct 文档读取、`doc_read`/`doc_search`、briefing 都会去掉它。 +- `protected=true` 的规则只能经人工路径修改、退役或删除。 +- 物理删除（purge）只针对已退役满一个发版周期的规则；被删除的 ID 记入同仓库的 + `_tombstones.yaml`，永不复用。 +- 尾注、`updated:`、`sources:`（只追加）、新页面的 `_index.md` 行与 tombstones 由知识服务 + 按 Knowledge Ops API 2.0 写入，并由 L1 门禁重算核对；手工改动也必须符合同一规则。 + +## Direct 路由表 + +每个仓库的 Direct owner 路由在 `repos//_routes.yaml`（`schema_version: 1`， +`owners[{owner, path, signals, scope_prefixes}]`，可选 `models{dir, page}`）。路由引用的页面 +必须存在；新增仓库只需提供它和 adapter，不改代码。 + ## 校验 ```bash diff --git a/src/infermatrix_copilot/adapters/base.py b/src/infermatrix_copilot/adapters/base.py index ffb97dd9..e785b474 100644 --- a/src/infermatrix_copilot/adapters/base.py +++ b/src/infermatrix_copilot/adapters/base.py @@ -16,6 +16,8 @@ import yaml +from ..knowledge_service.lifecycle import visible_text + HIGH_RISK_SECTIONS = ("push", "repo", "upstream", "rebase") @@ -87,7 +89,7 @@ def render_briefing_docs(root: Path | str, docs: list[str], *, header: str = "", if warnings is not None: warnings.append(f"briefing document missing or not Markdown: {rel}") continue - text = p.read_text(encoding="utf-8", errors="replace") + text = visible_text(p.read_text(encoding="utf-8", errors="replace")) parts.append(_without_frontmatter(text).strip()) if not parts: return "" diff --git a/src/infermatrix_copilot/knowledge_docs.py b/src/infermatrix_copilot/knowledge_docs.py index 9234dc10..5a05f178 100644 --- a/src/infermatrix_copilot/knowledge_docs.py +++ b/src/infermatrix_copilot/knowledge_docs.py @@ -7,6 +7,8 @@ from pathlib import Path from typing import Callable +from .knowledge_service.lifecycle import visible_text + class KnowledgeDocsError(ValueError): """A refused or invalid knowledge-base operation.""" @@ -94,7 +96,7 @@ def read(self, path: str, *, offset: int = 0, limit: int = 24_000) -> dict: raise KnowledgeDocsError("limit must be a positive integer") limit = min(limit, 65_536) target = self._resolve_doc(path) - data = target.read_text(encoding="utf-8", errors="replace") + data = visible_text(target.read_text(encoding="utf-8", errors="replace")) end = offset + limit return { "path": target.relative_to(self.root).as_posix(), @@ -121,8 +123,8 @@ def search(self, query: str, *, limit: int = 40) -> list[dict]: seen_files.add(target) self._checked(target) rel = target.relative_to(self.root).as_posix() - for lineno, line in enumerate( - target.read_text(encoding="utf-8", errors="replace").splitlines(), 1): + served = visible_text(target.read_text(encoding="utf-8", errors="replace")) + for lineno, line in enumerate(served.splitlines(), 1): line_folded = line.casefold() exact = folded in line_folded term_match = bool(terms) and all(t in line_folded for t in terms) diff --git a/src/infermatrix_copilot/knowledge_service/l1.py b/src/infermatrix_copilot/knowledge_service/l1.py new file mode 100644 index 00000000..53ea48cb --- /dev/null +++ b/src/infermatrix_copilot/knowledge_service/l1.py @@ -0,0 +1,493 @@ +"""L1: the deterministic half of the knowledge quality gate. + +Two entry points, both pure functions over ``{knowledge-relative path: text}`` +mappings so the service, the gate and the vendored verifier agree exactly: + +* ``check_tree(files)`` — structural issues in one tree (footers, supersede + links, tombstones, ``_routes.yaml``). +* ``check_changeset(base, head, changes)`` — what one change set may do: only + whitelisted files and modes, no new duplicate or reused rule IDs, no dangling + references to rules it retired or purged, no NEW tree issues, and a + classification of every change into blocks: + + ``rule`` a rule section added / edited / retired / superseded / purged. + Needs a per-block L2 verdict (keyed by ``block_id``). + ``prose`` any other wording change (page head, non-rule sections, + guides, ``_routes.yaml``). Needs an L2 prose verdict. + ``mechanical`` ``updated:``/``sources:`` frontmatter, index lines for new + pages, tombstones. Recomputed here; a mismatch is an issue, + never a block a verdict could cover. + +Standard library + PyYAML + ``lifecycle``/``ops`` only (vendored by kb-gate). +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from pathlib import PurePosixPath +from typing import Iterable, Mapping + +import yaml + +try: + from .lifecycle import ( + FRONTMATTER, LifecycleError, Page, Section, expected_sources, + ) + from .ops import ( + INDEX_NAME, TOMBSTONES_NAME, all_rule_ids, all_tombstoned_ids, + load_tombstones, page_over_capacity, render_tombstones, + ) +except ImportError: # pragma: no cover - vendored verifier bundle + from lifecycle import ( # type: ignore[no-redef] + FRONTMATTER, LifecycleError, Page, Section, expected_sources, + ) + from ops import ( # type: ignore[no-redef] + INDEX_NAME, TOMBSTONES_NAME, all_rule_ids, all_tombstoned_ids, + load_tombstones, page_over_capacity, render_tombstones, + ) + +ROUTES_NAME = "_routes.yaml" +KNOWLEDGE_PREFIX = "knowledge/" +_WHITELIST = ( + re.compile(r"knowledge/(?:repos/[^/]+|general)/(?:[^/]+/)*[^/]+\.md"), + re.compile(r"knowledge/repos/[^/]+/_routes\.yaml"), + re.compile(r"knowledge/(?:repos/[^/]+|general)/_tombstones\.yaml"), +) +_DENY = re.compile(r"knowledge/(?:skills|tools|\.claude)/|knowledge/[^/]+$") +_REGULAR_MODE = "100644" +_INDEX_LINE = re.compile(r"- \[[^\]\n]+\]\((?P[^)/\s]+\.md)\)") + + +@dataclass(frozen=True) +class Issue: + code: str + path: str + detail: str + + def to_dict(self) -> dict: + return {"code": self.code, "path": self.path, "detail": self.detail} + + +@dataclass(frozen=True) +class Change: + """One changed path as git reports it (repository-relative).""" + + path: str + status: str # A, M, D (renames are reported as D + A) + old_mode: str = "" + new_mode: str = "" + + +@dataclass(frozen=True) +class Block: + kind: str # rule | prose + path: str # knowledge-relative + rule_id: str # "" for prose + op: str # add | edit | retire | supersede | purge | prose + sha256: str # of the text the verdict judges + + @property + def block_id(self) -> str: + identity = json.dumps([self.kind, self.path, self.rule_id, self.op, self.sha256], + separators=(",", ":")) + return hashlib.sha256(identity.encode("utf-8")).hexdigest() + + def to_dict(self) -> dict: + return {"block_id": self.block_id, "kind": self.kind, "path": self.path, + "rule_id": self.rule_id, "op": self.op, "sha256": self.sha256} + + +@dataclass(frozen=True) +class ChangesetResult: + issues: tuple[Issue, ...] + blocks: tuple[Block, ...] + retired: tuple[str, ...] + purged: tuple[str, ...] + external_refs: tuple[tuple[str, str], ...] = () # (rule_id, repo path) + + @property + def ok(self) -> bool: + return not self.issues + + +def _sha(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _norm(text: str) -> str: + """Section text without trailing blank lines: appending a section adds a + separator to the previous one, which is layout, not a change of words.""" + return text.rstrip("\n") + "\n" + + +# -- tree-level ---------------------------------------------------------------- + +def _rule_pages(files: Mapping[str, str]) -> dict[str, Page]: + pages: dict[str, Page] = {} + for path, text in files.items(): + if path.endswith(".md"): + page = Page.parse(text) + if page.rules(): + pages[path] = page + return pages + + +def check_tree(files: Mapping[str, str]) -> list[Issue]: + issues: list[Issue] = [] + footers: dict[str, tuple[str, object]] = {} + for path, page in _rule_pages(files).items(): + for section in page.rules(): + try: + footer = section.footer + except LifecycleError as exc: + issues.append(Issue("footer_invalid", path, f"{section.rule_id}: {exc}")) + continue + footers.setdefault(section.rule_id, (path, footer)) + for rule_id, (path, footer) in footers.items(): + if footer.superseded_by: + target = footers.get(footer.superseded_by) + if target is None or target[1].supersedes != rule_id: + issues.append(Issue("supersede_link_broken", path, + f"{rule_id} -> {footer.superseded_by} has no matching supersedes")) + if footer.supersedes: + source = footers.get(footer.supersedes) + if source is not None and source[1].superseded_by != rule_id: + issues.append(Issue("supersede_link_broken", path, + f"{rule_id} supersedes {footer.supersedes}, which does not name it")) + try: + tombstoned = all_tombstoned_ids(files) + except (LifecycleError, yaml.YAMLError) as exc: + issues.append(Issue("tombstones_invalid", "", str(exc))) + tombstoned = set() + for rule_id in sorted(tombstoned & set(all_rule_ids(files))): + issues.append(Issue("tombstoned_id_reused", "", rule_id)) + for path, text in files.items(): + if PurePosixPath(path).name == ROUTES_NAME: + issues.extend(_check_routes(path, text, files)) + return issues + + +def _check_routes(path: str, text: str, files: Mapping[str, str]) -> list[Issue]: + try: + data = yaml.safe_load(text) + except yaml.YAMLError as exc: + return [Issue("routes_invalid", path, f"not YAML: {exc}")] + if not isinstance(data, dict) or data.get("schema_version") != 1: + return [Issue("routes_invalid", path, "schema_version must be 1")] + issues = [] + seen: set[str] = set() + for index, owner in enumerate(data.get("owners") or []): + if not isinstance(owner, dict) or not owner.get("owner") or not owner.get("path"): + issues.append(Issue("routes_invalid", path, f"owners[{index}] needs owner and path")) + continue + if owner["owner"] in seen: + issues.append(Issue("routes_invalid", path, f"duplicate owner {owner['owner']}")) + seen.add(owner["owner"]) + if str(owner["path"]) not in files: + issues.append(Issue("routes_page_missing", path, str(owner["path"]))) + for key in ("signals", "scope_prefixes"): + if not isinstance(owner.get(key) or [], list): + issues.append(Issue("routes_invalid", path, f"{owner['owner']}.{key} must be a list")) + models = data.get("models") + if models is not None and (not isinstance(models, dict) or not models.get("dir") or not models.get("page")): + issues.append(Issue("routes_invalid", path, "models needs dir and page")) + return issues + + +# -- change-set level ---------------------------------------------------------- + +def _whitelist_issues(changes: Iterable[Change]) -> list[Issue]: + issues = [] + for change in changes: + path = change.path + if not path.startswith(KNOWLEDGE_PREFIX) or _DENY.match(path) \ + or not any(p.fullmatch(path) for p in _WHITELIST): + issues.append(Issue("path_not_whitelisted", path, "only knowledge rule/guide pages, _routes.yaml and _tombstones.yaml auto-merge")) + continue + if change.status != "D" and change.new_mode != _REGULAR_MODE: + issues.append(Issue("mode_not_regular", path, f"mode {change.new_mode or '?'} (need 100644)")) + return issues + + +def _text_issues(path: str, text: str | None) -> list[Issue]: + if text is None: + return [] + if "\x00" in text or "�" in text: + return [Issue("not_utf8_text", path, "binary or non-UTF-8 content")] + return [] + + +def _frontmatter_mechanical(path: str, base: Page, head: Page) -> list[Issue]: + """Only ``updated:`` and ``sources:`` may change, and sources exactly as recomputed.""" + base_data, head_data = base.frontmatter_data(), head.frontmatter_data() + issues = [] + for key in sorted(set(base_data) | set(head_data)): + if key in ("updated", "sources"): + continue + if base_data.get(key) != head_data.get(key): + issues.append(Issue("frontmatter_changed", path, f"{key} may not change")) + if str(head_data.get("updated", "")) < str(base_data.get("updated", "")): + issues.append(Issue("frontmatter_changed", path, "updated went backwards")) + want = expected_sources(base.sources(), head) + if head.sources() != want: + issues.append(Issue("sources_mismatch", path, f"expected {want}, got {head.sources()}")) + return issues + + +def _new_page_mechanical(path: str, head: Page) -> list[Issue]: + data = head.frontmatter_data() + issues = [] + if data.get("type") != "rule": + issues.append(Issue("new_page_invalid", path, "a new page must be type: rule")) + if not str(data.get("title") or "").strip(): + issues.append(Issue("new_page_invalid", path, "a new page needs a title")) + want = expected_sources([], head) + if head.sources() != want: + issues.append(Issue("sources_mismatch", path, f"expected {want}, got {head.sources()}")) + return issues + + +def _classify_page(path: str, base_text: str | None, head_text: str | None, + blocks: list[Block], issues: list[Issue], + retired: list[str], purged: list[tuple[str, str]], + purge_retired_at: dict[str, str]) -> None: + base = Page.parse(base_text) if base_text is not None else None + head = Page.parse(head_text) if head_text is not None else None + if head is None: + issues.append(Issue("page_deleted", path, "pages are never deleted; purge rules instead")) + return + base_is_rule = base is not None and ( + base.frontmatter_data().get("type") == "rule" or bool(base.rules())) + is_rule_page = base_is_rule or head.frontmatter_data().get("type") == "rule" + if base_is_rule and head.frontmatter_data().get("type") != base.frontmatter_data().get("type"): + # a page may not leave the lifecycle by changing its type + issues.append(Issue("page_type_changed", path, "a rule page keeps its type")) + if not is_rule_page: + if base is None or base.render() != head.render(): + blocks.append(Block("prose", path, "", "prose", _sha(head.render()))) + return + if base is None: + issues.extend(_new_page_mechanical(path, head)) + base_rules: dict[str, Section] = {} + base_other: list[str] = [] + base_head = "" + else: + issues.extend(_frontmatter_mechanical(path, base, head)) + base_rules = {s.rule_id: s for s in base.rules()} + base_other = [_norm(s.text) for s in base.sections if not s.is_rule] + base_head = base.head + head_rules = {s.rule_id: s for s in head.rules()} + head_other = [_norm(s.text) for s in head.sections if not s.is_rule] + if (base is None and head.head.strip() != f"# {head.frontmatter_data().get('title', '')}") \ + or (base is not None and head.head != base_head) or head_other != base_other: + blocks.append(Block("prose", path, "", "prose", + _sha(head.head + "".join(head_other)))) + for rule_id, section in head_rules.items(): + old = base_rules.get(rule_id) + try: + footer = section.footer + old_footer = old.footer if old is not None else None + except LifecycleError as exc: + issues.append(Issue("footer_invalid", path, f"{rule_id}: {exc}")) + continue + if old is None: + if footer.status != "active": + issues.append(Issue("rule_added_retired", path, rule_id)) + blocks.append(Block("rule", path, rule_id, "add", _sha(_norm(section.text)))) + continue + if _norm(old.text) == _norm(section.text): + continue + if old_footer.status == "retired": + issues.append(Issue("retired_rule_changed", path, f"{rule_id} is retired and frozen")) + continue + if old_footer.protected and not footer.protected: + issues.append(Issue("protection_removed", path, rule_id)) + if footer.status == "retired": + if section.body_without_footer != old.body_without_footer: + issues.append(Issue("retire_changed_text", path, f"{rule_id}: retiring may not reword")) + if old_footer.protected or footer.protected: + issues.append(Issue("protected_rule_retired", path, rule_id)) + op = "supersede" if footer.reason == "superseded" else "retire" + retired.extend(dict.fromkeys((rule_id, *section.nested_rule_ids))) + blocks.append(Block("rule", path, rule_id, op, _sha(_norm(section.text)))) + else: + if footer != old_footer: + issues.append(Issue("footer_changed", path, f"{rule_id}: an edit keeps its footer")) + if old_footer.protected or footer.protected: + issues.append(Issue("protected_rule_edited", path, rule_id)) + if set(section.nested_rule_ids) != set(old.nested_rule_ids): + # a nested rule leaves only by retiring and purging its parent + issues.append(Issue("nested_rule_ids_changed", path, + f"{rule_id}: {sorted(set(old.nested_rule_ids) ^ set(section.nested_rule_ids))}")) + blocks.append(Block("rule", path, rule_id, "edit", _sha(_norm(section.text)))) + for rule_id, old in base_rules.items(): + if rule_id in head_rules: + continue + if old.footer.status != "retired": + issues.append(Issue("active_rule_removed", path, f"{rule_id}: retire before purging")) + continue + if old.footer.protected: + issues.append(Issue("protected_rule_purged", path, rule_id)) + purged.extend((rid, path) for rid in dict.fromkeys((rule_id, *old.nested_rule_ids))) + purge_retired_at[rule_id] = old.footer.retired_at + blocks.append(Block("rule", path, rule_id, "purge", _sha(_norm(old.text)))) + + +def _index_mechanical(path: str, base_text: str | None, head_text: str | None, + new_pages: set[str], issues: list[Issue], blocks: list[Block]) -> None: + if base_text is None or head_text is None: + issues.append(Issue("index_added_or_deleted", path, "index files are never created or deleted by a change set")) + return + directory = str(PurePosixPath(path).parent) + expected_new = sorted(PurePosixPath(p).name for p in new_pages + if str(PurePosixPath(p).parent) == directory + and p.endswith(".md") and PurePosixPath(p).name != INDEX_NAME) + for name in expected_new: # required whatever else the index change does + if f"]({name})" not in head_text: + issues.append(Issue("index_missing", path, f"new page {name} is not linked")) + if not head_text.startswith(base_text.rstrip("\n") + "\n") and head_text != base_text: + blocks.append(Block("prose", path, "", "prose", _sha(head_text))) + return + added = head_text[len(base_text.rstrip("\n") + "\n"):].splitlines() + files = [] + for line in added: + match = _INDEX_LINE.fullmatch(line.strip()) + if match is None: + blocks.append(Block("prose", path, "", "prose", _sha(head_text))) + return + files.append(match.group("file")) + if sorted(files) != expected_new: + issues.append(Issue("index_mismatch", path, f"expected index lines for {expected_new}, got {files}")) + + +def check_changeset( + base: Mapping[str, str], + head: Mapping[str, str], + changes: Iterable[Change], + *, + external_texts: Mapping[str, str] | None = None, + release: str = "", +) -> ChangesetResult: + """Gate one change set. ``base``/``head`` map knowledge-relative paths + (``repos/...``) to text; ``changes`` are repository-relative git changes. + ``external_texts`` (repo path -> text, outside ``knowledge/``) is scanned + for references to rules this change retires or purges. ``release`` is the + trusted current release (from the signed verdict); when given, purges must + carry it as ``purged_at``. A purge is always rejected in the release that + retired the rule.""" + changes = list(changes) + issues = _whitelist_issues(changes) + blocks: list[Block] = [] + retired: list[str] = [] + purged: list[tuple[str, str]] = [] # (rule id, page) + purge_retired_at: dict[str, str] = {} + knowledge_paths = [c.path[len(KNOWLEDGE_PREFIX):] for c in changes + if c.path.startswith(KNOWLEDGE_PREFIX)] + for path in knowledge_paths: + issues.extend(_text_issues(path, head.get(path))) + if path.endswith(".md") and head.get(path) is not None: + over = page_over_capacity(head[path]) + if over: + issues.append(Issue("page_over_capacity", path, over)) + new_pages = {p for p in knowledge_paths if p not in base and p in head} + for page_path in sorted(new_pages): + index = str(PurePosixPath(page_path).with_name(INDEX_NAME)) + if page_path.endswith(".md") and PurePosixPath(page_path).name != INDEX_NAME \ + and index not in knowledge_paths: + issues.append(Issue("index_missing", page_path, f"{index} must list the new page")) + tombstone_paths = [] + for path in sorted(knowledge_paths): + name = PurePosixPath(path).name + if name == INDEX_NAME: + _index_mechanical(path, base.get(path), head.get(path), new_pages, issues, blocks) + elif name == TOMBSTONES_NAME: + tombstone_paths.append(path) + elif name == ROUTES_NAME: + blocks.append(Block("prose", path, "", "prose", _sha(head.get(path) or ""))) + elif path.endswith(".md"): + try: + _classify_page(path, base.get(path), head.get(path), blocks, issues, + retired, purged, purge_retired_at) + except (LifecycleError, yaml.YAMLError) as exc: + issues.append(Issue("page_unparseable", path, str(exc))) + # tombstones: exactly the IDs purged in this change set, appended + for path in tombstone_paths: + try: + before = load_tombstones(base.get(path)) + after = load_tombstones(head.get(path)) + except (LifecycleError, yaml.YAMLError) as exc: + issues.append(Issue("tombstones_invalid", path, str(exc))) + continue + scope = str(PurePosixPath(path).parent) + purged_here = [rid for rid, page_path in purged if page_path.startswith(scope + "/")] + appended = after[len(before):] + if after[:len(before)] != before or sorted(i["id"] for i in appended) != sorted(purged_here): + issues.append(Issue("tombstones_mismatch", path, f"expected appended tombstones for {sorted(purged_here)}")) + for item in appended: + retired_at = purge_retired_at.get(str(item["id"])) + purged_at = str(item.get("purged_at") or "") + if retired_at is None: + continue # a nested ID; its parent carries the timing + if not purged_at or purged_at == retired_at: + issues.append(Issue("purge_too_early", path, + f"{item['id']}: purged in the release it was retired ({retired_at})")) + if release and purged_at != release: + issues.append(Issue("purge_release_mismatch", path, + f"{item['id']}: purged_at {purged_at} is not this release {release}")) + if head.get(path) != render_tombstones(after): + issues.append(Issue("tombstones_mismatch", path, "tombstones are not in canonical form")) + for rule_id, _page in purged: + if rule_id not in all_tombstoned_ids(head): + issues.append(Issue("purge_without_tombstone", "", rule_id)) + # no NEW duplicate IDs, no reuse of retired or tombstoned IDs + base_ids, head_ids = all_rule_ids(base), all_rule_ids(head) + for rule_id, count in head_ids.items(): + if count > base_ids.get(rule_id, 0) and count > 1: + issues.append(Issue("duplicate_rule_id", "", rule_id)) + base_tomb = all_tombstoned_ids(base) + for block in blocks: + if block.op == "add" and (block.rule_id in base_tomb or block.rule_id in base_ids): + issues.append(Issue("rule_id_reused", block.path, block.rule_id)) + # dangling references to what this change set retired or purged + gone = set(retired) | {rid for rid, _page in purged} + if gone: + issues.extend(_dangling_refs(head, gone)) + external = tuple( + (rule_id, path) + for path, text in sorted((external_texts or {}).items()) + for rule_id in sorted(gone) if re.search(rf"(? list[Issue]: + """References in ACTIVE text to rules that are now retired or purged. + Allowed: the rule's own section, lifecycle footers, tombstones.""" + issues = [] + for path, text in head.items(): + if not path.endswith(".md"): + continue + page = Page.parse(text) + chunks = [page.head] + [ + s.body_without_footer for s in page.sections + if s.rule_id not in gone and (not s.is_rule or _active(s)) + ] + body = "\n".join(chunks) + for rule_id in sorted(gone): + if re.search(rf"(? bool: + try: + return section.footer.status == "active" + except LifecycleError: + return True diff --git a/src/infermatrix_copilot/knowledge_service/lifecycle.py b/src/infermatrix_copilot/knowledge_service/lifecycle.py new file mode 100644 index 00000000..242d9410 --- /dev/null +++ b/src/infermatrix_copilot/knowledge_service/lifecycle.py @@ -0,0 +1,423 @@ +"""Rule pages as data: sections, lifecycle footers, and byte-preserving render. + +A rule page is frontmatter, a head (title + preamble), then sections. A section +starts at every ``## `` heading, and also at a ``### — `` heading +whose enclosing ``## `` section is not itself a rule (owner pages group rules +under topic headings). A section whose heading is ``## <ID> — <title>`` or +``### <ID> — <title>`` is a rule; ``###`` rules INSIDE a ``##`` rule stay part +of it as nested rules. Each rule may end +with one machine-readable footer, an HTML comment that renders invisibly:: + + <!-- kb:rule status=retired since=v0.30.0 retired_at=v0.31.0 reason=upstream-removed evidence="PR #8201" superseded_by=BENCH-1c --> + +No footer means ``status=active``, which is every rule written before the +lifecycle existed. Parsing then rendering an untouched page returns its exact +bytes, so an operation changes only the sections it targets. + +This module needs only the standard library and PyYAML and imports nothing +else from the package: the ``kb-gate`` verifier bundle vendors it verbatim. +""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass, field, replace +from typing import Iterable + +import yaml + +RULE_HEADING = re.compile(r"^(?P<level>#{2,3})\s+(?P<rule>[A-Za-z0-9][A-Za-z0-9-]{1,40})\s+[—-]\s+\S") +_TOP_RULE = re.compile(r"^##\s+[A-Za-z0-9][A-Za-z0-9-]{1,40}\s+[—-]\s+\S") +ANY_RULE_HEADING = re.compile( + r"^#{2,6}\s+(?P<rule>[A-Za-z0-9][A-Za-z0-9-]{1,40})(?=\s|$)", re.MULTILINE) +FOOTER = re.compile(r"^<!--\s*kb:rule\s+(?P<attrs>.*?)\s*-->\s*$") +_ATTR = re.compile(r'(?P<key>[a-z_]+)=(?:"(?P<quoted>[^"]*)"|(?P<bare>[^\s"]+))') +FRONTMATTER = re.compile(r"\A---\r?\n(?P<body>.*?)\r?\n---(?:\r?\n|\Z)", re.DOTALL) +CITATION = re.compile(r"\^\[(?P<ref>[^\]\n]{1,60})\]") + +STATUSES = ("active", "retired") +FOOTER_KEYS = ( + "status", "since", "retired_at", "reason", "evidence", + "supersedes", "superseded_by", "protected", +) +RETIRE_REASONS = ("upstream-removed", "superseded", "incorrect", "duplicate") + + +class LifecycleError(ValueError): + """A page, footer, or operation violates the rule lifecycle contract.""" + + +@dataclass(frozen=True) +class Footer: + status: str = "active" + since: str = "" + retired_at: str = "" + reason: str = "" + evidence: str = "" + supersedes: str = "" + superseded_by: str = "" + protected: bool = False + + @classmethod + def parse(cls, attrs: str) -> "Footer": + values: dict[str, str] = {} + consumed = 0 + for match in _ATTR.finditer(attrs): + if attrs[consumed:match.start()].strip(): + raise LifecycleError(f"malformed kb:rule footer: {attrs!r}") + key = match.group("key") + if key not in FOOTER_KEYS: + raise LifecycleError(f"unknown kb:rule footer key: {key}") + if key in values: + raise LifecycleError(f"repeated kb:rule footer key: {key}") + values[key] = match.group("quoted") if match.group("quoted") is not None \ + else match.group("bare") + consumed = match.end() + if attrs[consumed:].strip(): + raise LifecycleError(f"malformed kb:rule footer: {attrs!r}") + status = values.get("status", "active") + if status not in STATUSES: + raise LifecycleError(f"kb:rule status must be one of {STATUSES}: {status}") + protected = values.get("protected", "false") + if protected not in {"true", "false"}: + raise LifecycleError("kb:rule protected must be true or false") + footer = cls( + status=status, + since=values.get("since", ""), + retired_at=values.get("retired_at", ""), + reason=values.get("reason", ""), + evidence=values.get("evidence", ""), + supersedes=values.get("supersedes", ""), + superseded_by=values.get("superseded_by", ""), + protected=protected == "true", + ) + footer.check() + return footer + + def check(self) -> None: + if self.status == "retired": + missing = [key for key in ("retired_at", "reason", "evidence") + if not getattr(self, key)] + if missing: + raise LifecycleError(f"retired rule footer lacks {', '.join(missing)}") + if self.reason not in RETIRE_REASONS: + raise LifecycleError( + f"retire reason must be one of {RETIRE_REASONS}: {self.reason}") + if self.reason == "superseded" and not self.superseded_by: + raise LifecycleError("a superseded rule must name superseded_by") + elif self.retired_at or self.superseded_by: + raise LifecycleError("an active rule cannot carry retired_at/superseded_by") + + def render(self) -> str: + parts = [] + for key in FOOTER_KEYS: + value = getattr(self, key) + if key == "protected": + if value: + parts.append("protected=true") + continue + if key == "status": + parts.append(f"status={value}") + continue + if not value: + continue + if re.search(r'[\s"]', value): + if '"' in value: + raise LifecycleError(f"footer value may not contain a quote: {key}") + parts.append(f'{key}="{value}"') + else: + parts.append(f"{key}={value}") + return f"<!-- kb:rule {' '.join(parts)} -->" + + +@dataclass(frozen=True) +class Section: + """One level-two section, kept as its exact source text.""" + + text: str # includes the heading line and its trailing newline(s) + + @property + def heading(self) -> str: + return self.text.split("\n", 1)[0] + + @property + def rule_id(self) -> str: + match = RULE_HEADING.match(self.heading) + return match.group("rule") if match else "" + + @property + def level(self) -> int: + match = RULE_HEADING.match(self.heading) + return len(match.group("level")) if match else 2 + + @property + def is_rule(self) -> bool: + return bool(self.rule_id) + + def _footer_line(self) -> tuple[int, str] | None: + lines = self.text.rstrip("\n").split("\n") + for index in range(len(lines) - 1, 0, -1): + line = lines[index].strip() + if not line: + continue + return (index, line) if FOOTER.match(line) else None + return None + + @property + def footer(self) -> Footer: + found = self._footer_line() + if found is None: + return Footer() + return Footer.parse(FOOTER.match(found[1]).group("attrs")) + + @property + def has_footer(self) -> bool: + return self._footer_line() is not None + + @property + def body_without_footer(self) -> str: + """The rule's words: the section minus its footer, trailing blanks trimmed.""" + found = self._footer_line() + lines = self.text.rstrip("\n").split("\n") + if found is not None: + lines = lines[:found[0]] + return "\n".join(lines).rstrip() + "\n" + + @property + def content_sha256(self) -> str: + return hashlib.sha256(self.body_without_footer.encode("utf-8")).hexdigest() + + @property + def nested_rule_ids(self) -> tuple[str, ...]: + return tuple(m.group("rule") for m in ANY_RULE_HEADING.finditer(self.text)) + + @property + def citations(self) -> tuple[str, ...]: + return tuple(dict.fromkeys(m.group("ref").strip() for m in CITATION.finditer(self.text))) + + def with_footer(self, footer: Footer) -> "Section": + footer.check() + trailing = self.text[len(self.text.rstrip("\n")):] or "\n" + return Section(self.body_without_footer + "\n" + footer.render() + trailing) + + +@dataclass(frozen=True) +class Page: + frontmatter: str # the exact "---...---\n" block, or "" when absent + head: str # text between frontmatter and the first "## " section + sections: tuple[Section, ...] = field(default_factory=tuple) + + @classmethod + def parse(cls, text: str) -> "Page": + match = FRONTMATTER.match(text) + frontmatter = match.group(0) if match else "" + rest = text[len(frontmatter):] + chunks = _split_sections(rest) + head = chunks[0] + sections = tuple(Section(chunk) for chunk in chunks[1:]) + return cls(frontmatter, head, sections) + + def render(self) -> str: + return self.frontmatter + self.head + "".join(s.text for s in self.sections) + + # -- rules --------------------------------------------------------------- + + def rules(self) -> tuple[Section, ...]: + return tuple(s for s in self.sections if s.is_rule) + + def rule(self, rule_id: str) -> Section: + for section in self.sections: + if section.rule_id == rule_id: + return section + raise LifecycleError(f"rule {rule_id} is not on this page") + + def has_rule(self, rule_id: str) -> bool: + return any(section.rule_id == rule_id for section in self.sections) + + def replace_section(self, rule_id: str, new: Section | None) -> "Page": + out = [] + found = False + for section in self.sections: + if section.rule_id == rule_id: + found = True + if new is not None: + out.append(new) + else: + out.append(section) + if not found: + raise LifecycleError(f"rule {rule_id} is not on this page") + return replace(self, sections=tuple(out)) + + def append_section(self, section: Section) -> "Page": + sections = list(self.sections) + if sections and not sections[-1].text.endswith("\n\n"): + last = sections[-1].text + sections[-1] = Section(last.rstrip("\n") + "\n\n") + elif not sections and self.head and not self.head.endswith("\n\n"): + return replace(self, head=self.head.rstrip("\n") + "\n\n", sections=(section,)) + sections.append(section) + return replace(self, sections=tuple(sections)) + + # -- frontmatter --------------------------------------------------------- + + def frontmatter_field(self, key: str) -> str | None: + match = re.search(rf"(?m)^{re.escape(key)}:[ \t]*(?P<value>.*)$", self.frontmatter) + return match.group("value").strip() if match else None + + def with_frontmatter_field(self, key: str, value: str) -> "Page": + if not self.frontmatter: + raise LifecycleError("page has no frontmatter") + pattern = re.compile(rf"(?m)^{re.escape(key)}:[^\r\n]*$") + if not pattern.search(self.frontmatter): + raise LifecycleError(f"frontmatter has no {key} field") + return replace(self, frontmatter=pattern.sub(f"{key}: {value}", self.frontmatter, count=1)) + + def frontmatter_data(self) -> dict: + match = FRONTMATTER.match(self.frontmatter) + if not match: + return {} + try: + data = yaml.safe_load(match.group("body")) or {} + except yaml.YAMLError as exc: + raise LifecycleError(f"frontmatter is not valid YAML: {exc}") from exc + if not isinstance(data, dict): + raise LifecycleError("frontmatter must be a mapping") + return data + + def sources(self) -> list[str]: + value = self.frontmatter_data().get("sources") + if value is None: + return [] + if not isinstance(value, list): + raise LifecycleError("frontmatter sources must be a list") + return [str(item) for item in value] + + def with_sources(self, items: list[str]) -> "Page": + """Rewrite ``sources:`` in the page's own style (flow or block list).""" + match = re.search( + r"(?m)^sources:[ \t]*(?P<flow>\[.*\])?[ \t]*\r?\n(?P<block>(?:[ \t]+-[^\r\n]*\r?\n)*)", + self.frontmatter, + ) + if match is None: + raise LifecycleError("frontmatter has no sources field") + if match.group("flow") is not None or not match.group("block"): + rendered = f"sources: {render_flow_list(items)}\n" + else: + indent = re.match(r"[ \t]+", match.group("block")).group(0) + rendered = "sources:\n" + "".join( + f'{indent}- "{_quote_free(item)}"\n' for item in items) + return replace( + self, + frontmatter=self.frontmatter[:match.start()] + rendered + + self.frontmatter[match.end():], + ) + + +_FENCE = re.compile(r"^\s{0,3}(`{3,}|~{3,})") + + +def _split_sections(text: str) -> list[str]: + """Split at ``## `` headings, and at every ``### `` heading whose enclosing + ``## `` section is not a rule (so a level-three rule ends at its next + sibling); ``###`` inside a ``##`` rule stays nested. Fences never split.""" + chunks: list[str] = [] + current: list[str] = [] + fence: str | None = None + top_is_rule = False + for line in text.splitlines(keepends=True): + match = _FENCE.match(line) + if match: + marker = match.group(1) + if fence is None: + fence = marker[0] * len(marker) + elif marker[0] == fence[0] and len(marker) >= len(fence): + fence = None + split = False + if fence is None and not match: + if line.startswith("## "): + split = True + top_is_rule = bool(_TOP_RULE.match(line)) + elif line.startswith("### ") and not top_is_rule: + # every ### under a topic heading is its own section, so a + # level-three rule ends at its next sibling heading + split = True + if split: + chunks.append("".join(current)) + current = [line] + else: + current.append(line) + chunks.append("".join(current)) + return chunks + + +def parse_flow_list(raw: str) -> list[str]: + """Parse the YAML flow sequence the tree uses for ``sources:``/``tags:``.""" + raw = raw.strip() + if not (raw.startswith("[") and raw.endswith("]")): + raise LifecycleError(f"expected a [flow, list]: {raw!r}") + inner = raw[1:-1].strip() + if not inner: + return [] + items: list[str] = [] + for match in re.finditer(r'\s*(?:"(?P<q>[^"]*)"|(?P<b>[^,]+?))\s*(?:,|$)', inner): + value = match.group("q") if match.group("q") is not None else match.group("b") + if value is not None and value.strip(): + items.append(value.strip()) + return items + + +def _quote_free(item: str) -> str: + if '"' in item: + raise LifecycleError(f"list item may not contain a quote: {item!r}") + return item + + +def render_flow_list(items: Iterable[str]) -> str: + rendered = [] + for item in items: + if re.fullmatch(r"[A-Za-z0-9_./:@+-]+", item): + rendered.append(item) + else: + if '"' in item: + raise LifecycleError(f"list item may not contain a quote: {item!r}") + rendered.append(f'"{item}"') + return "[" + ", ".join(rendered) + "]" + + +def expected_sources(before: list[str], page: Page) -> list[str]: + """Deterministic ``sources:`` after a change: the old list, in order, plus + every citation an ACTIVE rule on the page makes that is not yet listed, in + order of first appearance. Nothing is ever removed (history stays cited).""" + out = list(before) + for section in page.rules(): + if section.footer.status != "active": + continue + for ref in section.citations: + if ref not in out: + out.append(ref) + return out + + +def visible_text(text: str) -> str: + """The page as reviewers and agents should see it: retired rules removed. + + A retired rule keeps its text on disk for one release cycle so the + retirement can be reviewed and reverted, but it must stop being SERVED at + once. Pages without retired rules are returned unchanged, byte for byte. + """ + if "kb:rule" not in text: + return text + page = Page.parse(text) + kept = [] + for section in page.sections: + try: + retired = section.is_rule and section.footer.status == "retired" + except LifecycleError: + retired = False # L1 rejects malformed footers; never hide on doubt + if not retired: + kept.append(section) + if len(kept) == len(page.sections): + return text + return replace(page, sections=tuple(kept)).render() + diff --git a/src/infermatrix_copilot/knowledge_service/ops.py b/src/infermatrix_copilot/knowledge_service/ops.py new file mode 100644 index 00000000..183cde9d --- /dev/null +++ b/src/infermatrix_copilot/knowledge_service/ops.py @@ -0,0 +1,375 @@ +"""Knowledge Ops API 2.0: typed changes to rule pages, with the lifecycle. + +The v1 curation path (``KnowledgeCurator``) can only append a section. The +knowledge service needs every change a maintainer makes by hand, expressed as a +typed operation the quality gate can judge one by one: + +==================== ========================================================= +``add`` append a new rule (optionally on a new ``rules-*.md`` page) +``edit_same_meaning`` reword an active rule under the same ID (gate must confirm + the meaning is unchanged) +``replace`` retire an active rule as ``superseded`` and add its + successor under a NEW ID, atomically +``retire`` retire an active rule (upstream removed it, it was wrong, + or it duplicates another); its text stays for one cycle +``purge`` delete a retired rule's text and tombstone its ID forever +==================== ========================================================= + +``apply_operations`` is a pure function over a mapping of knowledge-relative +paths to text, so the service, the gate and the verifier all compute exactly +the same result. After the operations it writes the MECHANICAL consequences the +verifier later recomputes: ``updated:``, ``sources:`` (append-only), an index +line for every new page, and tombstones for purged IDs. + +Standard library + PyYAML + ``lifecycle`` only: vendored by the verifier bundle. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from pathlib import PurePosixPath +from typing import Iterable, Mapping + +import yaml + +try: # package import; the verifier bundle vendors both modules side by side + from .lifecycle import ( + ANY_RULE_HEADING, RETIRE_REASONS, RULE_HEADING, Footer, LifecycleError, + Page, Section, expected_sources, + ) +except ImportError: # pragma: no cover - exercised by the vendored bundle + from lifecycle import ( # type: ignore[no-redef] + ANY_RULE_HEADING, RETIRE_REASONS, RULE_HEADING, Footer, LifecycleError, + Page, Section, expected_sources, + ) + +KNOWLEDGE_OPS_API_VERSION = "2.0.0" +OP_KINDS = ("add", "edit_same_meaning", "replace", "retire", "purge") +TOMBSTONES_NAME = "_tombstones.yaml" +INDEX_NAME = "_index.md" +_RULE_PAGE = re.compile(r"rules(?:-[a-z0-9][a-z0-9-]*)?\.md") +_DATE = re.compile(r"\d{4}-\d{2}-\d{2}") +_MAX_SECTION_CHARS = 16 * 1024 +# Mirrors knowledge/tools/check_knowledge_tree.py's split gate (>= 32 KiB or +# >= 500 non-empty lines fails the whole tree). +PAGE_MAX_BYTES = 32 * 1024 +PAGE_MAX_LINES = 500 + + +def page_over_capacity(text: str) -> str: + size = len(text.encode("utf-8")) + lines = sum(1 for line in text.splitlines() if line.strip()) + if size >= PAGE_MAX_BYTES or lines >= PAGE_MAX_LINES: + return f"{size} bytes / {lines} non-empty lines (limit < {PAGE_MAX_BYTES} / < {PAGE_MAX_LINES})" + return "" + + +@dataclass(frozen=True) +class KnowledgeOperation: + kind: str + page: str + rule_id: str + section_markdown: str = "" + new_rule_id: str = "" + new_page: str = "" + reason: str = "" + evidence: str = "" + page_title: str = "" + allow_protected: bool = False + + def to_dict(self) -> dict: + return {key: value for key, value in self.__dict__.items() if value not in ("", False)} + + @classmethod + def from_dict(cls, data: Mapping) -> "KnowledgeOperation": + unknown = set(data) - set(cls.__dataclass_fields__) + if unknown: + raise LifecycleError(f"unknown operation fields: {sorted(unknown)}") + return cls(**{key: data[key] for key in data}) + + +@dataclass(frozen=True) +class OperationsResult: + files: dict[str, str] # every created or changed path -> new text + touched_rules: tuple[str, ...] # rule IDs whose sections changed + created_pages: tuple[str, ...] = field(default_factory=tuple) + + +def repo_scope(page: str) -> str: + """``repos/<repo>`` or ``general`` for a knowledge-relative page path.""" + parts = PurePosixPath(page).parts + if len(parts) >= 2 and parts[0] == "repos": + return f"repos/{parts[1]}" + if parts and parts[0] == "general": + return "general" + raise LifecycleError(f"page is outside repos/<repo>/ and general/: {page}") + + +def tombstones_path(page: str) -> str: + return f"{repo_scope(page)}/{TOMBSTONES_NAME}" + + +def load_tombstones(text: str | None) -> list[dict]: + if not text: + return [] + data = yaml.safe_load(text) or {} + if not isinstance(data, dict) or data.get("schema_version") != 1: + raise LifecycleError("tombstones must be schema_version 1") + items = data.get("ids") or [] + if not isinstance(items, list) or any( + not isinstance(item, dict) or not item.get("id") for item in items + ): + raise LifecycleError("tombstones ids must be a list of {id, ...}") + return items + + +def render_tombstones(items: list[dict]) -> str: + lines = [ + "# IDs of purged knowledge rules. An ID listed here is never reused.", + "# Written by the knowledge service; recomputed by the kb-gate verifier.", + "schema_version: 1", + "ids:", + ] + for item in items: + lines.append( + f" - {{id: {item['id']}, purged_at: {item.get('purged_at', '')}, " + f"page: {item.get('page', '')}}}" + ) + return "\n".join(lines) + "\n" + + +def all_rule_ids(files: Mapping[str, str]) -> dict[str, int]: + """Heading IDs at any level across every rule page, with their counts.""" + counts: dict[str, int] = {} + for path, text in files.items(): + if not path.endswith(".md"): + continue + for match in ANY_RULE_HEADING.finditer(_without_fences(text)): + counts[match.group("rule")] = counts.get(match.group("rule"), 0) + 1 + return counts + + +def all_tombstoned_ids(files: Mapping[str, str]) -> set[str]: + out: set[str] = set() + for path, text in files.items(): + if PurePosixPath(path).name == TOMBSTONES_NAME: + out.update(str(item["id"]) for item in load_tombstones(text)) + return out + + +def _without_fences(text: str) -> str: + out, fence = [], None + for line in text.splitlines(keepends=True): + stripped = line.lstrip() + if stripped.startswith(("```", "~~~")): + marker = stripped[:3] + fence = None if fence == marker else (fence or marker) + out.append("\n") + continue + out.append("\n" if fence else line) + return "".join(out) + + +def _check_section(section_markdown: str, rule_id: str, level: int = 2) -> Section: + text = section_markdown.strip("\n") + "\n" + section = Section(text) + marks = "#" * level + if section.rule_id != rule_id or section.level != level: + raise LifecycleError(f"section heading must be '{marks} {rule_id} — <title>'") + if level == 2 and len(re.findall(r"(?m)^## ", text)) != 1: + raise LifecycleError("a rule section has exactly one level-two heading") + if level == 3 and (re.search(r"(?m)^## ", text) or len( + [m for m in re.finditer(r"(?m)^### ", text)]) != 1): + raise LifecycleError("a level-three rule section has exactly one heading and no ## heading") + if section.has_footer: + raise LifecycleError("the service writes lifecycle footers; do not include one") + if not 80 <= len(text) <= _MAX_SECTION_CHARS: + raise LifecycleError("section length is outside 80..16384 characters") + return section + + +def _new_page_text(page: str, title: str, today: str, template: Page | None) -> str: + tags = "[]" + if template is not None and template.frontmatter_field("tags") is not None: + tags = template.frontmatter_field("tags") + if not title or '"' in title: + raise LifecycleError("a new page needs a title without quotes") + return ( + "---\n" + f'title: "{title}"\n' + f"created: {today}\n" + f"updated: {today}\n" + "type: rule\n" + f"tags: {tags}\n" + "sources: []\n" + "---\n\n" + f"# {title}\n\n" + ) + + +def index_line(page: str, title: str) -> str: + return f"- [{title}]({PurePosixPath(page).name})\n" + + +def apply_operations( + files: Mapping[str, str], + operations: Iterable[KnowledgeOperation], + *, + release: str, + today: str, +) -> OperationsResult: + """Apply typed operations to ``files`` (knowledge-relative path -> text). + + Raises ``LifecycleError`` on the first invalid operation; nothing is + partially applied because the input mapping is never mutated. + """ + if not _DATE.fullmatch(today): + raise LifecycleError("today must be YYYY-MM-DD") + if not release or re.search(r'[\s"]', release): + raise LifecycleError("release must be a non-empty token") + work: dict[str, str] = dict(files) + original_ids = all_rule_ids(files) + tombstoned = all_tombstoned_ids(files) + added_ids: set[str] = set() + touched_pages: dict[str, list[str]] = {} # page -> sources before any change + created: list[str] = [] + touched_rules: list[str] = [] + purged: dict[str, list[dict]] = {} + + def page_of(path: str, *, create_title: str = "") -> Page: + if not _RULE_PAGE.fullmatch(PurePosixPath(path).name): + raise LifecycleError(f"not a rule page (rules.md / rules-<topic>.md): {path}") + repo_scope(path) + text = work.get(path) + if text is None: + if not create_title: + raise LifecycleError(f"page does not exist: {path}") + sibling = str(PurePosixPath(path).with_name("rules.md")) + template = Page.parse(work[sibling]) if sibling in work else None + index = str(PurePosixPath(path).with_name(INDEX_NAME)) + if index not in work: + raise LifecycleError(f"a new page's directory needs an {INDEX_NAME}: {index}") + text = _new_page_text(path, create_title, today, template) + work[path] = text + created.append(path) + work[index] = work[index].rstrip("\n") + "\n" + index_line(path, create_title) + page = Page.parse(text) + if page.frontmatter_field("type") != "rule": + raise LifecycleError(f"page is not type: rule: {path}") + touched_pages.setdefault(path, page.sources() if path not in created else []) + return page + + def fresh_id(rule_id: str) -> None: + if rule_id in original_ids or rule_id in added_ids: + raise LifecycleError(f"rule ID already exists in the tree: {rule_id}") + if rule_id in tombstoned: + raise LifecycleError(f"rule ID was purged and is reserved forever: {rule_id}") + added_ids.add(rule_id) + + def active(page: Page, rule_id: str, op: KnowledgeOperation) -> Section: + section = page.rule(rule_id) + footer = section.footer + if footer.status != "active": + raise LifecycleError(f"rule {rule_id} is not active") + if footer.protected and not op.allow_protected: + raise LifecycleError(f"rule {rule_id} is protected; only the human path may change it") + return section + + for op in operations: + if op.kind not in OP_KINDS: + raise LifecycleError(f"unknown operation kind: {op.kind}") + if op.kind == "add": + section = _check_section(op.section_markdown, op.rule_id) + fresh_id(op.rule_id) + for nested in section.nested_rule_ids: + if nested != op.rule_id: + fresh_id(nested) + page = page_of(op.page, create_title=op.page_title) + page = page.append_section(section.with_footer(Footer(status="active", since=release))) + work[op.page] = page.render() + touched_rules.append(op.rule_id) + elif op.kind == "edit_same_meaning": + page = page_of(op.page) + old = active(page, op.rule_id, op) + section = _check_section(op.section_markdown, op.rule_id, old.level) + if set(section.nested_rule_ids) != set(old.nested_rule_ids): + raise LifecycleError("an edit may not add or drop nested rule IDs") + trailing = old.text[len(old.text.rstrip("\n")):] + new = Section(section.text.rstrip("\n") + (trailing or "\n")) + if old.has_footer: + new = new.with_footer(old.footer) + work[op.page] = page.replace_section(op.rule_id, new).render() + touched_rules.append(op.rule_id) + elif op.kind == "replace": + if not op.new_rule_id or op.new_rule_id == op.rule_id: + raise LifecycleError("replace needs a different new_rule_id") + if not op.evidence: + raise LifecycleError("replace needs evidence") + page = page_of(op.page) + old = active(page, op.rule_id, op) + section = _check_section(op.section_markdown, op.new_rule_id) + fresh_id(op.new_rule_id) + for nested in section.nested_rule_ids: + if nested != op.new_rule_id: + fresh_id(nested) + retired = old.with_footer(Footer( + status="retired", since=old.footer.since, retired_at=release, + reason="superseded", evidence=op.evidence, + supersedes=old.footer.supersedes, # keep the chain A -> B -> C + superseded_by=op.new_rule_id, protected=old.footer.protected, + )) + work[op.page] = page.replace_section(op.rule_id, retired).render() + target = op.new_page or op.page + target_page = page_of(target, create_title=op.page_title) + successor = section.with_footer(Footer( + status="active", since=release, supersedes=op.rule_id)) + work[target] = target_page.append_section(successor).render() + touched_rules += [op.rule_id, op.new_rule_id] + elif op.kind == "retire": + if op.reason not in RETIRE_REASONS or op.reason == "superseded": + raise LifecycleError("retire reason must be upstream-removed, incorrect or duplicate; use replace for superseded") + if not op.evidence: + raise LifecycleError("retire needs evidence") + page = page_of(op.page) + old = active(page, op.rule_id, op) + retired = old.with_footer(Footer( + status="retired", since=old.footer.since, retired_at=release, + reason=op.reason, evidence=op.evidence, + supersedes=old.footer.supersedes, protected=old.footer.protected, + )) + work[op.page] = page.replace_section(op.rule_id, retired).render() + touched_rules.append(op.rule_id) + elif op.kind == "purge": + page = page_of(op.page) + old = page.rule(op.rule_id) + if old.footer.status != "retired": + raise LifecycleError(f"only a retired rule can be purged: {op.rule_id}") + if old.footer.protected and not op.allow_protected: + raise LifecycleError(f"rule {op.rule_id} is protected; only the human path may purge it") + if old.footer.retired_at == release: + raise LifecycleError("a rule is purged one full release after it was retired") + work[op.page] = page.replace_section(op.rule_id, None).render() + for rid in dict.fromkeys((op.rule_id, *old.nested_rule_ids)): + purged.setdefault(tombstones_path(op.page), []).append( + {"id": rid, "purged_at": release, "page": op.page}) + touched_rules.append(op.rule_id) + + for path, before_sources in touched_pages.items(): + page = Page.parse(work[path]) + page = page.with_frontmatter_field("updated", today) + page = page.with_sources(expected_sources(before_sources, page)) + work[path] = page.render() + for path, entries in purged.items(): + work[path] = render_tombstones(load_tombstones(work.get(path)) + entries) + for path in touched_pages: + over = page_over_capacity(work[path]) + if over: + raise LifecycleError(f"page full, add to a new rules-<topic>.md page instead: {path}: {over}") + + changed = {path: text for path, text in work.items() if files.get(path) != text} + return OperationsResult( + files=changed, + touched_rules=tuple(dict.fromkeys(touched_rules)), + created_pages=tuple(created), + ) diff --git a/src/infermatrix_copilot/sdk/v1/direct.py b/src/infermatrix_copilot/sdk/v1/direct.py index d14b90a4..e565a64b 100644 --- a/src/infermatrix_copilot/sdk/v1/direct.py +++ b/src/infermatrix_copilot/sdk/v1/direct.py @@ -177,6 +177,17 @@ def _document_path(self, document_id: str, view=None) -> Path: raise DocumentNotFoundError(f"unknown document_id: {document_id!r}") return view.path(value) + @staticmethod + def _served_bytes(path: Path) -> bytes: + """Document bytes as served: retired rules are withheld (their text + stays on disk for one release cycle but is no longer knowledge).""" + from ...knowledge_service.lifecycle import visible_text + + data = path.read_bytes() + if path.suffix != ".md" or b"kb:rule" not in data: + return data + return visible_text(data.decode("utf-8")).encode("utf-8") + def _document_id(self, path_value: str | Path, view=None) -> str: view = view if view is not None else self._view() path = Path(path_value).expanduser().resolve() @@ -190,7 +201,7 @@ def _document_id(self, path_value: str | Path, view=None) -> str: def _document_ref(self, path_value: str | Path, view=None) -> DocumentRef: view = view if view is not None else self._view() document_id = self._document_id(path_value, view) - data = self._document_path(document_id, view).read_bytes() + data = self._served_bytes(self._document_path(document_id, view)) excerpt_data = data[:_EXCERPT_BYTES] return DocumentRef( document_id=document_id, @@ -228,7 +239,7 @@ def read_document( "review_context_id was not issued by this DirectClient" ) view = issued.knowledge_view - data = self._document_path(document_id, view).read_bytes() + data = self._served_bytes(self._document_path(document_id, view)) page = data[offset:offset + max_bytes] next_offset = offset + len(page) if offset + len(page) < len(data) else None return DocumentPage( diff --git a/test/test_knowledge_ops_l1.py b/test/test_knowledge_ops_l1.py new file mode 100644 index 00000000..17c0914c --- /dev/null +++ b/test/test_knowledge_ops_l1.py @@ -0,0 +1,407 @@ +"""Knowledge Ops API 2.0 and the L1 gate, on a small synthetic tree.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from infermatrix_copilot.knowledge_service.l1 import ( + Change, check_changeset, check_tree, +) +from infermatrix_copilot.knowledge_service.lifecycle import ( + Footer, LifecycleError, Page, +) +from infermatrix_copilot.knowledge_service.ops import ( + KnowledgeOperation as Op, apply_operations, load_tombstones, +) + +TODAY = "2026-09-28" +REL = "v0.31.0" +PAGE = "repos/demo/core/rules.md" + + +def _rule(rule_id: str, extra: str = "", cite: str = "PR #10") -> str: + return ( + f"## {rule_id} — keep the demo queue bounded {extra}\n\n" + "- 触发：修改 demo queue 的容量或背压逻辑时。\n" + "- 强制：队列满时拒绝新请求并返回明确错误，不能无界增长。\n" + f"- 验收：测试覆盖满队列拒绝路径。 ^[{cite}]\n" + ) + + +def _tree() -> dict[str, str]: + return { + "repos/demo/_index.md": "# demo\n", + "repos/demo/core/_index.md": "# core\n\n- [rules](rules.md)\n", + PAGE: ( + "---\n" + 'title: "Demo core rules"\n' + "created: 2026-09-01\n" + "updated: 2026-09-01\n" + "type: rule\n" + "tags: [demo]\n" + 'sources: ["PR #10"]\n' + "---\n\n" + "# Demo core rules\n\n" + + _rule("DEMO-1a") + "\n" + _rule("DEMO-1b") + ), + "repos/demo/guide.md": "# guide\n\nSee DEMO-1b for the queue rule.\n", + } + + +def _changes(base: dict, head: dict) -> list[Change]: + out = [] + for path in sorted(set(base) | set(head)): + if base.get(path) == head.get(path): + continue + status = "A" if path not in base else "D" if path not in head else "M" + out.append(Change(f"knowledge/{path}", status, "100644", "" if status == "D" else "100644")) + return out + + +def _apply(base: dict, *ops: Op) -> dict: + result = apply_operations(base, ops, release=REL, today=TODAY) + return {**base, **result.files} + + +def test_parse_render_is_byte_exact_on_the_shipped_tree(): + root = Path(__file__).resolve().parents[1] / "knowledge" + for path in root.rglob("*.md"): + text = path.read_text(encoding="utf-8") + assert Page.parse(text).render() == text, path + + +def test_add_writes_footer_sources_and_updated(): + base = _tree() + head = _apply(base, Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", cite="PR #11"))) + page = Page.parse(head[PAGE]) + assert page.rule("DEMO-2a").footer == Footer(status="active", since=REL) + assert page.sources() == ["PR #10", "PR #11"] + assert page.frontmatter_field("updated") == TODAY + result = check_changeset(base, head, _changes(base, head)) + assert result.ok, result.issues + assert [(b.kind, b.rule_id, b.op) for b in result.blocks] == [("rule", "DEMO-2a", "add")] + + +def test_add_to_new_page_appends_index_line(): + base = _tree() + new = "repos/demo/core/rules-backpressure.md" + head = _apply(base, Op("add", new, "DEMO-3a", _rule("DEMO-3a"), page_title="Demo backpressure")) + assert head["repos/demo/core/_index.md"].endswith("- [Demo backpressure](rules-backpressure.md)\n") + result = check_changeset(base, head, _changes(base, head)) + assert result.ok, result.issues + assert {b.op for b in result.blocks} == {"add"} + + +def test_replace_supersedes_atomically_and_flags_dangling_guide(): + base = _tree() + head = _apply(base, Op("replace", PAGE, "DEMO-1b", _rule("DEMO-1c"), + new_rule_id="DEMO-1c", evidence="PR #12")) + page = Page.parse(head[PAGE]) + assert page.rule("DEMO-1b").footer.superseded_by == "DEMO-1c" + assert page.rule("DEMO-1c").footer.supersedes == "DEMO-1b" + result = check_changeset(base, head, _changes(base, head)) + assert ("dangling_reference", "repos/demo/guide.md") in {(i.code, i.path) for i in result.issues} + head["repos/demo/guide.md"] = "# guide\n\nSee DEMO-1c for the queue rule.\n" + result = check_changeset(base, head, _changes(base, head)) + assert result.ok, result.issues + ops = sorted((b.rule_id, b.op) for b in result.blocks if b.kind == "rule") + assert ops == [("DEMO-1b", "supersede"), ("DEMO-1c", "add")] + assert any(b.kind == "prose" and b.path == "repos/demo/guide.md" for b in result.blocks) + + +def test_retire_then_purge_tombstones_and_reserves_the_id(): + base = _tree() + retired = _apply(base, Op("retire", PAGE, "DEMO-1a", reason="upstream-removed", evidence="PR #13")) + assert check_changeset(base, retired, _changes(base, retired)).ok + with pytest.raises(LifecycleError, match="one full release"): + apply_operations(retired, [Op("purge", PAGE, "DEMO-1a")], release=REL, today=TODAY) + purged = {**retired, **apply_operations( + retired, [Op("purge", PAGE, "DEMO-1a")], release="v0.32.0", today=TODAY).files} + assert [t["id"] for t in load_tombstones(purged["repos/demo/_tombstones.yaml"])] == ["DEMO-1a"] + result = check_changeset(retired, purged, _changes(retired, purged)) + assert result.ok, result.issues + with pytest.raises(LifecycleError, match="reserved forever"): + apply_operations(purged, [Op("add", PAGE, "DEMO-1a", _rule("DEMO-1a"))], release="v0.32.0", today=TODAY) + + +def test_protected_rules_need_the_human_path(): + base = _tree() + page = Page.parse(base[PAGE]) + base[PAGE] = page.replace_section( + "DEMO-1a", page.rule("DEMO-1a").with_footer(Footer(protected=True))).render() + with pytest.raises(LifecycleError, match="protected"): + apply_operations(base, [Op("retire", PAGE, "DEMO-1a", reason="incorrect", evidence="x")], + release=REL, today=TODAY) + head = _apply(base, Op("retire", PAGE, "DEMO-1a", reason="incorrect", evidence="x", allow_protected=True)) + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert "protected_rule_retired" in codes + + +def test_edit_same_meaning_keeps_footer_and_id(): + base = _apply(_tree(), Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a"))) + head = _apply(base, Op("edit_same_meaning", PAGE, "DEMO-2a", _rule("DEMO-2a", extra="(clarified)"))) + assert Page.parse(head[PAGE]).rule("DEMO-2a").footer.since == REL + result = check_changeset(base, head, _changes(base, head)) + assert result.ok and [b.op for b in result.blocks] == ["edit"] + + +@pytest.mark.parametrize("mutate,code", [ + (lambda h: h.__setitem__(PAGE, h[PAGE].replace('sources: ["PR #10"]', 'sources: ["PR #99"]')), "sources_mismatch"), + (lambda h: h.__setitem__(PAGE, h[PAGE].replace("tags: [demo]", "tags: [other]")), "frontmatter_changed"), + (lambda h: h.__setitem__(PAGE, h[PAGE].replace(_rule("DEMO-1a"), "")), "active_rule_removed"), + (lambda h: h.__setitem__(PAGE, h[PAGE] + "\n" + _rule("DEMO-1b")), "duplicate_rule_id"), +]) +def test_forged_changes_are_issues(mutate, code): + base = _tree() + head = dict(base) + mutate(head) + assert code in {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + + +@pytest.mark.parametrize("path,mode,code", [ + ("knowledge/skills/x/run.py", "100644", "path_not_whitelisted"), + ("knowledge/tools/check.py", "100644", "path_not_whitelisted"), + ("knowledge/AGENTS.md", "100644", "path_not_whitelisted"), + ("src/infermatrix_copilot/x.py", "100644", "path_not_whitelisted"), + ("knowledge/repos/demo/core/rules.md", "100755", "mode_not_regular"), + ("knowledge/repos/demo/core/link.md", "120000", "mode_not_regular"), +]) +def test_whitelist_and_modes(path, mode, code): + result = check_changeset({}, {}, [Change(path, "M", "100644", mode)]) + assert code in {i.code for i in result.issues} + + +def test_prose_changes_become_prose_blocks(): + base = _tree() + head = dict(base) + head[PAGE] = head[PAGE].replace("# Demo core rules\n\n", "# Demo core rules\n\nNew intro.\n\n") + head["repos/demo/_routes.yaml"] = "schema_version: 1\nowners: []\n" + result = check_changeset(base, head, _changes(base, head)) + assert result.ok, result.issues + assert {(b.kind, b.path) for b in result.blocks} == {("prose", PAGE), ("prose", "repos/demo/_routes.yaml")} + + +def test_external_references_are_reported_for_companion_prs(): + base = _tree() + head = _apply(base, Op("retire", PAGE, "DEMO-1a", reason="incorrect", evidence="PR #14")) + result = check_changeset(base, head, _changes(base, head), + external_texts={"skills/x/SKILL.md": "apply DEMO-1a first"}) + assert result.external_refs == (("DEMO-1a", "skills/x/SKILL.md"),) + + +def test_shipped_tree_has_no_tree_level_issues(): + root = Path(__file__).resolve().parents[1] / "knowledge" + files = {p.relative_to(root).as_posix(): p.read_text(encoding="utf-8") + for p in root.rglob("*") if p.is_file() and p.suffix in {".md", ".yaml"}} + assert check_tree(files) == [] + + +def _copy_real_tree(tmp_path): + """knowledge/ + doc/knowledge/ in a fresh git repo: the validators follow + links into doc/ and ask git about local/.""" + import shutil + import subprocess + repo = Path(__file__).resolve().parents[1] + dst = tmp_path / "repo" + shutil.copytree(repo / "knowledge", dst / "knowledge", + ignore=shutil.ignore_patterns("__pycache__")) + shutil.copytree(repo / "doc" / "knowledge", dst / "doc" / "knowledge") + subprocess.run(["git", "init", "-q", str(dst)], check=True) + return dst / "knowledge" + + +def _load(root: Path) -> dict[str, str]: + return {p.relative_to(root).as_posix(): p.read_text(encoding="utf-8") + for p in root.rglob("*") if p.is_file() and p.suffix in {".md", ".yaml"} + and "tools" not in p.relative_to(root).parts} + + +def test_real_tree_operations_pass_both_validators(tmp_path): + import subprocess + import sys + + root = _copy_real_tree(tmp_path) + files = _load(root) + page = "repos/vllm-omni/components/scheduler/rules.md" + first = Page.parse(files[page]).rules()[0].rule_id + second = Page.parse(files[page]).rules()[1].rule_id + new_page = "repos/vllm-omni/components/scheduler/rules-kb-test.md" + ops = [ + Op("add", new_page, "SCHEDKB-1a", _rule("SCHEDKB-1a", cite="PR #8201"), page_title="Scheduler KB test"), + Op("retire", page, second, reason="incorrect", evidence="PR #8202"), + Op("replace", page, first, _rule("SCHEDKB-1b", cite="PR #8203"), + new_rule_id="SCHEDKB-1b", new_page=new_page, evidence="PR #8203"), + ] + result = apply_operations(files, ops, release="v0.31.0", today=TODAY) + head = {**files, **result.files} + # active pages still citing the retired IDs are flagged for the change set + # to fix; the point here is that both validators accept the written tree + for path, text in result.files.items(): + (root / path).parent.mkdir(parents=True, exist_ok=True) + (root / path).write_text(text, encoding="utf-8") + changes = _changes(files, head) + check = check_changeset(files, head, changes) + codes = {i.code for i in check.issues} + assert codes <= {"dangling_reference"}, check.issues + for tool in ("check_knowledge_tree.py", "check_wiki_lint.py"): + run = subprocess.run([sys.executable, str(root / "tools" / tool)], + capture_output=True, text=True, cwd=root.parent) + assert run.returncode == 0, run.stdout + run.stderr + + +def test_full_page_is_refused_with_a_split_hint(): + base = _tree() + base[PAGE] = base[PAGE] + "\n" + ("filler line\n" * 3000) + with pytest.raises(LifecycleError, match="page full"): + apply_operations(base, [Op("add", PAGE, "DEMO-9a", _rule("DEMO-9a"))], release=REL, today=TODAY) + + +def test_retired_rules_are_not_served(tmp_path, monkeypatch): + from infermatrix_copilot.adapters.base import render_briefing_docs + from infermatrix_copilot.knowledge_docs import KnowledgeDocs + from infermatrix_copilot.knowledge_service.lifecycle import visible_text + + base = _tree() + head = _apply(base, Op("retire", PAGE, "DEMO-1a", reason="incorrect", evidence="PR #15")) + served = visible_text(head[PAGE]) + assert "DEMO-1a" not in served and "DEMO-1b" in served + assert visible_text(base[PAGE]) == base[PAGE] + root = tmp_path / "k" + for path, text in head.items(): + (root / path).parent.mkdir(parents=True, exist_ok=True) + (root / path).write_text(text, encoding="utf-8") + docs = KnowledgeDocs(root, "repos/demo") + assert "DEMO-1a" not in docs.read(PAGE)["content"] + assert not [h for h in docs.search("DEMO-1a") if h["path"] == PAGE] + assert "DEMO-1a" not in render_briefing_docs(root, [PAGE]) + + +def _nested_rule(rule_id: str, nested: str) -> str: + return _rule(rule_id).rstrip("\n") + f"\n\n### {nested} — nested detail\n\n- 细节：保持同一队列约束。\n" + + +def test_review_regressions_page_type_protection_chain_nested_index(): + # 1. leaving the lifecycle by changing page type + base = _tree() + head = dict(base) + head[PAGE] = base[PAGE].replace("type: rule", "type: guide").replace(_rule("DEMO-1a"), "") + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert {"page_type_changed", "active_rule_removed"} <= codes + # 2. dropping protected=true while retiring + base = _tree() + page = Page.parse(base[PAGE]) + base[PAGE] = page.replace_section("DEMO-1a", page.rule("DEMO-1a").with_footer(Footer(protected=True))).render() + head = dict(base) + page = Page.parse(base[PAGE]) + head[PAGE] = page.replace_section("DEMO-1a", page.rule("DEMO-1a").with_footer(Footer( + status="retired", retired_at=REL, reason="incorrect", evidence="x"))).render() + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert {"protected_rule_retired", "protection_removed"} <= codes + # 3. A -> B -> C keeps B's backlink + base = _tree() + b = _apply(base, Op("replace", PAGE, "DEMO-1a", _rule("DEMO-1c"), new_rule_id="DEMO-1c", evidence="PR #1")) + c = _apply(b, Op("replace", PAGE, "DEMO-1c", _rule("DEMO-1d"), new_rule_id="DEMO-1d", evidence="PR #2")) + assert Page.parse(c[PAGE]).rule("DEMO-1c").footer.supersedes == "DEMO-1a" + assert check_tree(c) == [] + # 4. nested IDs retire and purge with their parent + base = _tree() + base[PAGE] = base[PAGE] + "\n" + _nested_rule("DEMO-5a", "DEMO-5a1") + base["repos/demo/guide.md"] += "Also DEMO-5a1.\n" + retired = _apply(base, Op("retire", PAGE, "DEMO-5a", reason="incorrect", evidence="PR #3")) + result = check_changeset(base, retired, _changes(base, retired)) + assert "DEMO-5a1" in result.retired + assert ("dangling_reference", "repos/demo/guide.md") in {(i.code, i.path) for i in result.issues} + purged = {**retired, **apply_operations(retired, [Op("purge", PAGE, "DEMO-5a")], + release="v0.32.0", today=TODAY).files} + purged["repos/demo/guide.md"] = "# guide\n" + retired["repos/demo/guide.md"] = "# guide\n" + result = check_changeset(retired, purged, _changes(retired, purged)) + assert result.ok, result.issues + assert set(result.purged) == {"DEMO-5a", "DEMO-5a1"} + # 5. a new page without its index line + base = _tree() + head = _apply(base, Op("add", "repos/demo/core/rules-x.md", "DEMO-6a", _rule("DEMO-6a"), page_title="X")) + head["repos/demo/core/_index.md"] = base["repos/demo/core/_index.md"] + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert "index_missing" in codes + + +def test_review_regressions_nested_edit_purge_timing_replace_nested(): + base = _tree() + base[PAGE] = base[PAGE] + "\n" + _nested_rule("DEMO-7a", "DEMO-7a1") + head = dict(base) + head[PAGE] = base[PAGE].replace("\n### DEMO-7a1 — nested detail\n\n- 细节：保持同一队列约束。\n", "\n") + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert "nested_rule_ids_changed" in codes + retired = _apply(_tree(), Op("retire", PAGE, "DEMO-1a", reason="incorrect", evidence="PR #4")) + forged = {**retired, **apply_operations(retired, [Op("purge", PAGE, "DEMO-1a")], + release="v0.32.0", today=TODAY).files} + forged["repos/demo/_tombstones.yaml"] = forged["repos/demo/_tombstones.yaml"].replace("v0.32.0", REL) + codes = {i.code for i in check_changeset(retired, forged, _changes(retired, forged)).issues} + assert "purge_too_early" in codes + ok = {**retired, **apply_operations(retired, [Op("purge", PAGE, "DEMO-1a")], + release="v0.32.0", today=TODAY).files} + assert "purge_release_mismatch" in {i.code for i in check_changeset( + retired, ok, _changes(retired, ok), release="v0.33.0").issues} + assert check_changeset(retired, ok, _changes(retired, ok), release="v0.32.0").ok + with pytest.raises(LifecycleError, match="already exists"): + apply_operations(_tree(), [Op("replace", PAGE, "DEMO-1a", _nested_rule("DEMO-1e", "DEMO-1b"), + new_rule_id="DEMO-1e", evidence="PR #5")], release=REL, today=TODAY) + + +def test_level_three_rules_under_topic_headings_are_rules(tmp_path): + root = Path(__file__).resolve().parents[1] / "knowledge" + page = "repos/vllm-omni/components/serving/rules-engine-lifecycle.md" + files = {page: (root / page).read_text(encoding="utf-8")} + parsed = Page.parse(files[page]) + assert "SERV-5r" in [s.rule_id for s in parsed.rules()] + assert parsed.rule("SERV-5r").level == 3 + result = apply_operations(files, [Op("retire", page, "SERV-5r", reason="incorrect", evidence="PR #9")], + release=REL, today=TODAY) + head = {**files, **result.files} + assert Page.parse(head[page]).rule("SERV-5r").footer.status == "retired" + from infermatrix_copilot.knowledge_service.lifecycle import visible_text + assert "### SERV-5r —" not in visible_text(head[page]) + assert "### SERV-5a —" in visible_text(head[page]) + deleted = dict(files) + deleted[page] = Page.parse(files[page]).replace_section("SERV-5r", None).render() + codes = {i.code for i in check_changeset(files, deleted, _changes(files, deleted)).issues} + assert "active_rule_removed" in codes + + +def test_new_page_link_required_even_when_index_is_prose(): + base = _tree() + head = _apply(base, Op("add", "repos/demo/core/rules-y.md", "DEMO-8a", _rule("DEMO-8a"), page_title="Y")) + head["repos/demo/core/_index.md"] = "# core, rewritten\n" + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert "index_missing" in codes + + +def test_level_three_rule_ends_at_its_sibling_heading(): + text = ( + "---\ntitle: \"t\"\ncreated: 2026-09-01\nupdated: 2026-09-01\ntype: rule\ntags: []\nsources: []\n---\n\n" + "# t\n\n## Topic\n\n" + _rule("DEMO-1a").replace("## ", "### ", 1) + + "\n### General guidance\n\nkeep this visible\n\n## DEMO-2a — top rule here\n\n" + "- body line one.\n\n### DEMO-2a1 — nested stays with its parent\n\n- nested.\n" + ) + page = Page.parse(text) + assert page.render() == text + assert "General guidance" not in page.rule("DEMO-1a").text + assert "DEMO-2a1" in page.rule("DEMO-2a").nested_rule_ids + files = {PAGE: text, "repos/demo/core/_index.md": "# core\n"} + head = apply_operations(files, [Op("retire", PAGE, "DEMO-1a", reason="incorrect", evidence="x")], + release=REL, today=TODAY).files[PAGE] + from infermatrix_copilot.knowledge_service.lifecycle import visible_text + served = visible_text(head) + assert "keep this visible" in served and "### DEMO-1a —" not in served + + +def test_new_metadata_files_need_no_index_line(): + base = _tree() + head = dict(base) + head["repos/demo/_routes.yaml"] = "schema_version: 1\nowners: []\n" + head["repos/demo/_index.md"] = "# demo, retitled\n" + codes = {i.code for i in check_changeset(base, head, _changes(base, head)).issues} + assert "index_missing" not in codes