The required check failed. Review the analysis workflow logs.
\n"""
+ (report_dir / "index.html").write_text(page)
+ PY
+
+ - name: Add bounded annotations and job summary
+ if: always()
+ uses: actions/github-script@60a0d83039c74a4aee543508d2ffcb1c3799cdea # v7.0.1
+ env:
+ REPORT_JSON: ${{ runner.temp }}/slop-cop-report/report.json
+ with:
+ script: |
+ const fs = require('fs');
+ const report = JSON.parse(fs.readFileSync(process.env.REPORT_JSON, 'utf8'));
+ const findings = (report.files || []).flatMap((file) => {
+ const chargedRuleIds = new Set((file.rule_costs || [])
+ .filter((cost) => Number(cost.charged_cost) > 0)
+ .map((cost) => cost.rule_id));
+ return (file.findings || [])
+ .filter((finding) => !finding.suppressed &&
+ (finding.blocking || (finding.chargeable && chargedRuleIds.has(finding.rule_id))))
+ .map((finding) => ({ file, finding }));
+ });
+ for (const { file, finding } of findings.slice(0, 50)) {
+ core.warning(`${finding.rule_id}: ${String(finding.advice || '').slice(0, 300)}`, {
+ file: file.path,
+ startLine: finding.line || 1,
+ startColumn: finding.column || 1,
+ });
+ }
+ await core.summary
+ .addHeading('Slop Cop')
+ .addRaw(`**${String(report.decision).toUpperCase()}** — score ${report.score ?? '—'}, threshold ${report.threshold}\n\n`)
+ .addTable([
+ [{data: 'Path', header: true}, {data: 'Head', header: true}, {data: 'Base', header: true}],
+ ...(report.files || []).map((file) => [file.path, String(file.score ?? '—'), String(file.base?.score ?? '—')]),
+ ])
+ .write();
+
+ - name: Upload report
+ if: always()
+ uses: actions/upload-artifact@65462800fd760344b1a7b4382951275a0abb4808 # v4.3.3
+ with:
+ name: slop-cop-pr-${{ steps.inputs.outputs.number || github.event.pull_request.number || 'main' }}-${{ steps.inputs.outputs.head_sha || github.event.pull_request.head.sha || github.sha }}
+ path: ${{ runner.temp }}/slop-cop-report
+ if-no-files-found: error
+ retention-days: 14
+
+ - name: Enforce saved result
+ if: always()
+ env:
+ REPORT_JSON: ${{ runner.temp }}/slop-cop-report/report.json
+ run: |
+ python - <<'PY'
+ import json
+ import os
+ from pathlib import Path
+
+ path = Path(os.environ["REPORT_JSON"])
+ if not path.is_file():
+ raise SystemExit("Slop Cop did not produce report.json")
+ result = json.loads(path.read_text())
+ if result.get("decision") not in {"pass", "overridden", "not_applicable"}:
+ raise SystemExit(f"Slop Cop decision: {result.get('decision')}")
+ PY
diff --git a/.gitignore b/.gitignore
index 8e2e0357..f2abeaad 100644
--- a/.gitignore
+++ b/.gitignore
@@ -48,6 +48,7 @@ temp/
*.temp
*.bak
.scratch/
+plans/
# Python
__pycache__/
diff --git a/AGENTS.md b/AGENTS.md
index 85d4fe96..b602af49 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -9,20 +9,24 @@ documentation for work that uses OpenShell as its runtime.
- Put self-contained implementations and experiments in `projects//`.
- Put project-specific user guides and references in `projects//docs/`.
+- Put repository development and CI tools in `dev-tools//`, with
+ tool-specific documentation beside the tool.
- Put cross-project user-facing documentation in `docs/documentation/`.
- Put Dev Notes (human-written technical notes) `docs/dev-notes/`.
- Put agent-facing repository maintenance workflows in `docs/development/`.
-Before changing a project, read that project's `README.md` and `pyproject.toml`;
-projects are self-contained and may have different platforms, dependencies, and
-validation commands. Before changing anything under `docs/` or `zensical.toml`,
-read `docs/development/index.md`.
+Before changing a project or development tool, read its `README.md` and
+`pyproject.toml`; each is self-contained and may have different platforms,
+dependencies, and validation commands. Before changing anything under `docs/`
+or `zensical.toml`, read `docs/development/index.md`.
## Repository rules
- Make the smallest change that satisfies the task and preserve unrelated work.
- Prefer explicit, clear names and language over concise but ambiguous
alternatives. Value concision when it does not reduce clarity.
+- Use absolute imports in handwritten Python package code. Generated sources
+ may retain the import style produced by their generator.
- Use `uv` for Python dependency management, environments, locking, builds, and
command execution unless a project explicitly documents an exception. Treat
`pyproject.toml` and the committed `uv.lock` as the dependency sources of truth.
@@ -41,8 +45,8 @@ read `docs/development/index.md`.
## Validation
-- For project changes, run the checks documented by that project's README from
- the project directory.
+- For project and development-tool changes, run the checks documented by its
+ README from that directory.
- For documentation changes, run `python3 tests/test_render_dev_notes.py` and
`scripts/build-docs.sh`, then serve the generated site as described in
`docs/development/index.md`.
diff --git a/dev-tools/README.md b/dev-tools/README.md
new file mode 100644
index 00000000..6d0992a5
--- /dev/null
+++ b/dev-tools/README.md
@@ -0,0 +1,9 @@
+# Development tools
+
+This directory contains repository-owned tooling for development, validation,
+and continuous integration. Each tool has its own dependencies, documentation,
+and validation commands.
+
+Current tools:
+
+- `slop-cop`: Editorial policy checks and CI reporting for Dev Notes.
diff --git a/dev-tools/slop-cop/AGENTS.md b/dev-tools/slop-cop/AGENTS.md
new file mode 100644
index 00000000..5a287ce2
--- /dev/null
+++ b/dev-tools/slop-cop/AGENTS.md
@@ -0,0 +1,17 @@
+# Slop Cop development instructions
+
+- Keep documentation, comments, names, tests, and output focused on current
+ behavior and repository requirements. Do not include design history or
+ comparisons with other tools.
+- Preserve the three rule extension paths: declarative phrase rules,
+ declarative bounded-regex rules, and Python rules exported explicitly from
+ `slop_cop.rules.custom.CUSTOM_RULES`.
+- Keep detection separate from scoring. Rules return bounded signals; only the
+ scorer determines points and policy decisions.
+- Preserve exact source offsets through Markdown projection. Add focused
+ positive, counterexample, masking, and boundary tests with every change.
+- Use `uv` for dependency management and execution. Run Ruff, mypy, and pytest
+ before handing off changes.
+- Do not add package discovery, entry-point loading, provider-specific model
+ integrations, automatic rewriting, or persisted generated reports.
+
diff --git a/dev-tools/slop-cop/README.md b/dev-tools/slop-cop/README.md
new file mode 100644
index 00000000..4e20cfd7
--- /dev/null
+++ b/dev-tools/slop-cop/README.md
@@ -0,0 +1,63 @@
+
Slop Cop
+
+Slop Cop reviews Dev Notes for configured editorial signals. It reports precise
+findings, calculates a transparent score, and enforces the repository threshold.
+The score is not an estimate of who or what wrote the document.
+
+## Run locally
+
+Install the locked environment from this directory:
+
+```bash
+uv sync --locked
+```
+
+Check one or more Dev Notes:
+
+```bash
+uv run slop-cop check \
+ --config slop-cop.toml \
+ ../../docs/dev-notes/posts/example.md
+```
+
+Create the same machine and HTML reports used in CI:
+
+```bash
+uv run slop-cop check \
+ --config slop-cop.toml \
+ --html-dir /tmp/slop-cop-report \
+ --json /tmp/slop-cop-report/report.json \
+ ../../docs/dev-notes/posts/example.md
+```
+
+Inspect and test rules:
+
+```bash
+uv run slop-cop list-rules
+uv run slop-cop explain rhetoric.not-just
+uv run slop-cop validate-rules
+uv run slop-cop benchmark --repository-root ../..
+uv run slop-cop check --only-rule rhetoric.not-just path/to/fixture.md
+```
+
+`check` returns `0` when all files pass, `1` when analysis completes but policy
+fails, and `2` for input, configuration, Markdown-projection, or
+report-generation failures.
+Requested reports are written before a policy-failure exit.
+
+## Documentation
+
+- [Rules and custom logic](docs/rules.md)
+- [Scoring](docs/scoring.md)
+- [CI, artifacts, suppressions, and overrides](docs/ci.md)
+
+Run the project checks with:
+
+```bash
+uv run pytest
+uv run ruff check .
+uv run ruff format --check .
+uv run mypy src tests
+uv run slop-cop validate-rules
+uv run slop-cop benchmark --repository-root ../..
+```
diff --git a/dev-tools/slop-cop/benchmarks/dev-note-history.toml b/dev-tools/slop-cop/benchmarks/dev-note-history.toml
new file mode 100644
index 00000000..251964df
--- /dev/null
+++ b/dev-tools/slop-cop/benchmarks/dev-note-history.toml
@@ -0,0 +1,93 @@
+[[benchmark]]
+name = "2026-07-13 initial draft"
+revision = "2b8ae6ef4b74a1eeafa767f9d9f0238af17bdcd5"
+path = "docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+source_url = "https://github.com/NVIDIA/OpenShell-Research/blob/2b8ae6ef4b74a1eeafa767f9d9f0238af17bdcd5/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+expected_decision = "fail"
+min_score = 70
+max_score = 79
+
+[[benchmark]]
+name = "2026-07-14 revised draft"
+revision = "808336e88d418e36364d57d6fda87e47e21dba82"
+path = "docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+source_url = "https://github.com/NVIDIA/OpenShell-Research/blob/808336e88d418e36364d57d6fda87e47e21dba82/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+expected_decision = "pass"
+min_score = 88
+max_score = 93
+
+[[benchmark]]
+name = "2026-07-16 revised draft"
+revision = "95a17f58c65df19166d92e70346cb7574ec84871"
+path = "docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+source_url = "https://github.com/NVIDIA/OpenShell-Research/blob/95a17f58c65df19166d92e70346cb7574ec84871/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+expected_decision = "pass"
+min_score = 84
+max_score = 91
+
+[[benchmark]]
+name = "2026-07-17 revised draft"
+revision = "be515c9c2684e1e3febec058f1a9e0e90da16a72"
+path = "docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+source_url = "https://github.com/NVIDIA/OpenShell-Research/blob/be515c9c2684e1e3febec058f1a9e0e90da16a72/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+expected_decision = "pass"
+min_score = 88
+max_score = 94
+
+[[benchmark]]
+name = "2026-07-18 revised draft"
+revision = "069696b0a28ffd0cd77e1e043a532f81b9cec3e2"
+path = "docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+source_url = "https://github.com/NVIDIA/OpenShell-Research/blob/069696b0a28ffd0cd77e1e043a532f81b9cec3e2/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md"
+expected_decision = "pass"
+min_score = 94
+max_score = 100
+
+[[benchmark]]
+name = "2026-07-20 published draft"
+revision = "b0d481796b8a0492053c7b3cac0c65444a2e99be"
+path = "docs/dev-notes/posts/2026-07-20-policy-controlling-reachy-mini-with-openshell.md"
+source_url = "https://github.com/NVIDIA/OpenShell-Research/blob/b0d481796b8a0492053c7b3cac0c65444a2e99be/docs/dev-notes/posts/2026-07-20-policy-controlling-reachy-mini-with-openshell.md"
+expected_decision = "pass"
+min_score = 97
+max_score = 100
+
+[[benchmark]]
+name = "clean technical prose"
+path = "docs/dev-notes/posts/benchmark-clean.md"
+fixture_path = "dev-tools/slop-cop/benchmarks/fixtures/clean-technical.md"
+expected_decision = "pass"
+min_score = 100
+max_score = 100
+
+[[benchmark]]
+name = "legitimate contrast and citation"
+path = "docs/dev-notes/posts/benchmark-contrast.md"
+fixture_path = "dev-tools/slop-cop/benchmarks/fixtures/legitimate-contrast-and-citation.md"
+expected_decision = "pass"
+min_score = 100
+max_score = 100
+
+[[benchmark]]
+name = "normal technical structure"
+path = "docs/dev-notes/posts/benchmark-structure.md"
+fixture_path = "dev-tools/slop-cop/benchmarks/fixtures/normal-technical-structure.md"
+expected_decision = "pass"
+min_score = 100
+max_score = 100
+
+[[benchmark]]
+name = "dense multi-family prose"
+path = "docs/dev-notes/posts/benchmark-dense.md"
+fixture_path = "dev-tools/slop-cop/benchmarks/fixtures/dense-formulaic.md"
+expected_decision = "fail"
+min_score = 0
+max_score = 79
+
+[[benchmark]]
+name = "blocking assistant residue"
+path = "docs/dev-notes/posts/benchmark-blocking.md"
+fixture_path = "dev-tools/slop-cop/benchmarks/fixtures/blocking-residue.md"
+expected_decision = "fail"
+min_score = 0
+max_score = 0
diff --git a/dev-tools/slop-cop/benchmarks/fixtures/blocking-residue.md b/dev-tools/slop-cop/benchmarks/fixtures/blocking-residue.md
new file mode 100644
index 00000000..d94e7ece
--- /dev/null
+++ b/dev-tools/slop-cop/benchmarks/fixtures/blocking-residue.md
@@ -0,0 +1,4 @@
+# Generated response
+
+As an AI language model, I cannot verify the implementation. Let me know if
+you would like me to expand this into a complete Dev Note.
diff --git a/dev-tools/slop-cop/benchmarks/fixtures/clean-technical.md b/dev-tools/slop-cop/benchmarks/fixtures/clean-technical.md
new file mode 100644
index 00000000..f87db6b5
--- /dev/null
+++ b/dev-tools/slop-cop/benchmarks/fixtures/clean-technical.md
@@ -0,0 +1,6 @@
+# Request enforcement
+
+The gateway parses each request once and passes an immutable representation to
+the configured gates. A denied request returns a structured error before any
+upstream connection is opened. Tests cover malformed input, allowed traffic,
+and denial responses.
diff --git a/dev-tools/slop-cop/benchmarks/fixtures/dense-formulaic.md b/dev-tools/slop-cop/benchmarks/fixtures/dense-formulaic.md
new file mode 100644
index 00000000..764d5f9b
--- /dev/null
+++ b/dev-tools/slop-cop/benchmarks/fixtures/dense-formulaic.md
@@ -0,0 +1,13 @@
+# System overview
+
+Clients receive data now. Clients receive output later. Clients receive errors immediately. It is not just a wrapper, but a policy boundary. This applies in the realm of systems.
+
+Clients receive data now. Clients receive output later. Clients receive errors immediately. It is not just a wrapper, but a policy boundary. This applies in the realm of systems.
+
+Clients receive data now. Clients receive output later. Clients receive errors immediately. It is not just a wrapper, but a policy boundary. This applies in the realm of systems.
+
+Clients receive data now. Clients receive output later. Clients receive errors immediately. It is not just a wrapper, but a policy boundary. This applies in the realm of systems.
+
+Clients receive data now. Clients receive output later. Clients receive errors immediately. It is not just a wrapper, but a policy boundary. This applies in the realm of systems.
+
+Clients receive data now. Clients receive output later. Clients receive errors immediately. It is not just a wrapper, but a policy boundary. This applies in the realm of systems.
diff --git a/dev-tools/slop-cop/benchmarks/fixtures/legitimate-contrast-and-citation.md b/dev-tools/slop-cop/benchmarks/fixtures/legitimate-contrast-and-citation.md
new file mode 100644
index 00000000..8803f8a6
--- /dev/null
+++ b/dev-tools/slop-cop/benchmarks/fixtures/legitimate-contrast-and-citation.md
@@ -0,0 +1,6 @@
+# Policy boundary
+
+The adapter is not just a transport wrapper; it is the authority that constrains
+device operations. This distinction matters because the sandbox never receives
+the hardware handle. A recent study reports the same separation between request
+selection and physical authority ([paper](https://example.test/paper)).
diff --git a/dev-tools/slop-cop/benchmarks/fixtures/normal-technical-structure.md b/dev-tools/slop-cop/benchmarks/fixtures/normal-technical-structure.md
new file mode 100644
index 00000000..02446007
--- /dev/null
+++ b/dev-tools/slop-cop/benchmarks/fixtures/normal-technical-structure.md
@@ -0,0 +1,9 @@
+# Validation sequence
+
+---
+
+**Parse the request.** Reject malformed input before gate evaluation.
+
+**Apply policy.** Evaluate the configured gates in a stable order.
+
+**Return the result.** Report the first denial with its concrete reason.
diff --git a/dev-tools/slop-cop/docs/ci.md b/dev-tools/slop-cop/docs/ci.md
new file mode 100644
index 00000000..4fb2d249
--- /dev/null
+++ b/dev-tools/slop-cop/docs/ci.md
@@ -0,0 +1,97 @@
+# CI integration
+
+The `Slop Cop` workflow analyzes Dev Notes and uploads `index.html` and
+`report.json` as a 14-day artifact. The `Slop Cop report` workflow validates that
+artifact and updates one sticky PR comment. Generated reports are not committed.
+
+## Enforcement boundary
+
+For a pull request, the required analysis runs Slop Cop code, configuration,
+and dependencies from the PR base revision against the candidate Dev Notes.
+The separate `Slop Cop candidate` workflow tests candidate Slop Cop changes
+without credentials and scans the complete candidate Dev Note corpus as a
+non-authoritative preview. Candidate code cannot access or publish artifacts in
+the trusted analysis workflow run.
+
+The `pull_request` event can use workflow orchestration changed by the pull
+request. Base-revision analyzer selection does not protect that orchestration by
+itself. Enforce `.github/workflows/slop-cop.yml` as an organization or
+repository-ruleset required workflow using its default-branch definition. Add
+all three Slop Cop workflow files to the repository's protected workflow paths and
+require designated review for changes to them. A branch-protection check name
+without this ruleset protection does not establish the same boundary.
+
+The analysis workflow has read-only permissions. It runs for every PR so the
+required check is present even when no Dev Note changed. A non-applicable run
+has a null score and creates no misleading clean score.
+
+The trusted reporting workflow runs from the default branch after analysis. It
+never executes PR-supplied code. It accepts exactly one artifact associated with
+the completed run, checks the PR number and current head SHA, revalidates any
+override through the GitHub API, rejects unknown JSON schemas and oversized or
+malformed data, and treats every report string as untrusted before creating
+Markdown. Stale runs do not update the comment.
+
+## Artifact and comment
+
+The artifact is named `slop-cop-pr--` and contains:
+
+```text
+slop-cop-report/
+├── index.html
+└── report.json
+```
+
+Download the artifact from the sticky PR comment or the analysis run. Open
+`index.html` locally; it is self-contained and requires no network or
+JavaScript. The HTML report shows score deductions first. Signals covered by a
+document allowance and advisory-only signals are grouped separately with exact,
+expandable source context. The PR comment shows the decision, threshold,
+per-file scores, base deltas, contributing findings, override details, analyzed
+revision, and expiration note.
+
+## Suppress one finding
+
+Place a standalone directive immediately before the next scanned prose block:
+
+```html
+
+```
+
+Name stable rule IDs and supply a concrete reason. The directive suppresses the
+first named finding in that block. Unknown IDs, missing reasons, unused
+directives, and file-wide wildcards are errors. Suppressions remain visible in
+terminal, JSON, and HTML output.
+
+## Override one revision
+
+An authorized maintainer can override a result by approving the exact current
+head revision with this line in the review body:
+
+```text
+Slop-Cop-Override:
+```
+
+The reviewer must currently have write, maintain, or admin permission. A new
+commit makes the review stale. The override changes the CI decision to
+`OVERRIDDEN`; it does not change the score, hide findings, or mark an incomplete
+analysis complete. The report records the reviewer, reason, review, and head
+revision.
+
+## External rules
+
+Built-in and declarative rules are offline. A configured custom rule can send
+selected projected prose to its named service. Required external rules fail
+closed when a credential or service is unavailable. Ordinary Actions secrets
+are unavailable to fork PR workflows, so enabling a required external rule
+must account for that behavior. The workflow never executes candidate rule code
+with repository secrets.
+
+## Required check rollout
+
+The introducing PR has neither a base-revision analyzer nor a default-branch
+required-workflow definition, so it cannot enforce itself. Validate it with the
+complete local suite and designated workflow review. After merge, configure the
+ruleset required workflow and protected workflow paths, require a successful
+full-corpus push run, then exercise passing, failing, stale, override, rename,
+deletion, and non-applicable PR revisions before treating the check as enforced.
diff --git a/dev-tools/slop-cop/docs/rules.md b/dev-tools/slop-cop/docs/rules.md
new file mode 100644
index 00000000..bc36efe5
--- /dev/null
+++ b/dev-tools/slop-cop/docs/rules.md
@@ -0,0 +1,215 @@
+# Rules
+
+Rules detect editorial signals. Repository configuration determines whether a
+signal is advisory, chargeable, or blocking and how it affects the score.
+
+## Manage declarative rules
+
+Use a declarative rule for a literal phrase or bounded regular expression. Add
+the rule block to `slop-cop.toml` and add positive and counterexample cases to
+`tests/rule_cases.toml`. No engine or renderer change is required.
+
+Phrase rule:
+
+```toml
+[[custom_rules.phrase]]
+id = "custom.empty-intensifier"
+version = 1
+category = "vocabulary"
+severity = "warning"
+title = "Empty intensifier"
+rationale = "The phrase asserts importance without naming an effect."
+advice = "Name the concrete effect."
+phrases = ["deeply transformative"]
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 2
+repeat_cost = 1
+cap = 5
+```
+
+Regular-expression rule:
+
+```toml
+[[custom_rules.regex]]
+id = "custom.generic-promise"
+version = 1
+category = "rhetoric"
+severity = "warning"
+title = "Generic promise"
+rationale = "The sentence promises unspecified later detail."
+advice = "Name the follow-up topic or remove the promise."
+pattern = '''\bmore on (?:that|this) (?:later|soon)\b'''
+flags = ["IGNORECASE"]
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 2
+repeat_cost = 1
+cap = 5
+```
+
+Each rule needs cases that execute its match and a legitimate nonmatch:
+
+```toml
+[[case]]
+name = "empty-intensifier-positive"
+rule_id = "custom.empty-intensifier"
+kind = "positive"
+source = "This is deeply transformative work."
+
+[[case]]
+name = "empty-intensifier-counterexample"
+rule_id = "custom.empty-intensifier"
+kind = "counterexample"
+source = "This changes request routing."
+```
+
+Each rule has a stable ID, positive integer version, category, title, rationale,
+advice, severity, allowance, and explicit scoring values. Increment the version
+when matching behavior or external-response interpretation changes.
+
+To disable a rule, set `enabled = false`. To remove one, remove its definition,
+cases, exceptions, and source suppressions in the same change. Run
+`slop-cop validate-rules` to find dangling references and missing coverage.
+
+Declarative regular expressions are limited to 500 characters. They may use the
+configured `IGNORECASE`, `MULTILINE`, `DOTALL`, and `VERBOSE` flags. Inline
+flags, backreferences, recursive constructs, conditionals, and lookbehind are
+rejected. Matching has a bounded timeout.
+
+## Add custom Python logic
+
+Use a custom Python rule when detection requires an algorithm or an external
+service. Create one module under `src/slop_cop/rules/custom/`, export one rule
+instance, and add it to `CUSTOM_RULES` in that directory's `__init__.py`.
+
+```python
+from slop_cop.rules.api import (
+ FunctionRule,
+ RuleContext,
+ RuleEvaluation,
+ RuleMetadata,
+ RuleSignal,
+)
+from slop_cop.runtime import RuleRuntime
+
+METADATA = RuleMetadata(
+ id="custom.repeated-claim-opener",
+ version=1,
+ category="repetition",
+ title="Repeated claim opener",
+ rationale="Repeated claim openings make consecutive paragraphs formulaic.",
+ advice="Vary the paragraph structure or combine the claims.",
+)
+
+
+async def evaluate(context: RuleContext, runtime: RuleRuntime) -> RuleEvaluation:
+ signals = tuple(
+ RuleSignal(start=item.start, end=item.end, key=item.normalized)
+ for item in context.repeated_sentence_starts(minimum_count=3)
+ )
+ return RuleEvaluation(signals=signals)
+
+
+RULE = FunctionRule(metadata=METADATA, evaluator=evaluate)
+```
+
+Register it explicitly:
+
+```python
+from .repeated_claim_opener import RULE as REPEATED_CLAIM_OPENER
+
+CUSTOM_RULES = (REPEATED_CLAIM_OPENER,)
+```
+
+Add the scoring policy separately:
+
+```toml
+[rules."custom.repeated-claim-opener"]
+enabled = true
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 3
+repeat_cost = 2
+cap = 9
+on_error = "fail"
+```
+
+The engine supplies Markdown projection, source mapping, segmentation,
+suppressions, scoring, and reporting. The custom evaluator only returns bounded
+signals and optional audit data. It cannot set the score, threshold, decision,
+or override state.
+
+`context.projected_prose`, `context.tokens`, `context.sentences`, and
+`context.paragraphs` contain reader-visible prose with code, comments, link
+destinations, image alt text, and HTML tag attributes removed. Repetition rules
+should use `context.repetition_prose`, `context.repetition_tokens`,
+`context.repetition_sentences`, and `context.repetition_paragraphs`. That view
+also removes headings, link labels, and figure captions, where repeated wording
+is commonly structural or descriptive. Both views preserve source length and
+line endings, so returned spans use the original source offsets.
+
+Add positive and counterexample entries to `tests/rule_cases.toml`. The shared
+contract suite validates IDs, metadata, result bounds, spans, ordering, and
+fixture coverage for built-in, declarative, and custom rules.
+
+## Call an external judge
+
+An external custom rule declares a named service in its metadata and obtains it
+through `RuleRuntime`:
+
+```python
+response = await runtime.service().post_json(
+ {"schema_version": 1, "prose": context.projected_prose}
+)
+result = JudgeResponse.model_validate(response.data)
+if result.judge_revision != runtime.settings["required_judge_revision"]:
+ raise ValueError("judge revision does not match configured revision")
+signal = RuleSignal.document(
+ key=result.label,
+ units=result.strength,
+ detail=result.explanation,
+ evidence=context.map_exact_quotes(result.evidence),
+)
+return RuleEvaluation(
+ signals=(signal,),
+ audit={**response.audit, "judge_revision": result.judge_revision},
+)
+```
+
+Configure the service in trusted repository configuration:
+
+```toml
+[services.editorial_judge]
+url = "https://judge.internal.example/v1/evaluate"
+token_env = "SLOP_COP_EDITORIAL_JUDGE_TOKEN"
+timeout_seconds = 20
+max_response_bytes = 65536
+max_attempts = 1
+
+[rules."custom.editorial-judge"]
+severity = "warning"
+service = "editorial_judge"
+max_signal_units = 5
+fixed_allowance = 0
+first_cost = 4
+repeat_cost = 2
+cap = 12
+settings = { required_judge_revision = "editorial-v1" }
+```
+
+The custom rule owns its request data, strict response model, and conversion to
+signals. The runtime owns the allowed origin, authentication, deadlines,
+redirect rejection, response limit, idempotency key, and content-safe audit
+record. A required CI rule must use this runtime instead of direct sockets,
+subprocesses, SDK transports, or an unconfigured destination.
+
+Document-scoped signals do not participate in passage-density calculations.
+Emit exact source spans when local concentration matters. Evidence quotations
+are display evidence and do not add scoring units.
+
+Use `on_error = "fail"` for an external rule that affects the required CI
+threshold. Use `advisory` only for a zero-point experiment. Tests must use a fake
+transport and cover clean, chargeable, malformed, oversized, wrong-revision,
+timeout, and transport outcomes.
diff --git a/dev-tools/slop-cop/docs/scoring.md b/dev-tools/slop-cop/docs/scoring.md
new file mode 100644
index 00000000..bee1d678
--- /dev/null
+++ b/dev-tools/slop-cop/docs/scoring.md
@@ -0,0 +1,121 @@
+# Scoring
+
+Slop Cop starts each file at 100 and subtracts bounded category costs. The
+default Dev Notes threshold is 80. A score describes configured editorial
+signals; it is not a probability, authorship judgment, or factuality grade.
+
+## Rule cost
+
+Rules emit signals. After suppressions and overlap deduplication, Slop Cop sums
+their bounded units and applies the rule allowance:
+
+```text
+document_excess = max(0, units - allowance(document))
+
+base_cost = 0 when excess is 0
+base_cost = first_cost + repeat_cost * (excess - 1) otherwise
+```
+
+An allowance may combine a fixed count with a document-density allowance based
+on the rule's natural opportunity: words, sentences, or paragraphs. A phrase
+rule and a sentence-opening rule therefore do not share an arbitrary universal
+denominator.
+
+## Passage density
+
+Selected rules and categories also inspect rolling word, sentence, or paragraph
+windows. Slop Cop finds the single densest window and charges a bounded density
+cost from its peak excess:
+
+```text
+peak_excess = max(0, signals_in_window - allowed_units)
+
+density_cost = 0
+ when peak_excess is 0
+else min(density_cap,
+ density_first_cost + density_repeat_cost * (peak_excess - 1))
+```
+
+Only primary, unsuppressed, exact source spans enter these windows. The peak is
+charged once even when overlapping windows contain the same findings. Appending
+clean prose cannot lower an existing peak. Document-scoped external judgments
+do not enter passage-density calculations.
+
+The final rule cost is capped:
+
+```text
+rule_cost = min(rule_cap, base_cost + density_cost)
+```
+
+This permits two occurrences spread across a long note while still charging a
+cluster in three neighboring paragraphs when that rule's policy says
+concentration matters.
+
+## Category and file decisions
+
+Related rules share a category cap. An optional category-density cost can
+capture a cluster of several distinct weak signals:
+
+```text
+category_cost = min(category_cap,
+ sum(rule_costs) + category_density_cost)
+score = max(0, round(100 - sum(category_costs)))
+```
+
+A file passes only when its score meets the threshold, it has no unsuppressed
+blocking finding, and required analysis completed. An advisory external error
+marks the analysis incomplete without manufacturing a penalty. A required
+external error fails the analysis.
+
+Each changed Dev Note is scored independently. The PR score is the lowest file
+score, and every changed file must pass. A clean second file cannot conceal a
+failing first file. Base scores and finding changes appear for comparison, but
+only the head result controls enforcement.
+
+## Tune policy
+
+Tune allowances, costs, density windows, and caps in `slop-cop.toml`; do not put
+score calculations inside detector code. Every scored change needs fixtures at
+the zero, first, repeat, density, and cap boundaries. Confirm that:
+
+- a common isolated construction does not fail a clean note;
+- a concentrated passage costs more than the same findings spread apart;
+- clean appended prose does not dilute an existing density cost;
+- overlapping detectors do not double charge one source span;
+- one weak category cannot exceed its cap;
+- the accepted Dev Note remains at or above 80 without a blanket exception;
+- dense multi-category fixtures remain below 80.
+
+Keep unresolved or context-sensitive rules advisory with zero points until
+their counterexamples and score boundaries are reliable.
+
+## Calibration benchmarks
+
+`benchmarks/dev-note-history.toml` records immutable Git revisions and focused
+repository fixtures with expected decisions and acceptable score ranges.
+`slop-cop benchmark` scores each source with the active configuration and fails
+when a score or decision leaves its declared range. This makes calibration drift
+a tested policy change instead of an incidental result of editing rule weights.
+
+The current references follow one Dev Note through revision and publication:
+
+| Reference | Expected score | Baseline score |
+| --- | ---: | ---: |
+| [2026-07-13 initial draft](https://github.com/NVIDIA/OpenShell-Research/blob/2b8ae6ef4b74a1eeafa767f9d9f0238af17bdcd5/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md) | 70–79, fail | 79 |
+| [2026-07-14 revised draft](https://github.com/NVIDIA/OpenShell-Research/blob/808336e88d418e36364d57d6fda87e47e21dba82/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md) | 88–93, pass | 91 |
+| [2026-07-16 revised draft](https://github.com/NVIDIA/OpenShell-Research/blob/95a17f58c65df19166d92e70346cb7574ec84871/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md) | 84–91, pass | 91 |
+| [2026-07-17 revised draft](https://github.com/NVIDIA/OpenShell-Research/blob/be515c9c2684e1e3febec058f1a9e0e90da16a72/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md) | 88–94, pass | 91 |
+| [2026-07-18 revised draft](https://github.com/NVIDIA/OpenShell-Research/blob/069696b0a28ffd0cd77e1e043a532f81b9cec3e2/docs/dev-notes/posts/2026-07-13-policy-controlling-reachy-mini-with-openshell.md) | 94–100, pass | 100 |
+| [2026-07-20 published draft](https://github.com/NVIDIA/OpenShell-Research/blob/b0d481796b8a0492053c7b3cac0c65444a2e99be/docs/dev-notes/posts/2026-07-20-policy-controlling-reachy-mini-with-openshell.md) | 97–100, pass | 100 |
+| Clean technical fixture | 100, pass | 100 |
+| Legitimate contrast and citation fixture | 100, pass | 100 |
+| Normal technical structure fixture | 100, pass | 100 |
+| Dense multi-family fixture | 0–79, fail | 62 |
+| Blocking assistant residue fixture | 0, fail | 0 |
+
+Historical revisions detect drift against real prose; their chronology is not a
+quality label. Focused fixtures define the intended treatment of clean prose,
+legitimate constructions, dense independent signals, and blocking residue. Use
+ranges wide enough for targeted detector improvements but narrow enough to catch
+a material scoring regression. Update a range only when the new result is an
+intentional editorial-policy change supported by the source.
diff --git a/dev-tools/slop-cop/pyproject.toml b/dev-tools/slop-cop/pyproject.toml
new file mode 100644
index 00000000..6583df23
--- /dev/null
+++ b/dev-tools/slop-cop/pyproject.toml
@@ -0,0 +1,48 @@
+[project]
+name = "slop-cop"
+version = "0.1.0"
+description = "Editorial policy checks for OpenShell Dev Notes"
+license = "Apache-2.0"
+requires-python = ">=3.12"
+dependencies = [
+ "httpx>=0.28.1,<0.29",
+ "pydantic>=2.11.0,<3",
+ "regex>=2025.7.34",
+]
+
+[project.scripts]
+slop-cop = "slop_cop.cli:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[dependency-groups]
+dev = [
+ "mypy>=1.17.1,<2",
+ "pytest>=8.4.1,<9",
+ "pytest-asyncio>=1.1.0,<2",
+ "ruff>=0.12.8,<0.13",
+ "types-regex>=2025.7.34.20250731",
+]
+
+[tool.pytest.ini_options]
+addopts = "-q"
+testpaths = ["tests"]
+asyncio_mode = "auto"
+
+[tool.ruff]
+line-length = 100
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "SIM", "RUF", "TID252"]
+allowed-confusables = ["’", "”"]
+
+[tool.ruff.lint.flake8-tidy-imports]
+ban-relative-imports = "all"
+
+[tool.mypy]
+python_version = "3.12"
+strict = true
+packages = ["slop_cop"]
diff --git a/dev-tools/slop-cop/slop-cop.toml b/dev-tools/slop-cop/slop-cop.toml
new file mode 100644
index 00000000..149e42d9
--- /dev/null
+++ b/dev-tools/slop-cop/slop-cop.toml
@@ -0,0 +1,433 @@
+schema_version = 1
+profile = "dev-notes"
+threshold = 80
+paths = ["docs/dev-notes/posts/*.md"]
+
+[contexts]
+scan_blockquotes = false
+scan_headings = true
+scan_captions = true
+
+[categories.artifact]
+cap = 100
+
+[categories.rhetoric]
+cap = 25
+
+[categories.rhetoric.density]
+unit = "paragraph"
+window = 3
+allowed_units = 3
+first_cost = 2
+repeat_cost = 1
+cap = 5
+
+[categories.vocabulary]
+cap = 15
+
+[categories.vocabulary.density]
+unit = "paragraph"
+window = 3
+allowed_units = 3
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[categories.repetition]
+cap = 25
+
+[categories.repetition.density]
+unit = "paragraph"
+window = 3
+allowed_units = 4
+first_cost = 2
+repeat_cost = 1
+cap = 5
+
+[categories.attribution]
+cap = 15
+
+[categories.ending]
+cap = 15
+
+[categories.structure]
+cap = 0
+
+[rules."artifact.ai-disclosure"]
+severity = "error"
+blocking = true
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 100
+repeat_cost = 100
+cap = 100
+
+[rules."artifact.chat-preamble"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 15
+repeat_cost = 10
+cap = 35
+
+[rules."artifact.continuation-offer"]
+severity = "error"
+blocking = true
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 100
+repeat_cost = 100
+cap = 100
+
+[rules."artifact.placeholder"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 8
+repeat_cost = 5
+cap = 20
+
+[rules."artifact.instruction-residue"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 10
+repeat_cost = 6
+cap = 25
+
+[rules."rhetoric.not-but"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 3
+repeat_cost = 2
+cap = 9
+
+[rules."rhetoric.not-but".document_density]
+unit = "word"
+interval = 1000
+allowed_units = 1
+
+[rules."rhetoric.not-but".density]
+unit = "paragraph"
+window = 3
+allowed_units = 1
+first_cost = 2
+repeat_cost = 2
+cap = 6
+
+[rules."rhetoric.not-just"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 3
+repeat_cost = 2
+cap = 9
+
+[rules."rhetoric.not-just".document_density]
+unit = "word"
+interval = 1000
+allowed_units = 1
+
+[rules."rhetoric.not-just".density]
+unit = "paragraph"
+window = 3
+allowed_units = 1
+first_cost = 2
+repeat_cost = 2
+cap = 6
+
+[rules."rhetoric.no-no"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 3
+repeat_cost = 2
+cap = 9
+
+[rules."rhetoric.no-no".density]
+unit = "paragraph"
+window = 3
+allowed_units = 1
+first_cost = 2
+repeat_cost = 2
+cap = 6
+
+[rules."rhetoric.imperative-reframe"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 3
+repeat_cost = 2
+cap = 9
+
+[rules."rhetoric.imperative-reframe".density]
+unit = "paragraph"
+window = 3
+allowed_units = 1
+first_cost = 2
+repeat_cost = 2
+cap = 6
+
+[rules."rhetoric.dramatic-fragment"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."rhetoric.throat-clearing"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."rhetoric.testament"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."rhetoric.crucial-role"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."rhetoric.landscape-metaphor"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."rhetoric.formulaic-certainty"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."rhetoric.parallel-negation"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."vocabulary.stock.delves"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[rules."vocabulary.stock.tapestry"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[rules."vocabulary.stock.realm"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[rules."vocabulary.stock.underscore"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[rules."vocabulary.stock.navigate"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[rules."vocabulary.stock.multifaceted"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 1
+repeat_cost = 1
+cap = 4
+
+[rules."repetition.ngram"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 2
+first_cost = 3
+repeat_cost = 2
+cap = 10
+
+[rules."repetition.ngram".density]
+unit = "paragraph"
+window = 5
+allowed_units = 2
+first_cost = 2
+repeat_cost = 1
+cap = 4
+
+[rules."repetition.sentence-opener"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 6
+repeat_cost = 3
+cap = 15
+
+[rules."repetition.sentence-opener".density]
+unit = "sentence"
+window = 5
+allowed_units = 2
+first_cost = 2
+repeat_cost = 1
+cap = 6
+
+[rules."repetition.template-shape"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 6
+repeat_cost = 3
+cap = 15
+
+[rules."repetition.template-shape".density]
+unit = "paragraph"
+window = 3
+allowed_units = 2
+first_cost = 3
+repeat_cost = 2
+cap = 6
+
+[rules."repetition.emphatic-fragments"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 6
+repeat_cost = 3
+cap = 15
+
+[rules."repetition.emphatic-fragments".density]
+unit = "sentence"
+window = 5
+allowed_units = 2
+first_cost = 3
+repeat_cost = 2
+cap = 6
+
+[rules."repetition.hedge-stack"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 2
+repeat_cost = 2
+cap = 8
+
+[rules."repetition.question-answer"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 2
+repeat_cost = 2
+cap = 8
+
+[rules."attribution.vague-authority"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 2
+repeat_cost = 2
+cap = 8
+
+[rules."attribution.citationless-study"]
+severity = "warning"
+max_signal_units = 1
+fixed_allowance = 1
+first_cost = 2
+repeat_cost = 2
+cap = 8
+
+[rules."ending.participial-tail"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."ending.generic-explanation"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."structure.bold-leadins"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."structure.triad"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."structure.bullet-run"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."structure.horizontal-rules"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[rules."structure.duplicate-title"]
+severity = "info"
+max_signal_units = 1
+fixed_allowance = 0
+first_cost = 0
+repeat_cost = 0
+cap = 0
+
+[vocabulary]
+allowed_terms = ["OpenShell", "Dev Note"]
+
+[custom_rules]
+
+[services]
diff --git a/dev-tools/slop-cop/src/slop_cop/__init__.py b/dev-tools/slop-cop/src/slop_cop/__init__.py
new file mode 100644
index 00000000..88f73165
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/__init__.py
@@ -0,0 +1,10 @@
+"""Editorial policy checks for OpenShell Dev Notes."""
+
+from importlib.metadata import PackageNotFoundError, version
+
+try:
+ __version__ = version("slop-cop")
+except PackageNotFoundError:
+ __version__ = "0.1.0"
+
+__all__ = ["__version__"]
diff --git a/dev-tools/slop-cop/src/slop_cop/assets/slop-cop.png b/dev-tools/slop-cop/src/slop_cop/assets/slop-cop.png
new file mode 100644
index 00000000..e4435c10
Binary files /dev/null and b/dev-tools/slop-cop/src/slop_cop/assets/slop-cop.png differ
diff --git a/dev-tools/slop-cop/src/slop_cop/benchmarks.py b/dev-tools/slop-cop/src/slop_cop/benchmarks.py
new file mode 100644
index 00000000..5f20e444
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/benchmarks.py
@@ -0,0 +1,184 @@
+"""Score calibration references."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import tomllib
+from collections.abc import Callable
+from pathlib import Path, PurePosixPath
+from typing import Literal
+from urllib.parse import urlsplit
+
+from pydantic import Field, field_validator, model_validator
+
+from slop_cop.config import SlopCopConfig, StrictModel
+from slop_cop.document import build_document
+from slop_cop.engine import analyze_document
+from slop_cop.findings import Decision
+from slop_cop.rules.registry import RuleRegistry
+
+_REVISION = re.compile(r"^[0-9a-f]{40}$")
+
+
+class BenchmarkReference(StrictModel):
+ """One immutable document revision and its acceptable result."""
+
+ name: str = Field(min_length=1, max_length=100)
+ revision: str | None = None
+ path: str = Field(min_length=1, max_length=4_096)
+ source_url: str | None = Field(default=None, min_length=1, max_length=4_096)
+ fixture_path: str | None = Field(default=None, min_length=1, max_length=4_096)
+ expected_decision: Literal["pass", "fail"]
+ min_score: int = Field(ge=0, le=100)
+ max_score: int = Field(ge=0, le=100)
+
+ @field_validator("revision")
+ @classmethod
+ def validate_revision(cls, value: str | None) -> str | None:
+ if value is None:
+ return None
+ if not _REVISION.fullmatch(value):
+ raise ValueError("benchmark revision must be a 40-character lowercase commit SHA")
+ return value
+
+ @field_validator("path")
+ @classmethod
+ def validate_path(cls, value: str) -> str:
+ path = PurePosixPath(value)
+ if path.is_absolute() or ".." in path.parts or path.suffix.casefold() != ".md":
+ raise ValueError("benchmark path must be a repository-relative Markdown path")
+ return value
+
+ @field_validator("source_url")
+ @classmethod
+ def validate_source_url(cls, value: str | None) -> str | None:
+ if value is None:
+ return None
+ parsed = urlsplit(value)
+ if parsed.scheme != "https" or parsed.hostname != "github.com":
+ raise ValueError("benchmark source_url must be an HTTPS GitHub URL")
+ return value
+
+ @field_validator("fixture_path")
+ @classmethod
+ def validate_fixture_path(cls, value: str | None) -> str | None:
+ if value is None:
+ return None
+ path = PurePosixPath(value)
+ if path.is_absolute() or ".." in path.parts or path.suffix.casefold() != ".md":
+ raise ValueError("fixture_path must be a repository-relative Markdown path")
+ return value
+
+ @model_validator(mode="after")
+ def validate_range(self) -> BenchmarkReference:
+ if self.min_score > self.max_score:
+ raise ValueError("benchmark min_score cannot exceed max_score")
+ historical = self.revision is not None or self.source_url is not None
+ fixture = self.fixture_path is not None
+ if historical == fixture:
+ raise ValueError("benchmark must select one historical revision or fixture_path")
+ if historical and (
+ self.revision is None
+ or self.source_url is None
+ or f"/blob/{self.revision}/{self.path}" not in self.source_url
+ ):
+ raise ValueError("benchmark source_url must identify its revision and path")
+ return self
+
+
+class BenchmarkManifest(StrictModel):
+ """Strict calibration manifest."""
+
+ benchmark: tuple[BenchmarkReference, ...] = Field(min_length=1)
+
+ @field_validator("benchmark", mode="before")
+ @classmethod
+ def convert_toml_array(cls, value: object) -> object:
+ if isinstance(value, list):
+ return tuple(value)
+ return value
+
+
+class BenchmarkResult(StrictModel):
+ """Observed result for one calibration reference."""
+
+ reference: BenchmarkReference
+ score: int
+ decision: Decision
+ within_range: bool
+
+
+SourceLoader = Callable[[BenchmarkReference], bytes]
+
+
+def load_benchmark_manifest(path: Path) -> BenchmarkManifest:
+ """Load and validate a benchmark manifest."""
+
+ return BenchmarkManifest.model_validate(tomllib.loads(path.read_text(encoding="utf-8")))
+
+
+def git_source_loader(repository_root: Path) -> SourceLoader:
+ """Return a loader for committed history and repository fixtures."""
+
+ def load(reference: BenchmarkReference) -> bytes:
+ if reference.fixture_path is not None:
+ root = repository_root.resolve(strict=True)
+ fixture = (root / reference.fixture_path).resolve(strict=True)
+ try:
+ fixture.relative_to(root)
+ except ValueError as error:
+ raise ValueError(
+ f"benchmark fixture is outside the repository: {reference.fixture_path}"
+ ) from error
+ if fixture.is_symlink() or not fixture.is_file():
+ raise ValueError(f"benchmark fixture is not a regular file: {fixture}")
+ return fixture.read_bytes()
+ assert reference.revision is not None
+ completed = subprocess.run(
+ ["git", "show", f"{reference.revision}:{reference.path}"],
+ cwd=repository_root,
+ check=False,
+ capture_output=True,
+ )
+ if completed.returncode != 0:
+ detail = completed.stderr.decode("utf-8", errors="replace").strip()
+ raise ValueError(f"cannot load benchmark {reference.name!r}: {detail}")
+ return completed.stdout
+
+ return load
+
+
+async def evaluate_benchmarks(
+ manifest: BenchmarkManifest,
+ *,
+ config: SlopCopConfig,
+ registry: RuleRegistry,
+ source_loader: SourceLoader,
+) -> tuple[BenchmarkResult, ...]:
+ """Score every reference and compare it with the declared calibration range."""
+
+ results: list[BenchmarkResult] = []
+ for reference in manifest.benchmark:
+ document = build_document(
+ reference.path,
+ source_loader(reference),
+ contexts=config.contexts,
+ max_source_bytes=config.source_max_bytes,
+ )
+ observed = (await analyze_document(document, registry, config)).file_result
+ if observed.score is None:
+ raise ValueError(f"benchmark {reference.name!r} produced no score")
+ within_range = (
+ reference.min_score <= observed.score <= reference.max_score
+ and observed.decision.value == reference.expected_decision
+ )
+ results.append(
+ BenchmarkResult(
+ reference=reference,
+ score=observed.score,
+ decision=observed.decision,
+ within_range=within_range,
+ )
+ )
+ return tuple(results)
diff --git a/dev-tools/slop-cop/src/slop_cop/cli.py b/dev-tools/slop-cop/src/slop_cop/cli.py
new file mode 100644
index 00000000..1d88f3dd
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/cli.py
@@ -0,0 +1,524 @@
+"""Command-line interface for Slop Cop."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import re
+import subprocess
+import sys
+from collections import defaultdict, deque
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Any
+
+from pydantic import ValidationError
+
+from slop_cop import __version__
+from slop_cop.benchmarks import (
+ evaluate_benchmarks,
+ git_source_loader,
+ load_benchmark_manifest,
+)
+from slop_cop.config import SlopCopConfig, load_config
+from slop_cop.document import Document, build_document
+from slop_cop.engine import EngineOutput, analyze_document
+from slop_cop.findings import (
+ AnalysisState,
+ BaseComparison,
+ Decision,
+ ExternalAudit,
+ Finding,
+ FindingChange,
+ OverrideRecord,
+ RuleExecutionError,
+ RunResult,
+)
+from slop_cop.report import (
+ ReportError,
+ terminal_report,
+ write_json_report,
+ write_report_directory,
+)
+from slop_cop.rules.registry import RuleKind, RuleRegistry, build_registry
+
+EXIT_OK = 0
+EXIT_POLICY = 1
+EXIT_ERROR = 2
+_REVISION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,255}$")
+
+
+def _default_config() -> Path:
+ return Path(__file__).resolve().parents[2] / "slop-cop.toml"
+
+
+def _default_benchmarks() -> Path:
+ return Path(__file__).resolve().parents[2] / "benchmarks" / "dev-note-history.toml"
+
+
+def _parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="slop-cop",
+ description="Review Dev Notes for configured editorial signals.",
+ )
+ parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
+ subparsers = parser.add_subparsers(dest="command", required=True)
+
+ def add_config(command: argparse.ArgumentParser) -> None:
+ command.add_argument("--config", type=Path, default=_default_config())
+
+ check = subparsers.add_parser("check", help="Analyze one or more Dev Notes.")
+ add_config(check)
+ check.add_argument("paths", nargs="*", help="Markdown paths beneath the repository root.")
+ check.add_argument("--repository-root", type=Path)
+ baseline = check.add_mutually_exclusive_group()
+ baseline.add_argument("--baseline-ref")
+ baseline.add_argument("--baseline-root", type=Path)
+ check.add_argument("--html-dir", type=Path)
+ check.add_argument("--json", dest="json_path", type=Path)
+ check.add_argument("--only-rule")
+ check.add_argument("--repository")
+ check.add_argument("--pull-request-number", type=int)
+ check.add_argument("--base-sha")
+ check.add_argument("--head-sha")
+ check.add_argument("--override-json", type=Path)
+
+ list_rules = subparsers.add_parser("list-rules", help="List configured rules.")
+ add_config(list_rules)
+ list_rules.add_argument("--kind", choices=("builtin", "declarative", "custom"))
+
+ explain = subparsers.add_parser("explain", help="Explain one configured rule.")
+ add_config(explain)
+ explain.add_argument("rule_id")
+
+ validate = subparsers.add_parser(
+ "validate-rules", help="Validate configuration and rule registry."
+ )
+ add_config(validate)
+ validate.add_argument("--cases", type=Path)
+
+ benchmark = subparsers.add_parser(
+ "benchmark", help="Check scores against historical calibration references."
+ )
+ add_config(benchmark)
+ benchmark.add_argument("--manifest", type=Path, default=_default_benchmarks())
+ benchmark.add_argument("--repository-root", type=Path)
+ return parser
+
+
+def _repository_root(value: Path | None) -> Path:
+ if value is not None:
+ root = value.resolve(strict=True)
+ if not root.is_dir():
+ raise ValueError(f"repository root is not a directory: {root}")
+ return root
+ current = Path.cwd().resolve()
+ for candidate in (current, *current.parents):
+ if (candidate / ".git").exists():
+ return candidate
+ return current
+
+
+def _resolve_input(root: Path, value: str) -> tuple[str, Path]:
+ supplied = Path(value)
+ unresolved = supplied if supplied.is_absolute() else Path.cwd() / supplied
+ lexical = Path(os.path.abspath(unresolved))
+ try:
+ lexical_relative = lexical.relative_to(root)
+ except ValueError as error:
+ raise ValueError(f"input is outside the repository root: {value}") from error
+ current = root
+ for part in lexical_relative.parts:
+ current /= part
+ if current.is_symlink():
+ raise ValueError(f"input must not contain symlinks: {value}")
+ physical = lexical.resolve(strict=True)
+ try:
+ relative = physical.relative_to(root)
+ except ValueError as error:
+ raise ValueError(f"input is outside the repository root: {value}") from error
+ if physical.is_symlink() or not physical.is_file():
+ raise ValueError(f"input must be a regular non-symlink file: {value}")
+ if relative.suffix.casefold() != ".md":
+ raise ValueError(f"input must be a Markdown file: {value}")
+ return relative.as_posix(), physical
+
+
+def _load_head_document(config: SlopCopConfig, logical: str, physical: Path) -> Document:
+ return build_document(
+ logical,
+ physical.read_bytes(),
+ contexts=config.contexts,
+ max_source_bytes=config.source_max_bytes,
+ )
+
+
+def _load_base_document(
+ config: SlopCopConfig,
+ logical: str,
+ *,
+ root: Path,
+ baseline_root: Path | None,
+ baseline_ref: str | None,
+) -> Document | None:
+ content: bytes | None = None
+ if baseline_root is not None:
+ if baseline_root.is_symlink():
+ raise ValueError(f"baseline root must not be a symlink: {baseline_root}")
+ base_root = baseline_root.resolve(strict=True)
+ unresolved_candidate = base_root / logical
+ current = base_root
+ for part in Path(logical).parts:
+ current /= part
+ if current.is_symlink():
+ raise ValueError(f"baseline input must not contain symlinks: {logical}")
+ candidate = unresolved_candidate.resolve(strict=False)
+ try:
+ candidate.relative_to(base_root)
+ except ValueError as error:
+ raise ValueError(f"baseline path escapes its root: {logical}") from error
+ if candidate.exists():
+ if candidate.is_symlink() or not candidate.is_file():
+ raise ValueError(f"baseline input is not a regular non-symlink file: {logical}")
+ content = candidate.read_bytes()
+ elif baseline_ref is not None:
+ if (
+ not _REVISION.fullmatch(baseline_ref)
+ or ".." in baseline_ref
+ or baseline_ref.startswith("-")
+ ):
+ raise ValueError("baseline ref contains unsupported characters")
+ completed = subprocess.run(
+ ["git", "show", f"{baseline_ref}:{logical}"],
+ cwd=root,
+ check=False,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.DEVNULL,
+ )
+ if completed.returncode == 0:
+ content = completed.stdout
+ if content is None:
+ return None
+ return build_document(
+ logical,
+ content,
+ contexts=config.contexts,
+ max_source_bytes=config.source_max_bytes,
+ )
+
+
+def _finding_changes(base: tuple[Finding, ...], head: tuple[Finding, ...]) -> FindingChange:
+ def key(finding: Finding) -> tuple[str, str, str]:
+ return finding.rule_id, finding.normalized_key, finding.score_group
+
+ base_by_key: dict[tuple[str, str, str], deque[Finding]] = defaultdict(deque)
+ for finding in base:
+ base_by_key[key(finding)].append(finding)
+ added: list[Finding] = []
+ persistent: list[Finding] = []
+ for finding in head:
+ matches = base_by_key[key(finding)]
+ if matches:
+ matches.popleft()
+ persistent.append(finding)
+ else:
+ added.append(finding)
+ removed = [finding for matches in base_by_key.values() for finding in matches]
+ return FindingChange(added=tuple(added), removed=tuple(removed), persistent=tuple(persistent))
+
+
+async def _analyze(
+ document: Document,
+ registry: RuleRegistry,
+ config: SlopCopConfig,
+ base_document: Document | None,
+) -> tuple[EngineOutput, Document]:
+ output = await analyze_document(document, registry, config)
+ if base_document is None:
+ return output, document
+ try:
+ base = await analyze_document(base_document, registry, config)
+ comparison = BaseComparison(
+ score=base.file_result.score,
+ delta=(
+ output.file_result.score - base.file_result.score
+ if output.file_result.score is not None and base.file_result.score is not None
+ else None
+ ),
+ analysis_state=base.file_result.analysis_state,
+ findings=_finding_changes(base.file_result.findings, output.file_result.findings),
+ errors=base.file_result.errors,
+ )
+ except Exception as error:
+ comparison = BaseComparison(
+ analysis_state=AnalysisState.INCOMPLETE,
+ errors=(
+ RuleExecutionError(
+ source_path=document.path,
+ error_code="base_analysis_failed",
+ message=f"base comparison could not complete: {type(error).__name__}",
+ fatal=False,
+ ),
+ ),
+ )
+ return EngineOutput(
+ file_result=output.file_result.model_copy(update={"base": comparison}),
+ external_audits=output.external_audits,
+ ), document
+
+
+def _external_audit(value: Mapping[str, Any]) -> ExternalAudit:
+ return ExternalAudit(
+ rule_id=str(value["rule_id"]),
+ rule_version=int(value["rule_version"]),
+ service=str(value.get("service", "unknown")),
+ endpoint_hostname=str(value.get("endpoint_hostname") or value.get("hostname") or "unknown"),
+ content_digest=str(value.get("content_digest") or value.get("request_content_hash")),
+ request_schema_version=str(value.get("request_schema_version", "1")),
+ response_schema_version=(
+ str(value["response_schema_version"])
+ if value.get("response_schema_version") is not None
+ else None
+ ),
+ service_request_id=(
+ str(value.get("service_request_id") or value.get("request_id"))
+ if value.get("service_request_id") or value.get("request_id")
+ else None
+ ),
+ judge_revision=(
+ str(value["judge_revision"]) if value.get("judge_revision") is not None else None
+ ),
+ attempts=int(value.get("attempts", 1)),
+ latency_ms=round(float(value.get("latency_ms", 0))),
+ outcome=str(value.get("outcome", "unknown")),
+ response_digest=(
+ str(value["response_digest"]) if value.get("response_digest") is not None else None
+ ),
+ )
+
+
+def _load_override(path: Path | None, head_sha: str | None) -> OverrideRecord | None:
+ if path is None:
+ return None
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ override = OverrideRecord.model_validate(value)
+ except (OSError, UnicodeError, json.JSONDecodeError, ValidationError) as error:
+ raise ValueError(f"invalid override record: {error}") from error
+ if not head_sha or override.head_sha != head_sha:
+ raise ValueError("override review is not attached to the analyzed head revision")
+ return override
+
+
+async def _check(args: argparse.Namespace) -> int:
+ config = load_config(args.config)
+ if not args.paths:
+ result = RunResult(
+ analysis_state=AnalysisState.NOT_APPLICABLE,
+ decision=Decision.NOT_APPLICABLE,
+ score=None,
+ threshold=config.threshold,
+ repository=args.repository,
+ pull_request_number=args.pull_request_number,
+ base_sha=args.base_sha,
+ head_sha=args.head_sha,
+ tool_version=__version__,
+ config_digest=config.digest,
+ )
+ sys.stdout.write(terminal_report(result))
+ if args.html_dir:
+ write_report_directory(result, args.html_dir)
+ if args.json_path:
+ write_json_report(result, args.json_path)
+ return EXIT_OK
+ root = _repository_root(args.repository_root)
+ registry = build_registry(config)
+ if args.only_rule:
+ registry = RuleRegistry((registry.by_id(args.only_rule),))
+ resolved = [_resolve_input(root, value) for value in args.paths]
+ baseline_root = args.baseline_root.resolve(strict=True) if args.baseline_root else None
+ tasks = []
+ sources: dict[str, str] = {}
+ projections: dict[str, str] = {}
+ for logical, physical in resolved:
+ document = _load_head_document(config, logical, physical)
+ sources[logical] = document.source
+ projections[logical] = document.prose_projection
+ base = _load_base_document(
+ config,
+ logical,
+ root=root,
+ baseline_root=baseline_root,
+ baseline_ref=args.baseline_ref,
+ )
+ tasks.append(_analyze(document, registry, config, base))
+ analyzed = await asyncio.gather(*tasks)
+ file_results = tuple(output.file_result for output, _ in analyzed)
+ scores = [result.score for result in file_results if result.score is not None]
+ score = min(scores) if scores else None
+ if any(result.analysis_state is AnalysisState.ERROR for result in file_results):
+ analysis_state = AnalysisState.ERROR
+ elif any(result.analysis_state is AnalysisState.INCOMPLETE for result in file_results):
+ analysis_state = AnalysisState.INCOMPLETE
+ else:
+ analysis_state = AnalysisState.COMPLETE
+ decision = (
+ Decision.PASS
+ if all(result.decision is Decision.PASS for result in file_results)
+ else Decision.FAIL
+ )
+ override = _load_override(args.override_json, args.head_sha)
+ if (
+ override is not None
+ and analysis_state is AnalysisState.COMPLETE
+ and decision is Decision.FAIL
+ ):
+ decision = Decision.OVERRIDDEN
+ else:
+ override = None
+ audits = tuple(
+ _external_audit(audit) for output, _ in analyzed for audit in output.external_audits
+ )
+ result = RunResult(
+ analysis_state=analysis_state,
+ decision=decision,
+ score=score,
+ threshold=config.threshold,
+ repository=args.repository,
+ pull_request_number=args.pull_request_number,
+ base_sha=args.base_sha,
+ head_sha=args.head_sha,
+ tool_version=__version__,
+ config_digest=config.digest,
+ files=file_results,
+ external_audits=audits,
+ override=override,
+ )
+ sys.stdout.write(terminal_report(result))
+ if args.html_dir:
+ write_report_directory(
+ result,
+ args.html_dir,
+ sources=sources,
+ projections=projections,
+ )
+ if args.json_path:
+ write_json_report(result, args.json_path)
+ return EXIT_OK if result.decision in {Decision.PASS, Decision.OVERRIDDEN} else EXIT_POLICY
+
+
+def _list_rules(config: SlopCopConfig, kind: RuleKind | None) -> int:
+ registry = build_registry(config)
+ for configured in registry.list(kind):
+ metadata = configured.metadata
+ state = "enabled" if configured.policy.enabled else "disabled"
+ network = ""
+ if metadata.execution_kind == "external":
+ service = configured.policy.service or "unconfigured"
+ service_config = config.services.get(service)
+ host = service_config.url if service_config is not None else "unconfigured"
+ network = f"; sends selected prose to {host}"
+ print(f"{metadata.id}\t{configured.kind}\t{state}\t{metadata.title}{network}")
+ return EXIT_OK
+
+
+def _explain(config: SlopCopConfig, rule_id: str) -> int:
+ configured = build_registry(config).by_id(rule_id)
+ metadata = configured.metadata
+ print(f"{metadata.id} (version {metadata.version})")
+ print(f"Title: {metadata.title}")
+ print(f"Category: {metadata.category}")
+ print(f"Kind: {configured.kind}; execution: {metadata.execution_kind}")
+ print(f"Rationale: {metadata.rationale}")
+ print(f"Action: {metadata.advice}")
+ print("Policy:")
+ print(json.dumps(configured.policy.model_dump(mode="json"), indent=2, sort_keys=True))
+ if metadata.execution_kind == "external":
+ service = configured.policy.service
+ endpoint = config.services[service].url if service else "unconfigured"
+ print(f"Content transfer: sends selected prose to {endpoint}")
+ return EXIT_OK
+
+
+def _validate_cases(registry: RuleRegistry, path: Path) -> None:
+ import tomllib
+
+ if not path.exists():
+ raise ValueError(f"rule case file does not exist: {path}")
+ values = tomllib.loads(path.read_text(encoding="utf-8")).get("case", [])
+ coverage: dict[str, set[str]] = defaultdict(set)
+ for value in values:
+ rule_id = str(value.get("rule_id", ""))
+ registry.by_id(rule_id)
+ coverage[rule_id].add(str(value.get("kind", "")))
+ for configured in registry.enabled():
+ if configured.policy.cap > 0 and coverage[configured.metadata.id] < {
+ "positive",
+ "counterexample",
+ }:
+ raise ValueError(
+ f"scored rule {configured.metadata.id!r} lacks positive and counterexample cases"
+ )
+
+
+def _validate(config: SlopCopConfig, config_path: Path, cases: Path | None) -> int:
+ registry = build_registry(config)
+ effective_cases = cases
+ if effective_cases is None:
+ candidate = config_path.resolve().parent / "tests" / "rule_cases.toml"
+ if candidate.exists():
+ effective_cases = candidate
+ if effective_cases is not None:
+ _validate_cases(registry, effective_cases)
+ print(f"Valid: {len(registry.rules)} rules; configuration {config.digest}")
+ return EXIT_OK
+
+
+async def _benchmark(config: SlopCopConfig, args: argparse.Namespace) -> int:
+ root = _repository_root(args.repository_root)
+ manifest = load_benchmark_manifest(args.manifest)
+ results = await evaluate_benchmarks(
+ manifest,
+ config=config,
+ registry=build_registry(config),
+ source_loader=git_source_loader(root),
+ )
+ print("score expected decision reference")
+ for result in results:
+ reference = result.reference
+ status = "ok" if result.within_range else "DRIFT"
+ expected = f"{reference.min_score}-{reference.max_score}"
+ print(
+ f"{result.score:>5} {expected:>8} {result.decision.value:<8} "
+ f"{status:<5} {reference.name}"
+ )
+ return EXIT_OK if all(result.within_range for result in results) else EXIT_POLICY
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+ args = _parser().parse_args(argv)
+ try:
+ if args.command == "check":
+ return asyncio.run(_check(args))
+ config = load_config(args.config)
+ if args.command == "list-rules":
+ return _list_rules(config, args.kind)
+ if args.command == "explain":
+ return _explain(config, args.rule_id)
+ if args.command == "validate-rules":
+ return _validate(config, args.config, args.cases)
+ if args.command == "benchmark":
+ return asyncio.run(_benchmark(config, args))
+ except KeyError as error:
+ print(f"slop-cop: unknown rule {error.args[0]!r}", file=sys.stderr)
+ return EXIT_ERROR
+ except (OSError, UnicodeError, ValueError, ValidationError, ReportError) as error:
+ print(f"slop-cop: {error}", file=sys.stderr)
+ return EXIT_ERROR
+ return EXIT_ERROR
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/dev-tools/slop-cop/src/slop_cop/config.py b/dev-tools/slop-cop/src/slop_cop/config.py
new file mode 100644
index 00000000..0af99e6e
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/config.py
@@ -0,0 +1,377 @@
+"""Validated Slop Cop configuration."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import tomllib
+from enum import StrEnum
+from pathlib import Path, PurePosixPath
+from typing import Annotated, Any, Literal
+from urllib.parse import urlsplit
+
+import regex
+from pydantic import (
+ BaseModel,
+ ConfigDict,
+ Field,
+ StringConstraints,
+ TypeAdapter,
+ field_validator,
+ model_validator,
+)
+
+MAX_SOURCE_BYTES = 1_048_576
+MAX_SIGNALS_PER_RULE_FILE = 5_000
+MAX_REGEX_PATTERN_LENGTH = 500
+MAX_EXTERNAL_RESPONSE_BYTES = 1_048_576
+DEFAULT_EXTERNAL_RESPONSE_BYTES = 65_536
+MAX_EXTERNAL_CONCURRENCY = 4
+MAX_EXTERNAL_FILE_SECONDS = 60.0
+
+RuleId = Annotated[
+ str,
+ StringConstraints(
+ strip_whitespace=True,
+ min_length=3,
+ max_length=128,
+ pattern=r"^[a-z][a-z0-9]*(?:[.-][a-z0-9]+)+$",
+ ),
+]
+CategoryId = Annotated[
+ str,
+ StringConstraints(
+ strip_whitespace=True,
+ min_length=2,
+ max_length=64,
+ pattern=r"^[a-z][a-z0-9-]*$",
+ ),
+]
+ServiceName = Annotated[
+ str,
+ StringConstraints(
+ strip_whitespace=True,
+ min_length=1,
+ max_length=64,
+ pattern=r"^[a-z][a-z0-9_]*$",
+ ),
+]
+ShortText = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=1_000)]
+
+
+class StrictModel(BaseModel):
+ """Base for immutable configuration records with no ignored keys."""
+
+ model_config = ConfigDict(extra="forbid", frozen=True, strict=True)
+
+
+class Severity(StrEnum):
+ INFO = "info"
+ WARNING = "warning"
+ ERROR = "error"
+
+
+class ErrorPolicy(StrEnum):
+ FAIL = "fail"
+ ADVISORY = "advisory"
+
+
+class DensityUnit(StrEnum):
+ WORD = "word"
+ SENTENCE = "sentence"
+ PARAGRAPH = "paragraph"
+
+
+class ContextConfig(StrictModel):
+ scan_blockquotes: bool = False
+ scan_headings: bool = True
+ scan_captions: bool = True
+
+
+class DocumentDensityPolicy(StrictModel):
+ unit: DensityUnit
+ interval: int = Field(ge=1, le=1_000_000)
+ allowed_units: int = Field(ge=0, le=MAX_SIGNALS_PER_RULE_FILE)
+
+
+class PassageDensityPolicy(StrictModel):
+ unit: DensityUnit
+ window: int = Field(ge=1, le=100_000)
+ allowed_units: int = Field(ge=0, le=MAX_SIGNALS_PER_RULE_FILE)
+ first_cost: float = Field(ge=0, le=100)
+ repeat_cost: float = Field(ge=0, le=100)
+ cap: float = Field(ge=0, le=100)
+
+ @model_validator(mode="after")
+ def validate_costs(self) -> PassageDensityPolicy:
+ if self.cap == 0 and (self.first_cost or self.repeat_cost):
+ raise ValueError("density costs must be zero when density cap is zero")
+ if self.cap > 0 and self.first_cost > self.cap:
+ raise ValueError("density first_cost cannot exceed density cap")
+ return self
+
+
+class CategoryPolicy(StrictModel):
+ cap: float = Field(ge=0, le=100)
+ density: PassageDensityPolicy | None = None
+
+ @model_validator(mode="after")
+ def validate_density_cap(self) -> CategoryPolicy:
+ if self.density is not None and self.density.cap > self.cap:
+ raise ValueError("category density cap cannot exceed category cap")
+ return self
+
+
+class RulePolicy(StrictModel):
+ enabled: bool = True
+ severity: Severity
+ blocking: bool = False
+ on_error: ErrorPolicy = ErrorPolicy.FAIL
+ service: ServiceName | None = None
+ max_signal_units: int = Field(ge=1, le=MAX_SIGNALS_PER_RULE_FILE)
+ fixed_allowance: int = Field(ge=0, le=MAX_SIGNALS_PER_RULE_FILE)
+ first_cost: float = Field(ge=0, le=100)
+ repeat_cost: float = Field(ge=0, le=100)
+ cap: float = Field(ge=0, le=100)
+ document_density: DocumentDensityPolicy | None = None
+ density: PassageDensityPolicy | None = None
+ settings: dict[str, Any] = Field(default_factory=dict)
+
+ @model_validator(mode="after")
+ def validate_policy(self) -> RulePolicy:
+ if self.first_cost > self.cap:
+ raise ValueError("first_cost cannot exceed rule cap")
+ if self.severity is Severity.INFO and (self.blocking or self.cap != 0):
+ raise ValueError("info rules must be advisory with zero cost")
+ if self.severity is Severity.ERROR and not self.blocking:
+ raise ValueError("error rules must be blocking")
+ if self.blocking and self.severity is not Severity.ERROR:
+ raise ValueError("blocking rules must use error severity")
+ if self.cap == 0 and (self.first_cost or self.repeat_cost):
+ raise ValueError("rule costs must be zero when rule cap is zero")
+ if self.density is not None and self.density.cap > self.cap:
+ raise ValueError("density cap cannot exceed rule cap")
+ return self
+
+
+class ServiceConfig(StrictModel):
+ url: str = Field(min_length=1, max_length=2_048)
+ token_env: str = Field(pattern=r"^[A-Z][A-Z0-9_]{1,127}$")
+ timeout_seconds: float = Field(default=20.0, gt=0, le=60.0)
+ max_response_bytes: int = Field(
+ default=DEFAULT_EXTERNAL_RESPONSE_BYTES,
+ ge=1,
+ le=MAX_EXTERNAL_RESPONSE_BYTES,
+ )
+ max_attempts: int = Field(default=1, ge=1, le=3)
+
+ @field_validator("url")
+ @classmethod
+ def validate_url(cls, value: str) -> str:
+ parsed = urlsplit(value)
+ if parsed.username or parsed.password or parsed.fragment or not parsed.hostname:
+ raise ValueError(
+ "service URL must be an absolute origin without credentials or fragment"
+ )
+ loopback = parsed.hostname in {"127.0.0.1", "::1", "localhost"}
+ if parsed.scheme != "https" and not (parsed.scheme == "http" and loopback):
+ raise ValueError("service URL must use HTTPS; HTTP is allowed only for loopback tests")
+ return value
+
+
+class VocabularyConfig(StrictModel):
+ allowed_terms: tuple[str, ...] = ()
+
+ @field_validator("allowed_terms")
+ @classmethod
+ def validate_terms(cls, values: tuple[str, ...]) -> tuple[str, ...]:
+ cleaned = tuple(value.strip() for value in values)
+ if any(not value or len(value) > 128 for value in cleaned):
+ raise ValueError("allowed terms must contain 1 to 128 characters")
+ if len({value.casefold() for value in cleaned}) != len(cleaned):
+ raise ValueError("allowed terms must be unique without regard to case")
+ return cleaned
+
+
+class DeclarativeRuleBase(StrictModel):
+ id: RuleId
+ version: int = Field(ge=1)
+ category: CategoryId
+ severity: Severity
+ title: ShortText
+ rationale: ShortText
+ advice: ShortText
+ enabled: bool = True
+ blocking: bool = False
+ on_error: ErrorPolicy = ErrorPolicy.FAIL
+ max_signal_units: int = Field(ge=1, le=MAX_SIGNALS_PER_RULE_FILE)
+ fixed_allowance: int = Field(ge=0, le=MAX_SIGNALS_PER_RULE_FILE)
+ first_cost: float = Field(ge=0, le=100)
+ repeat_cost: float = Field(ge=0, le=100)
+ cap: float = Field(ge=0, le=100)
+ document_density: DocumentDensityPolicy | None = None
+ density: PassageDensityPolicy | None = None
+ score_group: str | None = Field(default=None, max_length=128)
+
+ @model_validator(mode="after")
+ def validate_scoring(self) -> DeclarativeRuleBase:
+ RulePolicy.model_validate(
+ {
+ "enabled": self.enabled,
+ "severity": self.severity,
+ "blocking": self.blocking,
+ "on_error": self.on_error,
+ "max_signal_units": self.max_signal_units,
+ "fixed_allowance": self.fixed_allowance,
+ "first_cost": self.first_cost,
+ "repeat_cost": self.repeat_cost,
+ "cap": self.cap,
+ "document_density": self.document_density,
+ "density": self.density,
+ }
+ )
+ return self
+
+
+class PhraseRuleConfig(DeclarativeRuleBase):
+ phrases: tuple[str, ...] = Field(min_length=1, max_length=100)
+
+ @field_validator("phrases")
+ @classmethod
+ def validate_phrases(cls, values: tuple[str, ...]) -> tuple[str, ...]:
+ cleaned = tuple(value.strip() for value in values)
+ if any(not value or len(value) > 200 for value in cleaned):
+ raise ValueError("phrases must contain 1 to 200 characters")
+ if len({value.casefold() for value in cleaned}) != len(cleaned):
+ raise ValueError("phrases must be unique without regard to case")
+ return cleaned
+
+
+class RegexFlag(StrEnum):
+ IGNORECASE = "IGNORECASE"
+ MULTILINE = "MULTILINE"
+ DOTALL = "DOTALL"
+ VERBOSE = "VERBOSE"
+
+
+_UNSAFE_REGEX = re.compile(
+ r"(?:\\[1-9]|\\g[<{]|\(\?R|\(\?0|\(\?&|\(\?P>|\(\?\(|\(\?<=[^)]|\(\? str:
+ if _UNSAFE_REGEX.search(value):
+ raise ValueError("regex contains an unsupported construct")
+ try:
+ regex.compile(value)
+ except regex.error as error:
+ raise ValueError(f"invalid regex: {error}") from error
+ return value
+
+ @field_validator("flags")
+ @classmethod
+ def validate_flags(cls, values: tuple[RegexFlag, ...]) -> tuple[RegexFlag, ...]:
+ if len(set(values)) != len(values):
+ raise ValueError("regex flags must be unique")
+ return values
+
+
+class CustomRulesConfig(StrictModel):
+ phrase: tuple[PhraseRuleConfig, ...] = ()
+ regex: tuple[RegexRuleConfig, ...] = ()
+
+
+class SlopCopConfig(StrictModel):
+ schema_version: Literal[1]
+ profile: Literal["dev-notes"]
+ threshold: int = Field(ge=0, le=100)
+ paths: tuple[str, ...] = Field(min_length=1, max_length=32)
+ contexts: ContextConfig = ContextConfig()
+ categories: dict[CategoryId, CategoryPolicy]
+ rules: dict[RuleId, RulePolicy]
+ vocabulary: VocabularyConfig = VocabularyConfig()
+ custom_rules: CustomRulesConfig = CustomRulesConfig()
+ services: dict[ServiceName, ServiceConfig] = Field(default_factory=dict)
+ source_max_bytes: int = Field(default=MAX_SOURCE_BYTES, ge=1, le=MAX_SOURCE_BYTES)
+ external_concurrency: int = Field(
+ default=MAX_EXTERNAL_CONCURRENCY, ge=1, le=MAX_EXTERNAL_CONCURRENCY
+ )
+ external_file_timeout_seconds: float = Field(
+ default=MAX_EXTERNAL_FILE_SECONDS, gt=0, le=MAX_EXTERNAL_FILE_SECONDS
+ )
+
+ @field_validator("paths")
+ @classmethod
+ def validate_paths(cls, paths: tuple[str, ...]) -> tuple[str, ...]:
+ for path in paths:
+ pure = PurePosixPath(path)
+ if not path or pure.is_absolute() or ".." in pure.parts or "\\" in path:
+ raise ValueError("scan paths must be nonempty relative POSIX globs without '..'")
+ return paths
+
+ @model_validator(mode="after")
+ def validate_references(self) -> SlopCopConfig:
+ declarative = (*self.custom_rules.phrase, *self.custom_rules.regex)
+ all_ids = [*self.rules, *(rule.id for rule in declarative)]
+ if len(set(all_ids)) != len(all_ids):
+ raise ValueError("rule IDs must be unique across Python and declarative rules")
+ for rule in declarative:
+ if rule.category not in self.categories:
+ raise ValueError(f"rule {rule.id!r} references unknown category {rule.category!r}")
+ for rule_id, policy in self.rules.items():
+ if policy.service is not None and policy.service not in self.services:
+ raise ValueError(f"rule {rule_id!r} references unknown service {policy.service!r}")
+ return self
+
+ def canonical_json(self) -> str:
+ return json.dumps(
+ self.model_dump(mode="json"),
+ ensure_ascii=False,
+ separators=(",", ":"),
+ sort_keys=True,
+ )
+
+ @property
+ def digest(self) -> str:
+ return hashlib.sha256(self.canonical_json().encode()).hexdigest()
+
+
+_CONFIG_ADAPTER = TypeAdapter(SlopCopConfig)
+
+
+def load_config(path: str | Path) -> SlopCopConfig:
+ """Load and strictly validate a UTF-8 TOML configuration file."""
+
+ config_path = Path(path)
+ try:
+ data = tomllib.loads(config_path.read_text(encoding="utf-8"))
+ except (OSError, UnicodeError, tomllib.TOMLDecodeError) as error:
+ raise ValueError(f"cannot load configuration {config_path}: {error}") from error
+ # TOML has no tuple or enum scalar types. Allow those representation
+ # conversions while retaining the model's bounded values and forbidden keys.
+ return _CONFIG_ADAPTER.validate_python(data, strict=False)
+
+
+__all__ = [
+ "CategoryPolicy",
+ "ContextConfig",
+ "DensityUnit",
+ "DocumentDensityPolicy",
+ "ErrorPolicy",
+ "PassageDensityPolicy",
+ "PhraseRuleConfig",
+ "RegexFlag",
+ "RegexRuleConfig",
+ "RulePolicy",
+ "ServiceConfig",
+ "Severity",
+ "SlopCopConfig",
+ "load_config",
+]
diff --git a/dev-tools/slop-cop/src/slop_cop/document.py b/dev-tools/slop-cop/src/slop_cop/document.py
new file mode 100644
index 00000000..8c4b7b03
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/document.py
@@ -0,0 +1,785 @@
+"""Source-preserving Markdown projection and prose segmentation."""
+
+from __future__ import annotations
+
+import bisect
+import re
+import unicodedata
+from itertools import pairwise
+from pathlib import Path
+
+import regex
+from pydantic import Field, model_validator
+
+from slop_cop.config import MAX_SOURCE_BYTES, ContextConfig, RuleId, StrictModel
+
+_FENCE_OPEN = re.compile(
+ r"^(?P(?: {0,3}>[ \t]?)*)(?P {0,3})"
+ r"(?P`{3,}|~{3,})[^\r\n]*(?:\r?\n|$)",
+ re.MULTILINE,
+)
+_BLOCK_INTERRUPT = re.compile(
+ r"^(?: {4}|\t| {0,3}(?:#{1,6}(?:\s|$)|[-+*]\s|\d+[.)]\s|`{3,}|~{3,}|"
+ r"(?:\*\s*){3,}$|(?:-\s*){3,}$|(?:_\s*){3,}$|<[/!?A-Za-z]))"
+)
+_SUPPRESSION = re.compile(
+ r"^[ \t]*[ \t]*(?:\r?\n|$)',
+ re.MULTILINE,
+)
+_GENERATED_START = ""
+_GENERATED_END = ""
+_WORD = regex.compile(r"[\p{L}\p{N}](?:[\p{L}\p{N}\p{M}]|['’](?=[\p{L}\p{N}])|-(?=[\p{L}\p{N}]))*")
+_ABBREVIATIONS = frozenset({"e.g.", "i.e.", "mr.", "mrs.", "ms.", "dr.", "vs.", "etc."})
+
+
+class ProjectionError(ValueError):
+ """Raised when Markdown cannot be projected without ambiguous source ranges."""
+
+ def __init__(self, message: str, source: str, offset: int = 0) -> None:
+ line_starts = _line_starts(source)
+ line, column = line_column(line_starts, offset)
+ super().__init__(f"{message} at line {line}, column {column}")
+ self.offset = offset
+ self.line = line
+ self.column = column
+
+
+class Span(StrictModel):
+ start: int = Field(ge=0)
+ end: int = Field(ge=0)
+
+ @model_validator(mode="after")
+ def validate_order(self) -> Span:
+ if self.end < self.start:
+ raise ValueError("span end must not precede start")
+ return self
+
+
+class MaskedRange(Span):
+ reasons: tuple[str, ...] = Field(min_length=1)
+
+
+class ProseSegment(Span):
+ text: str
+ normalized: str
+
+
+class SuppressionDirective(StrictModel):
+ rule_ids: tuple[RuleId, ...] = Field(min_length=1)
+ reason: str = Field(min_length=1, max_length=500)
+ directive_span: Span
+ target_span: Span
+
+
+class DocumentMetrics(StrictModel):
+ source_bytes: int = Field(ge=0, le=MAX_SOURCE_BYTES)
+ source_code_points: int = Field(ge=0)
+ analyzable_words: int = Field(ge=0)
+ analyzable_sentences: int = Field(ge=0)
+ analyzable_paragraphs: int = Field(ge=0)
+ masked_code_points: int = Field(ge=0)
+
+
+class Document(StrictModel):
+ path: str = Field(min_length=1, max_length=4_096)
+ source: str
+ prose_projection: str
+ repetition_projection: str
+ line_starts: tuple[int, ...]
+ front_matter: tuple[tuple[str, str], ...] = ()
+ masked_ranges: tuple[MaskedRange, ...]
+ suppressions: tuple[SuppressionDirective, ...]
+ tokens: tuple[ProseSegment, ...]
+ sentences: tuple[ProseSegment, ...]
+ paragraphs: tuple[ProseSegment, ...]
+ repetition_tokens: tuple[ProseSegment, ...]
+ repetition_sentences: tuple[ProseSegment, ...]
+ repetition_paragraphs: tuple[ProseSegment, ...]
+ metrics: DocumentMetrics
+
+ @model_validator(mode="after")
+ def validate_projection(self) -> Document:
+ if len(self.source) != len(self.prose_projection):
+ raise ValueError("prose projection must have the same length as source")
+ if len(self.source) != len(self.repetition_projection):
+ raise ValueError("repetition projection must have the same length as source")
+ for index, character in enumerate(self.source):
+ if character in "\r\n" and self.prose_projection[index] != character:
+ raise ValueError("prose projection must preserve line endings")
+ if character in "\r\n" and self.repetition_projection[index] != character:
+ raise ValueError("repetition projection must preserve line endings")
+ if self.line_starts != _line_starts(self.source):
+ raise ValueError("line-start index does not match source")
+ return self
+
+ def line_column(self, offset: int) -> tuple[int, int]:
+ if not 0 <= offset <= len(self.source):
+ raise ValueError("offset lies outside source")
+ return line_column(self.line_starts, offset)
+
+ def source_span(self, span: Span) -> str:
+ if span.end > len(self.source):
+ raise ValueError("span lies outside source")
+ return self.source[span.start : span.end]
+
+
+def normalize_key(value: str) -> str:
+ return unicodedata.normalize("NFKC", value).casefold()
+
+
+def line_column(line_starts: tuple[int, ...], offset: int) -> tuple[int, int]:
+ line_index = bisect.bisect_right(line_starts, offset) - 1
+ return line_index + 1, offset - line_starts[line_index] + 1
+
+
+def _line_starts(source: str) -> tuple[int, ...]:
+ return (0, *(match.end() for match in re.finditer("\n", source)))
+
+
+def _line_end(source: str, offset: int) -> int:
+ newline = source.find("\n", offset)
+ return len(source) if newline < 0 else newline + 1
+
+
+def _covered(offset: int, ranges: list[tuple[int, int, str]]) -> bool:
+ return any(start <= offset < end for start, end, _ in ranges)
+
+
+def _strip_blockquote_markers(text: str) -> tuple[str, int]:
+ depth = 0
+ while (marker := re.match(r"^ {0,3}>[ \t]?", text)) is not None:
+ text = text[marker.end() :]
+ depth += 1
+ return text, depth
+
+
+def _add_front_matter(
+ source: str, ranges: list[tuple[int, int, str]]
+) -> tuple[tuple[str, str], ...]:
+ first_end = _line_end(source, 0)
+ if source[:first_end].rstrip("\r\n") != "---":
+ return ()
+ cursor = first_end
+ end = None
+ while cursor < len(source):
+ next_end = _line_end(source, cursor)
+ if source[cursor:next_end].rstrip("\r\n") in {"---", "..."}:
+ end = next_end
+ break
+ cursor = next_end
+ if end is None:
+ raise ProjectionError("unterminated YAML front matter", source)
+ ranges.append((0, end, "front-matter"))
+ metadata: list[tuple[str, str]] = []
+ for line in source[first_end:cursor].splitlines():
+ match = re.match(r"^([A-Za-z][A-Za-z0-9_-]*):\s*(.*?)\s*$", line)
+ if match and match.group(1) in {"title", "description", "date", "author"}:
+ metadata.append((match.group(1), match.group(2).strip("'\"")))
+ return tuple(metadata)
+
+
+def _add_generated_ranges(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ cursor = 0
+ while True:
+ start = source.find(_GENERATED_START, cursor)
+ end_only = source.find(_GENERATED_END, cursor)
+ if start < 0:
+ if end_only >= 0:
+ raise ProjectionError("generated byline end marker has no start", source, end_only)
+ return
+ if end_only >= 0 and end_only < start:
+ raise ProjectionError("generated byline end marker has no start", source, end_only)
+ end_marker = source.find(_GENERATED_END, start + len(_GENERATED_START))
+ if end_marker < 0:
+ raise ProjectionError("unterminated generated byline range", source, start)
+ nested = source.find(_GENERATED_START, start + len(_GENERATED_START), end_marker)
+ if nested >= 0:
+ raise ProjectionError("nested generated byline start marker", source, nested)
+ end = end_marker + len(_GENERATED_END)
+ ranges.append((start, end, "generated-byline"))
+ cursor = end
+
+
+def _add_fenced_code(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ cursor = 0
+ while cursor < len(source):
+ match = _FENCE_OPEN.search(source, cursor)
+ if match is None:
+ return
+ cursor = match.end()
+ if _covered(match.start(), ranges):
+ continue
+ fence = match.group("fence")
+ quote_depth = match.group("quote").count(">")
+ closer = re.compile(
+ rf"^(?: {{0,3}}>[ \t]?){{{quote_depth}}} {{0,3}}"
+ rf"{re.escape(fence[0])}{{{len(fence)},}}[ \t]*(?:\r?\n|$)",
+ re.MULTILINE,
+ ).search(source, match.end())
+ if closer is None:
+ raise ProjectionError("unterminated fenced code block", source, match.start())
+ ranges.append((match.start(), closer.end(), "fenced-code"))
+ cursor = closer.end()
+
+
+def _add_indented_code(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ lines = list(re.finditer(r"^.*(?:\r?\n|$)", source, re.MULTILINE))
+ in_block = False
+ block_start = 0
+ block_end = 0
+ previous_blank = True
+ for line in lines:
+ text = line.group(0).rstrip("\r\n")
+ content, _ = _strip_blockquote_markers(text)
+ blank = not content.strip()
+ indented = content.startswith(" ") or content.startswith("\t")
+ if in_block:
+ if indented or blank:
+ block_end = line.end()
+ else:
+ ranges.append((block_start, block_end, "indented-code"))
+ in_block = False
+ if not in_block and indented and previous_blank and not _covered(line.start(), ranges):
+ in_block = True
+ block_start = line.start()
+ block_end = line.end()
+ previous_blank = blank
+ if in_block:
+ ranges.append((block_start, block_end, "indented-code"))
+
+
+def _add_suppressions(
+ source: str, ranges: list[tuple[int, int, str]]
+) -> list[tuple[tuple[RuleId, ...], str, Span]]:
+ parsed: list[tuple[tuple[RuleId, ...], str, Span]] = []
+ matches = list(_SUPPRESSION.finditer(source))
+ valid_starts = {match.start() for match in matches}
+ for occurrence in re.finditer(r"", start + 4)
+ if end < 0:
+ if not _covered(start, ranges):
+ raise ProjectionError("unterminated HTML comment", source, start)
+ return
+ end += 3
+ if not _covered(start, ranges):
+ ranges.append((start, end, "html-comment"))
+ cursor = end
+
+
+def _add_html_nonprose_blocks(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ opening = r"<(script|style|template|pre|code|svg|math)\b(?:[^>'\"]|'[^']*'|\"[^\"]*\")*>"
+ for match in re.finditer(opening, source, re.IGNORECASE):
+ if _covered(match.start(), ranges):
+ continue
+ tag = match.group(1)
+ close = re.search(rf"{re.escape(tag)}[ \t]*>", source[match.end() :], re.IGNORECASE)
+ if close is not None:
+ ranges.append((match.start(), match.end() + close.end(), "html-nonprose"))
+
+
+def _add_inline_code(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ cursor = 0
+ while cursor < len(source):
+ start = source.find("`", cursor)
+ if start < 0:
+ return
+ if _covered(start, ranges):
+ cursor = start + 1
+ continue
+ run_end = start + 1
+ while run_end < len(source) and source[run_end] == "`":
+ run_end += 1
+ delimiter = source[start:run_end]
+ end = source.find(delimiter, run_end)
+ if end < 0:
+ raise ProjectionError("unterminated inline code span", source, start)
+ ranges.append((start, end + len(delimiter), "inline-code"))
+ cursor = end + len(delimiter)
+
+
+def _add_html_tags(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ cursor = 0
+ while cursor < len(source):
+ start = source.find("<", cursor)
+ if start < 0:
+ return
+ if _covered(start, ranges):
+ cursor = start + 1
+ continue
+ if (
+ start + 1 >= len(source)
+ or source[start + 1] not in "/!?ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+ ):
+ cursor = start + 1
+ continue
+ quote: str | None = None
+ end = start + 1
+ while end < len(source):
+ char = source[end]
+ if quote is not None:
+ if char == quote:
+ quote = None
+ elif char in "'\"":
+ quote = char
+ elif char == ">":
+ ranges.append((start, end + 1, "html-tag"))
+ cursor = end + 1
+ break
+ elif char == "<":
+ cursor = start + 1
+ break
+ end += 1
+ else:
+ cursor = start + 1
+
+
+def _find_closing_paren(source: str, start: int) -> int | None:
+ depth = 1
+ escaped = False
+ for index in range(start, len(source)):
+ char = source[index]
+ if escaped:
+ escaped = False
+ elif char == "\\":
+ escaped = True
+ elif char == "(":
+ depth += 1
+ elif char == ")":
+ depth -= 1
+ if depth == 0:
+ return index
+ elif char in "\r\n" and depth == 1:
+ return None
+ return None
+
+
+def _find_closing_bracket(source: str, start: int) -> int | None:
+ depth = 1
+ escaped = False
+ for index in range(start, len(source)):
+ char = source[index]
+ if escaped:
+ escaped = False
+ elif char == "\\":
+ escaped = True
+ elif char == "[":
+ depth += 1
+ elif char == "]":
+ depth -= 1
+ if depth == 0:
+ return index
+ return None
+
+
+def _add_links_and_images(
+ source: str,
+ ranges: list[tuple[int, int, str]],
+ repetition_ranges: list[tuple[int, int, str]],
+) -> None:
+ reference = re.compile(r"^ {0,3}\[[^\]\r\n]+\]:[^\r\n]*(?:\r?\n|$)", re.MULTILINE)
+ for match in reference.finditer(source):
+ if not _covered(match.start(), ranges):
+ ranges.append((match.start(), match.end(), "link-destination"))
+
+ opener = re.compile(r"!?\[")
+ cursor = 0
+ while (opener_match := opener.search(source, cursor)) is not None:
+ start = opener_match.start()
+ cursor = opener_match.end()
+ if _covered(start, ranges):
+ continue
+ label_end = _find_closing_bracket(source, opener_match.end())
+ if label_end is None:
+ continue
+ is_image = source[start] == "!"
+ if is_image:
+ ranges.append((start, label_end + 1, "image"))
+ else:
+ repetition_ranges.append((opener_match.end(), label_end, "repetition-link-label"))
+ if label_end + 1 < len(source) and source[label_end + 1] == "(":
+ close = _find_closing_paren(source, label_end + 2)
+ if close is None:
+ raise ProjectionError(
+ "unterminated Markdown link destination", source, label_end + 1
+ )
+ if is_image:
+ ranges.append((start, close + 1, "image"))
+ else:
+ ranges.extend(
+ (
+ (start, opener_match.end(), "link-markup"),
+ (label_end, close + 1, "link-destination"),
+ )
+ )
+ cursor = close + 1
+ elif label_end + 1 < len(source) and source[label_end + 1] == "[":
+ ref_end = source.find("]", label_end + 2)
+ if ref_end >= 0:
+ if is_image:
+ ranges.append((start, ref_end + 1, "image"))
+ else:
+ ranges.extend(
+ (
+ (start, opener_match.end(), "link-markup"),
+ (label_end, ref_end + 1, "link-destination"),
+ )
+ )
+ cursor = ref_end + 1
+
+
+def _add_repetition_contexts(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ for match in re.finditer(r"^ {0,3}#{1,6}(?:\s+|$)[^\r\n]*(?:\r?\n|$)", source, re.MULTILINE):
+ ranges.append((match.start(), match.end(), "repetition-heading"))
+ for match in re.finditer(
+ r"^.*(?:\r?\n)(?: {0,3})(?:=+|-+)[ \t]*(?:\r?\n|$)", source, re.MULTILINE
+ ):
+ ranges.append((match.start(), match.end(), "repetition-heading"))
+ for match in re.finditer(
+ r"'\"]|'[^']*'|\"[^\"]*\")*>(.*?)",
+ source,
+ re.IGNORECASE | re.DOTALL,
+ ):
+ ranges.append((match.start(1), match.end(1), "repetition-caption"))
+ for match in re.finditer(
+ r"'\"]|'[^']*'|\"[^\"]*\")*>(.*?)",
+ source,
+ re.IGNORECASE | re.DOTALL,
+ ):
+ ranges.append((match.start(1), match.end(1), "repetition-link-label"))
+
+
+def _add_blockquotes(source: str, ranges: list[tuple[int, int, str]]) -> None:
+ active = False
+ lazy_continuation = False
+ start = end = 0
+ for line in re.finditer(r"^.*(?:\r?\n|$)", source, re.MULTILINE):
+ text = line.group(0).rstrip("\r\n")
+ marker = re.match(r"^ {0,3}>[ \t]?", text)
+ blank = not text.strip()
+ if marker is not None and not _covered(line.start(), ranges):
+ if not active:
+ start = line.start()
+ active = True
+ end = line.end()
+ content, _ = _strip_blockquote_markers(text)
+ lazy_continuation = bool(content.strip()) and not _BLOCK_INTERRUPT.match(content)
+ elif active and lazy_continuation and not blank and not _BLOCK_INTERRUPT.match(text):
+ end = line.end()
+ elif active:
+ ranges.append((start, end, "blockquote"))
+ active = False
+ lazy_continuation = False
+ if active:
+ ranges.append((start, end, "blockquote"))
+
+
+def _add_disabled_visible_contexts(
+ source: str,
+ ranges: list[tuple[int, int, str]],
+ contexts: ContextConfig,
+) -> None:
+ if not contexts.scan_headings:
+ for match in re.finditer(r"^ {0,3}#{1,6}\s+[^\r\n]*(?:\r?\n|$)", source, re.MULTILINE):
+ if not _covered(match.start(), ranges):
+ ranges.append((match.start(), match.end(), "heading"))
+ if not contexts.scan_captions:
+ for match in re.finditer(
+ r"]*>.*?",
+ source,
+ re.IGNORECASE | re.DOTALL,
+ ):
+ if not _covered(match.start(), ranges):
+ ranges.append((match.start(), match.end(), "caption"))
+
+
+def _merge_ranges(source: str, ranges: list[tuple[int, int, str]]) -> tuple[MaskedRange, ...]:
+ boundaries: list[tuple[int, int, str]] = []
+ for start, end, reason in ranges:
+ if not 0 <= start <= end <= len(source):
+ raise ProjectionError("invalid masked range", source, max(0, start))
+ if start != end:
+ boundaries.append((start, end, reason))
+ if not boundaries:
+ return ()
+ points = sorted({point for start, end, _ in boundaries for point in (start, end)})
+ pieces: list[MaskedRange] = []
+ for start, end in pairwise(points):
+ reasons = tuple(
+ sorted({reason for left, right, reason in boundaries if left < end and right > start})
+ )
+ if not reasons:
+ continue
+ if pieces and pieces[-1].end == start and pieces[-1].reasons == reasons:
+ previous = pieces.pop()
+ pieces.append(MaskedRange(start=previous.start, end=end, reasons=reasons))
+ else:
+ pieces.append(MaskedRange(start=start, end=end, reasons=reasons))
+ return tuple(pieces)
+
+
+def _project(source: str, masked: tuple[MaskedRange, ...]) -> str:
+ characters = list(source)
+ for item in masked:
+ for index in range(item.start, item.end):
+ if characters[index] not in "\r\n":
+ characters[index] = " "
+ return "".join(characters)
+
+
+def _segment_tokens(projection: str) -> tuple[ProseSegment, ...]:
+ return tuple(
+ ProseSegment(
+ start=match.start(),
+ end=match.end(),
+ text=match.group(),
+ normalized=normalize_key(match.group()),
+ )
+ for match in _WORD.finditer(projection)
+ )
+
+
+def _paragraph_spans(projection: str) -> list[tuple[int, int]]:
+ spans: list[tuple[int, int]] = []
+ current_start: int | None = None
+ cursor = 0
+ for line in projection.splitlines(keepends=True):
+ content = line.rstrip("\r\n")
+ stripped = content.strip()
+ is_boundary = bool(re.match(r" {0,3}(?:#{1,6}\s|[-+*]\s|\d+[.)]\s)", content))
+ if not stripped:
+ if current_start is not None:
+ spans.append((current_start, cursor))
+ current_start = None
+ elif is_boundary:
+ if current_start is not None:
+ spans.append((current_start, cursor))
+ spans.append((cursor, cursor + len(content)))
+ current_start = None
+ elif current_start is None:
+ current_start = cursor
+ cursor += len(line)
+ if current_start is not None:
+ spans.append((current_start, len(projection)))
+ return [(start, end) for start, end in spans if projection[start:end].strip()]
+
+
+def _segment_paragraphs(projection: str) -> tuple[ProseSegment, ...]:
+ return tuple(
+ ProseSegment(
+ start=start,
+ end=end,
+ text=projection[start:end],
+ normalized=normalize_key(projection[start:end].strip()),
+ )
+ for start, end in _paragraph_spans(projection)
+ )
+
+
+def _segment_sentences(
+ projection: str, paragraphs: tuple[ProseSegment, ...]
+) -> tuple[ProseSegment, ...]:
+ sentences: list[ProseSegment] = []
+ for paragraph in paragraphs:
+ start = paragraph.start
+ cursor = start
+ while cursor < paragraph.end:
+ while cursor < paragraph.end and projection[cursor].isspace():
+ cursor += 1
+ if cursor >= paragraph.end:
+ break
+ end = paragraph.end
+ for match in re.finditer(r"[.!?]+(?:[\"'”’)]*)", projection[cursor : paragraph.end]):
+ candidate_end = cursor + match.end()
+ token_start = projection.rfind(" ", cursor, candidate_end) + 1
+ candidate = projection[token_start:candidate_end].casefold()
+ if candidate in _ABBREVIATIONS:
+ continue
+ if candidate_end == paragraph.end or projection[candidate_end].isspace():
+ end = candidate_end
+ break
+ text = projection[cursor:end]
+ if text.strip():
+ sentences.append(
+ ProseSegment(
+ start=cursor, end=end, text=text, normalized=normalize_key(text.strip())
+ )
+ )
+ cursor = end
+ return tuple(sentences)
+
+
+def _bind_suppressions(
+ source: str,
+ projection: str,
+ parsed: list[tuple[tuple[RuleId, ...], str, Span]],
+ paragraphs: tuple[ProseSegment, ...],
+) -> tuple[SuppressionDirective, ...]:
+ directives: list[SuppressionDirective] = []
+ for ids, reason, span in parsed:
+ target = next((paragraph for paragraph in paragraphs if paragraph.start >= span.end), None)
+ if target is None:
+ raise ProjectionError("suppression has no following prose block", source, span.start)
+ if projection[span.end : target.start].strip():
+ raise ProjectionError(
+ "suppression is not immediately before a prose block", source, span.start
+ )
+ directives.append(
+ SuppressionDirective(
+ rule_ids=ids,
+ reason=reason,
+ directive_span=span,
+ target_span=Span(start=target.start, end=target.end),
+ )
+ )
+ return tuple(directives)
+
+
+def build_document(
+ path: str | Path,
+ content: bytes | str,
+ *,
+ contexts: ContextConfig | None = None,
+ max_source_bytes: int = MAX_SOURCE_BYTES,
+) -> Document:
+ """Build an immutable source-mapped prose view from one Markdown document."""
+
+ if not 1 <= max_source_bytes <= MAX_SOURCE_BYTES:
+ raise ValueError(f"max_source_bytes must be between 1 and {MAX_SOURCE_BYTES}")
+ if isinstance(content, bytes):
+ raw = content
+ if len(raw) > max_source_bytes:
+ raise ProjectionError(f"source exceeds {max_source_bytes} bytes", "")
+ try:
+ source = raw.decode("utf-8", errors="strict")
+ except UnicodeDecodeError as error:
+ raise ProjectionError("source is not valid UTF-8", "", error.start) from error
+ else:
+ source = content
+ raw = source.encode("utf-8")
+ if len(raw) > max_source_bytes:
+ raise ProjectionError(f"source exceeds {max_source_bytes} bytes", source)
+ nul = source.find("\0")
+ if nul >= 0:
+ raise ProjectionError("source contains a NUL character", source, nul)
+
+ effective_contexts = contexts or ContextConfig()
+ ranges: list[tuple[int, int, str]] = []
+ repetition_ranges: list[tuple[int, int, str]] = []
+ front_matter = _add_front_matter(source, ranges)
+ _add_generated_ranges(source, ranges)
+ if not effective_contexts.scan_blockquotes:
+ _add_blockquotes(source, ranges)
+ _add_fenced_code(source, ranges)
+ _add_indented_code(source, ranges)
+ parsed_suppressions = _add_suppressions(source, ranges)
+ _add_html_comments(source, ranges)
+ _add_html_nonprose_blocks(source, ranges)
+ _add_inline_code(source, ranges)
+ _add_html_tags(source, ranges)
+ _add_links_and_images(source, ranges, repetition_ranges)
+ _add_disabled_visible_contexts(source, ranges, effective_contexts)
+ _add_repetition_contexts(source, repetition_ranges)
+
+ masked = _merge_ranges(source, ranges)
+ projection = _project(source, masked)
+ repetition_masked = _merge_ranges(source, [*ranges, *repetition_ranges])
+ repetition_projection = _project(source, repetition_masked)
+ tokens = _segment_tokens(projection)
+ paragraphs = _segment_paragraphs(projection)
+ sentences = _segment_sentences(projection, paragraphs)
+ repetition_tokens = _segment_tokens(repetition_projection)
+ repetition_paragraphs = _segment_paragraphs(repetition_projection)
+ repetition_sentences = _segment_sentences(repetition_projection, repetition_paragraphs)
+ suppressions = _bind_suppressions(source, projection, parsed_suppressions, paragraphs)
+ masked_count = sum(
+ 1
+ for item in masked
+ for character in source[item.start : item.end]
+ if character not in "\r\n"
+ )
+ metrics = DocumentMetrics(
+ source_bytes=len(raw),
+ source_code_points=len(source),
+ analyzable_words=len(tokens),
+ analyzable_sentences=len(sentences),
+ analyzable_paragraphs=len(paragraphs),
+ masked_code_points=masked_count,
+ )
+ return Document(
+ path=Path(path).as_posix(),
+ source=source,
+ prose_projection=projection,
+ repetition_projection=repetition_projection,
+ line_starts=_line_starts(source),
+ front_matter=front_matter,
+ masked_ranges=masked,
+ suppressions=suppressions,
+ tokens=tokens,
+ sentences=sentences,
+ paragraphs=paragraphs,
+ repetition_tokens=repetition_tokens,
+ repetition_sentences=repetition_sentences,
+ repetition_paragraphs=repetition_paragraphs,
+ metrics=metrics,
+ )
+
+
+def load_document(
+ path: str | Path,
+ *,
+ contexts: ContextConfig | None = None,
+ max_source_bytes: int = MAX_SOURCE_BYTES,
+) -> Document:
+ """Read a regular Markdown file and build its source-mapped prose view."""
+
+ source_path = Path(path)
+ if source_path.is_symlink() or not source_path.is_file():
+ raise ValueError(f"input must be a regular non-symlink file: {source_path}")
+ return build_document(
+ source_path,
+ source_path.read_bytes(),
+ contexts=contexts,
+ max_source_bytes=max_source_bytes,
+ )
+
+
+__all__ = [
+ "Document",
+ "DocumentMetrics",
+ "MaskedRange",
+ "ProjectionError",
+ "ProseSegment",
+ "Span",
+ "SuppressionDirective",
+ "build_document",
+ "line_column",
+ "load_document",
+ "normalize_key",
+]
diff --git a/dev-tools/slop-cop/src/slop_cop/engine.py b/dev-tools/slop-cop/src/slop_cop/engine.py
new file mode 100644
index 00000000..67159b70
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/engine.py
@@ -0,0 +1,369 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Mapping
+from contextlib import nullcontext
+from dataclasses import dataclass
+from typing import Any
+
+from slop_cop.config import ErrorPolicy, Severity, SlopCopConfig
+from slop_cop.document import Document, Span
+from slop_cop.findings import (
+ AnalysisState,
+ AppliedSuppression,
+ Decision,
+ FileResult,
+ Finding,
+ RuleExecutionError,
+)
+from slop_cop.rules.api import RuleContext, RuleEvaluation, validate_evaluation
+from slop_cop.rules.registry import ConfiguredRule, RuleRegistry
+from slop_cop.runtime import RuntimeManager
+from slop_cop.scoring import score_findings
+
+
+@dataclass(frozen=True, slots=True)
+class EngineOutput:
+ file_result: FileResult
+ external_audits: tuple[Mapping[str, Any], ...] = ()
+
+
+def _excerpt(source: str, start: int, end: int, limit: int = 1_000) -> str:
+ value = " ".join(source[start:end].split())
+ return value if len(value) <= limit else value[: limit - 1] + "…"
+
+
+def _finding(
+ document: Document,
+ configured: ConfiguredRule,
+ signal: Any,
+) -> Finding:
+ metadata = configured.metadata
+ policy = configured.policy
+ span = None
+ line = column = None
+ excerpt = signal.detail or ""
+ if signal.scope == "span":
+ span = Span(start=signal.start, end=signal.end)
+ line, column = document.line_column(signal.start)
+ excerpt = _excerpt(document.prose_projection, signal.start, signal.end)
+ advisory = policy.severity is Severity.INFO or policy.cap == 0
+ return Finding(
+ rule_id=metadata.id,
+ category=metadata.category,
+ severity=policy.severity,
+ source_path=document.path,
+ span=span,
+ line=line,
+ column=column,
+ excerpt=excerpt[:1000],
+ normalized_key=signal.key,
+ score_group=metadata.score_group or metadata.id,
+ explanation=metadata.rationale,
+ advice=metadata.advice,
+ units=signal.units,
+ advisory=advisory,
+ chargeable=not advisory,
+ blocking=policy.blocking,
+ )
+
+
+def _apply_suppressions(
+ document: Document,
+ findings: tuple[Finding, ...],
+ registry: RuleRegistry,
+) -> tuple[tuple[Finding, ...], tuple[RuleExecutionError, ...]]:
+ values = list(findings)
+ errors: list[RuleExecutionError] = []
+ known_ids = {configured.metadata.id for configured in registry}
+ for directive in document.suppressions:
+ for rule_id in directive.rule_ids:
+ if rule_id not in known_ids:
+ errors.append(
+ RuleExecutionError(
+ rule_id=None,
+ source_path=document.path,
+ error_code="unknown_suppression_rule",
+ message=f"suppression references unknown rule {rule_id!r}",
+ fatal=True,
+ )
+ )
+ continue
+ selected = next(
+ (
+ index
+ for index, finding in enumerate(values)
+ if finding.rule_id == rule_id
+ and not finding.suppressed
+ and finding.span is not None
+ and directive.target_span.start
+ <= finding.span.start
+ < directive.target_span.end
+ ),
+ None,
+ )
+ if selected is None:
+ errors.append(
+ RuleExecutionError(
+ rule_id=rule_id,
+ source_path=document.path,
+ error_code="unused_suppression",
+ message=f"suppression for {rule_id!r} did not match the next prose block",
+ fatal=True,
+ )
+ )
+ continue
+ finding = values[selected]
+ values[selected] = finding.model_copy(
+ update={
+ "suppressed": True,
+ "suppression_reason": directive.reason,
+ "chargeable": False,
+ "advisory": True,
+ "blocking": False,
+ }
+ )
+ return tuple(values), tuple(errors)
+
+
+def _materially_overlaps(left: Finding, right: Finding) -> bool:
+ if left.span is None or right.span is None:
+ return False
+ if left.score_group != right.score_group:
+ return False
+ overlap = min(left.span.end, right.span.end) - max(left.span.start, right.span.start)
+ if overlap <= 0:
+ return False
+ shorter = min(left.span.end - left.span.start, right.span.end - right.span.start)
+ return overlap == shorter or overlap * 2 >= shorter
+
+
+def deduplicate_findings(
+ findings: tuple[Finding, ...], registry: RuleRegistry
+) -> tuple[Finding, ...]:
+ order = {configured.metadata.id: configured.order for configured in registry}
+ policies = {configured.metadata.id: configured.policy for configured in registry}
+ priorities = {
+ configured.metadata.id: configured.metadata.overlap_priority for configured in registry
+ }
+ values = list(findings)
+ parent = list(range(len(values)))
+
+ def find(index: int) -> int:
+ while parent[index] != index:
+ parent[index] = parent[parent[index]]
+ index = parent[index]
+ return index
+
+ def union(left: int, right: int) -> None:
+ left_root, right_root = find(left), find(right)
+ if left_root != right_root:
+ parent[right_root] = left_root
+
+ mapped = [index for index, finding in enumerate(values) if finding.span is not None]
+ mapped.sort(key=lambda index: (values[index].span.start, values[index].span.end)) # type: ignore[union-attr]
+ for position, left_index in enumerate(mapped):
+ left = values[left_index]
+ assert left.span is not None
+ for right_index in mapped[position + 1 :]:
+ right = values[right_index]
+ assert right.span is not None
+ if right.span.start >= left.span.end:
+ break
+ if _materially_overlaps(left, right):
+ union(left_index, right_index)
+
+ clusters: dict[int, list[int]] = {}
+ for index in range(len(values)):
+ clusters.setdefault(find(index), []).append(index)
+ severity_rank = {Severity.ERROR: 0, Severity.WARNING: 1, Severity.INFO: 2}
+
+ def span_length(finding: Finding) -> int:
+ if finding.span is None:
+ return 1_000_000
+ return finding.span.end - finding.span.start
+
+ for indexes in clusters.values():
+ if len(indexes) < 2:
+ continue
+ primary = min(
+ indexes,
+ key=lambda index: (
+ values[index].suppressed,
+ not values[index].blocking,
+ severity_rank[values[index].severity],
+ -priorities[values[index].rule_id],
+ policies[values[index].rule_id].fixed_allowance,
+ -policies[values[index].rule_id].first_cost,
+ order.get(values[index].rule_id, 1_000_000),
+ span_length(values[index]),
+ ),
+ )
+ related = tuple(sorted({values[index].rule_id for index in indexes if index != primary}))
+ values[primary] = values[primary].model_copy(update={"related_rule_ids": related})
+ for index in indexes:
+ if index == primary or values[index].suppressed:
+ continue
+ values[index] = values[index].model_copy(
+ update={"chargeable": False, "advisory": True, "blocking": False}
+ )
+ return tuple(
+ sorted(
+ values,
+ key=lambda finding: (
+ finding.span.start if finding.span is not None else len(values) + 1,
+ order.get(finding.rule_id, 1_000_000),
+ finding.normalized_key,
+ ),
+ )
+ )
+
+
+async def _evaluate_rule(
+ configured: ConfiguredRule,
+ context: RuleContext,
+ manager: RuntimeManager,
+ timeout: float,
+) -> tuple[ConfiguredRule, RuleEvaluation | None, Exception | None]:
+ try:
+ async with asyncio.timeout(timeout):
+ evaluation = await configured.rule.evaluate(
+ context, manager.for_rule(configured.metadata, configured.policy)
+ )
+ evaluation = validate_evaluation(
+ configured.metadata,
+ evaluation,
+ document_length=len(context.source),
+ max_signal_units=configured.policy.max_signal_units,
+ )
+ return configured, evaluation, None
+ except Exception as error:
+ return configured, None, error
+
+
+async def analyze_document(
+ document: Document,
+ registry: RuleRegistry,
+ config: SlopCopConfig,
+ *,
+ runtime_manager: RuntimeManager | None = None,
+) -> EngineOutput:
+ if document.metrics.analyzable_words == 0:
+ initial_error = RuleExecutionError(
+ source_path=document.path,
+ error_code="no_analyzable_prose",
+ message="the document contains no analyzable prose",
+ fatal=True,
+ )
+ result = FileResult(
+ path=document.path,
+ analysis_state=AnalysisState.ERROR,
+ decision=Decision.FAIL,
+ score=None,
+ threshold=config.threshold,
+ hard_fail=False,
+ metrics=document.metrics,
+ errors=(initial_error,),
+ )
+ return EngineOutput(result)
+
+ context = RuleContext(
+ document=document,
+ repository_terms=frozenset(term.casefold() for term in config.vocabulary.allowed_terms),
+ )
+
+ manager_scope = nullcontext(runtime_manager) if runtime_manager else RuntimeManager(config)
+ async with manager_scope as manager:
+ async with asyncio.TaskGroup() as group:
+ tasks = [
+ group.create_task(
+ _evaluate_rule(
+ configured,
+ context,
+ manager,
+ config.external_file_timeout_seconds,
+ )
+ )
+ for configured in registry.enabled()
+ ]
+ evaluated = [task.result() for task in tasks]
+
+ evaluated.sort(key=lambda row: row[0].order)
+ findings: list[Finding] = []
+ errors: list[RuleExecutionError] = []
+ audits: list[Mapping[str, Any]] = []
+ for configured, evaluation, execution_error in evaluated:
+ if execution_error is not None:
+ fatal = configured.policy.on_error is ErrorPolicy.FAIL
+ errors.append(
+ RuleExecutionError(
+ rule_id=configured.metadata.id,
+ source_path=document.path,
+ error_code="rule_execution_failed",
+ message=f"rule {configured.metadata.id!r} could not complete",
+ fatal=fatal,
+ )
+ )
+ continue
+ assert evaluation is not None
+ findings.extend(_finding(document, configured, signal) for signal in evaluation.signals)
+ if evaluation.audit:
+ audits.append(
+ {
+ "rule_id": configured.metadata.id,
+ "rule_version": configured.metadata.version,
+ **evaluation.audit,
+ }
+ )
+
+ suppressed, suppression_errors = _apply_suppressions(document, tuple(findings), registry)
+ errors.extend(suppression_errors)
+ deduplicated = deduplicate_findings(suppressed, registry)
+ scored = score_findings(document, deduplicated, registry, config)
+ fatal = any(error.fatal for error in errors)
+ incomplete = any(not error.fatal for error in errors)
+ decision = (
+ Decision.PASS
+ if scored.score >= config.threshold and not scored.hard_fail and not fatal
+ else Decision.FAIL
+ )
+ state = (
+ AnalysisState.ERROR
+ if fatal
+ else AnalysisState.INCOMPLETE
+ if incomplete
+ else AnalysisState.COMPLETE
+ )
+ suppression_records = tuple(
+ AppliedSuppression(
+ rule_ids=directive.rule_ids,
+ reason=directive.reason,
+ directive_span=directive.directive_span,
+ target_span=directive.target_span,
+ suppressed_finding_ids=tuple(
+ f"{finding.rule_id}:{finding.line}:{finding.column}"
+ for finding in deduplicated
+ if finding.suppressed
+ and finding.suppression_reason == directive.reason
+ and finding.rule_id in directive.rule_ids
+ ),
+ )
+ for directive in document.suppressions
+ )
+ return EngineOutput(
+ FileResult(
+ path=document.path,
+ analysis_state=state,
+ decision=decision,
+ score=scored.score,
+ threshold=config.threshold,
+ hard_fail=scored.hard_fail,
+ metrics=document.metrics,
+ findings=deduplicated,
+ suppressions=suppression_records,
+ rule_costs=scored.rule_costs,
+ category_costs=scored.category_costs,
+ errors=tuple(errors),
+ ),
+ tuple(audits),
+ )
diff --git a/dev-tools/slop-cop/src/slop_cop/findings.py b/dev-tools/slop-cop/src/slop_cop/findings.py
new file mode 100644
index 00000000..b3afc43c
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/findings.py
@@ -0,0 +1,247 @@
+"""Immutable analysis and report records."""
+
+from __future__ import annotations
+
+from enum import StrEnum
+from typing import Annotated, Literal
+
+from pydantic import Field, StringConstraints, model_validator
+
+from slop_cop.config import CategoryId, RuleId, Severity, StrictModel
+from slop_cop.document import DocumentMetrics, Span
+
+BoundedText = Annotated[str, StringConstraints(max_length=1_000)]
+Digest = Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{64}$")]
+Revision = Annotated[str, StringConstraints(min_length=7, max_length=64, pattern=r"^[0-9a-f]+$")]
+
+
+class AnalysisState(StrEnum):
+ COMPLETE = "complete"
+ INCOMPLETE = "incomplete"
+ ERROR = "error"
+ NOT_APPLICABLE = "not_applicable"
+
+
+class Decision(StrEnum):
+ PASS = "pass"
+ FAIL = "fail"
+ OVERRIDDEN = "overridden"
+ NOT_APPLICABLE = "not_applicable"
+
+
+class Finding(StrictModel):
+ rule_id: RuleId
+ category: CategoryId
+ severity: Severity
+ source_path: str = Field(min_length=1, max_length=4_096)
+ span: Span | None
+ line: int | None = Field(default=None, ge=1)
+ column: int | None = Field(default=None, ge=1)
+ excerpt: BoundedText = ""
+ normalized_key: str = Field(min_length=1, max_length=256)
+ score_group: str = Field(min_length=1, max_length=128)
+ explanation: BoundedText
+ advice: BoundedText
+ units: int = Field(default=1, ge=1, le=5_000)
+ advisory: bool = False
+ chargeable: bool = True
+ suppressed: bool = False
+ blocking: bool = False
+ suppression_reason: BoundedText | None = None
+ related_rule_ids: tuple[RuleId, ...] = ()
+
+ @model_validator(mode="after")
+ def validate_location_and_state(self) -> Finding:
+ if self.span is None and (self.line is not None or self.column is not None):
+ raise ValueError("document-scoped findings cannot have line or column")
+ if self.span is not None and (self.line is None or self.column is None):
+ raise ValueError("source-mapped findings require line and column")
+ if self.suppressed and not self.suppression_reason:
+ raise ValueError("suppressed findings require a reason")
+ if not self.suppressed and self.suppression_reason is not None:
+ raise ValueError("unsuppressed findings cannot have a suppression reason")
+ if self.advisory and self.chargeable:
+ raise ValueError("advisory findings cannot be chargeable")
+ return self
+
+
+class DensityMeasurement(StrictModel):
+ unit: Literal["word", "sentence", "paragraph"]
+ window: int = Field(ge=1)
+ allowed_units: int = Field(ge=0)
+ peak_units: int = Field(ge=0)
+ peak_excess: int = Field(ge=0)
+ cost: float = Field(ge=0, le=100)
+ window_span: Span | None = None
+
+
+class RuleCost(StrictModel):
+ rule_id: RuleId
+ deduplicated_units: int = Field(ge=0)
+ allowance: int = Field(ge=0)
+ document_excess: int = Field(ge=0)
+ base_cost: float = Field(ge=0, le=100)
+ density: DensityMeasurement | None = None
+ cap: float = Field(ge=0, le=100)
+ charged_cost: float = Field(ge=0, le=100)
+
+ @model_validator(mode="after")
+ def validate_arithmetic_bounds(self) -> RuleCost:
+ if self.document_excess != max(0, self.deduplicated_units - self.allowance):
+ raise ValueError("document_excess does not match units and allowance")
+ if self.charged_cost > self.cap:
+ raise ValueError("charged rule cost cannot exceed its cap")
+ return self
+
+
+class CategoryCost(StrictModel):
+ category: CategoryId
+ rule_cost: float = Field(ge=0, le=100)
+ density: DensityMeasurement | None = None
+ cap: float = Field(ge=0, le=100)
+ charged_cost: float = Field(ge=0, le=100)
+
+ @model_validator(mode="after")
+ def validate_cap(self) -> CategoryCost:
+ if self.charged_cost > self.cap:
+ raise ValueError("charged category cost cannot exceed its cap")
+ return self
+
+
+class RuleExecutionError(StrictModel):
+ rule_id: RuleId | None = None
+ source_path: str | None = Field(default=None, max_length=4_096)
+ error_code: str = Field(min_length=1, max_length=64, pattern=r"^[a-z][a-z0-9_-]*$")
+ message: BoundedText
+ fatal: bool
+
+
+class AppliedSuppression(StrictModel):
+ rule_ids: tuple[RuleId, ...] = Field(min_length=1)
+ reason: BoundedText
+ directive_span: Span
+ target_span: Span
+ suppressed_finding_ids: tuple[str, ...] = ()
+
+
+class ExternalAudit(StrictModel):
+ rule_id: RuleId
+ rule_version: int = Field(ge=1)
+ service: str = Field(min_length=1, max_length=64)
+ endpoint_hostname: str = Field(min_length=1, max_length=253)
+ content_digest: Digest
+ request_schema_version: str = Field(min_length=1, max_length=64)
+ response_schema_version: str | None = Field(default=None, max_length=64)
+ service_request_id: str | None = Field(default=None, max_length=256)
+ judge_revision: str | None = Field(default=None, max_length=128)
+ attempts: int = Field(ge=1, le=3)
+ latency_ms: int = Field(ge=0, le=600_000)
+ outcome: str = Field(min_length=1, max_length=64)
+ response_digest: Digest | None = None
+
+
+class FindingChange(StrictModel):
+ added: tuple[Finding, ...] = ()
+ removed: tuple[Finding, ...] = ()
+ persistent: tuple[Finding, ...] = ()
+
+
+class BaseComparison(StrictModel):
+ score: int | None = Field(default=None, ge=0, le=100)
+ delta: int | None = Field(default=None, ge=-100, le=100)
+ analysis_state: AnalysisState
+ findings: FindingChange = FindingChange()
+ errors: tuple[RuleExecutionError, ...] = ()
+
+
+class FileResult(StrictModel):
+ path: str = Field(min_length=1, max_length=4_096)
+ analysis_state: AnalysisState
+ decision: Decision
+ score: int | None = Field(default=None, ge=0, le=100)
+ threshold: int = Field(ge=0, le=100)
+ hard_fail: bool = False
+ metrics: DocumentMetrics
+ findings: tuple[Finding, ...] = ()
+ suppressions: tuple[AppliedSuppression, ...] = ()
+ rule_costs: tuple[RuleCost, ...] = ()
+ category_costs: tuple[CategoryCost, ...] = ()
+ errors: tuple[RuleExecutionError, ...] = ()
+ base: BaseComparison | None = None
+
+ @model_validator(mode="after")
+ def validate_decision(self) -> FileResult:
+ if self.analysis_state is AnalysisState.NOT_APPLICABLE:
+ if self.decision is not Decision.NOT_APPLICABLE or self.score is not None:
+ raise ValueError("not-applicable file results require no score and no decision")
+ elif self.analysis_state is not AnalysisState.ERROR and self.score is None:
+ raise ValueError("analyzed file results require a score")
+ return self
+
+
+class OverrideRecord(StrictModel):
+ reviewer: str = Field(min_length=1, max_length=256)
+ reason: BoundedText
+ review_id: int = Field(ge=1)
+ review_url: str = Field(min_length=1, max_length=2_048)
+ head_sha: Revision
+
+
+class RunResult(StrictModel):
+ schema_version: Literal[1] = 1
+ analysis_state: AnalysisState
+ decision: Decision
+ score: int | None = Field(default=None, ge=0, le=100)
+ threshold: int = Field(ge=0, le=100)
+ repository: str | None = Field(default=None, max_length=256)
+ pull_request_number: int | None = Field(default=None, ge=1)
+ base_sha: Revision | None = None
+ head_sha: Revision | None = None
+ tool_version: str = Field(min_length=1, max_length=64)
+ config_digest: Digest
+ files: tuple[FileResult, ...] = ()
+ rule_errors: tuple[RuleExecutionError, ...] = ()
+ external_audits: tuple[ExternalAudit, ...] = ()
+ override: OverrideRecord | None = None
+
+ @model_validator(mode="after")
+ def validate_aggregate(self) -> RunResult:
+ if self.analysis_state is AnalysisState.NOT_APPLICABLE:
+ if self.decision is not Decision.NOT_APPLICABLE or self.score is not None or self.files:
+ raise ValueError("not-applicable runs require no score, files, or policy decision")
+ return self
+ if not self.files:
+ raise ValueError("applicable runs require at least one file result")
+ scores = [item.score for item in self.files if item.score is not None]
+ if scores and self.score != min(scores):
+ raise ValueError("run score must be the minimum scored file result")
+ if not scores and self.score is not None:
+ raise ValueError("a run without scored file results cannot have a score")
+ if self.decision is Decision.OVERRIDDEN and self.override is None:
+ raise ValueError("overridden decisions require override metadata")
+ if self.decision is not Decision.OVERRIDDEN and self.override is not None:
+ raise ValueError("override metadata requires an overridden decision")
+ if self.decision is Decision.OVERRIDDEN:
+ if self.analysis_state is not AnalysisState.COMPLETE:
+ raise ValueError("overrides require complete analysis")
+ if not any(item.decision is Decision.FAIL for item in self.files):
+ raise ValueError("overrides require a policy failure")
+ return self
+
+
+__all__ = [
+ "AnalysisState",
+ "AppliedSuppression",
+ "BaseComparison",
+ "CategoryCost",
+ "Decision",
+ "DensityMeasurement",
+ "ExternalAudit",
+ "FileResult",
+ "Finding",
+ "FindingChange",
+ "OverrideRecord",
+ "RuleCost",
+ "RuleExecutionError",
+ "RunResult",
+]
diff --git a/dev-tools/slop-cop/src/slop_cop/report.py b/dev-tools/slop-cop/src/slop_cop/report.py
new file mode 100644
index 00000000..f28461b4
--- /dev/null
+++ b/dev-tools/slop-cop/src/slop_cop/report.py
@@ -0,0 +1,919 @@
+"""Render Slop Cop results without recalculating analysis or scoring."""
+
+from __future__ import annotations
+
+import html
+import json
+from base64 import b64encode
+from collections.abc import Mapping, Sequence
+from functools import cache
+from importlib.resources import files
+from pathlib import Path
+from typing import Any, cast
+
+from pydantic import ValidationError
+
+from slop_cop.findings import RunResult
+
+JSON_SCHEMA_VERSION = 1
+MAX_HTML_BYTES = 10 * 1024 * 1024
+_CSP = "default-src 'none'; style-src 'unsafe-inline'; img-src data:"
+
+
+class ReportError(ValueError):
+ """Raised when a result cannot be serialized safely."""
+
+
+def result_data(result: RunResult | Mapping[str, Any]) -> dict[str, Any]:
+ """Validate and serialize one canonical run result."""
+ try:
+ validated = (
+ result
+ if isinstance(result, RunResult)
+ else RunResult.model_validate(result, strict=False)
+ )
+ except ValidationError as error:
+ raise ReportError(f"Invalid run result: {error}") from error
+ return validated.model_dump(mode="json")
+
+
+def json_report(result: RunResult | Mapping[str, Any]) -> str:
+ """Serialize a result with stable key ordering and a final newline."""
+ data = result_data(result)
+ data.setdefault("schema_version", JSON_SCHEMA_VERSION)
+ try:
+ return json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+ except (TypeError, ValueError) as error:
+ raise ReportError(f"The run result is not JSON serializable: {error}") from error
+
+
+def write_json_report(result: RunResult | Mapping[str, Any], destination: str | Path) -> Path:
+ """Write the canonical machine report."""
+ path = Path(destination)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(json_report(result), encoding="utf-8")
+ return path
+
+
+def terminal_report(result: RunResult | Mapping[str, Any]) -> str:
+ """Render a compact, stable terminal summary."""
+ data = result_data(result)
+ state = _upper(data.get("decision") or data.get("analysis_state") or "error")
+ threshold = data.get("threshold")
+ score = data.get("score")
+ lines = [f"Slop Cop: {state} score={_display(score)} threshold={_display(threshold)}"]
+ for file_result in _items(data.get("files")):
+ path = _text(file_result.get("path"))
+ charged_rule_ids = _charged_rule_ids(file_result)
+ file_score = file_result.get("score")
+ file_state = _upper(
+ file_result.get("decision") or file_result.get("analysis_state") or "complete"
+ )
+ base_score = _base_score(file_result)
+ delta = _delta(file_score, base_score)
+ comparison = (
+ "" if base_score is None else f" base={_display(base_score)} delta={_signed(delta)}"
+ )
+ lines.append(f"{path}: {file_state} score={_display(file_score)}{comparison}")
+ for category in _category_rows(file_result):
+ lines.append(f" category {category[0]}: -{_display(category[1])} points")
+ for rule in _items(file_result.get("rule_costs")):
+ charged = rule.get("charged_cost", 0)
+ density = rule.get("density")
+ density_cost = density.get("cost", 0) if isinstance(density, Mapping) else 0
+ if _number(charged) or _number(density_cost):
+ lines.append(
+ f" rule {_text(rule.get('rule_id') or 'unknown')}: "
+ f"units={_display(rule.get('deduplicated_units'))} "
+ f"allowance={_display(rule.get('allowance'))} "
+ f"base=-{_display(rule.get('base_cost'))} "
+ f"density=-{_display(density_cost)} "
+ f"charged=-{_display(charged)}"
+ )
+ if isinstance(density, Mapping) and density.get("peak_excess"):
+ density_unit = _text(density.get("unit") or "units")
+ lines.append(
+ f" peak={_display(density.get('peak_units'))} in "
+ f"{_display(density.get('window'))} {density_unit}; "
+ f"excess={_display(density.get('peak_excess'))}"
+ )
+ for finding in _findings(file_result):
+ if finding.get("suppressed") or finding.get("advisory"):
+ continue
+ if finding.get("rule_id") not in charged_rule_ids and not finding.get("blocking"):
+ continue
+ location = _finding_location(path, finding)
+ rule_id = _text(finding.get("rule_id") or "unknown")
+ excerpt = _bounded(_text(finding.get("excerpt")), 120)
+ lines.append(f" {location} [{rule_id}] {excerpt}".rstrip())
+ advice = _text(finding.get("advice"))
+ if advice:
+ lines.append(f" {advice}")
+ findings = _findings(file_result)
+ within_allowance = sum(
+ 1
+ for finding in findings
+ if finding.get("chargeable")
+ and not finding.get("suppressed")
+ and not finding.get("blocking")
+ and finding.get("rule_id") not in charged_rule_ids
+ )
+ advisory = sum(1 for finding in findings if finding.get("advisory"))
+ suppressed = sum(1 for finding in findings if finding.get("suppressed"))
+ if within_allowance or advisory or suppressed:
+ lines.append(
+ " unscored signals: "
+ f"within_allowance={within_allowance} advisory={advisory} "
+ f"suppressed={suppressed}"
+ )
+ for error in _items(file_result.get("errors")):
+ lines.append(
+ f" analysis error [{_text(error.get('error_code') or 'unknown')}]: "
+ f"{_bounded(_text(error.get('message') or ''), 300)}"
+ )
+ for error in _items(data.get("rule_errors")):
+ lines.append(f"Rule error: {_bounded(_text(error.get('message') or error), 300)}")
+ override = data.get("override")
+ if isinstance(override, Mapping):
+ lines.append(
+ f"Override: {_text(override.get('reviewer') or 'unknown')} - "
+ f"{_bounded(_text(override.get('reason') or ''), 300)}"
+ )
+ for audit in _items(data.get("external_audits")):
+ lines.append(
+ f"External rule {_text(audit.get('rule_id') or 'unknown')} sent selected prose to "
+ f"{_text(audit.get('endpoint_hostname') or audit.get('service') or 'unknown')}; "
+ f"outcome={_text(audit.get('outcome') or 'unknown')}"
+ )
+ return "\n".join(lines) + "\n"
+
+
+def html_report(
+ result: RunResult | Mapping[str, Any],
+ *,
+ sources: Mapping[str, str] | None = None,
+ projections: Mapping[str, str] | None = None,
+) -> str:
+ """Render a self-contained HTML report with all result text escaped."""
+ data = result_data(result)
+ state = _upper(data.get("decision") or data.get("analysis_state") or "error")
+ score = data.get("score")
+ threshold = data.get("threshold")
+ head_sha = _text(data.get("head_sha") or "")
+ files = _items(data.get("files"))
+ body: list[str] = [
+ '',
+ f'',
+ '',
+ "Slop Cop report",
+ f"",
+ '
The score measures configured editorial signals. '
+ "It does not identify the author or determine whether a model wrote the text.
",
+ ]
+ override = data.get("override")
+ if isinstance(override, Mapping):
+ body.append(_render_override(override))
+ if not files:
+ message = (
+ "No changed Dev Note required analysis."
+ if state == "NOT APPLICABLE"
+ else "No file results were produced."
+ )
+ body.append(f"
Files
{_h(message)}
")
+ else:
+ if len(files) > 1:
+ body.append(_render_file_table(files))
+ for index, file_result in enumerate(files, 1):
+ body.append(_render_file(file_result, index, sources=sources, projections=projections))
+ body.append(_render_rule_errors(data))
+ body.append(_render_external_audits(data))
+ body.append(_render_provenance(data))
+ body.append("\n")
+ rendered = "".join(body)
+ if len(rendered.encode("utf-8")) > MAX_HTML_BYTES:
+ raise ReportError(f"HTML report exceeds the {MAX_HTML_BYTES}-byte limit.")
+ return rendered
+
+
+def write_html_report(
+ result: RunResult | Mapping[str, Any],
+ destination: str | Path,
+ *,
+ sources: Mapping[str, str] | None = None,
+ projections: Mapping[str, str] | None = None,
+) -> Path:
+ """Write the self-contained HTML report."""
+ path = Path(destination)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(html_report(result, sources=sources, projections=projections), encoding="utf-8")
+ return path
+
+
+def write_report_directory(
+ result: RunResult | Mapping[str, Any],
+ destination: str | Path,
+ *,
+ sources: Mapping[str, str] | None = None,
+ projections: Mapping[str, str] | None = None,
+) -> tuple[Path, Path]:
+ """Write ``index.html`` and ``report.json`` into one artifact directory."""
+ directory = Path(destination)
+ directory.mkdir(parents=True, exist_ok=True)
+ return (
+ write_html_report(
+ result,
+ directory / "index.html",
+ sources=sources,
+ projections=projections,
+ ),
+ write_json_report(result, directory / "report.json"),
+ )
+
+
+def _render_file_table(files: list[dict[str, Any]]) -> str:
+ rows = []
+ for item in files:
+ path = _text(item.get("path"))
+ score = item.get("score")
+ base = _base_score(item)
+ state = _upper(item.get("decision") or item.get("analysis_state") or "complete")
+ findings = _findings(item)
+ charged_rule_ids = _charged_rule_ids(item)
+ scored = sum(
+ 1
+ for finding in findings
+ if finding.get("rule_id") in charged_rule_ids
+ and finding.get("chargeable")
+ and not finding.get("suppressed")
+ )
+ unscored = sum(
+ 1
+ for finding in findings
+ if not finding.get("blocking")
+ and not finding.get("suppressed")
+ and finding.get("rule_id") not in charged_rule_ids
+ )
+ rows.append(
+ "
'
+ ]
+ raw_metrics = item.get("metrics")
+ metrics: Mapping[str, Any] = raw_metrics if isinstance(raw_metrics, Mapping) else {}
+ source = sources.get(path) if sources is not None else None
+ projection = projections.get(path) if projections is not None else None
+ charged_rule_ids = _charged_rule_ids(item)
+ scored = sum(
+ 1
+ for finding in findings
+ if finding.get("rule_id") in charged_rule_ids
+ and finding.get("chargeable")
+ and not finding.get("suppressed")
+ )
+ unscored = sum(
+ 1
+ for finding in findings
+ if not finding.get("blocking")
+ and not finding.get("suppressed")
+ and finding.get("rule_id") not in charged_rule_ids
+ )
+ sections.append('