From b991bc5554ac464a4a4e01b7bee5d85bd8486b52 Mon Sep 17 00:00:00 2001 From: Scott Wofford Date: Mon, 6 Jul 2026 23:11:01 -0700 Subject: [PATCH 1/3] chore: set objective to generate policy YAML from CLAUDE.md From 638ec4c229728c8fb299cef16089ba0240c9e6a1 Mon Sep 17 00:00:00 2001 From: Scott Wofford Date: Mon, 6 Jul 2026 23:15:49 -0700 Subject: [PATCH 2/3] feat: generate policy YAML from CLAUDE.md Adds luthien_proxy.policy_generation.claude_md: extracts enforceable behavioral rules from a CLAUDE.md/AGENTS.md with a deterministic heuristic (normative-language filter, code fences and tables skipped), emits a SimpleLLMPolicy YAML with every rule tagged [:], and round-trip validates the output through load_policy_from_yaml. Run: uv run python -m luthien_proxy.policy_generation.claude_md CLAUDE.md Trello: https://trello.com/c/j9DbbAaC --- README.md | 8 + changelog.d/policy-from-claude-md.md | 6 + docs/policies.md | 32 ++ .../policy_generation/__init__.py | 12 + .../policy_generation/__main__.py | 6 + .../policy_generation/claude_md.py | 334 ++++++++++++++++++ .../policy_generation/test_claude_md.py | 202 +++++++++++ 7 files changed, 600 insertions(+) create mode 100644 changelog.d/policy-from-claude-md.md create mode 100644 src/luthien_proxy/policy_generation/__init__.py create mode 100644 src/luthien_proxy/policy_generation/__main__.py create mode 100644 src/luthien_proxy/policy_generation/claude_md.py create mode 100644 tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py diff --git a/README.md b/README.md index 003e607d2..933abf40c 100644 --- a/README.md +++ b/README.md @@ -243,6 +243,14 @@ policy: max_tokens: 256 ``` +Already have a CLAUDE.md? Generate a policy from it — the generator extracts your behavioral rules (skipping build commands and repo trivia) and emits a ready-to-load `SimpleLLMPolicy` config, with each rule tagged with its source line: + +```bash +uv run python -m luthien_proxy.policy_generation.claude_md CLAUDE.md -o config/claude_md_policy.yaml +``` + +See [Generate a Policy from Your CLAUDE.md](docs/policies.md#generate-a-policy-from-your-claudemd) for details. + ### Built-in Presets Ready-to-use policies in `src/luthien_proxy/policies/presets/` — no configuration needed. diff --git a/changelog.d/policy-from-claude-md.md b/changelog.d/policy-from-claude-md.md new file mode 100644 index 000000000..5f208affc --- /dev/null +++ b/changelog.d/policy-from-claude-md.md @@ -0,0 +1,6 @@ +--- +category: Features +pr: 802 +--- + +**Generate a policy from CLAUDE.md**: new `uv run python -m luthien_proxy.policy_generation.claude_md ` command extracts enforceable behavioral rules from an existing CLAUDE.md / AGENTS.md and emits a ready-to-load `SimpleLLMPolicy` YAML, with every rule tagged with its source line and the output round-trip validated through the policy loader diff --git a/docs/policies.md b/docs/policies.md index 20d7979d6..7f38c454a 100644 --- a/docs/policies.md +++ b/docs/policies.md @@ -17,6 +17,38 @@ policy: --- +## Generate a Policy from Your CLAUDE.md + +If your project already has a CLAUDE.md (or AGENTS.md), you can turn its behavioral rules into a working policy in one step: + +```bash +uv run python -m luthien_proxy.policy_generation.claude_md CLAUDE.md -o config/claude_md_policy.yaml +export POLICY_CONFIG=config/claude_md_policy.yaml +``` + +The generator extracts enforceable rules (lines with normative language like "never", "always", "must", "avoid", "prefer") while skipping code blocks, build commands, and repo trivia. It emits a `SimpleLLMPolicy` config whose judge checks every response against those rules. Extraction is deterministic — no LLM call, no credentials needed at generation time. + +Every rule in the generated YAML is tagged with its source line, so judge decisions trace back to your CLAUDE.md: + +```yaml +instructions: |- + ... + 1. [CLAUDE.md:171] Formatting via Ruff: double quotes, spaces for indent. + 2. [CLAUDE.md:208] IMPORTANT: Always write unit tests when adding or significantly modifying code. + ... +``` + +Options: + +- `-o / --output` — write to a file (default: stdout) +- `--model` — judge model (default: `claude-haiku-4-5`) +- `--on-error pass|block` — what happens when the judge call fails (default: `pass`) +- `--no-validate` — skip the round-trip check through the policy loader + +The generated file is a starting point — edit the instructions freely; it's plain `SimpleLLMPolicy` YAML. + +--- + ## Quick Start Presets Ready-to-use policies with zero configuration. Each wraps `SimpleLLMPolicy` with hardcoded instructions — just set the class and go. diff --git a/src/luthien_proxy/policy_generation/__init__.py b/src/luthien_proxy/policy_generation/__init__.py new file mode 100644 index 000000000..181e8bc27 --- /dev/null +++ b/src/luthien_proxy/policy_generation/__init__.py @@ -0,0 +1,12 @@ +"""Utilities that generate Luthien policy YAML from external sources. + +Currently supports generating a `SimpleLLMPolicy` configuration from a +project's CLAUDE.md / AGENTS.md file (`claude_md` module). Run it with: + + uv run python -m luthien_proxy.policy_generation.claude_md path/to/CLAUDE.md + +Note: this package intentionally avoids importing submodules at package level +so `python -m luthien_proxy.policy_generation.claude_md` runs without a +double-import warning. Import from `luthien_proxy.policy_generation.claude_md` +directly. +""" diff --git a/src/luthien_proxy/policy_generation/__main__.py b/src/luthien_proxy/policy_generation/__main__.py new file mode 100644 index 000000000..13f50f0e8 --- /dev/null +++ b/src/luthien_proxy/policy_generation/__main__.py @@ -0,0 +1,6 @@ +"""Allow `python -m luthien_proxy.policy_generation ` as a shorthand.""" + +from luthien_proxy.policy_generation.claude_md import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/luthien_proxy/policy_generation/claude_md.py b/src/luthien_proxy/policy_generation/claude_md.py new file mode 100644 index 000000000..725728c0b --- /dev/null +++ b/src/luthien_proxy/policy_generation/claude_md.py @@ -0,0 +1,334 @@ +"""Generate a Luthien policy YAML from a CLAUDE.md / AGENTS.md file. + +Reads a CLAUDE.md, extracts enforceable behavioral rules with a deterministic +heuristic (no LLM call), and emits a `SimpleLLMPolicy` configuration that +`luthien_proxy.config.load_policy_from_yaml` accepts. Every extracted rule is +tagged with its source line number so judge decisions stay traceable back to +the originating CLAUDE.md text. + +Why heuristic extraction: it is deterministic (same input -> same policy), +needs no credentials at generation time, and keeps rule provenance exact. +LLM-assisted extraction can layer on top later without changing the output +format. + +Usage: + + uv run python -m luthien_proxy.policy_generation.claude_md CLAUDE.md + uv run python -m luthien_proxy.policy_generation.claude_md CLAUDE.md -o config/claude_md_policy.yaml +""" + +from __future__ import annotations + +import argparse +import hashlib +import re +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path + +import yaml + +from luthien_proxy.config import load_policy_from_yaml + +DEFAULT_MODEL = "claude-haiku-4-5" + +# A candidate line must contain at least one normative marker to count as an +# enforceable rule (vs. repo trivia like directory listings or build commands). +_NORMATIVE_PATTERN = re.compile( + r""" + \b( + never + | always + | must(\ not)? + | do\ not + | don'?t + | avoid + | prefer(red)? + | required? + | forbidden + | disallowed + | ban(ned)? + | should(\ not)? + | ensure + | instead\ of + | only\ (use|if|when) + )\b + """, + re.IGNORECASE | re.VERBOSE, +) + +_BULLET_PATTERN = re.compile(r"^(\s*)(?:[-*+]|\d+\.)\s+(.*)$") +_FENCE_PATTERN = re.compile(r"^\s*(```|~~~)") +_HEADING_PATTERN = re.compile(r"^\s*#{1,6}\s") +_LINK_PATTERN = re.compile(r"\[([^\]]*)\]\([^)]*\)") + +# Rules shorter than this (after markdown stripping) are fragments, not rules. +_MIN_RULE_CHARS = 12 +# Rules longer than this are prose sections, not individually enforceable rules. +_MAX_RULE_CHARS = 400 + + +@dataclass(frozen=True) +class ExtractedRule: + """A behavioral rule extracted from a CLAUDE.md file. + + Attributes: + text: The rule text with markdown decoration stripped. + line: 1-based line number in the source file where the rule starts. + """ + + text: str + line: int + + +def _strip_markdown(text: str) -> str: + """Remove markdown decoration, keeping the readable text.""" + text = _LINK_PATTERN.sub(r"\1", text) + text = text.replace("**", "") + text = text.replace("`", "") + return re.sub(r"\s+", " ", text).strip() + + +def _looks_like_command(text: str) -> bool: + """Reject lines that are shell invocations rather than behavioral rules.""" + stripped = text.strip() + return stripped.startswith(("$", ">", "cd ", "git ", "uv ", "npm ", "./", "export ")) + + +@dataclass +class _Candidate: + """A candidate rule being accumulated across continuation lines.""" + + first_line: int + parts: list[str] + + def text(self) -> str: + return _strip_markdown(" ".join(self.parts)) + + +def _is_candidate_break(line: str) -> bool: + """True when a line terminates the current candidate paragraph/bullet.""" + stripped = line.strip() + return not stripped or bool(_HEADING_PATTERN.match(line)) or stripped.startswith("|") + + +def extract_rules(markdown: str) -> list[ExtractedRule]: + """Extract enforceable behavioral rules from CLAUDE.md content. + + Walks the document line by line, skipping fenced code blocks, headings, + and tables. Bullets and short paragraphs qualify as rules when they carry + a normative marker (never / always / must / avoid / prefer / ...). Rules + keep the 1-based line number where they start. + + Args: + markdown: Full text of a CLAUDE.md / AGENTS.md file. + + Returns: + Rules in document order, deduplicated case-insensitively. + """ + rules: list[ExtractedRule] = [] + seen: set[str] = set() + in_fence = False + candidate: _Candidate | None = None + + def flush(current: _Candidate | None) -> None: + if current is None: + return + text = current.text() + if not (_MIN_RULE_CHARS <= len(text) <= _MAX_RULE_CHARS): + return + if not _NORMATIVE_PATTERN.search(text): + return + if _looks_like_command(text): + return + key = text.casefold() + if key in seen: + return + seen.add(key) + rules.append(ExtractedRule(text=text, line=current.first_line)) + + for lineno, line in enumerate(markdown.splitlines(), start=1): + if _FENCE_PATTERN.match(line): + flush(candidate) + candidate = None + in_fence = not in_fence + continue + if in_fence: + continue + + if _is_candidate_break(line): + flush(candidate) + candidate = None + continue + + bullet_match = _BULLET_PATTERN.match(line) + if bullet_match: + flush(candidate) + candidate = _Candidate(first_line=lineno, parts=[bullet_match.group(2)]) + continue + + if candidate is not None: + candidate.parts.append(line.strip()) + else: + candidate = _Candidate(first_line=lineno, parts=[line.strip()]) + + flush(candidate) + return rules + + +def _build_instructions(rules: list[ExtractedRule], source_name: str) -> str: + """Compose judge instructions from extracted rules, tagged with source lines.""" + numbered = "\n".join(f"{i}. [{source_name}:{rule.line}] {rule.text}" for i, rule in enumerate(rules, start=1)) + return ( + "You are reviewing responses from an AI coding assistant. The project's " + f"{source_name} defines behavioral rules the assistant must follow. " + "Evaluate each content block against these rules (each rule is tagged " + "with the source line it came from):\n\n" + f"{numbered}\n\n" + "If a block complies with every rule, return it unchanged. If a block " + "violates a rule, rewrite it minimally so it complies. If a violation " + "cannot be fixed by rewriting, replace the block with a brief note " + "naming the violated rule and its source tag." + ) + + +class _LiteralDumper(yaml.SafeDumper): + """SafeDumper that renders multiline strings as literal blocks (|-).""" + + +def _represent_multiline_str(dumper: yaml.SafeDumper, data: str) -> yaml.ScalarNode: + if "\n" in data: + return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|") + return dumper.represent_scalar("tag:yaml.org,2002:str", data) + + +_LiteralDumper.add_representer(str, _represent_multiline_str) + + +def generate_policy_yaml( + rules: list[ExtractedRule], + source_path: Path, + *, + model: str = DEFAULT_MODEL, + on_error: str = "pass", +) -> str: + """Render a SimpleLLMPolicy YAML document from extracted rules. + + Args: + rules: Rules extracted from the source file (must be non-empty). + source_path: The CLAUDE.md file the rules came from (for provenance). + model: Judge model identifier. + on_error: Judge failure behavior ("pass" or "block"). + + Returns: + A YAML string loadable by `luthien_proxy.config.load_policy_from_yaml`. + + Raises: + ValueError: If `rules` is empty. + """ + if not rules: + raise ValueError("Cannot generate a policy from zero rules") + + source_text = source_path.read_text(encoding="utf-8") + digest = hashlib.sha256(source_text.encode("utf-8")).hexdigest()[:12] + + document = { + "policy": { + "class": "luthien_proxy.policies.simple_llm_policy:SimpleLLMPolicy", + "config": { + "model": model, + "on_error": on_error, + "inference_provider": "user_credentials", + "instructions": _build_instructions(rules, source_path.name), + }, + } + } + body = yaml.dump(document, Dumper=_LiteralDumper, sort_keys=False, width=100, allow_unicode=True) + header = ( + f"# Luthien policy generated from {source_path.name}\n" + f"# Source: {source_path} (sha256 {digest})\n" + f"# Rules extracted: {len(rules)} (each tagged [{source_path.name}:] below)\n" + "# Regenerate: uv run python -m luthien_proxy.policy_generation.claude_md " + f"{source_path}\n" + ) + return header + body + + +def _validate_policy_yaml(yaml_text: str) -> None: + """Round-trip the generated YAML through the real policy loader. + + Raises: + Exception: Whatever `load_policy_from_yaml` raises for an invalid config. + """ + with tempfile.NamedTemporaryFile("w", suffix=".yaml", encoding="utf-8", delete=False) as handle: + handle.write(yaml_text) + temp_path = handle.name + try: + load_policy_from_yaml(temp_path) + finally: + Path(temp_path).unlink(missing_ok=True) + + +def main(argv: list[str] | None = None) -> int: + """CLI entry point: generate and validate a policy YAML from a CLAUDE.md.""" + parser = argparse.ArgumentParser( + prog="python -m luthien_proxy.policy_generation.claude_md", + description="Generate a Luthien SimpleLLMPolicy YAML from a CLAUDE.md file.", + ) + parser.add_argument("input", type=Path, help="Path to CLAUDE.md / AGENTS.md") + parser.add_argument( + "-o", + "--output", + type=Path, + default=None, + help="Write the policy YAML here (default: print to stdout)", + ) + parser.add_argument("--model", default=DEFAULT_MODEL, help=f"Judge model (default: {DEFAULT_MODEL})") + parser.add_argument( + "--on-error", + choices=("pass", "block"), + default="pass", + help="Judge failure behavior: pass content with a warning, or block it (default: pass)", + ) + parser.add_argument( + "--no-validate", + action="store_true", + help="Skip round-trip validation through the policy loader", + ) + args = parser.parse_args(argv) + + if not args.input.is_file(): + print(f"error: {args.input} is not a file", file=sys.stderr) + return 1 + + rules = extract_rules(args.input.read_text(encoding="utf-8")) + if not rules: + print( + f"error: no enforceable rules found in {args.input}. " + "The extractor looks for bullets/paragraphs with normative language " + "(never / always / must / avoid / prefer / ...).", + file=sys.stderr, + ) + return 1 + + yaml_text = generate_policy_yaml(rules, args.input, model=args.model, on_error=args.on_error) + + if not args.no_validate: + try: + _validate_policy_yaml(yaml_text) + except Exception as exc: + print(f"error: generated YAML failed policy-loader validation: {exc}", file=sys.stderr) + return 2 + + if args.output is None: + print(yaml_text, end="") + else: + args.output.write_text(yaml_text, encoding="utf-8") + print(f"Wrote {args.output} ({len(rules)} rules extracted from {args.input})", file=sys.stderr) + print(f"Activate it with: export POLICY_CONFIG={args.output}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py b/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py new file mode 100644 index 000000000..bf1ebaea5 --- /dev/null +++ b/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py @@ -0,0 +1,202 @@ +# ABOUTME: Tests for CLAUDE.md -> policy YAML generation +# ABOUTME: Covers rule extraction heuristics, YAML rendering, round-trip loading, and the CLI + +"""Tests for luthien_proxy.policy_generation.claude_md.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from luthien_proxy.config import load_policy_from_yaml +from luthien_proxy.policies.simple_llm_policy import SimpleLLMPolicy +from luthien_proxy.policy_generation.claude_md import ( + ExtractedRule, + extract_rules, + generate_policy_yaml, + main, +) + +REPO_ROOT = Path(__file__).parents[4] + +SAMPLE_CLAUDE_MD = """\ +# Project Guidelines + +## Setup + +- Run `npm install` to get started. +- The dev server lives at localhost:3000. + +## Coding Rules + +- Never commit secrets or API keys. +- Always write unit tests when adding new code. +- Prefer f-strings over .format() for readability. +- Nice weather today. + +```bash +# Never run this inside a fence — it should be skipped +always_skip_me --must +``` + +**Do not edit generated files by hand.** They are rebuilt on every release +and manual edits will be lost. + +| Column | Never used | +|--------|------------| +| a | must skip | +""" + + +class TestExtractRules: + def test_extracts_normative_bullets(self): + rules = extract_rules(SAMPLE_CLAUDE_MD) + texts = [r.text for r in rules] + + assert "Never commit secrets or API keys." in texts + assert "Always write unit tests when adding new code." in texts + assert "Prefer f-strings over .format() for readability." in texts + + def test_skips_non_normative_and_trivia_lines(self): + rules = extract_rules(SAMPLE_CLAUDE_MD) + texts = [r.text for r in rules] + + assert not any("npm install" in t for t in texts) + assert not any("localhost:3000" in t for t in texts) + assert not any("Nice weather" in t for t in texts) + + def test_skips_fenced_code_blocks(self): + rules = extract_rules(SAMPLE_CLAUDE_MD) + assert not any("always_skip_me" in r.text for r in rules) + + def test_skips_table_rows(self): + rules = extract_rules(SAMPLE_CLAUDE_MD) + assert not any("must skip" in r.text for r in rules) + + def test_paragraph_rules_join_continuation_lines(self): + rules = extract_rules(SAMPLE_CLAUDE_MD) + para = next(r for r in rules if r.text.startswith("Do not edit generated files")) + assert "manual edits will be lost" in para.text + + def test_line_numbers_are_one_based_and_correct(self): + rules = extract_rules(SAMPLE_CLAUDE_MD) + never_rule = next(r for r in rules if r.text.startswith("Never commit secrets")) + lines = SAMPLE_CLAUDE_MD.splitlines() + assert "Never commit secrets" in lines[never_rule.line - 1] + + def test_deduplicates_case_insensitively(self): + doc = "- Never push to main.\n\n- never push to MAIN.\n" + rules = extract_rules(doc) + assert len(rules) == 1 + + def test_strips_markdown_decoration(self): + doc = "- **Always** use [uv](https://docs.astral.sh/uv/) and `pytest` for tests.\n" + rules = extract_rules(doc) + assert rules[0].text == "Always use uv and pytest for tests." + + def test_empty_document_yields_no_rules(self): + assert extract_rules("") == [] + assert extract_rules("# Just a heading\n\nSome plain prose.\n") == [] + + def test_overlong_paragraphs_are_skipped(self): + doc = "- Never " + "x" * 500 + "\n" + assert extract_rules(doc) == [] + + +class TestGeneratePolicyYaml: + def _rules(self) -> list[ExtractedRule]: + return [ + ExtractedRule(text="Never commit secrets.", line=7), + ExtractedRule(text="Always write tests.", line=9), + ] + + def test_rejects_empty_rules(self, tmp_path: Path): + source = tmp_path / "CLAUDE.md" + source.write_text("# empty\n") + with pytest.raises(ValueError): + generate_policy_yaml([], source) + + def test_yaml_structure_and_traceability(self, tmp_path: Path): + source = tmp_path / "CLAUDE.md" + source.write_text(SAMPLE_CLAUDE_MD) + text = generate_policy_yaml(self._rules(), source) + + parsed = yaml.safe_load(text) + policy = parsed["policy"] + assert policy["class"] == "luthien_proxy.policies.simple_llm_policy:SimpleLLMPolicy" + instructions = policy["config"]["instructions"] + assert "1. [CLAUDE.md:7] Never commit secrets." in instructions + assert "2. [CLAUDE.md:9] Always write tests." in instructions + + def test_model_and_on_error_options(self, tmp_path: Path): + source = tmp_path / "CLAUDE.md" + source.write_text(SAMPLE_CLAUDE_MD) + text = generate_policy_yaml(self._rules(), source, model="claude-sonnet-4-5", on_error="block") + + parsed = yaml.safe_load(text) + assert parsed["policy"]["config"]["model"] == "claude-sonnet-4-5" + assert parsed["policy"]["config"]["on_error"] == "block" + + def test_round_trip_through_policy_loader(self, tmp_path: Path): + source = tmp_path / "CLAUDE.md" + source.write_text(SAMPLE_CLAUDE_MD) + out = tmp_path / "policy.yaml" + out.write_text(generate_policy_yaml(extract_rules(SAMPLE_CLAUDE_MD), source)) + + policy = load_policy_from_yaml(str(out)) + + assert isinstance(policy, SimpleLLMPolicy) + assert "[CLAUDE.md:" in policy._config.instructions + + +class TestAgainstRepoAgentsMd: + """Realistic example: the generator run on this repo's own AGENTS.md.""" + + def test_repo_agents_md_generates_loadable_policy(self, tmp_path: Path): + source = REPO_ROOT / "AGENTS.md" + rules = extract_rules(source.read_text(encoding="utf-8")) + + # Loose bounds: AGENTS.md evolves, but it is rule-dense. + assert len(rules) >= 10 + + out = tmp_path / "policy.yaml" + out.write_text(generate_policy_yaml(rules, source)) + policy = load_policy_from_yaml(str(out)) + + assert isinstance(policy, SimpleLLMPolicy) + # Every rule is tagged back to a source line. + assert policy._config.instructions.count("[AGENTS.md:") == len(rules) + + +class TestCli: + def test_writes_output_file(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + source = tmp_path / "CLAUDE.md" + source.write_text(SAMPLE_CLAUDE_MD) + out = tmp_path / "policy.yaml" + + exit_code = main([str(source), "-o", str(out)]) + + assert exit_code == 0 + assert isinstance(load_policy_from_yaml(str(out)), SimpleLLMPolicy) + + def test_stdout_mode_prints_yaml(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + source = tmp_path / "CLAUDE.md" + source.write_text(SAMPLE_CLAUDE_MD) + + exit_code = main([str(source)]) + + assert exit_code == 0 + printed = capsys.readouterr().out + assert yaml.safe_load(printed)["policy"]["class"].endswith("SimpleLLMPolicy") + + def test_missing_input_fails(self, tmp_path: Path): + assert main([str(tmp_path / "nope.md")]) == 1 + + def test_no_rules_found_fails(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + source = tmp_path / "CLAUDE.md" + source.write_text("# heading only\n\nplain prose without normative language\n") + + assert main([str(source)]) == 1 + assert "no enforceable rules" in capsys.readouterr().err From a5717019c6da790d48c25e087807d05ff742adf0 Mon Sep 17 00:00:00 2001 From: Scott Wofford Date: Mon, 6 Jul 2026 23:25:41 -0700 Subject: [PATCH 3/3] =?UTF-8?q?refactor:=20address=20review=20=E2=80=94=20?= =?UTF-8?q?blockquote=20rules,=20skip-count=20reporting,=20public-API=20te?= =?UTF-8?q?sts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove _looks_like_command filter (false positives on rules like 'git rebase should never be run interactively'); the normative-language gate already excludes plain commands - Extract rules from blockquote callouts (strip leading '>') - Return ExtractionResult with skipped_too_long count; CLI reports skips on stderr and takes --max-rule-chars to override the cap - Header comments use the file name, not the absolute path - generate_policy_yaml accepts source_text to avoid a second file read - Tests assert via parsed YAML instead of SimpleLLMPolicy._config; add CLI coverage for --model/--on-error/--no-validate/validation-failure - Docs: --max-rule-chars flag + trust note on regeneration --- docs/policies.md | 3 + .../policy_generation/claude_md.py | 90 +++++++++--- .../policy_generation/test_claude_md.py | 134 ++++++++++++++---- 3 files changed, 175 insertions(+), 52 deletions(-) diff --git a/docs/policies.md b/docs/policies.md index 7f38c454a..6ddf1631f 100644 --- a/docs/policies.md +++ b/docs/policies.md @@ -43,10 +43,13 @@ Options: - `-o / --output` — write to a file (default: stdout) - `--model` — judge model (default: `claude-haiku-4-5`) - `--on-error pass|block` — what happens when the judge call fails (default: `pass`) +- `--max-rule-chars` — skip rules longer than this many characters (default: 400; skips are reported on stderr) - `--no-validate` — skip the round-trip check through the policy loader The generated file is a starting point — edit the instructions freely; it's plain `SimpleLLMPolicy` YAML. +**Trust note:** extracted rules go verbatim into the judge's instructions, so the generated policy is only as trustworthy as the CLAUDE.md it came from. Review CLAUDE.md changes with the same care as policy changes — text added to CLAUDE.md (e.g. via a malicious PR) becomes judge instructions the next time you regenerate. + --- ## Quick Start Presets diff --git a/src/luthien_proxy/policy_generation/claude_md.py b/src/luthien_proxy/policy_generation/claude_md.py index 725728c0b..af687901e 100644 --- a/src/luthien_proxy/policy_generation/claude_md.py +++ b/src/luthien_proxy/policy_generation/claude_md.py @@ -24,6 +24,7 @@ import re import sys import tempfile +from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path @@ -62,6 +63,7 @@ _FENCE_PATTERN = re.compile(r"^\s*(```|~~~)") _HEADING_PATTERN = re.compile(r"^\s*#{1,6}\s") _LINK_PATTERN = re.compile(r"\[([^\]]*)\]\([^)]*\)") +_BLOCKQUOTE_PATTERN = re.compile(r"^(\s*)>\s?") # Rules shorter than this (after markdown stripping) are fragments, not rules. _MIN_RULE_CHARS = 12 @@ -82,6 +84,20 @@ class ExtractedRule: line: int +@dataclass(frozen=True) +class ExtractionResult: + """Outcome of a rule-extraction pass. + + Attributes: + rules: Extracted rules in document order. + skipped_too_long: Normative candidates dropped for exceeding the + rule-length cap (surfaced so long rules never vanish silently). + """ + + rules: tuple[ExtractedRule, ...] + skipped_too_long: int + + def _strip_markdown(text: str) -> str: """Remove markdown decoration, keeping the readable text.""" text = _LINK_PATTERN.sub(r"\1", text) @@ -90,12 +106,6 @@ def _strip_markdown(text: str) -> str: return re.sub(r"\s+", " ", text).strip() -def _looks_like_command(text: str) -> bool: - """Reject lines that are shell invocations rather than behavioral rules.""" - stripped = text.strip() - return stripped.startswith(("$", ">", "cd ", "git ", "uv ", "npm ", "./", "export ")) - - @dataclass class _Candidate: """A candidate rule being accumulated across continuation lines.""" @@ -113,34 +123,41 @@ def _is_candidate_break(line: str) -> bool: return not stripped or bool(_HEADING_PATTERN.match(line)) or stripped.startswith("|") -def extract_rules(markdown: str) -> list[ExtractedRule]: +def extract_rules(markdown: str, *, max_rule_chars: int = _MAX_RULE_CHARS) -> ExtractionResult: """Extract enforceable behavioral rules from CLAUDE.md content. Walks the document line by line, skipping fenced code blocks, headings, - and tables. Bullets and short paragraphs qualify as rules when they carry - a normative marker (never / always / must / avoid / prefer / ...). Rules - keep the 1-based line number where they start. + and tables. Bullets, blockquotes, and short paragraphs qualify as rules + when they carry a normative marker (never / always / must / avoid / + prefer / ...). Rules keep the 1-based line number where they start. Args: markdown: Full text of a CLAUDE.md / AGENTS.md file. + max_rule_chars: Candidates longer than this (after markdown stripping) + are counted in `skipped_too_long` instead of extracted. Returns: - Rules in document order, deduplicated case-insensitively. + Extraction result with rules in document order (deduplicated + case-insensitively) plus a count of normative candidates skipped + for exceeding `max_rule_chars`. """ rules: list[ExtractedRule] = [] seen: set[str] = set() in_fence = False candidate: _Candidate | None = None + skipped_too_long = 0 def flush(current: _Candidate | None) -> None: + nonlocal skipped_too_long if current is None: return text = current.text() - if not (_MIN_RULE_CHARS <= len(text) <= _MAX_RULE_CHARS): + if len(text) < _MIN_RULE_CHARS: return if not _NORMATIVE_PATTERN.search(text): return - if _looks_like_command(text): + if len(text) > max_rule_chars: + skipped_too_long += 1 return key = text.casefold() if key in seen: @@ -148,8 +165,8 @@ def flush(current: _Candidate | None) -> None: seen.add(key) rules.append(ExtractedRule(text=text, line=current.first_line)) - for lineno, line in enumerate(markdown.splitlines(), start=1): - if _FENCE_PATTERN.match(line): + for lineno, raw_line in enumerate(markdown.splitlines(), start=1): + if _FENCE_PATTERN.match(raw_line): flush(candidate) candidate = None in_fence = not in_fence @@ -157,6 +174,9 @@ def flush(current: _Candidate | None) -> None: if in_fence: continue + # Blockquoted rules (callout style) participate like normal text. + line = _BLOCKQUOTE_PATTERN.sub(r"\1", raw_line) + if _is_candidate_break(line): flush(candidate) candidate = None @@ -174,10 +194,10 @@ def flush(current: _Candidate | None) -> None: candidate = _Candidate(first_line=lineno, parts=[line.strip()]) flush(candidate) - return rules + return ExtractionResult(rules=tuple(rules), skipped_too_long=skipped_too_long) -def _build_instructions(rules: list[ExtractedRule], source_name: str) -> str: +def _build_instructions(rules: Sequence[ExtractedRule], source_name: str) -> str: """Compose judge instructions from extracted rules, tagged with source lines.""" numbered = "\n".join(f"{i}. [{source_name}:{rule.line}] {rule.text}" for i, rule in enumerate(rules, start=1)) return ( @@ -207,11 +227,12 @@ def _represent_multiline_str(dumper: yaml.SafeDumper, data: str) -> yaml.ScalarN def generate_policy_yaml( - rules: list[ExtractedRule], + rules: Sequence[ExtractedRule], source_path: Path, *, model: str = DEFAULT_MODEL, on_error: str = "pass", + source_text: str | None = None, ) -> str: """Render a SimpleLLMPolicy YAML document from extracted rules. @@ -220,6 +241,8 @@ def generate_policy_yaml( source_path: The CLAUDE.md file the rules came from (for provenance). model: Judge model identifier. on_error: Judge failure behavior ("pass" or "block"). + source_text: The source file's content, if the caller already read it + (avoids a second read). Read from `source_path` when None. Returns: A YAML string loadable by `luthien_proxy.config.load_policy_from_yaml`. @@ -230,7 +253,8 @@ def generate_policy_yaml( if not rules: raise ValueError("Cannot generate a policy from zero rules") - source_text = source_path.read_text(encoding="utf-8") + if source_text is None: + source_text = source_path.read_text(encoding="utf-8") digest = hashlib.sha256(source_text.encode("utf-8")).hexdigest()[:12] document = { @@ -247,10 +271,10 @@ def generate_policy_yaml( body = yaml.dump(document, Dumper=_LiteralDumper, sort_keys=False, width=100, allow_unicode=True) header = ( f"# Luthien policy generated from {source_path.name}\n" - f"# Source: {source_path} (sha256 {digest})\n" + f"# Source: {source_path.name} (sha256 {digest})\n" f"# Rules extracted: {len(rules)} (each tagged [{source_path.name}:] below)\n" "# Regenerate: uv run python -m luthien_proxy.policy_generation.claude_md " - f"{source_path}\n" + f"{source_path.name}\n" ) return header + body @@ -296,13 +320,27 @@ def main(argv: list[str] | None = None) -> int: action="store_true", help="Skip round-trip validation through the policy loader", ) + parser.add_argument( + "--max-rule-chars", + type=int, + default=_MAX_RULE_CHARS, + help=f"Skip rules longer than this many characters (default: {_MAX_RULE_CHARS})", + ) args = parser.parse_args(argv) if not args.input.is_file(): print(f"error: {args.input} is not a file", file=sys.stderr) return 1 - rules = extract_rules(args.input.read_text(encoding="utf-8")) + source_text = args.input.read_text(encoding="utf-8") + result = extract_rules(source_text, max_rule_chars=args.max_rule_chars) + rules = result.rules + if result.skipped_too_long: + print( + f"note: skipped {result.skipped_too_long} rule candidate(s) longer than " + f"{args.max_rule_chars} characters (raise with --max-rule-chars)", + file=sys.stderr, + ) if not rules: print( f"error: no enforceable rules found in {args.input}. " @@ -312,7 +350,13 @@ def main(argv: list[str] | None = None) -> int: ) return 1 - yaml_text = generate_policy_yaml(rules, args.input, model=args.model, on_error=args.on_error) + yaml_text = generate_policy_yaml( + rules, + args.input, + model=args.model, + on_error=args.on_error, + source_text=source_text, + ) if not args.no_validate: try: diff --git a/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py b/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py index bf1ebaea5..3a96c10f6 100644 --- a/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py +++ b/tests/luthien_proxy/unit_tests/policy_generation/test_claude_md.py @@ -12,6 +12,7 @@ from luthien_proxy.config import load_policy_from_yaml from luthien_proxy.policies.simple_llm_policy import SimpleLLMPolicy +from luthien_proxy.policy_generation import claude_md from luthien_proxy.policy_generation.claude_md import ( ExtractedRule, extract_rules, @@ -44,6 +45,8 @@ **Do not edit generated files by hand.** They are rebuilt on every release and manual edits will be lost. +> Never store credentials in a blockquote either. + | Column | Never used | |--------|------------| | a | must skip | @@ -52,57 +55,71 @@ class TestExtractRules: def test_extracts_normative_bullets(self): - rules = extract_rules(SAMPLE_CLAUDE_MD) - texts = [r.text for r in rules] + texts = [r.text for r in extract_rules(SAMPLE_CLAUDE_MD).rules] assert "Never commit secrets or API keys." in texts assert "Always write unit tests when adding new code." in texts assert "Prefer f-strings over .format() for readability." in texts def test_skips_non_normative_and_trivia_lines(self): - rules = extract_rules(SAMPLE_CLAUDE_MD) - texts = [r.text for r in rules] + texts = [r.text for r in extract_rules(SAMPLE_CLAUDE_MD).rules] assert not any("npm install" in t for t in texts) assert not any("localhost:3000" in t for t in texts) assert not any("Nice weather" in t for t in texts) def test_skips_fenced_code_blocks(self): - rules = extract_rules(SAMPLE_CLAUDE_MD) + rules = extract_rules(SAMPLE_CLAUDE_MD).rules assert not any("always_skip_me" in r.text for r in rules) def test_skips_table_rows(self): - rules = extract_rules(SAMPLE_CLAUDE_MD) + rules = extract_rules(SAMPLE_CLAUDE_MD).rules assert not any("must skip" in r.text for r in rules) + def test_extracts_blockquote_rules(self): + texts = [r.text for r in extract_rules(SAMPLE_CLAUDE_MD).rules] + assert "Never store credentials in a blockquote either." in texts + + def test_normative_rule_starting_with_command_word_is_kept(self): + doc = "git rebase should never be run interactively in this repo.\n" + texts = [r.text for r in extract_rules(doc).rules] + assert texts == ["git rebase should never be run interactively in this repo."] + def test_paragraph_rules_join_continuation_lines(self): - rules = extract_rules(SAMPLE_CLAUDE_MD) + rules = extract_rules(SAMPLE_CLAUDE_MD).rules para = next(r for r in rules if r.text.startswith("Do not edit generated files")) assert "manual edits will be lost" in para.text def test_line_numbers_are_one_based_and_correct(self): - rules = extract_rules(SAMPLE_CLAUDE_MD) + rules = extract_rules(SAMPLE_CLAUDE_MD).rules never_rule = next(r for r in rules if r.text.startswith("Never commit secrets")) lines = SAMPLE_CLAUDE_MD.splitlines() assert "Never commit secrets" in lines[never_rule.line - 1] def test_deduplicates_case_insensitively(self): doc = "- Never push to main.\n\n- never push to MAIN.\n" - rules = extract_rules(doc) - assert len(rules) == 1 + assert len(extract_rules(doc).rules) == 1 def test_strips_markdown_decoration(self): doc = "- **Always** use [uv](https://docs.astral.sh/uv/) and `pytest` for tests.\n" - rules = extract_rules(doc) + rules = extract_rules(doc).rules assert rules[0].text == "Always use uv and pytest for tests." def test_empty_document_yields_no_rules(self): - assert extract_rules("") == [] - assert extract_rules("# Just a heading\n\nSome plain prose.\n") == [] + assert extract_rules("").rules == () + assert extract_rules("# Just a heading\n\nSome plain prose.\n").rules == () + + def test_overlong_rules_are_counted_not_silently_dropped(self): + doc = "- Never " + "x" * 500 + "\n" + result = extract_rules(doc) + assert result.rules == () + assert result.skipped_too_long == 1 - def test_overlong_paragraphs_are_skipped(self): + def test_max_rule_chars_override_recovers_long_rules(self): doc = "- Never " + "x" * 500 + "\n" - assert extract_rules(doc) == [] + result = extract_rules(doc, max_rule_chars=1000) + assert len(result.rules) == 1 + assert result.skipped_too_long == 0 class TestGeneratePolicyYaml: @@ -130,6 +147,14 @@ def test_yaml_structure_and_traceability(self, tmp_path: Path): assert "1. [CLAUDE.md:7] Never commit secrets." in instructions assert "2. [CLAUDE.md:9] Always write tests." in instructions + def test_header_uses_file_name_not_absolute_path(self, tmp_path: Path): + source = tmp_path / "CLAUDE.md" + source.write_text(SAMPLE_CLAUDE_MD) + text = generate_policy_yaml(self._rules(), source) + + header = text.split("policy:")[0] + assert str(tmp_path) not in header + def test_model_and_on_error_options(self, tmp_path: Path): source = tmp_path / "CLAUDE.md" source.write_text(SAMPLE_CLAUDE_MD) @@ -139,16 +164,20 @@ def test_model_and_on_error_options(self, tmp_path: Path): assert parsed["policy"]["config"]["model"] == "claude-sonnet-4-5" assert parsed["policy"]["config"]["on_error"] == "block" + def test_source_text_param_avoids_reading_file(self, tmp_path: Path): + source = tmp_path / "CLAUDE.md" # never written to disk + text = generate_policy_yaml(self._rules(), source, source_text=SAMPLE_CLAUDE_MD) + assert yaml.safe_load(text)["policy"]["config"]["instructions"] + def test_round_trip_through_policy_loader(self, tmp_path: Path): source = tmp_path / "CLAUDE.md" source.write_text(SAMPLE_CLAUDE_MD) out = tmp_path / "policy.yaml" - out.write_text(generate_policy_yaml(extract_rules(SAMPLE_CLAUDE_MD), source)) + out.write_text(generate_policy_yaml(extract_rules(SAMPLE_CLAUDE_MD).rules, source)) policy = load_policy_from_yaml(str(out)) assert isinstance(policy, SimpleLLMPolicy) - assert "[CLAUDE.md:" in policy._config.instructions class TestAgainstRepoAgentsMd: @@ -156,7 +185,7 @@ class TestAgainstRepoAgentsMd: def test_repo_agents_md_generates_loadable_policy(self, tmp_path: Path): source = REPO_ROOT / "AGENTS.md" - rules = extract_rules(source.read_text(encoding="utf-8")) + rules = extract_rules(source.read_text(encoding="utf-8")).rules # Loose bounds: AGENTS.md evolves, but it is rule-dense. assert len(rules) >= 10 @@ -164,33 +193,80 @@ def test_repo_agents_md_generates_loadable_policy(self, tmp_path: Path): out = tmp_path / "policy.yaml" out.write_text(generate_policy_yaml(rules, source)) policy = load_policy_from_yaml(str(out)) - assert isinstance(policy, SimpleLLMPolicy) + # Every rule is tagged back to a source line. - assert policy._config.instructions.count("[AGENTS.md:") == len(rules) + instructions = yaml.safe_load(out.read_text())["policy"]["config"]["instructions"] + assert instructions.count("[AGENTS.md:") == len(rules) class TestCli: - def test_writes_output_file(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + def _write_sample(self, tmp_path: Path) -> Path: source = tmp_path / "CLAUDE.md" source.write_text(SAMPLE_CLAUDE_MD) - out = tmp_path / "policy.yaml" + return source - exit_code = main([str(source), "-o", str(out)]) + def test_writes_output_file(self, tmp_path: Path): + source = self._write_sample(tmp_path) + out = tmp_path / "policy.yaml" - assert exit_code == 0 + assert main([str(source), "-o", str(out)]) == 0 assert isinstance(load_policy_from_yaml(str(out)), SimpleLLMPolicy) def test_stdout_mode_prints_yaml(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): - source = tmp_path / "CLAUDE.md" - source.write_text(SAMPLE_CLAUDE_MD) - - exit_code = main([str(source)]) + source = self._write_sample(tmp_path) - assert exit_code == 0 + assert main([str(source)]) == 0 printed = capsys.readouterr().out assert yaml.safe_load(printed)["policy"]["class"].endswith("SimpleLLMPolicy") + def test_model_and_on_error_flags_reach_output(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + source = self._write_sample(tmp_path) + + assert main([str(source), "--model", "claude-sonnet-4-5", "--on-error", "block"]) == 0 + config = yaml.safe_load(capsys.readouterr().out)["policy"]["config"] + assert config["model"] == "claude-sonnet-4-5" + assert config["on_error"] == "block" + + def test_no_validate_skips_loader( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ): + source = self._write_sample(tmp_path) + + def boom(_yaml_text: str) -> None: + raise AssertionError("validation should not run with --no-validate") + + monkeypatch.setattr(claude_md, "_validate_policy_yaml", boom) + assert main([str(source), "--no-validate"]) == 0 + + def test_validation_failure_exits_2( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ): + source = self._write_sample(tmp_path) + + def boom(_yaml_text: str) -> None: + raise ValueError("synthetic loader failure") + + monkeypatch.setattr(claude_md, "_validate_policy_yaml", boom) + assert main([str(source)]) == 2 + assert "failed policy-loader validation" in capsys.readouterr().err + + def test_skipped_long_rules_reported_on_stderr(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + source = tmp_path / "CLAUDE.md" + source.write_text("- Never push to main.\n\n- Never " + "x" * 500 + "\n") + + assert main([str(source)]) == 0 + assert "skipped 1 rule candidate(s)" in capsys.readouterr().err + + def test_max_rule_chars_flag(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + source = tmp_path / "CLAUDE.md" + source.write_text("- Never " + "x" * 500 + "\n") + + assert main([str(source), "--max-rule-chars", "1000"]) == 0 + captured = capsys.readouterr() + assert "skipped" not in captured.err + assert yaml.safe_load(captured.out)["policy"]["config"]["instructions"] + def test_missing_input_fails(self, tmp_path: Path): assert main([str(tmp_path / "nope.md")]) == 1