diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e3213e65..8b58b480 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,10 @@ jobs: ruff check src/ tests/ ruff format --check src/ tests/ + # Three-valued: 0 clean · 1 a citation drifted (re-anchor it) · 2 the gate + # could not verify (an unregistered unverifiable citation, an absent cited + # source, a stale/over-ratchet citation-baseline.json, or zero citations + # graded). Both 1 and 2 fail the step; the script's report says which. - name: Protocol-doc citation drift check run: python scripts/check_protocol_doc_citations.py diff --git a/citation-baseline.json b/citation-baseline.json new file mode 100644 index 00000000..5f496b51 --- /dev/null +++ b/citation-baseline.json @@ -0,0 +1,21 @@ +{ + "frozen_count": 2, + "entries": [ + { + "doc_line": 179, + "path": "onboard/protocol.py", + "cited_start": 107, + "cited_end": 110, + "finding": "The doc section is 2.5 conversation_start and this line is its **Source (model)** pointer, but src/bonfire/onboard/protocol.py:107-110 covers the blank line before class AllScansComplete plus that class's header and the opening of its docstring. The ConversationStart model the sentence names lives at lines 120-123. Repair = re-point the range at 120-123.", + "reason": "hand-verified 2026-07-28 as genuinely drifted: the doc names the conversation_start model while the cited range covers AllScansComplete. Recorded so the gate can mount now; correcting the doc is owned by the doc. NOT registered as merely unmechanisable - the citation is wrong, and it reaches the unverified bucket only because every backticked hint in its context window is listed in _SKIP_HINTS and the cited line sits at module level, so neither resolution pass fires. Remove this entry when the doc range is repaired; the ratchet then owes a shrink." + }, + { + "doc_line": 266, + "path": "onboard/protocol.py", + "cited_start": 175, + "cited_end": 177, + "finding": "The doc sentence above this line reads 'The client-type registry has exactly one entry', but src/bonfire/onboard/protocol.py:175-177 covers the Parsing banner comment, a blank line, and the first line of _SERVER_TYPES - the SERVER registry, which has eight entries. The one-entry client registry, _CLIENT_TYPES, lives at lines 188-190. Repair = re-point the range at 188-190.", + "reason": "hand-verified 2026-07-28 as genuinely drifted: the doc says client while the cited range points at the server registry. Recorded so the gate can mount now; correcting the doc is owned by the doc. NOT registered as merely unmechanisable - the citation is wrong, and it reaches the unverified bucket only because no backticked identifier appears in the five doc lines before it and the cited line sits at module level, so neither resolution pass fires. Remove this entry when the doc range is repaired; the ratchet then owes a shrink." + } + ] +} diff --git a/scripts/check_protocol_doc_citations.py b/scripts/check_protocol_doc_citations.py index d118a4a7..4507b842 100644 --- a/scripts/check_protocol_doc_citations.py +++ b/scripts/check_protocol_doc_citations.py @@ -3,7 +3,7 @@ # Copyright 2026 BonfireAI r"""Verify ``docs/scan-front-door-protocol.md`` source citations stay anchored. -The Front Door protocol doc cites ~70 ``src/bonfire/...py:NN[-NN]`` +The Front Door protocol doc cites ~60 ``src/bonfire/...py:NN[-NN]`` line ranges. After any wave of insert-heavy edits in the cited modules (Wave 4 trust-triangle, Wave 9 Lane B oversize handling, Wave 10 vault_seed symlink hardening, …) the line numbers drift @@ -12,32 +12,59 @@ this doc as the authoritative third-party-client contract, so the citations are part of the contract surface, not just decoration. -This script: - -* Reads ``docs/scan-front-door-protocol.md``. -* Extracts every ``src/bonfire/.../.py:NN`` and - ``src/bonfire/.../.py:NN-NN`` citation (Python files only; - ``ui.html`` citations are skipped because they have no AST). -* Resolves each citation against the source AST in two passes: - 1. **Containment pass.** Find the innermost class/function/ - module-level assignment that *contains* the cited start line. - A citation that lands inside ``FrontDoorServer._ws_handler``'s - body is OK regardless of whether the surrounding doc text - names ``_ws_handler``, because the line still points at code - inside that symbol. - 2. **Hint pass (fallback).** If the cited line is in module - top-level whitespace/imports, walks the doc up to 5 lines back - looking for a backticked Python identifier - (``\`ConversationEngine.start\```, ``\`_SERVER_TYPES\```) - and asserts the symbol's actual start line is within - ``--tolerance`` of the cited start. -* Fails (exit 1) with one line per drifted citation. - -The check is intentionally conservative: citations whose symbol -cannot be matched mechanically AND don't land in any indexed body -are reported as ``unverified`` on stderr and do NOT fail the run. -The script's job is to catch silent drift on the citations we CAN -mechanise, not to gate the whole doc on perfect machine-readability. +This script reads the doc, extracts every ``src/bonfire/.../.py`` +citation (Python only; ``ui.html`` citations have no AST) and resolves +each one against the source AST in two passes: + +1. **Containment pass.** Find the innermost class/function/module-level + assignment that *contains* the cited start line. A citation landing + inside ``FrontDoorServer._ws_handler``'s body is OK whether or not the + surrounding doc text names ``_ws_handler`` — the line still points at + code inside that symbol. +2. **Hint pass (fallback).** If the cited line is at module level + (imports, blank lines, banner comments), walk the doc up to 5 lines + back for a backticked Python identifier (``\`ConversationEngine.start\```, + ``\`_SERVER_TYPES\```) and assert the symbol's actual start line is + within ``--tolerance`` of the cited start. + +The honest exit contract +------------------------ + +==== ===================================================================== +Exit Meaning +==== ===================================================================== +0 Clean. Every citation resolved (or is registered in + ``citation-baseline.json``) AND the run graded a NON-EMPTY set. +1 DRIFT. The gate DID its job: a citation resolved to a symbol and the + cited line is wrong. Re-anchor it. +2 COULD NOT VERIFY. The gate COULD NOT do its job: a citation it cannot + mechanically resolve and that no registry entry covers, a cited + source file absent or unparseable, an unreadable doc, a malformed or + over-ratchet registry, a stale registry entry, or a run that graded + ZERO citations. +==== ===================================================================== + +Exit 1 outranks exit 2 when both stand; every could-not-verify blocker is +still printed under the verdict, so nothing is masked. + +This REPLACES the fail-open behaviour the script shipped with, which +reported unresolvable citations as ``unverified`` on stderr and returned 0 +anyway. That made the gate MOST permissive exactly where its own +confidence was LOWEST, and the run that measured it found BOTH of its +``unverified`` citations were real drift, laundered into a category +structurally incapable of failing. ``unverified`` is no longer a pass. + +To land the strict gate ahead of the doc repair it demands, each +known-unverifiable citation is named — one at a time, with a written +finding and a written reason, under a ratchet — in +``citation-baseline.json`` (see ``scripts/citation_baseline.py``). The +registry covers the ``unverified`` bucket ONLY; it can never launder a +``drift``, and it can never cover a source file the gate could not read. + +Collaborators: ``citation_doc`` (the doc side), ``citation_source`` (the +source side), ``citation_baseline`` (the registry), ``citation_report`` +(the verdict, the report and the exit code). This file owns the +resolution rules and the CLI. Usage:: @@ -45,47 +72,47 @@ python scripts/check_protocol_doc_citations.py --json python scripts/check_protocol_doc_citations.py --tolerance 0 -Exits 0 when every verifiable citation resolves; 1 on any drift. +``--doc``, ``--source-root`` and ``--baseline`` exist so the control rods +can point the whole gate at a fixture tree. They relocate WHAT is graded; +none of them can switch a verdict off. """ from __future__ import annotations import argparse -import ast import json -import re import sys -from dataclasses import dataclass, field from pathlib import Path -# --------------------------------------------------------------------------- -# Paths -# --------------------------------------------------------------------------- +# ``scripts/`` is not a package, so the collaborator modules are imported +# by bare name. The interpreter already seeds ``sys.path[0]`` with this +# directory when the script runs as ``python scripts/.py``; the +# insert is what makes the imports work when a test loads this file by +# path (``importlib.util.spec_from_file_location``) from another cwd. +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from citation_baseline import BASELINE_FILENAME, CitationBaselineError, load_baseline +from citation_doc import ( + CONTEXT_BACK_LINES, + Citation, + HintProbe, + extract_citations, + hint_for_citation, +) +from citation_report import CheckResult, Graded, fail_to_run, report +from citation_source import ( + SourceIndexError, + SymbolIndex, + index_symbols, + innermost_containing, + neighbours, +) # Repo root is the parent of this script's parent (``scripts/``). _REPO_ROOT = Path(__file__).resolve().parent.parent _DOC_PATH = _REPO_ROOT / "docs" / "scan-front-door-protocol.md" - -# Matches ``src/bonfire/.py:NN`` or ``.py:NN-NN`` inside a -# markdown backtick span. We accept either an opening ``\``` directly -# before the path or surrounding text — the doc uses both shapes. -_CITATION_RE = re.compile( - r"src/bonfire/(?P[A-Za-z0-9_/]+\.py):(?P\d+)(?:-(?P\d+))?" -) - -# Matches an inline-code Python identifier. Used to find a symbol hint -# in the 5 lines preceding a citation. We capture dotted names -# (``ConversationEngine.start``) so we can resolve method-on-class -# citations as well as bare ``parse_server_message`` shapes. -_IDENT_RE = re.compile(r"`([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)`") - -# How many lines of doc context to scan backwards from a citation for -# a symbol hint. The doc's pattern is "**Source (model)**: -# `src/.../protocol.py:NN-NN`." with the symbol named within the -# preceding paragraph; 5 lines is enough for the longest case -# (multi-line bullet) but tight enough that we don't capture an -# unrelated symbol two paragraphs up. -_CONTEXT_BACK_LINES = 5 +_SOURCE_ROOT = _REPO_ROOT / "src" / "bonfire" +_BASELINE_PATH = _REPO_ROOT / BASELINE_FILENAME # Tolerance window (lines) for "cited start line is close to the # symbol's start line". Wave-to-wave inserts of 1-2 lines inside a @@ -96,179 +123,46 @@ # symbol's def/class line. _DEFAULT_TOLERANCE = 3 -# Identifiers that are doc-shorthand for line ranges we do NOT want -# to resolve mechanically — protocol modules expose them but they're -# referenced for context rather than as load-bearing pointers, and a -# false "drift" alert would be more noise than signal. -_SKIP_HINTS = frozenset( - { - # Generic terms that match many things or are doc verbs - "true", - "false", - "none", - "type", - "yes", - "no", - # Doc section-anchor backticks that aren't Python symbols - "narration", - "question", - "reflection", - "scan_start", - "scan_update", - "scan_complete", - "all_scans_complete", - "conversation_start", - "falcor_message", - "config_generated", - "user_message", - "server_error", - } -) - # --------------------------------------------------------------------------- -# Data types +# Resolution # --------------------------------------------------------------------------- -@dataclass -class Citation: - """One ``src/bonfire/...py:NN-NN`` reference in the doc.""" - - doc_line: int # 1-based line number in the doc - path: str # e.g. "onboard/protocol.py" - start: int # cited start line - end: int # cited end line (= start when single-line) - - -@dataclass -class SymbolIndex: - """Top-level + nested symbols extracted from a source file via ``ast``.""" - - # name -> (start_line, end_line). Includes both top-level and - # dotted ``Class.method`` entries so the doc's - # ``ConversationEngine.start`` shape resolves naturally. - by_name: dict[str, tuple[int, int]] = field(default_factory=dict) - - # Flat ordered list of (start_line, end_line, name) for the - # containment pass. Ordered by widening end-line so the FIRST - # match while iterating innermost-out gives the tightest - # enclosing symbol. - intervals: list[tuple[int, int, str]] = field(default_factory=list) - - -@dataclass -class CheckResult: - """Per-citation verdict.""" - - citation: Citation - status: str # "ok" | "drift" | "unverified" - resolved_symbol: str | None = None - expected_start: int | None = None - expected_end: int | None = None - detail: str = "" - - -# --------------------------------------------------------------------------- -# Source-file symbol indexing -# --------------------------------------------------------------------------- - - -def _index_symbols(source_path: Path) -> SymbolIndex: # noqa: C901 - """Return name -> line-range map for every def/class in *source_path*. - - Top-level functions/classes are keyed by their bare name. Methods - are also keyed dotted (``Class.method``). Module-level assignments - to ``UPPER_CASE`` or ``_underscore`` names (``_SERVER_TYPES``, - ``MAX_USER_MESSAGE_LEN``, etc.) are indexed too — the doc cites - those tables/constants by name and we want to detect when they - move. - """ - text = source_path.read_text(encoding="utf-8") - tree = ast.parse(text, filename=str(source_path)) - index = SymbolIndex() - - def _add(name: str, lineno: int, end_lineno: int | None) -> None: - if end_lineno is None: - end_lineno = lineno - # First-write-wins so a re-bound name doesn't clobber the - # earlier (and usually authoritative) definition. - index.by_name.setdefault(name, (lineno, end_lineno)) - index.intervals.append((lineno, end_lineno, name)) - - for node in tree.body: - if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef): - _add(node.name, node.lineno, node.end_lineno) - elif isinstance(node, ast.ClassDef): - _add(node.name, node.lineno, node.end_lineno) - # Walk class body for methods. - for child in node.body: - if isinstance(child, ast.FunctionDef | ast.AsyncFunctionDef): - _add(child.name, child.lineno, child.end_lineno) - _add(f"{node.name}.{child.name}", child.lineno, child.end_lineno) - elif isinstance(node, ast.Assign): - # Module-level constants / tables. - for target in node.targets: - if isinstance(target, ast.Name): - _add(target.id, node.lineno, node.end_lineno) - elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - _add(node.target.id, node.lineno, node.end_lineno) - - # Sort intervals so the tightest (smallest span) enclosing range - # comes first when we filter by ``start <= cited <= end``. - index.intervals.sort(key=lambda triple: triple[1] - triple[0]) - return index - - -def _innermost_containing(index: SymbolIndex, lineno: int) -> tuple[int, int, str] | None: - """Return the smallest indexed (start, end, name) that contains ``lineno``.""" - for start, end, name in index.intervals: - if start <= lineno <= end: - return (start, end, name) - return None - - -# --------------------------------------------------------------------------- -# Doc parsing -# --------------------------------------------------------------------------- - - -def _extract_citations(doc_text: str) -> list[Citation]: - """Return one ``Citation`` per ``src/bonfire/...py:NN[-NN]`` match.""" - citations: list[Citation] = [] - for doc_line_idx, line in enumerate(doc_text.splitlines(), start=1): - for match in _CITATION_RE.finditer(line): - path = match.group("path") - start = int(match.group("start")) - end_raw = match.group("end") - end = int(end_raw) if end_raw is not None else start - citations.append(Citation(doc_line=doc_line_idx, path=path, start=start, end=end)) - return citations - - -def _hint_for_citation(doc_lines: list[str], cite: Citation) -> str | None: - """Walk back up to ``_CONTEXT_BACK_LINES`` looking for a backticked symbol.""" - # doc_lines is 0-indexed; cite.doc_line is 1-based. - end_idx = cite.doc_line - 1 - start_idx = max(0, end_idx - _CONTEXT_BACK_LINES) - # Walk back from the citation's own line first so an "on the same - # line" hint wins ("``parse_server_message``: ``src/...:NN-NN``"). - for idx in range(end_idx, start_idx - 1, -1): - line = doc_lines[idx] - for hit in _IDENT_RE.findall(line): - head = hit.split(".", 1)[0] - if head.lower() in _SKIP_HINTS: - continue - # Skip pure module shorthands that aren't symbols. - if hit in {"flow.py", "server.py", "protocol.py", "scan.py", "ui.html"}: - continue - return hit - return None +def _rejected_hints_text(probe: HintProbe | None) -> str: + """Say what the hint pass found and threw away, or that it found nothing.""" + rejected = probe.rejected if probe is not None else () + if not rejected: + return ( + f"no backticked identifier found in the {CONTEXT_BACK_LINES} doc lines " + "before the citation" + ) + shown = ", ".join(f"`{ident}` ({why})" for ident, why in rejected) + return f"doc hints found and rejected: {shown}" -# --------------------------------------------------------------------------- -# Check logic -# --------------------------------------------------------------------------- +def _unverified_detail( + cite: Citation, + index: SymbolIndex, + hint: str | None, + probe: HintProbe | None, +) -> str: + """Say what the cited line actually IS, so a human can act on it.""" + above, below = neighbours(index, cite.start) + parts = [ + f"cited line {cite.start} is not inside any indexed symbol", + f"nearest symbol above: {above}", + f"nearest symbol below: {below}", + ] + if hint is None: + parts.append(_rejected_hints_text(probe)) + else: + parts.append(f"doc hint `{hint}` is not an indexed symbol in src/bonfire/{cite.path}") + parts.append( + "FIX: re-point the citation at the symbol the doc names, or name that symbol in " + "backticks within the preceding doc lines so the hint pass can resolve it" + ) + return "; ".join(parts) def _check_citation( @@ -277,6 +171,7 @@ def _check_citation( hint: str | None, index: SymbolIndex, tolerance: int, + probe: HintProbe | None = None, ) -> CheckResult: """Return the verdict for a single citation. @@ -288,7 +183,7 @@ def _check_citation( when the doc hint names a specific symbol AND that symbol exists AND the citation's start line is NOT inside that symbol (drift). """ - containing = _innermost_containing(index, cite.start) + containing = innermost_containing(index, cite.start) if containing is not None: c_start, c_end, c_name = containing # Containment wins. The hint heuristic is too loose (it walks @@ -317,19 +212,14 @@ def _check_citation( if hint and "." in hint: tail = hint.rsplit(".", 1)[1] if tail in index.by_name: - expected = index.by_name[tail] - return _verdict(cite, hint, expected, tolerance) + return _verdict(cite, hint, index.by_name[tail], tolerance) return CheckResult( citation=cite, status="unverified", - detail=( - f"cited line {cite.start} not inside any indexed symbol " - f"and no symbol hint resolvable (hint={hint!r})" - ), + detail=_unverified_detail(cite, index, hint, probe), ) - expected = index.by_name[hint] - return _verdict(cite, hint, expected, tolerance) + return _verdict(cite, hint, index.by_name[hint], tolerance) def _verdict( @@ -381,12 +271,64 @@ def _verdict( ) +def _grade(doc_text: str, source_root: Path, tolerance: int) -> Graded: + """Grade every citation in *doc_text* against *source_root*. + + A source file that cannot be read or parsed blocks only its own + citations; the rest of the doc is still graded, so one missing + module cannot shrink the report to nothing without saying so. + """ + doc_lines = doc_text.splitlines() + citations = extract_citations(doc_text) + indices: dict[str, SymbolIndex] = {} + blocked: dict[str, str] = {} + results: list[CheckResult] = [] + for cite in citations: + if cite.path not in indices and cite.path not in blocked: + try: + indices[cite.path] = index_symbols(source_root / cite.path) + except SourceIndexError as exc: + blocked[cite.path] = str(exc) + if cite.path in blocked: + results.append(CheckResult(citation=cite, status="blocked", detail=blocked[cite.path])) + continue + probe = hint_for_citation(doc_lines, cite) + results.append( + _check_citation( + cite, + hint=probe.symbol, + index=indices[cite.path], + tolerance=tolerance, + probe=probe, + ) + ) + return Graded(results=results, extracted=len(citations)) + + # --------------------------------------------------------------------------- # Driver # --------------------------------------------------------------------------- -def main(argv: list[str] | None = None) -> int: +def _emit_json(results: list[CheckResult]) -> None: + payload = [ + { + "status": r.status, + "doc_line": r.citation.doc_line, + "path": r.citation.path, + "cited_start": r.citation.start, + "cited_end": r.citation.end, + "symbol": r.resolved_symbol, + "expected_start": r.expected_start, + "expected_end": r.expected_end, + "detail": r.detail, + } + for r in results + ] + sys.stdout.write(json.dumps(payload, indent=2) + "\n") + + +def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument( "--tolerance", @@ -405,90 +347,35 @@ def main(argv: list[str] | None = None) -> int: default=_DOC_PATH, help="Path to the protocol doc (default: docs/scan-front-door-protocol.md).", ) - args = parser.parse_args(argv) - - doc_text = args.doc.read_text(encoding="utf-8") - doc_lines = doc_text.splitlines() - citations = _extract_citations(doc_text) - - # Cache: src-path-string -> SymbolIndex. - indices: dict[str, SymbolIndex] = {} - - results: list[CheckResult] = [] - for cite in citations: - if cite.path not in indices: - source_file = _REPO_ROOT / "src" / "bonfire" / cite.path - if not source_file.is_file(): - results.append( - CheckResult( - citation=cite, - status="unverified", - detail=f"source not found: {source_file}", - ) - ) - continue - indices[cite.path] = _index_symbols(source_file) - hint = _hint_for_citation(doc_lines, cite) - results.append( - _check_citation( - cite, - hint=hint, - index=indices[cite.path], - tolerance=args.tolerance, - ) - ) + parser.add_argument( + "--source-root", + type=Path, + default=_SOURCE_ROOT, + help="Root the citations' paths resolve against (default: src/bonfire).", + ) + parser.add_argument( + "--baseline", + type=Path, + default=_BASELINE_PATH, + help=f"Registry of reasoned unverifiable citations (default: {BASELINE_FILENAME}).", + ) + return parser - drifts = [r for r in results if r.status == "drift"] - unverified = [r for r in results if r.status == "unverified"] +def main(argv: list[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + try: + doc_text = args.doc.read_text(encoding="utf-8") + except OSError as exc: + return fail_to_run(f"cannot read the doc under test: {exc}") + try: + baseline = load_baseline(args.baseline) + except CitationBaselineError as exc: + return fail_to_run(f"{BASELINE_FILENAME} is unusable: {exc}") + graded = _grade(doc_text, args.source_root, args.tolerance) if args.json: - payload = [ - { - "status": r.status, - "doc_line": r.citation.doc_line, - "path": r.citation.path, - "cited_start": r.citation.start, - "cited_end": r.citation.end, - "symbol": r.resolved_symbol, - "expected_start": r.expected_start, - "expected_end": r.expected_end, - "detail": r.detail, - } - for r in results - ] - sys.stdout.write(json.dumps(payload, indent=2) + "\n") - - # Human report. - if drifts: - sys.stderr.write(f"\nDRIFT: {len(drifts)} citation(s) point at moved symbols:\n") - for r in drifts: - sys.stderr.write( - f" doc line {r.citation.doc_line}: " - f"src/bonfire/{r.citation.path}:{r.citation.start}" - f"{'-' + str(r.citation.end) if r.citation.end != r.citation.start else ''}" - f" symbol={r.resolved_symbol!r} " - f"expected={r.expected_start}-{r.expected_end}\n" - f" {r.detail}\n" - ) - if unverified: - sys.stderr.write( - f"\n{len(unverified)} citation(s) could not be mechanically verified " - "(no symbol hint in surrounding context — review manually):\n" - ) - for r in unverified: - sys.stderr.write( - f" doc line {r.citation.doc_line}: " - f"src/bonfire/{r.citation.path}:{r.citation.start} " - f"({r.detail})\n" - ) - - ok_count = sum(1 for r in results if r.status == "ok") - sys.stderr.write( - f"\nSummary: {ok_count} ok, {len(drifts)} drift, {len(unverified)} unverified, " - f"{len(results)} total citations checked.\n" - ) - - return 1 if drifts else 0 + _emit_json(graded.results) + return report(graded, baseline, args.doc) if __name__ == "__main__": diff --git a/scripts/citation_baseline.py b/scripts/citation_baseline.py new file mode 100644 index 00000000..d800646a --- /dev/null +++ b/scripts/citation_baseline.py @@ -0,0 +1,278 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 BonfireAI +"""The reasoned, ratcheted registry of citations the drift gate cannot mechanise. + +``check_protocol_doc_citations.py`` exits 2 when a citation cannot be +mechanically resolved: "unverified" is not a pass, because the run that +measured this gate found that BOTH of its unverified citations were real +drift, laundered into a category that could not fail. Making unverified +fail closes that hole and immediately raises the landing problem: the +two known-bad citations would hold the gate off ``main`` forever, and +repairing the doc's prose is owned by the doc, not by the gate. + +This module is the answer, modelled on the shared quality kit's +``exemptions.json``: a registry at ``citation-baseline.json`` where each +unverifiable citation is named, one at a time, with a written reason and +a written finding, under a ratchet. + +Shape:: + + {"frozen_count": 2, "entries": [{ + "doc_line": 179, # 1-based line of the citation IN THE DOC + "path": "onboard/protocol.py", # exactly one file, never a glob + "cited_start": 107, # the cited range, verbatim + "cited_end": 110, + "finding": "what is actually wrong, in repair terms", + "reason": "why it is registered instead of fixed here" + }]} + +Rules, all enforced here and reported loudly by the driver on EVERY run: + +1. **Exact keys.** All six, no others. A key this gate does not read + cannot carry a claim; a missing key means the entry cannot be tied to + one citation. +2. **One citation per entry.** The identity is + ``(doc_line, path, cited_start, cited_end)`` — no globs, no + wildcards, no file-level blessings. Move the citation, widen its + range, or move the doc paragraph and the entry stops matching. +3. **The ratchet.** Entries may be removed freely. Adding one requires + bumping ``frozen_count`` in the same commit, so growth is a visible + decision. ``frozen_count`` above the live count is printed as slack + that may shrink. +4. **Stale entries are violations.** An entry matching no unverified + citation is a registry grading less than it thinks it is; see + :func:`cover`. +5. **The registry never covers drift.** It is consulted for the + ``unverified`` bucket only. The driver decides drift before it opens + this file, and drift exits 1 whatever is written here. +""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +BASELINE_FILENAME = "citation-baseline.json" + +# The identity of one citation: doc line, source path, cited range. +# A tuple rather than a shared class so this module stays independent of +# the driver's record types (no import cycle, and the registry can be +# unit-tested without the gate). +CitationKey = tuple[int, str, int, int] + +_INT_KEYS = ("doc_line", "cited_start", "cited_end") +_TEXT_KEYS = ("path", "finding", "reason") +_REQUIRED_KEYS = frozenset(_INT_KEYS + _TEXT_KEYS) + + +class CitationBaselineError(Exception): + """The registry itself is unreadable or malformed. + + Typed so the driver reports "the gate could not do its job" (exit 2) + rather than crashing with a traceback or, worse, treating an + unparseable registry as an empty one. + """ + + +@dataclass(frozen=True) +class BaselineEntry: + """One registered citation: which one, what is wrong, why it is here.""" + + doc_line: int + path: str + cited_start: int + cited_end: int + finding: str + reason: str + + @property + def key(self) -> CitationKey: + """The exact citation this entry covers, and no other.""" + return (self.doc_line, self.path, self.cited_start, self.cited_end) + + def label(self) -> str: + """Human-facing one-liner naming the covered citation.""" + return ( + f"doc line {self.doc_line}: src/bonfire/{self.path}:{self.cited_start}-{self.cited_end}" + ) + + +@dataclass(frozen=True) +class Baseline: + """The committed registry: the live entries plus the ratchet number.""" + + frozen_count: int + entries: tuple[BaselineEntry, ...] + source: Path + + +@dataclass(frozen=True) +class Coverage: + """Which unverified citations the registry covers, and what went stale.""" + + covered: tuple[tuple[CitationKey, BaselineEntry], ...] + uncovered: tuple[CitationKey, ...] + stale: tuple[BaselineEntry, ...] + + +def _require_keys(raw: dict[str, object], where: str) -> None: + keys = set(raw) + if keys == _REQUIRED_KEYS: + return + missing = sorted(_REQUIRED_KEYS - keys) + unknown = sorted(keys - _REQUIRED_KEYS) + raise CitationBaselineError( + f"{where} must carry exactly {sorted(_REQUIRED_KEYS)} " + f"(missing={missing}, unknown={unknown}): a key this gate does not read " + "cannot carry a claim, and a missing key means the entry cannot be " + "tied to one citation" + ) + + +def _require_line_numbers(raw: dict[str, object], where: str) -> None: + for key in _INT_KEYS: + value = raw[key] + if not isinstance(value, int) or isinstance(value, bool) or value < 1: + raise CitationBaselineError( + f"{where}.{key} must be a positive line number, got {value!r}" + ) + + +def _require_prose(raw: dict[str, object], where: str) -> None: + for key in _TEXT_KEYS: + value = raw[key] + if not isinstance(value, str) or not value.strip(): + raise CitationBaselineError( + f"{where}.{key} must be a non-empty string, got {value!r} " + "(a blank key carries no claim, and an entry without a written " + "finding and reason is a silent suppression)" + ) + + +def _entry_from(raw: object, position: int) -> BaselineEntry: + """Validate one raw registry row into a :class:`BaselineEntry`.""" + where = f"entries[{position}]" + if not isinstance(raw, dict): + raise CitationBaselineError(f"{where} must be a JSON object, got {type(raw).__name__}") + _require_keys(raw, where) + _require_line_numbers(raw, where) + _require_prose(raw, where) + path: str = raw["path"] + if "*" in path or "?" in path: + raise CitationBaselineError( + f"{where}.path must name exactly one source file, not a glob: {path!r}" + ) + start: int = raw["cited_start"] + end: int = raw["cited_end"] + if end < start: + raise CitationBaselineError(f"{where}.cited_end {end} precedes cited_start {start}") + return BaselineEntry( + doc_line=raw["doc_line"], + path=path, + cited_start=start, + cited_end=end, + finding=raw["finding"], + reason=raw["reason"], + ) + + +def _require_no_duplicates(entries: Sequence[BaselineEntry], source: Path) -> None: + seen: set[CitationKey] = set() + for entry in entries: + if entry.key in seen: + raise CitationBaselineError( + f"{source}: two entries cover the same citation ({entry.label()}); " + "duplicate rows inflate the ratchet count while blessing one citation" + ) + seen.add(entry.key) + + +def _require_frozen_count(data: dict[str, object], source: Path) -> int: + frozen = data.get("frozen_count") + if not isinstance(frozen, int) or isinstance(frozen, bool) or frozen < 0: + raise CitationBaselineError( + f"{source}: frozen_count must be a non-negative integer, got {frozen!r}" + ) + return frozen + + +def load_baseline(path: Path) -> Baseline: + """Load the registry. A missing file is an empty (day-one) registry. + + Empty is safe: with nothing registered, every unverified citation is + unregistered, so the driver exits 2. Deleting the file cannot buy a + pass, only a louder failure. + """ + if not path.exists(): + return Baseline(frozen_count=0, entries=(), source=path) + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise CitationBaselineError(f"cannot read {path}: {exc}") from exc + if not isinstance(data, dict): + raise CitationBaselineError(f"{path} must hold a JSON object") + frozen = _require_frozen_count(data, path) + raw_entries = data.get("entries") + if not isinstance(raw_entries, list): + raise CitationBaselineError( + f"{path}: entries must be a JSON list, got {type(raw_entries).__name__}" + ) + entries = tuple(_entry_from(raw, position) for position, raw in enumerate(raw_entries)) + _require_no_duplicates(entries, path) + return Baseline(frozen_count=frozen, entries=entries, source=path) + + +def cover(keys: Sequence[CitationKey], baseline: Baseline) -> Coverage: + """Split *keys* (the unverified citations) into covered and uncovered. + + ``stale`` holds every entry that matched nothing. A stale entry is + reported by the driver as a violation, not a notice: the registry is + a per-citation claim, and one that no longer matches means either the + doc moved (so the blessing was never re-read against what it now + covers) or the citation was repaired (so the blessing is dead and the + ratchet owes a shrink). Both cases must be settled by a human before + the gate can be trusted to know what it is grading. Removing an + entry is free, so the cost of the strict reading is one deletion. + """ + index = {entry.key: entry for entry in baseline.entries} + covered = tuple((key, index[key]) for key in keys if key in index) + uncovered = tuple(key for key in keys if key not in index) + matched = {key for key, _ in covered} + stale = tuple(entry for entry in baseline.entries if entry.key not in matched) + return Coverage(covered=covered, uncovered=uncovered, stale=stale) + + +def ratchet(baseline: Baseline) -> tuple[list[str], list[str]]: + """Return (violations, notices) for the add-requires-a-bump ratchet. + + The first notice is unconditional: the count and the frozen number + are printed on every run, clean or not, so the size of the blessed + set is a visible decision rather than a fact nobody re-reads. + """ + count = len(baseline.entries) + headline = ( + f"citation baseline: {count} registered entr{_y(count)}, " + f"frozen_count={baseline.frozen_count} ({baseline.source})" + ) + notices = [headline] + violations: list[str] = [] + if count > baseline.frozen_count: + violations.append( + f"citation baseline holds {count} entr{_y(count)} but frozen_count is " + f"{baseline.frozen_count}: adding an entry is a decision and must bump " + "frozen_count in the same commit" + ) + elif count < baseline.frozen_count: + slack = baseline.frozen_count - count + notices.append( + f"citation baseline slack: frozen_count={baseline.frozen_count} exceeds the " + f"{count} live entr{_y(count)} by {slack} — it may shrink to {count}" + ) + return violations, notices + + +def _y(count: int) -> str: + """Pluralise ``entry``/``entries`` without a second format string.""" + return "y" if count == 1 else "ies" diff --git a/scripts/citation_doc.py b/scripts/citation_doc.py new file mode 100644 index 00000000..a2925f93 --- /dev/null +++ b/scripts/citation_doc.py @@ -0,0 +1,156 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 BonfireAI + +r"""What the DOC says: the citation record, and how citations are read out of it. + +One of four modules behind ``check_protocol_doc_citations.py``, split by +question rather than by size (the gate outgrew the 500-line file cap when +it acquired an honest three-valued exit contract): + +* this module — the doc side: what a citation IS, and what the doc says + near one; +* ``citation_source`` — the source side: what symbol lives at line N; +* ``citation_baseline`` — the registry of reasoned, ratcheted blessings; +* ``citation_report`` — the verdict record, the report, the exit code. + +The gate script itself keeps the resolution rules and the CLI. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +# Matches ``src/bonfire/.py:NN`` or ``.py:NN-NN`` inside a +# markdown backtick span. We accept either an opening ``\``` directly +# before the path or surrounding text — the doc uses both shapes. +CITATION_RE = re.compile( + r"src/bonfire/(?P[A-Za-z0-9_/]+\.py):(?P\d+)(?:-(?P\d+))?" +) + +# Matches an inline-code Python identifier. Used to find a symbol hint +# in the 5 lines preceding a citation. We capture dotted names +# (``ConversationEngine.start``) so we can resolve method-on-class +# citations as well as bare ``parse_server_message`` shapes. +IDENT_RE = re.compile(r"`([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)`") + +# How many lines of doc context to scan backwards from a citation for +# a symbol hint. The doc's pattern is "**Source (model)**: +# `src/.../protocol.py:NN-NN`." with the symbol named within the +# preceding paragraph; 5 lines is enough for the longest case +# (multi-line bullet) but tight enough that we don't capture an +# unrelated symbol two paragraphs up. +CONTEXT_BACK_LINES = 5 + +# Identifiers that are doc-shorthand for line ranges we do NOT want +# to resolve mechanically — protocol modules expose them but they're +# referenced for context rather than as load-bearing pointers, and a +# false "drift" alert would be more noise than signal. +# +# MEASURED SCAR: when EVERY hint in a citation's context window is on +# this list AND the cited line sits at module level, both resolution +# passes come up empty and the citation lands in ``unverified``. While +# ``unverified`` exited 0, that combination laundered real drift into a +# free pass — it is how two genuinely wrong citations sat unnoticed in +# the shipped doc. It cannot any more (the gate exits 2), and every +# rejection is now reported with this table named, so a citation in that +# position is legible instead of merely dismissed. +SKIP_HINTS = frozenset( + { + # Generic terms that match many things or are doc verbs + "true", + "false", + "none", + "type", + "yes", + "no", + # Doc section-anchor backticks that aren't Python symbols + "narration", + "question", + "reflection", + "scan_start", + "scan_update", + "scan_complete", + "all_scans_complete", + "conversation_start", + "falcor_message", + "config_generated", + "user_message", + "server_error", + } +) + +# Backticked spans that name a module, not a symbol: nothing to resolve. +MODULE_SHORTHANDS = frozenset({"flow.py", "server.py", "protocol.py", "scan.py", "ui.html"}) + +# Why a candidate hint was thrown away. Reported verbatim, with the +# table named, because "no symbol hint resolvable" is not actionable. +SKIP_HINTS_REASON = "listed in citation_doc.SKIP_HINTS" +SHORTHAND_REASON = "a module filename, not a symbol" + + +@dataclass +class Citation: + """One ``src/bonfire/...py:NN-NN`` reference in the doc.""" + + doc_line: int # 1-based line number in the doc + path: str # e.g. "onboard/protocol.py" + start: int # cited start line + end: int # cited end line (= start when single-line) + + +@dataclass(frozen=True) +class HintProbe: + """What the hint pass found in the doc lines before a citation. + + ``rejected`` carries ``(identifier, why)`` for every backticked + candidate the pass threw away. Reporting it is the difference + between "no symbol hint resolvable" and "the three hints in range + are all in SKIP_HINTS" — only the second can be acted on. + """ + + symbol: str | None = None + rejected: tuple[tuple[str, str], ...] = () + + +def extract_citations(doc_text: str) -> list[Citation]: + """Return one ``Citation`` per ``src/bonfire/...py:NN[-NN]`` match.""" + citations: list[Citation] = [] + for doc_line_idx, line in enumerate(doc_text.splitlines(), start=1): + for match in CITATION_RE.finditer(line): + end_raw = match.group("end") + start = int(match.group("start")) + citations.append( + Citation( + doc_line=doc_line_idx, + path=match.group("path"), + start=start, + end=int(end_raw) if end_raw is not None else start, + ) + ) + return citations + + +def hint_for_citation(doc_lines: list[str], cite: Citation) -> HintProbe: + """Walk back up to ``CONTEXT_BACK_LINES`` looking for a backticked symbol.""" + # doc_lines is 0-indexed; cite.doc_line is 1-based. + end_idx = cite.doc_line - 1 + start_idx = max(0, end_idx - CONTEXT_BACK_LINES) + rejected: list[tuple[str, str]] = [] + # Walk back from the citation's own line first so an "on the same + # line" hint wins ("``parse_server_message``: ``src/...:NN-NN``"). + for idx in range(end_idx, start_idx - 1, -1): + for hit in IDENT_RE.findall(doc_lines[idx]): + if hit.split(".", 1)[0].lower() in SKIP_HINTS: + rejected.append((hit, SKIP_HINTS_REASON)) + elif hit in MODULE_SHORTHANDS: + rejected.append((hit, SHORTHAND_REASON)) + else: + return HintProbe(symbol=hit, rejected=tuple(rejected)) + return HintProbe(symbol=None, rejected=tuple(rejected)) + + +def cited_range(cite: Citation) -> str: + """``src/bonfire/:NN`` or ``:NN-NN``, as the doc wrote it.""" + span = f"{cite.start}-{cite.end}" if cite.end != cite.start else str(cite.start) + return f"src/bonfire/{cite.path}:{span}" diff --git a/scripts/citation_report.py b/scripts/citation_report.py new file mode 100644 index 00000000..ebac5f15 --- /dev/null +++ b/scripts/citation_report.py @@ -0,0 +1,270 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 BonfireAI + +"""The VERDICT side: what one citation graded to, how a run reports, what it exits. + +One of four modules behind ``check_protocol_doc_citations.py`` (see +``citation_doc`` for the split). This module owns the exit contract's +arithmetic and its prose: + +* ``0`` — every citation resolved or is registered, AND the graded set + was non-empty. +* ``1`` — DRIFT: the gate DID its job and found a citation pointing at + moved code. +* ``2`` — COULD NOT VERIFY: the gate COULD NOT do its job. + +Exit 1 outranks exit 2 when both stand, because drift is the more +actionable signal — and every could-not-verify blocker is still printed +under the verdict, so nothing is masked. Everything the gate refuses on +is printed BEFORE the verdict, and every pass bought by the registry is +printed WITH its reason: a blessing nobody reads is a silent one. +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass +from pathlib import Path + +from citation_baseline import ( + BASELINE_FILENAME, + Baseline, + BaselineEntry, + CitationKey, + Coverage, + cover, + ratchet, +) +from citation_doc import Citation, cited_range + + +@dataclass +class CheckResult: + """Per-citation verdict. + + ``status`` is one of: + + * ``ok`` — resolved, and the cited line points at the symbol. + * ``drift`` — resolved, and the cited line is wrong. Exit 1. + * ``unverified`` — the source was read and indexed, but this + citation could not be resolved mechanically. Exit 2 unless a + ``citation-baseline.json`` entry covers it. NOT a pass on its own. + * ``blocked`` — the cited source could not be read or parsed, so the + gate never got to look. Always exit 2, and NEVER registrable: a + registry able to bless a missing file would bless the gate's own + blindness. + """ + + citation: Citation + status: str # "ok" | "drift" | "unverified" | "blocked" + resolved_symbol: str | None = None + expected_start: int | None = None + expected_end: int | None = None + detail: str = "" + + +@dataclass +class Graded: + """What one run extracted and what it actually graded. + + Both counts are load-bearing: the verdict asserts they are equal and + non-zero, so a gate that selected an empty set cannot report clean. + """ + + results: list[CheckResult] + extracted: int + + +def citation_key(cite: Citation) -> CitationKey: + """The registry identity of a citation: doc line, path, cited range.""" + return (cite.doc_line, cite.path, cite.start, cite.end) + + +def _write(text: str) -> None: + sys.stderr.write(text) + + +def _report_notices(notices: list[str]) -> None: + for line in notices: + _write(f"\n{line}\n") + + +def _report_drift(drifts: list[CheckResult]) -> None: + if not drifts: + return + _write(f"\nDRIFT: {len(drifts)} citation(s) point at moved symbols:\n") + for r in drifts: + _write( + f" doc line {r.citation.doc_line}: {cited_range(r.citation)}" + f" symbol={r.resolved_symbol!r} expected={r.expected_start}-{r.expected_end}\n" + f" {r.detail}\n" + ) + + +def _report_registered( + covered: tuple[tuple[CitationKey, BaselineEntry], ...], + by_key: dict[CitationKey, CheckResult], +) -> None: + """Print every registry-bought pass with its reason. Loud, never silent.""" + if not covered: + return + _write( + f"\n{len(covered)} unverifiable citation(s) are REGISTERED in " + f"{BASELINE_FILENAME} and pass loudly:\n" + ) + for key, entry in covered: + _write( + f" {entry.label()}\n" + f" finding: {entry.finding}\n" + f" reason: {entry.reason}\n" + f" gate saw: {by_key[key].detail}\n" + ) + + +def _report_unregistered( + uncovered: tuple[CitationKey, ...], + by_key: dict[CitationKey, CheckResult], +) -> None: + if not uncovered: + return + _write( + f"\n{len(uncovered)} citation(s) could not be mechanically verified and are NOT " + f"registered in {BASELINE_FILENAME}:\n" + ) + for key in uncovered: + result = by_key[key] + _write( + f" doc line {result.citation.doc_line}: {cited_range(result.citation)}\n" + f" {result.detail}\n" + ) + + +def _report_blocked(blocked: list[CheckResult]) -> None: + if not blocked: + return + _write(f"\n{len(blocked)} citation(s) name source the gate could not read:\n") + for r in blocked: + _write(f" doc line {r.citation.doc_line}: {cited_range(r.citation)}\n {r.detail}\n") + + +def _vacuity_blockers(graded: Graded, doc: Path) -> list[str]: + """The control rod on the gate itself: a graded set of zero is a failure.""" + if graded.extracted == 0: + empty = ( + f"the gate GRADED NOTHING: 0 citations extracted from {doc}. A gate that grades an " + "empty set reports clean having checked nothing, so an empty set is a failure, not " + "a pass — check the doc path and the citation shape (src/bonfire/.py:NN[-NN])." + ) + return [empty] + if len(graded.results) != graded.extracted: + shortfall = ( + f"the gate graded {len(graded.results)} of the {graded.extracted} citations it " + "extracted: every extracted citation must produce a verdict, so a shortfall means " + "the grading loop dropped work." + ) + return [shortfall] + return [] + + +def _blockers( + graded: Graded, + coverage: Coverage, + blocked: list[CheckResult], + ratchet_violations: list[str], + doc: Path, +) -> list[str]: + """Every reason the gate could not do its job, in exit-2 terms.""" + reasons = list(ratchet_violations) + reasons.extend( + f"unregistered unverifiable citation: doc line {key[0]}, src/bonfire/{key[1]}:" + f"{key[2]}-{key[3]} — resolve it, or register it with a reason and bump frozen_count" + for key in coverage.uncovered + ) + reasons.extend( + f"cited source the gate could not read: doc line {r.citation.doc_line} " + f"({cited_range(r.citation)}) — {r.detail}" + for r in blocked + ) + reasons.extend( + f"stale {BASELINE_FILENAME} entry: {entry.label()} matches no unverifiable citation — " + "the citation moved or was repaired, so the blessing must be re-read or removed" + for entry in coverage.stale + ) + reasons.extend(_vacuity_blockers(graded, doc)) + return reasons + + +def _write_summary( + graded: Graded, + drifts: list[CheckResult], + coverage: Coverage, + blocked: list[CheckResult], +) -> None: + ok_count = sum(1 for r in graded.results if r.status == "ok") + checked = len(graded.results) + _write( + f"\nSummary: {ok_count} ok, {len(drifts)} drift, " + f"{len(coverage.covered)} registered-unverifiable, " + f"{len(coverage.uncovered)} unregistered-unverifiable, {len(blocked)} blocked, " + f"checked={checked} of {graded.extracted} extracted.\n" + ) + passed = checked > 0 and checked == graded.extracted + _write(f"Non-vacuity: {'PASS' if passed else 'FAIL'} — checked={checked} (> 0 required).\n") + + +def _write_blockers(blockers: list[str]) -> None: + for reason in blockers: + _write(f" - {reason}\n") + + +def _decide(drifts: list[CheckResult], blockers: list[str]) -> int: + """Write the verdict block and return the exit code.""" + if drifts: + _write( + f"\nVerdict: exit 1 — DRIFT. The gate DID its job and found {len(drifts)} " + "citation(s) pointing at moved code. Re-anchor them.\n" + ) + if blockers: + _write( + f" ({len(blockers)} could-not-verify blocker(s) also stand; exit 1 outranks " + "exit 2 because drift is the more actionable signal:)\n" + ) + _write_blockers(blockers) + return 1 + if blockers: + _write( + "\nVerdict: exit 2 — COULD NOT VERIFY. This is NOT 'the gate did its job and " + "found drift' (exit 1); it is 'the gate could not do its job':\n" + ) + _write_blockers(blockers) + return 2 + _write( + "\nVerdict: exit 0 — every citation resolved or is registered, and the graded " + "set is non-empty.\n" + ) + return 0 + + +def report(graded: Graded, baseline: Baseline, doc: Path) -> int: + """Print the whole board, then decide once. Returns the exit code.""" + drifts = [r for r in graded.results if r.status == "drift"] + unverified = [r for r in graded.results if r.status == "unverified"] + blocked = [r for r in graded.results if r.status == "blocked"] + by_key = {citation_key(r.citation): r for r in unverified} + coverage = cover(list(by_key), baseline) + ratchet_violations, ratchet_notices = ratchet(baseline) + + _report_notices(ratchet_notices) + _report_drift(drifts) + _report_registered(coverage.covered, by_key) + _report_unregistered(coverage.uncovered, by_key) + _report_blocked(blocked) + _write_summary(graded, drifts, coverage, blocked) + return _decide(drifts, _blockers(graded, coverage, blocked, ratchet_violations, doc)) + + +def fail_to_run(message: str) -> int: + """Report a structural inability to run and return the reserved code.""" + verdict = "\nVerdict: exit 2 — COULD NOT VERIFY. The gate could not do its job:\n" + _write(f"{verdict} - {message}\n") + return 2 diff --git a/scripts/citation_source.py b/scripts/citation_source.py new file mode 100644 index 00000000..efee7d61 --- /dev/null +++ b/scripts/citation_source.py @@ -0,0 +1,137 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 BonfireAI +"""What the SOURCE says: the AST symbol index the citation gate resolves against. + +One of four modules behind ``check_protocol_doc_citations.py`` (see +``citation_doc`` for the split). This one answers a single question: +"what symbol, if any, lives at line N of this file, and what are its +nearest neighbours?". The driver keeps the resolution rules and the CLI. + +Every failure here is a typed :class:`SourceIndexError`. A cited source +file that cannot be read or parsed means the gate COULD NOT DO ITS JOB +(the driver turns that into exit 2), never a silently skipped citation. +""" + +from __future__ import annotations + +import ast +from dataclasses import dataclass, field +from pathlib import Path + + +class SourceIndexError(Exception): + """A cited source file could not be read or parsed. + + Typed rather than a bare raise so the driver can tell "this citation + is unverifiable" apart from "this gate crashed". + """ + + +@dataclass +class SymbolIndex: + """Top-level + nested symbols extracted from a source file via ``ast``.""" + + # name -> (start_line, end_line). Includes both top-level and + # dotted ``Class.method`` entries so the doc's + # ``ConversationEngine.start`` shape resolves naturally. + by_name: dict[str, tuple[int, int]] = field(default_factory=dict) + + # Flat ordered list of (start_line, end_line, name) for the + # containment pass. Ordered by widening span so the FIRST match + # while iterating gives the tightest enclosing symbol. + intervals: list[tuple[int, int, str]] = field(default_factory=list) + + +def index_symbols(source_path: Path) -> SymbolIndex: # noqa: C901 + """Return name -> line-range map for every def/class in *source_path*. + + Top-level functions/classes are keyed by their bare name. Methods + are also keyed dotted (``Class.method``). Module-level assignments + to ``UPPER_CASE`` or ``_underscore`` names (``_SERVER_TYPES``, + ``MAX_USER_MESSAGE_LEN``, etc.) are indexed too — the doc cites + those tables/constants by name and we want to detect when they + move. + + Raises :class:`SourceIndexError` when the file cannot be read or + parsed. A gate that cannot read the code it grades has not passed. + """ + try: + text = source_path.read_text(encoding="utf-8") + except OSError as exc: + raise SourceIndexError(f"cannot read cited source {source_path}: {exc}") from exc + try: + tree = ast.parse(text, filename=str(source_path)) + except (SyntaxError, ValueError) as exc: + raise SourceIndexError(f"cannot parse cited source {source_path}: {exc}") from exc + index = SymbolIndex() + + def _add(name: str, lineno: int, end_lineno: int | None) -> None: + if end_lineno is None: + end_lineno = lineno + # First-write-wins so a re-bound name doesn't clobber the + # earlier (and usually authoritative) definition. + index.by_name.setdefault(name, (lineno, end_lineno)) + index.intervals.append((lineno, end_lineno, name)) + + for node in tree.body: + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef): + _add(node.name, node.lineno, node.end_lineno) + elif isinstance(node, ast.ClassDef): + _add(node.name, node.lineno, node.end_lineno) + # Walk class body for methods. + for child in node.body: + if isinstance(child, ast.FunctionDef | ast.AsyncFunctionDef): + _add(child.name, child.lineno, child.end_lineno) + _add(f"{node.name}.{child.name}", child.lineno, child.end_lineno) + elif isinstance(node, ast.Assign): + # Module-level constants / tables. + for target in node.targets: + if isinstance(target, ast.Name): + _add(target.id, node.lineno, node.end_lineno) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + _add(node.target.id, node.lineno, node.end_lineno) + + # Sort intervals so the tightest (smallest span) enclosing range + # comes first when we filter by ``start <= cited <= end``. + index.intervals.sort(key=lambda triple: triple[1] - triple[0]) + return index + + +def innermost_containing(index: SymbolIndex, lineno: int) -> tuple[int, int, str] | None: + """Return the smallest indexed (start, end, name) that contains ``lineno``.""" + for start, end, name in index.intervals: + if start <= lineno <= end: + return (start, end, name) + return None + + +def _describe(interval: tuple[int, int, str] | None, empty: str) -> str: + """Render one neighbour as ``name (lines A-B)``, or *empty* when absent.""" + if interval is None: + return empty + start, end, name = interval + return f"{name} (lines {start}-{end})" + + +def neighbours(index: SymbolIndex, lineno: int) -> tuple[str, str]: + """Describe the nearest indexed symbol above and below *lineno*. + + This is the actionable half of an "unverified" report: knowing that + the cited line sits in the gap between ``UserMessage`` and + ``_SERVER_TYPES`` is what lets a human see, in one read, which + symbol the doc probably meant and which one it actually points near. + """ + above = max( + (iv for iv in index.intervals if iv[1] < lineno), + key=lambda iv: iv[1], + default=None, + ) + below = min( + (iv for iv in index.intervals if iv[0] > lineno), + key=lambda iv: iv[0], + default=None, + ) + return ( + _describe(above, "nothing indexed above it"), + _describe(below, "nothing indexed below it"), + ) diff --git a/tests/scripts/test_check_protocol_doc_citations.py b/tests/scripts/test_check_protocol_doc_citations.py index 8b982222..57af773f 100644 --- a/tests/scripts/test_check_protocol_doc_citations.py +++ b/tests/scripts/test_check_protocol_doc_citations.py @@ -3,29 +3,38 @@ """Tests for the protocol-doc citation drift checker. -Covers: - -1. The checker resolves every current citation in - ``docs/scan-front-door-protocol.md`` against the current source — - the doc and the code are in sync as of this commit, so a clean run - must report 0 drift (the W11 Lane C postcondition). -2. A deliberately drifted citation (line number bumped past every - indexed symbol and with no recoverable hint) is flagged as - ``unverified`` — the checker does not silently treat unknown lines - as OK. -3. A citation whose doc-hint names a real symbol that has moved is - flagged as ``drift`` — the checker actually catches the failure - mode it exists to catch. -4. A citation whose cited line sits inside a real - class/function/assignment is reported as ``ok`` with the resolved - symbol name — the containment pass works. +The checker's contract is a three-valued exit code: + +* **0** — every citation resolved (or is registered in + ``citation-baseline.json``) AND the run graded a non-empty set. +* **1** — DRIFT: the gate did its job and a citation points at moved code. +* **2** — COULD NOT VERIFY: the gate could not do its job (an + unregistered unverifiable citation, an absent/unparseable cited + source, an unreadable doc, a malformed or over-ratchet registry, a + stale registry entry, or a run that graded zero citations). + +Every branch of that contract has a control rod below: a fixture built +in ``tmp_path`` with its own tiny source tree, driven through the real +``main()`` via ``--doc`` / ``--source-root`` / ``--baseline``. Rods that +must go RED assert the exit code AND the text a human would act on. +The GREEN rods are the counterfactual, so a rod going red for the wrong +reason cannot hide: the same fixture harness exits 0 on a resolvable +citation, and the real repo doc exits 0 having graded a non-empty set. + +Unit-level coverage of the resolution passes (containment, hint, +drift, unverified) sits at the bottom. """ from __future__ import annotations import importlib.util +import json +import re import sys from pathlib import Path +from typing import Any + +import pytest _SCRIPT_PATH = Path(__file__).resolve().parents[2] / "scripts" / "check_protocol_doc_citations.py" _spec = importlib.util.spec_from_file_location("check_protocol_doc_citations", _SCRIPT_PATH) @@ -34,21 +43,348 @@ sys.modules["check_protocol_doc_citations"] = _module _spec.loader.exec_module(_module) +# The script puts its own ``scripts/`` directory on sys.path and imports +# its siblings by bare name, so executing it above registered them. Read +# them out of sys.modules rather than importing here: an import +# statement after module-level code is E402, and the script is the only +# thing that legitimately owns that path insert. +_source = sys.modules["citation_source"] +_baseline_mod = sys.modules["citation_baseline"] +_doc_mod = sys.modules["citation_doc"] + _REPO = Path(__file__).resolve().parents[2] _PROTOCOL_PY = _REPO / "src" / "bonfire" / "onboard" / "protocol.py" +# The fixture source tree every rod resolves its citations against. +# Line map, load-bearing for every cited line number below: +# 1 module docstring · 3 CONSTANT · 6 class Widget · 9 def spin +# 10 return (the last line inside Widget.spin) +# Lines 2, 4, 5 are module-level gaps: cited there, a citation is inside +# no indexed symbol, which is what forces the hint pass. +_FIXTURE_SOURCE = '''"""Fixture module.""" + +CONSTANT = 1 + + +class Widget: + """A widget.""" + + def spin(self) -> int: + return CONSTANT +''' + +# A citation whose hint (``Widget``) resolves to a real symbol at line 6 +# while the cited line is 20: real drift, and far enough from the symbol +# that the rod does not depend on the value of ``--tolerance``. +_DOC_DRIFT = """# Fixture protocol doc + +The `Widget` class: `src/bonfire/fixture/sample.py:20`. +""" + +# Cited line 4 is a module-level gap and no backticked identifier appears +# in the preceding lines, so neither pass resolves: unverifiable. +_DOC_UNVERIFIED = """# Fixture protocol doc + +The registry table has exactly one entry: +`src/bonfire/fixture/sample.py:4`. +""" + +# Same, but the one hint in range is on the checker's SKIP_HINTS list. +_DOC_SKIPPED_HINT = """# Fixture protocol doc + +The `scan_update` frame is described above. +`src/bonfire/fixture/sample.py:4`. +""" + +# Cited line 10 is inside ``Widget.spin``: resolves by containment. +_DOC_CLEAN = """# Fixture protocol doc + +The spin method: `src/bonfire/fixture/sample.py:10`. +""" + +_DOC_ABSENT_SOURCE = """# Fixture protocol doc + +Absent module: `src/bonfire/fixture/ghost.py:5`. +""" + +_DOC_NO_CITATIONS = """# Fixture protocol doc + +This doc cites no source lines at all. +""" + + +def _entry(doc_line: int, cited_start: int, cited_end: int) -> dict[str, Any]: + """One well-formed registry row for the fixture tree.""" + return { + "doc_line": doc_line, + "path": "fixture/sample.py", + "cited_start": cited_start, + "cited_end": cited_end, + "finding": "fixture citation: the doc names the registry table, the range covers a gap", + "reason": "fixture entry for the control rods; never a real blessing", + } + + +def _argv( + tmp_path: Path, + doc_body: str, + *, + baseline: dict[str, Any] | None = None, +) -> list[str]: + """Write a fixture doc + source tree + registry; return the CLI argv. + + ``--baseline`` always points inside ``tmp_path``, even when no + registry is written, so no rod can accidentally read (or be rescued + by) the repository's real ``citation-baseline.json``. + """ + doc = tmp_path / "doc.md" + doc.write_text(doc_body, encoding="utf-8") + source_root = tmp_path / "srcroot" + (source_root / "fixture").mkdir(parents=True) + (source_root / "fixture" / "sample.py").write_text(_FIXTURE_SOURCE, encoding="utf-8") + registry = tmp_path / "citation-baseline.json" + if baseline is not None: + registry.write_text(json.dumps(baseline, indent=2), encoding="utf-8") + return [ + "--doc", + str(doc), + "--source-root", + str(source_root), + "--baseline", + str(registry), + ] + + +# --------------------------------------------------------------------------- +# RED rods — each proves the gate refuses +# --------------------------------------------------------------------------- + + +def test_drifted_citation_exits_one( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 1: a citation resolving to a moved symbol is DRIFT -> exit 1.""" + rc = _module.main(_argv(tmp_path, _DOC_DRIFT)) + err = capsys.readouterr().err + assert rc == 1, f"drift must exit 1, got {rc}; report was:\n{err}" + assert "DRIFT" in err + assert "Verdict: exit 1" in err + -def test_current_doc_has_no_drift() -> None: - """The doc shipped in this commit must match current source exactly.""" +def test_unregistered_unverifiable_citation_exits_two( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 2: an unverifiable citation nobody registered -> exit 2, not 0. + + This is the defect this gate was rebuilt to close: ``unverified`` + used to print on stderr and return 0, so the gate was most + permissive exactly where its confidence was lowest. The report must + also be actionable — it names the neighbouring symbols and says why + no hint was usable. + """ + rc = _module.main(_argv(tmp_path, _DOC_UNVERIFIED)) + err = capsys.readouterr().err + assert rc == 2, f"an unregistered unverifiable citation must exit 2, got {rc}:\n{err}" + assert "COULD NOT VERIFY" in err + assert "unregistered unverifiable citation" in err + assert "nearest symbol above: CONSTANT (lines 3-3)" in err + assert "nearest symbol below: Widget (lines 6-10)" in err + assert "no backticked identifier found in the 5 doc lines" in err + + +def test_unverifiable_report_names_the_skip_hints_rejection( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """A hint rejected by SKIP_HINTS is reported by name, not silently dropped.""" + rc = _module.main(_argv(tmp_path, _DOC_SKIPPED_HINT)) + err = capsys.readouterr().err + assert rc == 2 + assert "`scan_update` (listed in citation_doc.SKIP_HINTS)" in err + assert "scan_update" in _doc_mod.SKIP_HINTS, "the rod must name a real SKIP_HINTS member" + + +def test_doc_with_zero_citations_is_not_clean( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 3: a graded set of zero must NOT report clean -> exit 2. + + The control rod on the gate itself. A gate that selects the set it + grades must assert that set is non-empty, or it reports success + having checked nothing. + """ + rc = _module.main(_argv(tmp_path, _DOC_NO_CITATIONS)) + err = capsys.readouterr().err + assert rc == 2, f"a doc with no citations must not pass, got {rc}:\n{err}" + assert "GRADED NOTHING" in err + assert "0 citations extracted" in err + assert "Non-vacuity: FAIL — checked=0" in err + + +def test_absent_cited_source_exits_two( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 4: a citation naming a file that does not exist -> exit 2.""" + rc = _module.main(_argv(tmp_path, _DOC_ABSENT_SOURCE)) + err = capsys.readouterr().err + assert rc == 2, f"an absent cited source must exit 2, got {rc}:\n{err}" + assert "could not read" in err + assert "ghost.py" in err + + +def test_registry_cannot_launder_a_drift( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 5: a registry entry aimed at a DRIFT still exits 1. + + The registry covers the ``unverified`` bucket only. A citation that + resolves and is wrong fails no matter what is written about it. + """ + baseline = {"frozen_count": 1, "entries": [_entry(doc_line=3, cited_start=20, cited_end=20)]} + rc = _module.main(_argv(tmp_path, _DOC_DRIFT, baseline=baseline)) + err = capsys.readouterr().err + assert rc == 1, f"a registered drift must still exit 1, got {rc}:\n{err}" + assert "DRIFT" in err + assert "0 registered-unverifiable" in err + assert "REGISTERED" not in err, "a drift must never be reported as a registered pass" + + +def test_added_entry_without_a_frozen_count_bump_exits_two( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 6: the ratchet — an entry added without bumping frozen_count is red.""" + baseline = {"frozen_count": 0, "entries": [_entry(doc_line=4, cited_start=4, cited_end=4)]} + rc = _module.main(_argv(tmp_path, _DOC_UNVERIFIED, baseline=baseline)) + err = capsys.readouterr().err + assert rc == 2, f"an unbumped ratchet must exit 2, got {rc}:\n{err}" + assert "must bump frozen_count in the same commit" in err + + +def test_stale_registry_entry_exits_two( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """An entry matching no unverifiable citation is a violation, not a notice.""" + baseline = {"frozen_count": 1, "entries": [_entry(doc_line=99, cited_start=4, cited_end=4)]} + rc = _module.main(_argv(tmp_path, _DOC_CLEAN, baseline=baseline)) + err = capsys.readouterr().err + assert rc == 2, f"a stale entry must exit 2, got {rc}:\n{err}" + assert "stale citation-baseline.json entry" in err + + +def test_malformed_registry_exits_two( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """A registry entry missing its reason cannot buy a pass.""" + broken = _entry(doc_line=4, cited_start=4, cited_end=4) + del broken["reason"] + baseline = {"frozen_count": 1, "entries": [broken]} + rc = _module.main(_argv(tmp_path, _DOC_UNVERIFIED, baseline=baseline)) + err = capsys.readouterr().err + assert rc == 2, f"a malformed registry must exit 2, got {rc}:\n{err}" + assert "citation-baseline.json is unusable" in err + + +def test_unreadable_doc_exits_two( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """A doc that cannot be read is a structural inability to run -> exit 2.""" + rc = _module.main(["--doc", str(tmp_path / "nope.md")]) + err = capsys.readouterr().err + assert rc == 2, f"an unreadable doc must exit 2, got {rc}:\n{err}" + assert "cannot read the doc under test" in err + + +# --------------------------------------------------------------------------- +# GREEN rods — the counterfactual, so a red rod cannot be red for free +# --------------------------------------------------------------------------- + + +def test_real_doc_and_real_registry_exit_zero_on_a_nonempty_graded_set( + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 7: the shipped doc + shipped registry exit 0, having graded > 0. + + BEHAVIOUR CHANGE (this assertion replaces the old + ``test_current_doc_has_no_drift``): exit 0 used to mean only "no + drift", and the run that measured this gate reached 0 with two + ``unverified`` citations that were both real drift — the pass was + partly manufactured by a category that could not fail. Exit 0 now + means every citation resolved OR is named in + ``citation-baseline.json`` with a written reason, AND the graded set + was non-empty. The ``checked > 0`` assertion is the other half: a + doc whose citations stopped being extractable would have satisfied + the old assertion by grading nothing. + """ rc = _module.main([]) - assert rc == 0, "protocol doc citations have drifted; re-anchor them" + err = capsys.readouterr().err + assert rc == 0, f"the shipped doc + registry must exit 0, got {rc}:\n{err}" + match = re.search(r"checked=(\d+) of (\d+) extracted", err) + assert match is not None, f"the summary must report the graded count:\n{err}" + checked, extracted = int(match.group(1)), int(match.group(2)) + assert checked > 0, "a clean run must have graded a non-empty set of citations" + assert checked == extracted, "every extracted citation must have been graded" + assert "Non-vacuity: PASS" in err + + +def test_registered_unverifiable_citation_passes_loudly_with_its_reason( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Rod 8: a registered unverifiable citation exits 0 and prints its reason.""" + entry = _entry(doc_line=4, cited_start=4, cited_end=4) + baseline = {"frozen_count": 1, "entries": [entry]} + rc = _module.main(_argv(tmp_path, _DOC_UNVERIFIED, baseline=baseline)) + err = capsys.readouterr().err + assert rc == 0, f"a registered unverifiable citation must exit 0, got {rc}:\n{err}" + assert "REGISTERED" in err + assert entry["reason"] in err + assert entry["finding"] in err + assert "citation baseline: 1 registered entry, frozen_count=1" in err + + +def test_clean_fixture_exits_zero_without_a_registry( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """The harness itself is not rigged red: a resolvable citation exits 0.""" + rc = _module.main(_argv(tmp_path, _DOC_CLEAN)) + err = capsys.readouterr().err + assert rc == 0, f"a containment-resolved citation must exit 0, got {rc}:\n{err}" + assert "Verdict: exit 0" in err + + +def test_removed_entries_leave_shrinkable_slack( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Entries may be removed freely; the leftover frozen_count prints as slack.""" + baseline: dict[str, Any] = {"frozen_count": 3, "entries": []} + rc = _module.main(_argv(tmp_path, _DOC_CLEAN, baseline=baseline)) + err = capsys.readouterr().err + assert rc == 0, f"slack is not a violation, got {rc}:\n{err}" + assert "citation baseline slack: frozen_count=3" in err + assert "it may shrink to 0" in err + + +# --------------------------------------------------------------------------- +# Resolution-pass units (against the real protocol module) +# --------------------------------------------------------------------------- def test_drift_is_flagged_when_hint_resolves_to_moved_symbol() -> None: """A citation that names a symbol but points at the wrong line is drift.""" - idx = _module._index_symbols(_PROTOCOL_PY) - cite = _module.Citation(doc_line=1, path="onboard/protocol.py", start=15, end=15) + idx = _source.index_symbols(_PROTOCOL_PY) + cite = _doc_mod.Citation(doc_line=1, path="onboard/protocol.py", start=15, end=15) result = _module._check_citation(cite, hint="ConversationStart", index=idx, tolerance=3) assert result.status == "drift" # The actual ConversationStart class is well-known to live somewhere @@ -60,21 +396,21 @@ def test_drift_is_flagged_when_hint_resolves_to_moved_symbol() -> None: def test_out_of_range_citation_is_unverified_not_silently_ok() -> None: """A cited line past EOF with no hint MUST be ``unverified``.""" - idx = _module._index_symbols(_PROTOCOL_PY) - cite = _module.Citation(doc_line=1, path="onboard/protocol.py", start=99_999, end=99_999) + idx = _source.index_symbols(_PROTOCOL_PY) + cite = _doc_mod.Citation(doc_line=1, path="onboard/protocol.py", start=99_999, end=99_999) result = _module._check_citation(cite, hint=None, index=idx, tolerance=3) assert result.status == "unverified" def test_containment_pass_resolves_in_body_citation() -> None: """A citation that lands inside a class body is OK + names the class.""" - idx = _module._index_symbols(_PROTOCOL_PY) + idx = _source.index_symbols(_PROTOCOL_PY) # ``ScanUpdate``'s ``detail`` field lives inside the ScanUpdate # class. The citation does not have to point at the class header # to count as OK. scan_update_range = idx.by_name["ScanUpdate"] in_body_line = scan_update_range[1] # last line of the class - cite = _module.Citation( + cite = _doc_mod.Citation( doc_line=1, path="onboard/protocol.py", start=in_body_line, @@ -83,3 +419,20 @@ def test_containment_pass_resolves_in_body_citation() -> None: result = _module._check_citation(cite, hint=None, index=idx, tolerance=3) assert result.status == "ok" assert result.resolved_symbol == "ScanUpdate" + + +def test_unparseable_source_raises_a_typed_error() -> None: + """The source indexer speaks a typed error, never a bare crash.""" + with pytest.raises(_source.SourceIndexError): + _source.index_symbols(_REPO / "docs" / "scan-front-door-protocol.md") + + +def test_shipped_registry_entries_each_name_one_citation() -> None: + """The shipped registry parses, and every entry is a single-citation claim.""" + baseline = _baseline_mod.load_baseline(_REPO / _baseline_mod.BASELINE_FILENAME) + assert len(baseline.entries) <= baseline.frozen_count + keys = [entry.key for entry in baseline.entries] + assert len(keys) == len(set(keys)), "two entries must never cover the same citation" + for entry in baseline.entries: + assert "*" not in entry.path and "?" not in entry.path + assert entry.finding.strip() and entry.reason.strip()