feat(29-01): deterministic peer critic pattern-scan engine

maf_starter/critic.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+import sys
+from typing import Literal
+
+from maf_starter.config import Settings
+
+
+@dataclass(frozen=True)
+class Finding:
+    check: str
+    severity: Literal["blocking", "advisory"]
+    file: str
+    line: int | None
+    detail: str
+
+
+@dataclass(frozen=True)
+class DiffFile:
+    path: str
+    added_lines: list[str]
+
+
+@dataclass(frozen=True)
+class CriticReport:
+    findings: list[Finding]
+    blocking_count: int
+    advisory_count: int
+    exit_code: int
+
+
+_BARE_EXCEPT_PATTERN = re.compile(r"^\s*except\s*:\s*$")
+_EXCEPTION_EXCEPT_PATTERN = re.compile(r"^\s*except\s+Exception(?:\s+as\s+\w+)?\s*:\s*$")
+_LOG_TOKENS = ("logger.", "logging.", "print(", "emit_failure_telemetry(", "telemetry.")
+_TYPED_FAILURE_TOKENS = ("FailureState", "McpException", "ValueError", "RuntimeError", "InvalidOperationException")
+
+
+def parse_unified_diff(diff_text: str) -> list[DiffFile]:
+    files: list[DiffFile] = []
+    current_path: str | None = None
+    current_added: list[str] = []
+
+    for line in diff_text.splitlines():
+        if line.startswith("diff --git "):
+            if current_path is not None:
+                files.append(DiffFile(path=current_path, added_lines=current_added))
+            current_path = None
+            current_added = []
+            continue
+        if line.startswith("+++ b/"):
+            current_path = line[6:]
+            continue
+        if current_path is None:
+            continue
+        if line.startswith("+") and not line.startswith("+++"):
+            current_added.append(line[1:])
+
+    if current_path is not None:
+        files.append(DiffFile(path=current_path, added_lines=current_added))
+
+    return files
+
+
+def check_bare_except(file_path: str, added_lines: list[str]) -> list[Finding]:
+    findings: list[Finding] = []
+    for index, line in enumerate(added_lines):
+        if not (_BARE_EXCEPT_PATTERN.match(line) or _EXCEPTION_EXCEPT_PATTERN.match(line)):
+            continue
+        window = added_lines[index + 1 : index + 5]
+        has_reraise = any("raise" in candidate for candidate in window)
+        has_logging = any(any(token in candidate for token in _LOG_TOKENS) for candidate in window)
+        has_typed_failure = any(any(token in candidate for token in _TYPED_FAILURE_TOKENS) for candidate in window)
+        if has_reraise and has_logging and has_typed_failure:
+            continue
+        findings.append(
+            Finding(
+                check="bare-except",
+                severity="blocking",
+                file=file_path,
+                line=index + 1,
+                detail="Broad exception handler added without a typed, logged re-raise path.",
+            )
+        )
+    return findings
+
+
+def check_missing_telemetry(file_path: str, added_lines: list[str]) -> list[Finding]:
+    findings: list[Finding] = []
+    for index, line in enumerate(added_lines):
+        if "except" not in line:
+            continue
+        window = added_lines[index + 1 : index + 5]
+        has_logging = any(any(token in candidate for token in _LOG_TOKENS) for candidate in window)
+        if has_logging:
+            continue
+        findings.append(
+            Finding(
+                check="missing-telemetry",
+                severity="advisory",
+                file=file_path,
+                line=index + 1,
+                detail="Exception path does not emit structured telemetry or logging.",
+            )
+        )
+    return findings
+
+
+def check_file_size_limit(file_path: str, added_line_count: int, limit: int = 400) -> list[Finding]:
+    if added_line_count <= limit:
+        return []
+    return [
+        Finding(
+            check="file-size-limit",
+            severity="advisory",
+            file=file_path,
+            line=None,
+            detail=f"Added line count {added_line_count} exceeds limit {limit}.",
+        )
+    ]
+
+
+def run_critic(diff_text: str, *, llm_pass: bool = False, settings: Settings | None = None) -> CriticReport:
+    if llm_pass:
+        raise NotImplementedError("LLM critic pass is reserved for a future phase.")
+
+    try:
+        _ = settings
+        files = parse_unified_diff(diff_text)
+        if not files:
+            raise ValueError("No unified diff content was found.")
+
+        findings: list[Finding] = []
+        for diff_file in files:
+            findings.extend(check_bare_except(diff_file.path, diff_file.added_lines))
+            findings.extend(check_missing_telemetry(diff_file.path, diff_file.added_lines))
+            findings.extend(check_file_size_limit(diff_file.path, len(diff_file.added_lines)))
+
+        blocking_count = sum(1 for finding in findings if finding.severity == "blocking")
+        advisory_count = sum(1 for finding in findings if finding.severity == "advisory")
+        return CriticReport(
+            findings=findings,
+            blocking_count=blocking_count,
+            advisory_count=advisory_count,
+            exit_code=1 if blocking_count > 0 else 0,
+        )
+    except Exception as exc:  # noqa: BLE001
+        print(f"critic error: {exc}", file=sys.stderr)
+        finding = Finding(
+            check="critic-error",
+            severity="advisory",
+            file="<critic>",
+            line=None,
+            detail=str(exc),
+        )
+        return CriticReport(findings=[finding], blocking_count=0, advisory_count=1, exit_code=0)

