diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 3513957c..43de1b98 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -1,6 +1,6 @@ # kb_service/ —— 规范 - + `知识服务核心：仓库配置、账本、outbox、CLI · refactor-status: new` @@ -291,3 +291,10 @@ diff 时保留原摘录（证据不会因此变得更"好过"）。只扩展带 `merged_pr` 事件的证据新增 `merge_commit_sha`（旧事件经 `pull()` 查询）。`run_gate(evidence_for=...)` 由 `gate_and_stage` 与事实证明共用同一个观测者（一次镜像同步）。首个线上运行中"所给 diff 在…之前截断"是 L2 判 unsure/fail 的主要原因， 而整批 10 个 PR 的证据（约 105 KB）对每条规则都重复发送；按规则取证据后约 11 KB + 所引 PR 的完整 diff。 + +## 2026-09-30 adapter 生命周期开关检查（kb init） +`lifecycle_flip.check_lifecycle_flip(base_yaml, head_yaml, allowed=)`：两版 manifest 解析后删去允许的键路径， +其余必须完全相等（比较解析值，注释与格式可保留）；`FLIP_KEYS` = `knowledge_lifecycle.enabled/mode`， +`CALIBRATION_KEYS` = `knowledge_lifecycle.calibration_set`。`check_flip_to_shadow` 另要求 head 为 +`enabled: true` 且 `mode` 为 shadow（缺省即 shadow）。知识 L1 的白名单只含 `knowledge/`，adapter 这一处改动由它单独校验； +调用方还需在 head adapter 上跑 `config.parse_lifecycle`。测试：`test_kb_lifecycle_flip.py`。 diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index e33bc2f9..bf4c3261 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 @@ -72,3 +72,9 @@ merge group 逐段验证、暂停清单与 `human-approved` 审批核对已于 2 `recheck(upstream, facts, observer)` 由发布器调用：仓库一致、SHA 为 40 位、每条事实重新观测后逐字段相等， 签名了不成立的必需事实或格式错误均为问题；上游不可读（未知 SHA、网络、API 错误）抛 `FactsError`，由调用方重试。 仅标准库。`gate_verifier.verify_change` 不再拒绝带事实的判定（由发布器的 `recheck` 复核）。 + +## 2026-09-30 钉点声明（pinned_claims） +`pinned_claims` 供 `kb init` 使用：`PinnedObserver` 在本地克隆上以固定钉点实现 `facts.Observer`（另有 +`is_ancestor`），`check_rules` 在钉点上观测规则声明且要求引用的 PR 已合并进钉点历史，`Evidence` / +`check_evidence` 把行区间绑定到内容哈希。服务与发布器的 `facts` 路径不变。详见 +[pinned_claims.md](knowledge_service/pinned_claims.md)。 diff --git a/doc/architecture/SPEC/knowledge_service/pinned_claims.md b/doc/architecture/SPEC/knowledge_service/pinned_claims.md new file mode 100644 index 00000000..7b6c4ac9 --- /dev/null +++ b/doc/architecture/SPEC/knowledge_service/pinned_claims.md @@ -0,0 +1,37 @@ +# knowledge_service/pinned_claims.py —— 规范 + + + +`LOC ~200 · 在钉住的上游 SHA 上核验规则声明与代码证据（kb init §9.1） · refactor-status: new` + +## 职责 +- `PinnedObserver(repo_dir, repository, pin, pull=)`：在本地 git 仓库（bare 或非 bare）上实现 + `facts.Observer`，`head()` 固定返回钉住的 SHA；`top_level` / `path_exists` / `file_text` 读 git + 对象，`pull(n)` 默认走 `gh api repos/{repo}/pulls/{n}`（可注入；任何查询失败——没有 `gh`、网络错误、非对象回答——都归一为 `FactsError`，原有的 `FactsError` 原样抛出）。另加协议之外的 + `is_ancestor(sha)`：`sha` 是钉点或其祖先（非 40 位 SHA、镜像中不存在的对象、非 commit 对象为 False；候选对象损坏（`cat-file --batch-check` 同样输出 missing 但 stderr 报解包错误），或 git 读不到钉点历史（`merge-base` 此时也以 1 退出、只在 stderr 报 `Could not read`）时抛 `FactsError`，不把读失败当成"合入在钉点之后"）。钉点必须是 40 位 + 且在仓库中存在的提交，否则构造即抛 `FactsError`。 +- `check_rules(rule_texts, observer)`：对每条（活跃）规则用 `facts.claims_in` 抽取声明，按 + `facts.observe` 在钉点观测；每条声明必须成立，`^[PR #N]` 还必须已合并进钉点历史 + （`is_ancestor(merge_commit_sha)`）。返回以规则 ID 开头的问题列表。 +- `Evidence(path, start, end, sha256)`、`evidence_for(observer, path, start, end)`、 + `check_evidence(entries, observer)`：把钉点上某文件的行区间（1 起、闭区间）绑定到内容哈希， + 并能复核：文件存在、区间在文件内、哈希一致。 + +## 公开契约 +哈希口径：`str.splitlines()` 切行（`\n`、`\r\n`、缺末尾换行视为相同），取 `start..end` 行以 +`\n` 连接再补一个 `\n`，UTF-8 编码后 SHA-256。`Evidence.to_dict()/from_dict()` 用于 init 记录。 + +## 不变量 +- 与服务端 `facts.attest` 的声明抽取完全一致（同一 `claims_in`），只是观测点从上游 head 换成钉点。 +- 不修改 `facts.Observer` 协议，不影响服务与发布器。 +- 上游不可读（git 失败、未知提交、PR 查询失败）抛 `FactsError`，不当作"声明不成立"。 + +## 边界 —— 不属于这里 +不写任何文件、不调用模型、不决定 init 各阶段如何处理问题（阻断或进清单由调用方决定）。 + +## 依赖（允许） +标准库 + `facts`；外部命令仅 `git` 与（默认 `pull`）`gh`。 + +## 测试 +`test_kb_pinned_claims.py`（临时 git 仓库：钉点前后文件、符号、PR 合并先后与未合并、证据区间与 +哈希、换行风格无关、非法钉点）。 diff --git a/src/infermatrix_copilot/kb_service/lifecycle_flip.py b/src/infermatrix_copilot/kb_service/lifecycle_flip.py new file mode 100644 index 00000000..2da7463a --- /dev/null +++ b/src/infermatrix_copilot/kb_service/lifecycle_flip.py @@ -0,0 +1,125 @@ +"""Check that an adapter manifest edit changes only named lifecycle keys. + +``kb init`` touches one file outside ``knowledge/``: the adapter manifest, and +only to switch the repository's knowledge service on (``enabled`` + ``mode``) +or to name its calibration set. The knowledge L1 never sees that file (its +whitelist is ``knowledge/``), so this check validates it on its own: both +documents are parsed, the allowed key paths are removed from both, and what is +left must be equal. Formatting and comments are free to stay as they are; the +comparison is on parsed values. + +The caller additionally runs ``config.parse_lifecycle`` on the head adapter +(it needs an adapter object, not text) so the new values are also valid. +""" + +from __future__ import annotations + +from typing import Any, Iterable + +import yaml + +LIFECYCLE = "knowledge_lifecycle" +FLIP_KEYS: frozenset[tuple[str, ...]] = frozenset({(LIFECYCLE, "enabled"), (LIFECYCLE, "mode")}) +CALIBRATION_KEYS: frozenset[tuple[str, ...]] = frozenset({(LIFECYCLE, "calibration_set")}) + + +def _load(text: str, which: str) -> tuple[Any, list[str]]: + try: + data = yaml.safe_load(text) + except yaml.YAMLError as exc: + return None, [f"{which} manifest is not valid YAML: {exc}"] + if not isinstance(data, dict): + return None, [f"{which} manifest is not a mapping"] + return data, [] + + +def _copy(value: Any, active: frozenset[int] = frozenset()) -> Any: + """A tree copy with YAML aliases UNshared: ``deepcopy`` keeps ``*alias`` + nodes shared, so removing an allowed key through one path would also hide + a change made through another. A recursive alias is refused.""" + if isinstance(value, (dict, list)): + if id(value) in active: + raise ValueError("manifest contains a recursive alias") + active = active | {id(value)} + if isinstance(value, dict): + return {key: _copy(item, active) for key, item in value.items()} + return [_copy(item, active) for item in value] + return value + + +def _without(data: dict, paths: Iterable[tuple[str, ...]]) -> dict: + out = _copy(data) + for path in paths: + node: Any = out + for key in path[:-1]: + node = node.get(key) if isinstance(node, dict) else None + if node is None: + break + if isinstance(node, dict): + node.pop(path[-1], None) + return out + + +def _typed(mapping: dict) -> dict[tuple[str, Any], Any]: + # 1, 1.0 and True are equal (and collide) as Python dict keys; YAML keeps them apart + return {(type(key).__name__, key): value for key, value in mapping.items()} + + +def _diff_keys(base: Any, head: Any, prefix: str = "") -> list[str]: + """Paths where ``base`` and ``head`` differ, type-sensitively: YAML ``1``, + ``1.0`` and ``true`` are different values even though Python compares + them equal (and ``parse_lifecycle`` would stringify them differently).""" + where = prefix or "" + if isinstance(base, dict) and isinstance(head, dict): + typed_base, typed_head = _typed(base), _typed(head) + out = [] + for key in sorted(set(typed_base) | set(typed_head), key=str): + child = f"{prefix}.{key[1]}" if prefix else str(key[1]) + if key not in typed_base or key not in typed_head: + out.append(child) + else: + out.extend(_diff_keys(typed_base[key], typed_head[key], child)) + return out + if isinstance(base, list) and isinstance(head, list): + if len(base) != len(head): + return [where] + out = [] + for index, (left, right) in enumerate(zip(base, head)): + out.extend(_diff_keys(left, right, f"{where}[{index}]")) + return out + same = type(base) is type(head) and (base == head or (base != base and head != head)) # .nan + return [] if same else [where] + + +def check_lifecycle_flip(base_yaml: str, head_yaml: str, *, + allowed: Iterable[tuple[str, ...]]) -> list[str]: + """Problems when the manifest changed anything besides the ``allowed`` + key paths (empty list: the edit is confined to them).""" + allowed = list(allowed) + base, problems = _load(base_yaml, "base") + head, more = _load(head_yaml, "head") + problems += more + if problems: + return problems + try: + changed = _diff_keys(_without(base, allowed), _without(head, allowed)) + except ValueError as exc: + return [str(exc)] + return [f"manifest key {key} changed; only " + f"{', '.join('.'.join(p) for p in sorted(allowed))} may change" for key in changed] + + +def check_flip_to_shadow(base_yaml: str, head_yaml: str) -> list[str]: + """The init lifecycle flip: only ``enabled``/``mode`` change, and the head + turns the service on in shadow mode.""" + problems = check_lifecycle_flip(base_yaml, head_yaml, allowed=FLIP_KEYS) + if problems: + return problems + section = (yaml.safe_load(head_yaml) or {}).get(LIFECYCLE) + if not isinstance(section, dict): + return [f"head manifest has no {LIFECYCLE} mapping"] + if section.get("enabled") is not True: + problems.append(f"{LIFECYCLE}.enabled must be true") + if section.get("mode", "shadow") != "shadow": # parse_lifecycle's default + problems.append(f"{LIFECYCLE}.mode must be shadow") + return problems diff --git a/src/infermatrix_copilot/knowledge_service/pinned_claims.py b/src/infermatrix_copilot/knowledge_service/pinned_claims.py new file mode 100644 index 00000000..4842bc2f --- /dev/null +++ b/src/infermatrix_copilot/knowledge_service/pinned_claims.py @@ -0,0 +1,248 @@ +"""Rule claims and code evidence checked at ONE pinned upstream SHA. + +``facts.attest`` observes the upstream's current head: right for the service, +which gates a change against the upstream as it is now. A bootstrap (``kb +init``) writes a whole knowledge base against one pinned baseline instead, and +must prove every rule it writes holds *at that pin* — including that a cited PR +was merged before it. This module does that from a local clone the caller +already has checked out: + +* ``PinnedObserver`` implements ``facts.Observer`` over a local git repository + (bare or not) with ``head()`` fixed to the pin, plus ``is_ancestor`` (which + the service's Observer does not need and does not have); +* ``check_rules`` extracts each rule's claims exactly as the service does + (``facts.claims_in``) and observes them at the pin; +* ``Evidence`` / ``evidence_for`` / ``check_evidence`` bind a line range of an + upstream file at the pin to a content hash, so the range can be re-verified. + +Evidence hash: the file text is split with ``str.splitlines()`` (so ``\\n``, +``\\r\\n`` and a missing final newline all count the same), lines ``start`` to +``end`` (1-based, inclusive) are joined with ``\\n`` plus one trailing ``\\n``, +and the UTF-8 bytes are hashed with SHA-256. + +Standard library only (plus ``facts``). Nothing here writes anywhere. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, Iterable, Mapping + +from .facts import FactsError, claims_in, describe, holds, merge_claims, observe + +_SHA = re.compile(r"[0-9a-f]{40}") + + +def _gh_pull(repository: str, number: int) -> dict: + proc = subprocess.run(["gh", "api", f"repos/{repository}/pulls/{number}"], + capture_output=True, text=True, check=False) + if proc.returncode != 0: + raise FactsError(f"{repository} PR #{number}: {proc.stderr.strip()[:300]}") + try: + data = json.loads(proc.stdout) + except ValueError as exc: + raise FactsError(f"{repository} PR #{number}: unexpected answer") from exc + if not isinstance(data, dict): + raise FactsError(f"{repository} PR #{number}: unexpected answer") + return data + + +class PinnedObserver: + """``facts.Observer`` over a local clone, answering for one pinned commit.""" + + def __init__(self, repo_dir: str | Path, repository: str, pin: str, *, + pull: Callable[[str, int], dict] | None = None): + self.repo_dir = Path(repo_dir) + self.repository = repository + if not _SHA.fullmatch(pin or ""): + raise FactsError(f"pin must be a 40-hex commit SHA, got {pin!r}") + self.pin = pin + self._pull = pull or _gh_pull + self._commits: set[str] = set() + self._require_commit(pin) + + def _git(self, *args: str, input: bytes | None = None) -> subprocess.CompletedProcess: + try: + return subprocess.run(["git", "-C", str(self.repo_dir), *args], + capture_output=True, check=False, input=input) + except OSError as exc: + raise FactsError(f"git unavailable: {exc}") from exc + + def _require_commit(self, sha: str) -> None: + if sha in self._commits: + return + proc = self._git("cat-file", "-t", sha) + if proc.returncode != 0 or proc.stdout.decode().strip() != "commit": + raise FactsError(f"{sha[:12]} is not a commit in {self.repo_dir}") + self._commits.add(sha) + + def head(self) -> str: + return self.pin + + def top_level(self, sha: str) -> set[str]: + self._require_commit(sha) + proc = self._git("ls-tree", "--name-only", sha) + if proc.returncode != 0: + raise FactsError(f"cannot list {sha[:12]}: {proc.stderr.decode()[:300]}") + return {line for line in proc.stdout.decode("utf-8", "replace").splitlines() if line} + + def _entry(self, sha: str, path: str) -> tuple[str, str] | None: + """``(type, object id)`` of ``path`` in ``sha``'s tree; None when the + tree has no such entry. Any git failure (unreadable tree, failed lazy + fetch) raises: it is not evidence that the path is absent.""" + self._require_commit(sha) + path = path.strip("/") + if not path: + return None + proc = self._git("ls-tree", "-z", sha, "--", path) + if proc.returncode != 0: + raise FactsError(f"cannot list {path} at {sha[:12]}: {proc.stderr.decode()[:300]}") + for record in proc.stdout.decode("utf-8", "replace").split("\0"): + meta, _, name = record.partition("\t") + parts = meta.split() + if name == path and len(parts) == 3: + return parts[1], parts[2] + return None + + def path_exists(self, sha: str, path: str) -> bool: + return self._entry(sha, path) is not None + + def file_text(self, sha: str, path: str) -> str | None: + """The file's text at ``sha``; None unless ``path`` is a file (blob). + An entry that exists but cannot be read raises ``FactsError``.""" + entry = self._entry(sha, path) + if entry is None or entry[0] != "blob": + return None # missing, or a directory / submodule: no file text + proc = self._git("cat-file", "blob", entry[1]) + if proc.returncode != 0: + raise FactsError(f"cannot read {path} at {sha[:12]}: {proc.stderr.decode()[:300]}") + return proc.stdout.decode("utf-8", "replace") + + def pull(self, number: int) -> dict: + """The PR as GitHub reports it. Every lookup failure (no ``gh``, a + network error, a malformed answer) is a ``FactsError``: the upstream + could not be read, so the caller retries instead of crashing.""" + try: + data = self._pull(self.repository, number) + except FactsError: + raise + except Exception as exc: # noqa: BLE001 - normalise any lookup failure + raise FactsError(f"{self.repository} PR #{number}: {type(exc).__name__}: {exc}"[:400]) from exc + if not isinstance(data, dict): + raise FactsError(f"{self.repository} PR #{number}: unexpected answer") + return data + + def is_ancestor(self, sha: str) -> bool: + """``sha`` is the pin or one of its ancestors (an unknown SHA is not). + Raises ``FactsError`` when git cannot answer, so a read failure is + retried rather than reported as a PR merged after the pin. git exits + 1 both for "not an ancestor" and when it could not read the history + walk (it then only prints ``error: Could not read``), so a 1 counts + as "no" only with a silent stderr.""" + if not _SHA.fullmatch(sha or ""): + return False + # --batch-check says " missing" with a silent stderr only when the + # object is absent; a corrupt object ALSO prints "missing" but reports + # the unpack error on stderr, and that is a read failure, not absence + check = self._git("cat-file", "--batch-check", input=f"{sha}\n".encode()) + answer = check.stdout.decode(errors="replace").split() + if check.returncode != 0 or check.stderr.strip(): + raise FactsError(f"cannot read {sha[:12]}: {check.stderr.decode(errors='replace')[:300]}") + if answer[1:2] == ["missing"]: + return False # a commit this mirror does not have: unknown, so not an ancestor + if answer[1:2] != ["commit"]: + return False # a blob/tree/tag id is not a merge commit + proc = self._git("merge-base", "--is-ancestor", sha, self.pin) + if proc.returncode == 0: + return True + if proc.returncode == 1 and not proc.stderr.strip(): + return False + raise FactsError(f"cannot decide whether {sha[:12]} is an ancestor of {self.pin[:12]}: " + f"{proc.stderr.decode(errors='replace')[:300]}") + + +def check_rules(rule_texts: Mapping[str, str], observer: PinnedObserver) -> list[str]: + """Problems with the claims of active rules (rule ID -> rule text) at the + pin: every claim must hold, and a cited PR must have been merged into the + pin's history. Raises ``FactsError`` when the upstream cannot be read.""" + pin = observer.head() + top = observer.top_level(pin) + problems: list[str] = [] + seen: dict[tuple, dict] = {} + for rule_id, text in rule_texts.items(): + for claim in merge_claims(claims_in(text, top, active=True)): + fact = seen.get(claim.key) + if fact is None: + fact = seen[claim.key] = observe(claim, observer, pin) + if not holds(fact): + problems.append(f"{rule_id}: {describe(fact)}") + elif claim.kind == "pr" and not observer.is_ancestor(fact.get("merge_commit_sha", "")): + problems.append(f"{rule_id}: {observer.repository} PR #{claim.pr} was merged " + f"after {pin[:12]} (not in the pinned history)") + return problems + + +@dataclass(frozen=True) +class Evidence: + """Lines ``start``..``end`` (1-based, inclusive) of ``path`` at the pin.""" + path: str + start: int + end: int + sha256: str + + def to_dict(self) -> dict: + return {"path": self.path, "start": self.start, "end": self.end, "sha256": self.sha256} + + @classmethod + def from_dict(cls, data: Mapping) -> "Evidence": + return cls(str(data["path"]), int(data["start"]), int(data["end"]), str(data["sha256"])) + + +def _range_text(text: str, start: int, end: int) -> str | None: + lines = text.splitlines() + if not 1 <= start <= end <= len(lines): + return None + return "\n".join(lines[start - 1:end]) + "\n" + + +def _digest(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def evidence_for(observer: PinnedObserver, path: str, start: int, end: int) -> Evidence: + """Bind a line range at the pin to its hash (raises ``FactsError`` when the + file is missing or the range does not fit).""" + text = observer.file_text(observer.head(), path) + if text is None: + raise FactsError(f"{path} does not exist at {observer.head()[:12]}") + chunk = _range_text(text, start, end) + if chunk is None: + raise FactsError(f"{path}:L{start}-L{end} is outside the file at {observer.head()[:12]}") + return Evidence(path, start, end, _digest(chunk)) + + +def check_evidence(entries: Iterable[Evidence], observer: PinnedObserver) -> list[str]: + """Problems with recorded evidence at the pin: the file must exist, the + range must fit, and the text in it must hash to the recorded value.""" + pin = observer.head() + problems: list[str] = [] + texts: dict[str, str | None] = {} + for entry in entries: + where = f"{entry.path}:L{entry.start}-L{entry.end}" + if entry.path not in texts: + texts[entry.path] = observer.file_text(pin, entry.path) + text = texts[entry.path] + if text is None: + problems.append(f"{where}: file does not exist at {pin[:12]}") + continue + chunk = _range_text(text, entry.start, entry.end) + if chunk is None: + problems.append(f"{where}: range is outside the file at {pin[:12]}") + elif _digest(chunk) != entry.sha256: + problems.append(f"{where}: content differs from the recorded evidence at {pin[:12]}") + return problems diff --git a/test/test_kb_lifecycle_flip.py b/test/test_kb_lifecycle_flip.py new file mode 100644 index 00000000..75f93464 --- /dev/null +++ b/test/test_kb_lifecycle_flip.py @@ -0,0 +1,82 @@ +"""An adapter manifest edit may change only the named lifecycle keys (kb init §9.3).""" + +from __future__ import annotations + +from infermatrix_copilot.kb_service.lifecycle_flip import ( + CALIBRATION_KEYS, FLIP_KEYS, check_flip_to_shadow, check_lifecycle_flip, +) + +BASE = """\ +repo: + full_name: o/r # the upstream +knowledge_lifecycle: + enabled: false + mode: shadow + knowledge_dir: repos/r + release: {trigger: none} +""" + + +def test_flip_to_shadow_is_accepted_with_comments_kept(): + head = BASE.replace("enabled: false", "enabled: true # kb init PR 3") + assert check_flip_to_shadow(BASE, head) == [] + + +def test_mode_may_be_left_to_its_default(): + base = BASE.replace(" mode: shadow\n", "") + assert check_flip_to_shadow(base, base.replace("enabled: false", "enabled: true")) == [] + + +def test_flip_must_turn_on_shadow(): + assert check_flip_to_shadow(BASE, BASE) == ["knowledge_lifecycle.enabled must be true"] + head = BASE.replace("enabled: false", "enabled: true").replace("mode: shadow", "mode: auto_merge") + assert check_flip_to_shadow(BASE, head) == ["knowledge_lifecycle.mode must be shadow"] + + +def test_any_other_change_is_refused(): + head = BASE.replace("enabled: false", "enabled: true").replace("trigger: none", "trigger: github_release") + problems = check_flip_to_shadow(BASE, head) + assert problems and "knowledge_lifecycle.release.trigger" in problems[0] + head = BASE.replace("enabled: false", "enabled: true") + "push: {allowed: true}\n" + assert "manifest key push changed" in check_flip_to_shadow(BASE, head)[0] + head = BASE.replace("full_name: o/r", "full_name: o/other") + assert "repo.full_name" in check_lifecycle_flip(BASE, head, allowed=FLIP_KEYS)[0] + + +def test_calibration_set_edit(): + head = BASE.replace(" knowledge_dir", " calibration_set: kb-calibration\n knowledge_dir") + assert check_lifecycle_flip(BASE, head, allowed=CALIBRATION_KEYS) == [] + assert check_lifecycle_flip(BASE, head, allowed=FLIP_KEYS) + + +def test_unparseable_manifests_are_problems(): + assert "not valid YAML" in check_lifecycle_flip(BASE, "a: [", allowed=FLIP_KEYS)[0] + assert "not a mapping" in check_lifecycle_flip("- 1\n", BASE, allowed=FLIP_KEYS)[0] + + +def test_yaml_aliases_cannot_hide_a_change(): + base = BASE + "other: &x {enabled: false}\nmirror: *x\n" + head = base.replace("enabled: false}", "enabled: true}") + problems = check_lifecycle_flip(base, head, allowed=FLIP_KEYS) + assert any("other.enabled" in p for p in problems) + shared = "knowledge_lifecycle: &lc {enabled: false, mode: shadow}\ncopy: *lc\n" + flipped = "knowledge_lifecycle: &lc {enabled: true, mode: shadow}\ncopy: *lc\n" + assert any("copy.enabled" in p for p in check_flip_to_shadow(shared, flipped)) + + +def test_value_types_are_compared_strictly(): + base = BASE.replace(" knowledge_dir", " calibration_set: 1\n knowledge_dir") + head = base.replace("enabled: false", "enabled: true").replace("calibration_set: 1", "calibration_set: true") + assert "knowledge_lifecycle.calibration_set" in check_flip_to_shadow(base, head)[0] + lists = "xs: [1, 2]\nys: {a: 1}\n" + assert "xs[1]" in check_lifecycle_flip(lists, "xs: [1, 2.0]\nys: {a: 1}\n", allowed=FLIP_KEYS)[0] + assert "xs" in check_lifecycle_flip(lists, "xs: [1]\nys: {a: 1}\n", allowed=FLIP_KEYS)[0] + keyed = "m: {1: a}\n" + assert check_lifecycle_flip(keyed, "m: {true: a}\n", allowed=FLIP_KEYS) + unchanged = "x: .nan\nxs: [1, {k: [true]}]\n" + assert check_lifecycle_flip(unchanged, unchanged, allowed=FLIP_KEYS) == [] + + +def test_recursive_alias_is_a_problem(): + loop = "a: &a [*a]\n" + assert "recursive alias" in check_lifecycle_flip(loop, loop, allowed=FLIP_KEYS)[0] diff --git a/test/test_kb_pinned_claims.py b/test/test_kb_pinned_claims.py new file mode 100644 index 00000000..247f32d0 --- /dev/null +++ b/test/test_kb_pinned_claims.py @@ -0,0 +1,208 @@ +"""Rule claims and code evidence checked at a pinned upstream SHA (kb init §9.1).""" + +from __future__ import annotations + +import subprocess + +import pytest + +from infermatrix_copilot.knowledge_service.facts import FactsError +from infermatrix_copilot.knowledge_service.pinned_claims import ( + Evidence, PinnedObserver, check_evidence, check_rules, evidence_for, +) + + +def _git(repo, *args) -> str: + return subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True, + text=True).stdout.strip() + + +def _commit(repo, files: dict[str, str], message: str) -> str: + for rel, text in files.items(): + path = repo / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + _git(repo, "add", "-A") + _git(repo, "-c", "user.name=t", "-c", "user.email=t@example.com", "commit", "-q", "-m", message) + return _git(repo, "rev-parse", "HEAD") + + +@pytest.fixture() +def upstream(tmp_path): + repo = tmp_path / "up" + repo.mkdir() + _git(repo, "init", "-q") + first = _commit(repo, {"pkg/core.py": "class Engine:\n def step(self):\n return 1\n", + "README.md": "hello\n"}, "first") + pin = _commit(repo, {"pkg/util.py": "def helper():\n pass\n"}, "second") + later = _commit(repo, {"pkg/late.py": "X = 1\n"}, "third") + return repo, first, pin, later + + +def _pulls(table): + def pull(repository, number): + if number not in table: + raise FactsError(f"{repository} PR #{number}: not found") + return table[number] + return pull + + +def test_observer_answers_at_the_pin(upstream): + repo, first, pin, later = upstream + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + assert obs.head() == pin + assert obs.top_level(pin) == {"pkg", "README.md"} + assert obs.path_exists(pin, "pkg/util.py") + assert not obs.path_exists(pin, "pkg/late.py") + assert obs.file_text(pin, "pkg/late.py") is None + assert obs.file_text(pin, "README.md") == "hello\n" + assert obs.is_ancestor(first) and obs.is_ancestor(pin) + assert not obs.is_ancestor(later) + assert not obs.is_ancestor("0" * 40) and not obs.is_ancestor("nope") + + +def test_observer_refuses_a_bad_pin(upstream): + repo, *_ = upstream + with pytest.raises(FactsError): + PinnedObserver(repo, "o/up", "abc") + with pytest.raises(FactsError): + PinnedObserver(repo, "o/up", "1" * 40) + + +def test_rule_claims_hold_or_are_reported(upstream): + repo, first, pin, later = upstream + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({ + 1: {"merged": True, "merge_commit_sha": first}, + 2: {"merged": True, "merge_commit_sha": later}, + 3: {"merged": False}, + })) + ok = {"R-1": "Use `pkg/core.py::Engine.step` and `pkg/util.py` ^[PR #1]"} + assert check_rules(ok, obs) == [] + bad = { + "R-2": "See `pkg/late.py`", # exists only after the pin + "R-3": "Call `pkg/core.py::Engine.run`", # symbol not defined + "R-4": "Fixed in ^[PR #2]", # merged after the pin + "R-5": "Proposed in ^[PR #3]", # never merged + } + problems = check_rules(bad, obs) + assert len(problems) == 4 + assert problems[0].startswith("R-2:") and "pkg/late.py" in problems[0] + assert problems[1].startswith("R-3:") and "Engine.run" in problems[1] + assert problems[2].startswith("R-4:") and "merged after" in problems[2] + assert problems[3].startswith("R-5:") and "not merged" in problems[3] + + +def test_prose_words_are_not_claims(upstream): + repo, _, pin, _ = upstream + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + assert check_rules({"R-1": "Mention `tests` and `config/x.py` only"}, obs) == [] + + +def test_evidence_binds_a_range_to_its_text(upstream): + repo, _, pin, _ = upstream + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + entry = evidence_for(obs, "pkg/core.py", 2, 3) + assert check_evidence([entry], obs) == [] + assert Evidence.from_dict(entry.to_dict()) == entry + moved = Evidence("pkg/core.py", 1, 2, entry.sha256) + outside = Evidence("pkg/core.py", 2, 9, entry.sha256) + missing = Evidence("pkg/late.py", 1, 1, entry.sha256) + backwards = Evidence("pkg/core.py", 3, 2, entry.sha256) + problems = check_evidence([moved, outside, missing, backwards], obs) + assert [p.split(": ", 1)[1].split(" ")[0] for p in problems] == ["content", "range", "file", "range"] + with pytest.raises(FactsError): + evidence_for(obs, "pkg/core.py", 0, 1) + with pytest.raises(FactsError): + evidence_for(obs, "pkg/late.py", 1, 1) + + +def test_directories_are_paths_but_not_evidence(upstream): + repo, _, pin, _ = upstream + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + assert obs.path_exists(pin, "pkg") and obs.file_text(pin, "pkg") is None + assert check_rules({"R-1": "Lives under `pkg/`"}, obs) == [] + with pytest.raises(FactsError): + evidence_for(obs, "pkg", 1, 1) + assert "file does not exist" in check_evidence([Evidence("pkg", 1, 1, "0" * 64)], obs)[0] + + +def test_evidence_hash_ignores_line_ending_style(tmp_path): + repo = tmp_path / "crlf" + repo.mkdir() + _git(repo, "init", "-q") + _git(repo, "config", "core.autocrlf", "false") + a = _commit(repo, {"f.txt": "a\nb\nc"}, "lf") + b = _commit(repo, {"f.txt": "a\r\nb\r\nc\r\n"}, "crlf") + lf = evidence_for(PinnedObserver(repo, "o/x", a, pull=_pulls({})), "f.txt", 1, 3) + crlf = evidence_for(PinnedObserver(repo, "o/x", b, pull=_pulls({})), "f.txt", 1, 3) + assert lf.sha256 == crlf.sha256 + + +def test_an_unreadable_object_is_an_error_not_an_absence(upstream): + repo, _, pin, _ = upstream + blob = _git(repo, "rev-parse", f"{pin}:pkg/util.py") + (repo / ".git" / "objects" / blob[:2] / blob[2:]).unlink() # loose object gone + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + assert obs.path_exists(pin, "pkg/util.py") # the tree still names it + with pytest.raises(FactsError): + obs.file_text(pin, "pkg/util.py") + with pytest.raises(FactsError): + check_rules({"R-1": "Call `pkg/util.py::helper`"}, obs) + with pytest.raises(FactsError): + check_evidence([Evidence("pkg/util.py", 1, 1, "0" * 64)], obs) + + +def test_an_unreadable_tree_is_an_error(upstream): + repo, _, pin, _ = upstream + tree = _git(repo, "rev-parse", f"{pin}:pkg") + (repo / ".git" / "objects" / tree[:2] / tree[2:]).unlink() + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + with pytest.raises(FactsError): + obs.path_exists(pin, "pkg/core.py") + + +def test_an_unreadable_history_is_an_error_not_a_later_merge(upstream): + repo, first, pin, later = upstream + (repo / ".git" / "objects" / first[:2] / first[2:]).unlink() # the pin's parent is gone + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + with pytest.raises(FactsError): + obs.is_ancestor(later) # git cannot walk the pin's history: no answer, not "no" + assert not obs.is_ancestor("0" * 40) # an unknown commit is still simply not an ancestor + + +def test_a_corrupt_candidate_commit_is_an_error_not_unknown(upstream): + repo, _, pin, later = upstream + path = repo / ".git" / "objects" / later[:2] / later[2:] + path.chmod(0o644) + path.write_bytes(b"not a zlib stream") + obs = PinnedObserver(repo, "o/up", pin, pull=_pulls({})) + with pytest.raises(FactsError): + obs.is_ancestor(later) + + +@pytest.mark.parametrize("failure", [FileNotFoundError("gh"), ConnectionError("reset"), ValueError("bad json")]) +def test_a_pr_lookup_failure_is_a_facts_error(upstream, failure): + repo, _, pin, _ = upstream + + def pull(repository, number): + raise failure + + obs = PinnedObserver(repo, "o/up", pin, pull=pull) + with pytest.raises(FactsError): + obs.pull(7) + with pytest.raises(FactsError): + check_rules({"R-1": "Fixed in ^[PR #7]"}, obs) + + +def test_a_pr_lookup_keeps_its_own_facts_error_and_rejects_non_objects(upstream): + repo, _, pin, _ = upstream + original = FactsError("o/up PR #7: not found") + + def pull(repository, number): + raise original + + with pytest.raises(FactsError) as raised: + PinnedObserver(repo, "o/up", pin, pull=pull).pull(7) + assert raised.value is original + with pytest.raises(FactsError): + PinnedObserver(repo, "o/up", pin, pull=lambda r, n: ["not", "a", "pr"]).pull(7)