From 1d94d75f62e9f2adf2924500e5c9e063af928de6 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Mon, 28 Sep 2026 18:29:41 +0800 Subject: [PATCH 1/2] Add typed knowledge operations with a rule lifecycle and the L1 gate The v1 curation path can only append a rule, so nothing could ever retire, supersede or delete one, and the SDK never wrote `sources:` or index lines. This adds the Knowledge Ops API 2.0 the knowledge service and its quality gate build on. The v1 KnowledgeCurator (API 1.1.0) is unchanged. - lifecycle: rule pages parse into frontmatter, head and level-two sections (fence-aware) and render back byte for byte (checked over the whole tree). A rule may end with a `` footer: status, since, retired_at, reason, evidence, supersedes, superseded_by, protected. - ops: apply_operations() is pure. add (also to a new rules-.md page plus its index line), edit_same_meaning, replace (retire as superseded and add the successor atomically), retire, purge (retired for a full release; the ID goes to _tombstones.yaml and is reserved forever). It writes updated:, append-only sources:, index lines and tombstones, refuses protected rules without the human path, and refuses a page past the split line. - l1: check_tree() and check_changeset(). The latter enforces the path whitelist and 100644 mode, no new duplicate or reused IDs, no dangling references to retired or purged rules, and no new tree issues. It recomputes every mechanical change and splits the rest into rule and prose blocks with stable block IDs for per-block L2 verdicts. It also reports references outside knowledge/ for companion PRs. - Retired rules stop being served at once: Direct document reads, doc_read, doc_search and briefings drop them; pages without retired rules are served byte for byte as before. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/adapters/base.md | 5 +- doc/architecture/SPEC/knowledge_docs.md | 3 + doc/architecture/SPEC/knowledge_service/l1.md | 34 ++ .../SPEC/knowledge_service/lifecycle.md | 24 + .../SPEC/knowledge_service/ops.md | 27 + doc/architecture/SPEC/sdk.md | 3 + doc/knowledge/SCHEMA.md | 27 + src/infermatrix_copilot/adapters/base.py | 4 +- src/infermatrix_copilot/knowledge_docs.py | 8 +- .../knowledge_service/l1.py | 493 ++++++++++++++++++ .../knowledge_service/lifecycle.py | 423 +++++++++++++++ .../knowledge_service/ops.py | 375 +++++++++++++ src/infermatrix_copilot/sdk/v1/direct.py | 15 +- test/test_knowledge_ops_l1.py | 407 +++++++++++++++ 14 files changed, 1841 insertions(+), 7 deletions(-) create mode 100644 doc/architecture/SPEC/knowledge_service/l1.md create mode 100644 doc/architecture/SPEC/knowledge_service/lifecycle.md create mode 100644 doc/architecture/SPEC/knowledge_service/ops.md create mode 100644 src/infermatrix_copilot/knowledge_service/l1.py create mode 100644 src/infermatrix_copilot/knowledge_service/lifecycle.py create mode 100644 src/infermatrix_copilot/knowledge_service/ops.py create mode 100644 test/test_knowledge_ops_l1.py diff --git a/doc/architecture/SPEC/adapters/base.md b/doc/architecture/SPEC/adapters/base.md index 98d7c80eb..b50ec8ef7 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 509e50078..2fdc7af54 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/l1.md b/doc/architecture/SPEC/knowledge_service/l1.md new file mode 100644 index 000000000..630001db4 --- /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 000000000..fcc9996cb --- /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 000000000..78ed3dcc1 --- /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 15feb6c86..66c6802bb 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 c12538752..50758fa31 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 ffb97dd96..e785b4748 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 9234dc108..5a05f178f 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 000000000..53ea48cb8 --- /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 000000000..242d94105 --- /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 000000000..183cde9d5 --- /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 d14b90a4f..e565a64bc 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 000000000..17c0914c1 --- /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 From ced168f5c67adf0a670f6da1184740d271115a60 Mon Sep 17 00:00:00 2001 From: tzhouam <tzhouam@connect.ust.hk> Date: Mon, 28 Sep 2026 18:45:41 +0800 Subject: [PATCH 2/2] Describe Knowledge Ops API 2.0 in the knowledge_service SPEC The package page was not re-verified when lifecycle, ops and l1 were added, so the strict SPEC freshness gate failed in CI. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --- doc/architecture/SPEC/knowledge_service.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index 9df7c42c2..f2723dd02 100644 --- a/doc/architecture/SPEC/knowledge_service.md +++ b/doc/architecture/SPEC/knowledge_service.md @@ -1,6 +1,6 @@ # knowledge_service/ — provider curation components -<!-- verified-against: 2026-09-26 --> +<!-- verified-against: 2026-09-28 --> 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.