maf_starter/critic_cli.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import sys
+
+from maf_starter.critic import run_critic
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="critic", description="Run the Resilience First peer critic against a unified diff")
+    parser.add_argument("--diff", required=True, help="Diff file path or '-' to read from stdin")
+    parser.add_argument(
+        "--severity-gate",
+        choices=("blocking", "advisory"),
+        default="blocking",
+        help="Exit non-zero on blocking findings only, or on any finding in advisory mode",
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+
+    try:
+        if args.diff == "-":
+            diff_text = sys.stdin.read()
+        else:
+            diff_text = Path(args.diff).read_text(encoding="utf-8", errors="replace")
+    except (FileNotFoundError, OSError) as exc:
+        print(f"critic: unable to read diff: {exc}", file=sys.stderr)
+        return 2
+
+    report = run_critic(diff_text)
+    for finding in report.findings:
+        print(f"{finding.file}:{finding.line or '-'} [{finding.severity}] {finding.check} - {finding.detail}")
+    print(f"critic: {report.blocking_count} blocking, {report.advisory_count} advisory")
+
+    if args.severity_gate == "advisory":
+        return 1 if (report.blocking_count + report.advisory_count) > 0 else 0
+    return report.exit_code
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_critic.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import io
+from pathlib import Path
+import textwrap
+
+import pytest
+
+from maf_starter.critic import (
+    check_bare_except,
+    check_file_size_limit,
+    check_missing_telemetry,
+    parse_unified_diff,
+    run_critic,
+)
+from maf_starter.critic_cli import build_parser, main
+
+
+DIRTY_DIFF = textwrap.dedent(
+    """\
+    diff --git a/app.py b/app.py
+    +++ b/app.py
+    @@
+    +try:
+    +    do_work()
+    +except Exception:
+    +    pass
+    """
+)
+
+DIRTY_DIFF_WITH_TELEMETRY_GAP = textwrap.dedent(
+    """\
+    diff --git a/app.py b/app.py
+    +++ b/app.py
+    @@
+    +try:
+    +    do_work()
+    +except ValueError:
+    +    recover()
+    """
+)
+
+CLEAN_DIFF = textwrap.dedent(
+    """\
+    diff --git a/app.py b/app.py
+    +++ b/app.py
+    @@
+    +try:
+    +    do_work()
+    +except Exception as exc:
+    +    logger.error("failed", exc_info=exc)
+    +    raise RuntimeError("typed failure") from exc
+    """
+)
+
+
+def test_parse_unified_diff_returns_changed_files_and_added_lines() -> None:
+    diff_text = textwrap.dedent(
+        """\
+        diff --git a/app.py b/app.py
+        +++ b/app.py
+        @@
+        +first
+        diff --git a/worker.py b/worker.py
+        +++ b/worker.py
+        @@
+        +second
+        """
+    )
+
+    files = parse_unified_diff(diff_text)
+
+    assert len(files) == 2
+    assert files[0].path == "app.py"
+    assert files[0].added_lines == ["first"]
+    assert files[1].path == "worker.py"
+    assert files[1].added_lines == ["second"]
+
+
+def test_check_bare_except_blocks_untyped_unlogged_handler() -> None:
+    findings = check_bare_except("app.py", ["except Exception:", "    pass"])
+
+    assert len(findings) == 1
+    assert findings[0].check == "bare-except"
+    assert findings[0].severity == "blocking"
+
+
+def test_check_bare_except_allows_logged_typed_reraise() -> None:
+    findings = check_bare_except(
+        "app.py",
+        [
+            "except Exception as exc:",
+            '    logger.error("boom", exc_info=exc)',
+            '    raise RuntimeError("typed failure") from exc',
+        ],
+    )
+
+    assert findings == []
+
+
+def test_check_missing_telemetry_returns_advisory_only() -> None:
+    findings = check_missing_telemetry("app.py", ["except ValueError:", "    recover()"])
+
+    assert len(findings) == 1
+    assert findings[0].check == "missing-telemetry"
+    assert findings[0].severity == "advisory"
+
+
+def test_check_file_size_limit_flags_large_files() -> None:
+    findings = check_file_size_limit("big.py", 401, limit=400)
+
+    assert len(findings) == 1
+    assert findings[0].check == "file-size-limit"
+    assert findings[0].severity == "advisory"
+
+
+def test_run_critic_blocks_dirty_diff() -> None:
+    report = run_critic(DIRTY_DIFF)
+
+    assert report.exit_code == 1
+    assert report.blocking_count == 1
+    assert any(finding.check == "bare-except" for finding in report.findings)
+
+
+def test_run_critic_passes_clean_diff() -> None:
+    report = run_critic(CLEAN_DIFF)
+
+    assert report.exit_code == 0
+    assert report.blocking_count == 0
+    assert report.advisory_count == 0
+
+
+def test_run_critic_returns_advisory_error_on_malformed_input() -> None:
+    report = run_critic("not a diff at all")
+
+    assert report.exit_code == 0
+    assert report.blocking_count == 0
+    assert report.advisory_count == 1
+    assert report.findings[0].check == "critic-error"
+
+
+def test_build_parser_defaults_to_blocking_gate() -> None:
+    args = build_parser().parse_args(["--diff", "-"])
+
+    assert args.diff == "-"
+    assert args.severity_gate == "blocking"
+
+
+def test_main_returns_zero_for_clean_diff_file(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    diff_path = tmp_path / "clean.diff"
+    diff_path.write_text(CLEAN_DIFF, encoding="utf-8")
+
+    exit_code = main(["--diff", str(diff_path)])
+
+    out = capsys.readouterr().out
+    assert exit_code == 0
+    assert "critic: 0 blocking, 0 advisory" in out
+
+
+def test_main_returns_one_for_blocking_diff_file(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    diff_path = tmp_path / "dirty.diff"
+    diff_path.write_text(DIRTY_DIFF, encoding="utf-8")
+
+    exit_code = main(["--diff", str(diff_path)])
+
+    out = capsys.readouterr().out
+    assert exit_code == 1
+    assert "[blocking] bare-except" in out
+
+
+def test_main_returns_two_for_missing_diff_path(capsys: pytest.CaptureFixture[str]) -> None:
+    exit_code = main(["--diff", "missing.diff"])
+
+    err = capsys.readouterr().err
+    assert exit_code == 2
+    assert "unable to read diff" in err
+
+
+def test_main_reads_diff_from_stdin_and_supports_advisory_gate(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setattr("sys.stdin", io.StringIO(DIRTY_DIFF_WITH_TELEMETRY_GAP))
+
+    exit_code = main(["--diff", "-", "--severity-gate", "advisory"])
+
+    out = capsys.readouterr().out
+    assert exit_code == 1
+    assert "critic: 0 blocking, 1 advisory" in out