diff --git a/.github/workflows/claim-vocabulary-check.yml b/.github/workflows/claim-vocabulary-check.yml new file mode 100644 index 00000000..4fe163cd --- /dev/null +++ b/.github/workflows/claim-vocabulary-check.yml @@ -0,0 +1,54 @@ +name: Claim Vocabulary Check + +# AAASM-5798 / AAASM-5599: ADR 0033's banned-absolute and undifferentiated-verb +# rules against this repo's reader-facing Markdown (README, CONTRIBUTING, +# SECURITY, docs/**). Ported from agent-assembly's check_claim_vocabulary.py +# (AAASM-5679) and docs Hub's port (AAASM-5797); per AAASM-5795's +# federation-by-convention decision this repo's copy is independent -- see +# scripts/check_claim_vocabulary.py's own docstring. +# +# Runs where CLAUDE.md's own "Docs-only PRs run NO CI" gap otherwise leaves +# docs/**/*.md and README.md changes with nothing but human review: this repo's +# main ci.yaml path filter excludes docs/** and *.md entirely. +# +# Full-tree, not diff-scoped: the tree's blocking baseline is zero as of this +# same PR (five real findings fixed alongside the port), so per +# claim-vocabulary.md's adoption sequence there is no backlog to work down +# with --diff-base. + +on: + pull_request: + paths: + - "README.md" + - "CONTRIBUTING.md" + - "SECURITY.md" + - "docs/**" + - "scripts/check_claim_vocabulary.py" + - ".github/workflows/claim-vocabulary-check.yml" + push: + branches: + - main + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + claim-vocabulary: + name: Check for banned-absolute and undifferentiated-verb claims + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + + - name: Run the claim-vocabulary self-test + run: python3 scripts/check_claim_vocabulary.py --selftest + + - name: Scan for banned-absolute and undifferentiated-verb claims + run: python3 scripts/check_claim_vocabulary.py --root . diff --git a/README.md b/README.md index ce84455a..e0328fa6 100644 --- a/README.md +++ b/README.md @@ -177,7 +177,7 @@ with init_assembly( What this does: -1. `init_assembly()` registers the agent with the gateway and auto-loads the LangChain adapter — every tool call from now on goes through the policy gate. +1. `init_assembly()` registers the agent with the gateway and auto-loads the LangChain adapter — each tool call the wrapped agent makes from now on goes through the policy gate. 2. The `FakeListLLM` replays canned responses so the example runs **offline** with no real LLM. 3. The `with` block tears down the gateway connection and unwinds adapter hooks on exit. diff --git a/docs/concepts/index.md b/docs/concepts/index.md index 6744049f..71e8435c 100644 --- a/docs/concepts/index.md +++ b/docs/concepts/index.md @@ -127,7 +127,7 @@ Two independent knobs control governance. It's worth keeping them straight: | --- | --- | | `auto` (default) | Picks the best available layer for the current platform (eBPF on Linux, else proxy). | | `sdk-only` | In-process only — framework adapters enforce on tool calls; no network sidecar. Most portable; best for tests. | -| `proxy` | Routes outbound traffic through the `aasm` sidecar proxy — network-egress policy with no code changes. | +| `proxy` | Routes outbound traffic through the `aasm` sidecar proxy — network-egress policy without modifying the agent's source. | | `ebpf` | Kernel-level interception via eBPF. **Linux only** — raises `ConfigurationError` elsewhere. | ### Enforcement modes diff --git a/docs/examples/langchain-basic-agent.md b/docs/examples/langchain-basic-agent.md index b38663e1..9af89fab 100644 --- a/docs/examples/langchain-basic-agent.md +++ b/docs/examples/langchain-basic-agent.md @@ -4,7 +4,7 @@ Integrate Agent Assembly with LangChain to enforce governance policy on tool cal ## What this example demonstrates -This example wires Agent Assembly governance into a LangChain agent so that every tool call is checked against policy *before* the tool runs. It covers: +This example wires Agent Assembly governance into a LangChain agent so that each tool call the agent makes is checked against policy *before* the tool runs. It covers: - Initializing Agent Assembly with `init_assembly()`. - Wrapping LangChain tools with `AssemblyCallbackHandler` + a governance interceptor. diff --git a/docs/examples/llamaindex-tool-policy.md b/docs/examples/llamaindex-tool-policy.md index fbe4a5d0..31d35526 100644 --- a/docs/examples/llamaindex-tool-policy.md +++ b/docs/examples/llamaindex-tool-policy.md @@ -1,6 +1,6 @@ # LlamaIndex -Integrates Agent Assembly with [LlamaIndex](https://docs.llamaindex.ai/) using the **native `LlamaIndexAdapter`**, so every tool call a LlamaIndex agent makes is governed automatically — no per-tool wrapper. +Integrates Agent Assembly with [LlamaIndex](https://docs.llamaindex.ai/) using the **native `LlamaIndexAdapter`**, so each tool call a LlamaIndex agent makes is governed automatically — no per-tool wrapper. ## What this example demonstrates diff --git a/docs/index.md b/docs/index.md index 72b07d43..2e667b5f 100644 --- a/docs/index.md +++ b/docs/index.md @@ -1,7 +1,7 @@ # Agent Assembly Python SDK **In plain terms:** this SDK is how a Python agent asks for permission before it acts. -You wrap your existing agent in one `init_assembly()` call, and from that point on every +You wrap your existing agent in one `init_assembly()` call, and from that point on each tool call your agent makes is checked against a governance policy — allowed or denied — without you rewriting a single line of the agent itself. Over a connected runtime the outcome of each call is also handed to the runtime's event channel (see below). diff --git a/scripts/check_claim_vocabulary.py b/scripts/check_claim_vocabulary.py new file mode 100644 index 00000000..3bb0a70a --- /dev/null +++ b/scripts/check_claim_vocabulary.py @@ -0,0 +1,711 @@ +#!/usr/bin/env python3 +"""Enforce ADR 0033's claim vocabulary and banned-absolutes rules against +python-sdk's reader-facing Markdown. AAASM-5798 / AAASM-5599. + +Adapted from agent-assembly's `scripts/check_claim_vocabulary.py` (AAASM-5679) +and docs Hub's port (AAASM-5797). Per AAASM-5795's federation-by-convention +decision this is a ONE-TIME PORT, not a shared dependency -- this file +diverges independently from the two it was ported from. What is kept +verbatim is kept because the underlying problem is identical (Markdown +source, the same ADR 0033 forbidden-design-7 absolutes, the same +false-positive shape from quoting a banned phrase to forbid it), not because +of a code-sharing obligation. + +File scope differs from both prior ports: `quickstart_snippets/` is excluded +(ruff.toml already excludes it for the same reason -- vendored, verbatim +copies from the `examples` repo, regenerated by `generate_quickstart_tabs.py` +into `docs/quick-start.md`; this repo does not author that prose and +flagging it would point an author at text they cannot fix here) and so is +`verification-reports/**` (its purpose is to quote overstatements in order +to disprove them, same as agent-assembly's own exclusion). + +Run against tracked Markdown source: + + python3 scripts/check_claim_vocabulary.py + python3 scripts/check_claim_vocabulary.py --diff-base remote/main + +Exit 0 = clean or --report-only, 1 = a blocking claim published, 2 = the +checker could not prove itself (see --selftest). + +Waivers: NONE of this file's rules are waivable. `claim-vocabulary.md` (in +agent-assembly) §7.1 states no waiver reaches a banned absolute +(`CLAIM-ABS-*`) or an undifferentiated verb (`CLAIM-VERB-01`) -- both are +unwaivable everywhere, not just in the repo that specification lives in. +""" + +from __future__ import annotations + +import argparse +import fnmatch +import re +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path + +# -------------------------------------------------------------------------- +# Macros. Each carries its own non-capturing group -- textual substitution of +# a bare alternation silently changes what the collocation rules mean. +# -------------------------------------------------------------------------- +MACROS: dict[str, str] = { + "SEP": r"(?:[-‑_\s]+)", + "DOC-NOUN": r"(?:reference|guide|list|example|walkthrough|inventory|history|re-audit|rewrite|set\b)", + "GOV-NOUN": ( + r"(?:coverage|protection|mediation|interception|enforcement|visibility" + r"|observability|monitoring|detection|inspection|audit(?:ing|s)?" + r"|governance|security|telemetry)" + ), + "SUBJ": ( + r"(?:Agent\s+Assembly|Assembly|the\s+(?:gateway|proxy|runtime|SDK|sandbox" + r"|platform|product|CLI|dashboard|policy\s+engine)|aa-[a-z-]+)" + ), +} + +NEG_PATTERN = re.compile( + r"(?:\bno\b|\bnot\b|\bnever\b|\bneither\b|\bnor\b|\bwithout\b|\bnothing\b" + r"|\bcannot\b|\bcan't\b|\bisn't\b|\baren't\b|\bdoesn't\b|\bdon't\b" + r"|\brather\s+than\b|\binstead\s+of\b|\bnon-|\bunder-|\bincomplete\b)", + re.IGNORECASE, +) +NEG_WINDOW_CHARS = 70 +CLAUSE_BOUNDARIES = set(".;!?\n") + +CFG_NOUN_PATTERN = re.compile( + r"\s*(?:entry|entries|rule|rules|pattern|handler|route|case|branch|glob" + r"|selector|wildcard|for\b)", + re.IGNORECASE, +) + +_FILLER = "\x00" +_NEWLINE_FILLER = "." + + +@dataclass(frozen=True) +class Rule: + rule_id: str + severity: str + pattern: str + guards: tuple[str, ...] + + +RULES: tuple[Rule, ...] = ( + Rule("CLAIM-ABS-01", "blocking", r"catch(?:es|ing)?everything", ("NEG",)), + Rule("CLAIM-ABS-02", "finding", r"catch[-‑_\s]?all", ("NEG", "CFG-NOUN")), + Rule( + "CLAIM-ABS-03", + "blocking", + r"(?:can\s?not|cannot|can't|couldnot)(?:be)?bypass(?:ed)?", + (), + ), + Rule("CLAIM-ABS-04", "blocking", r"un-?bypassable", ()), + Rule("CLAIM-ABS-05", "blocking", r"nowheretohide", ()), + Rule("CLAIM-ABS-06", "finding", r"everyaction", ("NEG",)), + Rule("CLAIM-ABS-07", "blocking", r"everytoolcalls?", ("NEG",)), + Rule("CLAIM-ABS-08", "blocking", r"nocodechanges?", ()), + Rule("CLAIM-ABS-09", "blocking", r"immutableaudit", ()), + Rule("CLAIM-ABS-10", "blocking", r"(?:full|whole)fleet", ()), + Rule( + "CLAIM-ABS-11", + "finding", + r"\b(?:complete|comprehensive|universal)\s+(?!)[^.;:!?]{0,40}?\b\b", + ("NEG",), + ), + Rule( + "CLAIM-ABS-12", + "finding", + r"\b\b[^.;:!?]{0,40}?\b(?:is|are|was|were|remains?)\b" + r"[^.;:!?]{0,15}?\b(?:complete|comprehensive|universal)\b", + ("NEG",), + ), + Rule( + "CLAIM-VERB-01", + "finding", + r"\b\b[^.;:!?]{0,30}?\b(?:protects|enforces|catches|prevents" + r"|guarantees|blocks|stops)\s+(?:the|a|an|all|every|any|its|their|each)?" + r"\s*[a-z][a-z-]{2,}", + ("NEG",), + ), +) + +QUOTE_RULE_ID = "CLAIM-QUOTE-01" + + +def _expand(pattern: str) -> str: + for name, expansion in MACROS.items(): + pattern = pattern.replace(f"<{name}>", expansion) + return pattern + + +COMPILED: tuple[tuple[Rule, re.Pattern[str]], ...] = tuple( + (rule, re.compile(_expand(rule.pattern), re.IGNORECASE)) for rule in RULES +) + +# -------------------------------------------------------------------------- +# File scope. +# -------------------------------------------------------------------------- +EXTENSIONS = {".md", ".markdown"} + +INCLUDE_GLOBS = ( + "docs/**", + "README.md", + "**/README.md", + "CONTRIBUTING.md", + "SECURITY.md", +) + +EXCLUDE_GLOBS = ( + # Vendored, verbatim copies from the `examples` repo -- this repo does + # not author this prose. ruff.toml excludes the same directory for the + # same reason. + "quickstart_snippets/**", + "verification-reports/**", + ".venv/**", + "node_modules/**", +) + + +def _matches_any(path: str, globs: tuple[str, ...]) -> bool: + for pattern in globs: + if fnmatch.fnmatch(path, pattern): + return True + if pattern.endswith("/**") and (path == pattern[:-3] or path.startswith(pattern[:-2])): + return True + return False + + +def in_scope(path: str) -> bool: + if _matches_any(path, EXCLUDE_GLOBS): + return False + if Path(path).suffix not in EXTENSIONS: + return False + return _matches_any(path, INCLUDE_GLOBS) + + +def _is_repo_relative(root: Path, rel: str) -> bool: + candidate = Path(rel) + if candidate.is_absolute(): + try: + candidate.relative_to(root) + except ValueError: + return False + return (root / rel).exists() or not candidate.is_absolute() + + +_SAFE_REF = re.compile(r"^[A-Za-z0-9._/~^@{}:-]+$") + + +def _changed_lines(root: Path, base: str, targets: list[str]) -> dict[str, set[int]]: + """Line numbers added or modified since `base`, per file. Raised, not + swallowed, on failure -- a silent "nothing changed" would pass everything. + + `base` is validated against a plain git-ref character set before it + reaches subprocess (defence in depth; list-form argv with no shell=True + is not exploitable regardless). + """ + if not _SAFE_REF.match(base): + raise ValueError(f"--diff-base {base!r} is not a plain git ref (letters, digits, '.', '_', '-', '/' only)") + result = subprocess.run( # NOSONAR (S4721): list-form argv, no shell=True; base validated above. + ["git", "-c", "core.quotepath=false", "diff", "-U0", "--no-color", "--merge-base", base, "--", *targets], + cwd=root, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"git diff against {base!r} failed ({result.returncode}): {result.stderr.strip()}") + + changed: dict[str, set[int]] = {} + current: str | None = None + hunk = re.compile(r"^@@ -\S+ \+(\d+)(?:,(\d+))? @@") + for line in result.stdout.split("\n"): + if line.startswith("+++ b/"): + current = line[6:] + changed.setdefault(current, set()) + elif current and (m := hunk.match(line)): + start = int(m.group(1)) + count = int(m.group(2) or 1) + changed[current].update(range(start, start + count)) + return changed + + +def tracked_files(root: Path) -> list[str]: + out = subprocess.run( + ["git", "ls-files", "-z"], + cwd=root, + capture_output=True, + text=True, + check=True, + ) + return [p for p in out.stdout.split("\0") if p] + + +# -------------------------------------------------------------------------- +# Normalisation. Every step preserves length, so an offset in the normalised +# text is an offset in the original file. +# -------------------------------------------------------------------------- +_BLOCK_START = re.compile( + r"""^(?: + \#{1,6}\ # ATX heading + | [-*+]\ # bullet + | \d+[.)]\ # ordered list + | \| # table row + | (?:`{3,}|~{3,}) # code fence + | (?:\*\s*){3,}$|(?:-\s*){3,}$|(?:_\s*){3,}$ # thematic break + | <[A-Za-z/!] # HTML block + | (?:\ {4}|\t)\S # indented code + )""", + re.VERBOSE, +) +_BQ = re.compile(r"^((?:\s*>\s?)*)") + + +def _blockquote_depth(line: str) -> tuple[int, str]: + match = _BQ.match(line) + marker = match.group(1) if match else "" + return marker.count(">"), line[len(marker) :] + + +_META_TAG = re.compile(r"]*>", re.IGNORECASE | re.DOTALL) +_SETEXT_UNDERLINE = re.compile(r"^\s{0,3}(?:=+|-+)\s*$") +_TITLE_DESC_LINE = re.compile(r"^\s*(?:title|description)\s*:", re.IGNORECASE) +_ATX_HEADING = re.compile(r"^\s{0,3}#{1,6}\s") + + +def _opens_front_matter(lines: list[str]) -> bool: + """True when a leading `---` is YAML front matter rather than a thematic + break. This repo's Markdown (unlike docs Hub's mdBook corpus) may carry + front matter, so -- unlike the docs Hub port -- this check is real, not + asserted away.""" + key = re.compile(r"^\s*[A-Za-z_][\w.-]*\s*:") + for line in lines[1:]: + if line.strip() in {"---", "..."}: + return True + if line.strip() and not key.match(line): + return False + return False + + +def _structural_ranges(text: str) -> list[tuple[int, int]]: + """Offsets where a backticked or quoted banned absolute keeps its own + severity: front matter, a heading, a `title:`/`description:` value, or a + `` tag -- each is quoted alone somewhere a reader never sees the + surrounding sentence's exemption label.""" + ranges: list[tuple[int, int]] = [] + offset = 0 + lines = text.split("\n") + offsets: list[int] = [] + running = 0 + for line in lines: + offsets.append(running) + running += len(line) + 1 + + in_front_matter = False + front_matter_lines: set[int] = set() + for index, line in enumerate(lines): + start, end = offset, offset + len(line) + offset = end + 1 + + if index == 0 and line.strip() == "---" and _opens_front_matter(lines): + in_front_matter = True + front_matter_lines.add(index) + continue + if in_front_matter: + front_matter_lines.add(index) + if line.strip() in {"---", "..."}: + in_front_matter = False + else: + ranges.append((start, end)) + continue + + if _ATX_HEADING.match(line): + ranges.append((start, end)) + continue + if m := _TITLE_DESC_LINE.match(line): + ranges.append((start + m.end(), end)) + continue + + for index in range(len(lines) - 1): + if index in front_matter_lines or (index + 1) in front_matter_lines: + continue + if not _SETEXT_UNDERLINE.match(lines[index + 1]): + continue + body = lines[index] + if not body.strip() or _BLOCK_START.match(body): + continue + start = offsets[index] + ranges.append((start, start + len(body))) + + for m in _META_TAG.finditer(text): + ranges.append((m.start(), m.end())) + return ranges + + +def _in_ranges(ranges: list[tuple[int, int]], offset: int) -> bool: + return any(lo <= offset < hi for lo, hi in ranges) + + +def _mask_code_regions(text: str, structural: list[tuple[int, int]] | None = None) -> str: + """Returns a same-length string with exempt regions replaced by `_FILLER`.""" + chars = list(text) + n = len(text) + + def blank(start: int, end: int) -> None: + for i in range(start, min(end, n)): + if chars[i] != "\n": + chars[i] = _FILLER + + for m in re.finditer(r"", text, re.DOTALL): + blank(m.start(), m.end()) + + lines_span: list[tuple[int, int, str]] = [] + offset = 0 + for line in text.splitlines(keepends=True): + lines_span.append((offset, offset + len(line), line)) + offset += len(line) + + fence: str | None = None + for start, end, line in lines_span: + stripped = line.lstrip() + m = re.match(r"(`{3,}|~{3,})", stripped) + if fence is None: + if m: + fence = m.group(1) + blank(start, end) + else: + blank(start, end) + if m and m.group(1)[0] == fence[0] and len(m.group(1)) >= len(fence): + fence = None + + for start, end, line in lines_span: + if re.match(r"^(?:\ {4}|\t)\S", line): + blank(start, end) + + masked = "".join(chars) + + structural = structural or [] + for m in re.finditer(r"(`+)[^`\n]*?\1", masked): + if _in_ranges(structural, m.start()): + continue + blank(m.start(), m.end()) + masked = "".join(chars) + + for m in re.finditer(r"\]\([^)]*\)", masked): + blank(m.start(), m.end()) + for m in re.finditer(r"]*>|https?://\S+", masked): + blank(m.start(), m.end()) + + return "".join(chars) + + +def _join_soft_wraps(masked: str, original: str) -> str: + orig_lines = original.split("\n") + out = list(masked) + offset = 0 + offsets: list[int] = [] + for line in orig_lines: + offsets.append(offset) + offset += len(line) + 1 + + for i in range(len(orig_lines) - 1): + first, second = orig_lines[i], orig_lines[i + 1] + if not first.strip() or not second.strip(): + continue + depth_a, body_a = _blockquote_depth(first) + depth_b, body_b = _blockquote_depth(second) + if depth_a != depth_b: + continue + if not body_a.strip() or not body_b.strip(): + continue + if _BLOCK_START.match(body_b): + continue + newline_at = offsets[i] + len(first) + if newline_at < len(out) and out[newline_at] == "\n": + out[newline_at] = " " + return "".join(out) + + +def _logical_line_bounds(text: str, index: int) -> tuple[int, int]: + start = text.rfind("\n", 0, index) + 1 + end = text.find("\n", index) + return start, (len(text) if end == -1 else end) + + +def _quoted_spans(segment: str) -> list[tuple[int, int]]: + spans: list[tuple[int, int]] = [] + for quote_open, quote_close in (('"', '"'), ("“", "”")): + pending: int | None = None + for i, ch in enumerate(segment): + if quote_open == quote_close: + if ch == quote_open: + if pending is None: + pending = i + else: + spans.append((pending, i + 1)) + pending = None + else: + if ch == quote_open: + pending = i + elif ch == quote_close and pending is not None: + spans.append((pending, i + 1)) + pending = None + return spans + + +def _neg_fires(text: str, match_start: int) -> bool: + window_start = max(0, match_start - NEG_WINDOW_CHARS) + i = match_start - 1 + while i >= window_start: + if text[i] in CLAUSE_BOUNDARIES: + window_start = i + 1 + break + i -= 1 + return bool(NEG_PATTERN.search(text[window_start:match_start])) + + +@dataclass(frozen=True) +class Diagnostic: + path: str + line: int + col: int + end_line: int + rule_id: str + severity: str + message: str + matched: str + + +def scan_text(path: str, original: str) -> list[Diagnostic]: + structural = _structural_ranges(original) + masked = _mask_code_regions(original, structural) + normalised = _join_soft_wraps(masked, original) + match_text = normalised.replace("\n", _NEWLINE_FILLER) + + line_starts = [0] + for i, ch in enumerate(original): + if ch == "\n": + line_starts.append(i + 1) + + def position(offset: int) -> tuple[int, int]: + lo, hi = 0, len(line_starts) - 1 + while lo < hi: + mid = (lo + hi + 1) // 2 + if line_starts[mid] <= offset: + lo = mid + else: + hi = mid - 1 + return lo + 1, offset - line_starts[lo] + 1 + + diagnostics: list[Diagnostic] = [] + for rule, compiled in COMPILED: + for m in compiled.finditer(match_text): + if "NEG" in rule.guards and _neg_fires(match_text, m.start()): + continue + if "CFG-NOUN" in rule.guards and CFG_NOUN_PATTERN.match(match_text, m.end()): + continue + + line, col = position(m.start()) + end_line, _ = position(max(m.start(), m.end() - 1)) + matched = re.sub(r"\s+", " ", original[m.start() : m.end()]).strip() + + start, end = _logical_line_bounds(normalised, m.start()) + rel = m.start() - start + in_quote = any(lo <= rel < hi for lo, hi in _quoted_spans(match_text[start:end])) and not _in_ranges( + structural, m.start() + ) + if in_quote: + diagnostics.append( + Diagnostic( + path, + line, + col, + end_line, + QUOTE_RULE_ID, + "info", + f"{rule.rule_id} phrase inside a quoted span (negative example)", + matched, + ) + ) + else: + diagnostics.append( + Diagnostic( + path, + line, + col, + end_line, + rule.rule_id, + rule.severity, + f"banned claim wording ({rule.rule_id})", + matched, + ) + ) + return sorted(diagnostics, key=lambda d: (d.path, d.line, d.col, d.rule_id)) + + +def _introduced(d: Diagnostic, touched: set[int]) -> bool: + return any(line in touched for line in range(d.line, d.end_line + 1)) + + +def scan_file(root: Path, rel: str) -> list[Diagnostic]: + try: + text = (root / rel).read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return [] + return scan_text(rel, text) + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("paths", nargs="*", help="explicit paths (default: repo scope)") + parser.add_argument("--root", default=".", help="repository root") + parser.add_argument( + "--report-only", + action="store_true", + help="never exit non-zero; used while the blocking baseline is non-empty", + ) + parser.add_argument("--selftest", action="store_true", help="run built-in fixtures") + parser.add_argument( + "--diff-base", + help="restrict BLOCKING diagnostics to lines added or modified since this ref", + ) + args = parser.parse_args(argv) + + if args.selftest: + return selftest() + + root = Path(args.root).resolve() + if args.paths: + targets = [] + for p in args.paths: + candidate = Path(p) + if candidate.is_absolute(): + try: + targets.append(str(candidate.relative_to(root))) + except ValueError: + targets.append(str(candidate)) + else: + targets.append(p) + dropped = [t for t in targets if _is_repo_relative(root, t) and not in_scope(t)] + if dropped: + for t in dropped: + print(f"skipped (outside scope): {t}") + targets = [t for t in targets if t not in set(dropped)] + else: + targets = [p for p in tracked_files(root) if in_scope(p)] + + changed_lines: dict[str, set[int]] | None = None + if args.diff_base: + changed_lines = _changed_lines(root, args.diff_base, targets) + + diagnostics: list[Diagnostic] = [] + for rel in targets: + for d in scan_file(root, rel): + touched = changed_lines.get(d.path, set()) if changed_lines is not None else set() + if changed_lines is not None and d.severity == "blocking" and not _introduced(d, touched): + d = Diagnostic( + d.path, + d.line, + d.col, + d.end_line, + d.rule_id, + "pre-existing", + d.message + " (pre-existing; not introduced by this change)", + d.matched, + ) + diagnostics.append(d) + + counts = {"blocking": 0, "finding": 0, "info": 0, "pre-existing": 0} + for d in diagnostics: + counts[d.severity] = counts.get(d.severity, 0) + 1 + print(f"{d.path}:{d.line}:{d.col} {d.rule_id} {d.severity} {d.message} — {d.matched!r}") + + print( + f"\ncheck_claim_vocabulary: {len(targets)} file(s) scanned; " + f"{counts['blocking']} blocking, {counts['finding']} finding, " + f"{counts['info']} info, {counts['pre-existing']} pre-existing." + ) + + if args.report_only: + return 0 + return 1 if counts["blocking"] else 0 + + +# -------------------------------------------------------------------------- +# Self-test. Each rule gets a positive case; every guard gets a case where it +# must suppress. +# -------------------------------------------------------------------------- +SELFTEST_CASES: tuple[tuple[str, str | None], ...] = ( + ("eBPF catches everything else, including bypass attempts.", "CLAIM-ABS-01"), + ("It does not catch everything.", None), + ("The gateway cannot be bypassed by an agent.", "CLAIM-ABS-03"), + ("An unbypassable control.", "CLAIM-ABS-04"), + ("With all three layers there is nowhere to hide.", "CLAIM-ABS-05"), + ("Checked before every tool call.", "CLAIM-ABS-07"), + ("Deploy with no code changes.", "CLAIM-ABS-08"), + ("Recorded in an immutable audit trail.", "CLAIM-ABS-09"), + ("Rolled out across the whole fleet.", "CLAIM-ABS-10"), + ("Governs the full fleet today.", "CLAIM-ABS-10"), + ("complete coverage of agent traffic", "CLAIM-ABS-11"), + ("There is no claim of complete detection.", None), # NEG guard + ("catch-all rules are configured here", None), # CFG-NOUN guard + ("a catch-all promise", "CLAIM-ABS-02"), + ("`immutable audit` is a banned phrase", None), # inline code + ('The bug was "an immutable audit trail" on the front page.', QUOTE_RULE_ID), + ("Checked before every action the agent takes.", "CLAIM-ABS-06"), + ("Our coverage is complete.", "CLAIM-ABS-12"), + ("Agent Assembly enforces a zero-trust posture.", "CLAIM-VERB-01"), + ("```\nimmutable audit\n```", None), + (" immutable audit trail\n", None), + ("", None), + ("See [the docs](https://example.com/immutable-audit-trail).", None), + ("## Agent Assembly `catches everything` on your fleet", "CLAIM-ABS-01"), + ('## The "immutable audit" trail', "CLAIM-ABS-09"), + ('', "CLAIM-ABS-09"), + ("---\ntitle: An `immutable audit` trail\n---\n", "CLAIM-ABS-09"), + ("description: keeps an `immutable audit` trail\n", "CLAIM-ABS-09"), + ("The phrase `catches everything` is banned in body copy.", None), + ("It does not, per `RFC-1`, catch everything.", None), + ("It does not, per RFC-1, catch everything.", None), + ("This is not a claim of `verified` complete coverage.", None), + ("a don`t\nkept in an immutable audit trail\na won`t\n", "CLAIM-ABS-09"), + ("a complete guide to coverage", None), + ("complete mediation of agent traffic", "CLAIM-ABS-11"), + ("---\nsummary: kept in an `immutable audit` trail\nowner: platform\n---\n", "CLAIM-ABS-09"), + ("Agent Assembly `catches everything`\n=====\n", "CLAIM-ABS-01"), + ("---\n\nIt is not an `immutable audit` trail.\n", None), + ("It keeps an `immutable audit` trail\n---\n", "CLAIM-ABS-09"), +) + + +def selftest() -> int: + failures: list[str] = [] + for i, (text, expected) in enumerate(SELFTEST_CASES): + diags = scan_text(f"", text) + ids = sorted({d.rule_id for d in diags if d.severity != "pre-existing"}) + if expected is None: + if ids: + failures.append(f"case {i} {text!r}: expected none, got {ids}") + elif expected not in ids: + failures.append(f"case {i} {text!r}: expected {expected}, got {ids or 'NOTHING'}") + + if not in_scope("docs/quick-start.md"): + failures.append("in_scope() rejects a real docs/ page") + if in_scope("quickstart_snippets/langchain/basic.py"): + failures.append("in_scope() accepts a vendored quickstart snippet") + if in_scope("verification-reports/AAASM-2714.md"): + failures.append("in_scope() accepts a verification report") + if in_scope("docs/index.txt"): + failures.append("in_scope() accepts a non-Markdown extension") + + if not _SAFE_REF.match("remote/main") or not _SAFE_REF.match("HEAD~1"): + failures.append("_SAFE_REF rejects an ordinary git ref") + if _SAFE_REF.match("; rm -rf /") or _SAFE_REF.match("$(whoami)"): + failures.append("_SAFE_REF accepts a shell-metacharacter string as a git ref") + + total = len(SELFTEST_CASES) + 6 + print(f"check_claim_vocabulary selftest: {total - len(failures)}/{total} checks passed") + for f in failures: + print(" FAIL:", f) + return 0 if not failures else 2 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:]))