feat: migrate LLM rule evaluation to evidence-based validation model

DOCS/research/evidence_evaluation.md

@@ -0,0 +1,28 @@
+# Evidence-Based Governance Rule Evaluation
+
+Anchor's rule engine has been updated to separate the **evidence collection** (pattern matching) phase from the **policy evaluation** (rule verification) phase.
+
+Rather than immediately generating violations when a pattern match occurs, the engine collects matching occurrences as **Candidate Evidence** and passes them to a specialized validation pipeline.
+
+```
+Pattern Match
+      ↓
+Candidate Evidence
+      ↓
+Rule Evaluation
+      ↓
+Finding (Violation)
+```
+
+## ALN-001 / MIT-003-A (LLM Output Without Validation)
+
+The primary driver for this feature is `ALN-001` (`LLM Output Without Validation`). Previously, any match on an LLM API call immediately raised a violation, producing high false positive rates for libraries, wrappers, and validated code.
+
+Under the new pipeline:
+1. LLM API calls are registered as candidate evidence.
+2. The specialized `ALN-001` / `MIT-003-A` evaluator scans the file for validation markers:
+   * **Validation Frameworks**: Imports of `pydantic`, `instructor`, `guardrails`, `marshmallow`, or `jsonschema`.
+   * **Class Structures**: Use of `BaseModel` schemas.
+   * **Validation API Calls**: `.validate()`, `.parse_obj()`, or `json.loads()`.
+   * **Explicit Markers**: The `# anchor: validate` instruction comment.
+3. If validation markers are present, the candidate is discarded, eliminating false positives for safe code.

anchor/core/engine.py

@@ -557,7 +557,53 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
                         "severity": rule.get("severity", "error")
                     })
 
-        return {"violations": violations, "suppressed": suppressed}
+        evaluated_violations = self.evaluate_candidates(violations, content, file_path)
+        return {"violations": evaluated_violations, "suppressed": suppressed}
+
+    def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes, file_path: str) -> List[Dict[str, Any]]:
+        """
+        Runs candidate findings through an evidence-based rule evaluation pipeline.
+        Promotes candidates to final violations only if conditions are met.
+        """
+        violations = []
+        try:
+            content_str = content.decode("utf-8", errors="ignore")
+        except Exception:
+            content_str = ""
+
+        for candidate in candidates:
+            rule_id = candidate.get("id", "")
+
+            # Specialized evaluator for ALN-001 / MIT-003-A (LLM Output Without Validation)
+            if "ALN-001" in rule_id or "MIT-003-A" in rule_id:
+                import re
+                # Check for common validation frameworks, helper functions, and custom schemas
+                validation_patterns = [
+                    r"\bimport\s+(pydantic|instructor|guardrails|marshmallow|jsonschema)\b",
+                    r"\bfrom\s+(pydantic|instructor|guardrails|marshmallow|jsonschema)\b",
+                    r"\bBaseModel\b",
+                    r"\.validate\s*\(",
+                    r"\.parse_obj\s*\(",
+                    r"\bjson\.loads\s*\(",
+                    r"\bvalidate_\w+",
+                    r"\bcheck_\w+",
+                    r"#\s*anchor:\s*validate\b"
+                ]
+
+                has_validation = False
+                for pat in validation_patterns:
+                    if re.search(pat, content_str):
+                        has_validation = True
+                        break
+
+                if has_validation:
+                    if self.verbose:
+                        import click
+                        click.secho(f"    🛡️ [EVALUATOR] ALN-001 candidate at line {candidate['line']} discarded (validation markers found).", fg="green", dim=True)
+                    continue
+
+            violations.append(candidate)
+        return violations
 
     def _execute_query(self, root_node, adapter: LanguageAdapter, s_expr: str) -> List[Dict]:
         """Standardized query execution returning all capture groups for verification."""

tests/unit/test_evaluation.py

@@ -0,0 +1,98 @@
+import pytest
+from anchor.core.engine import PolicyEngine
+from anchor.adapters.python import PythonAdapter
+
+def test_aln_001_without_validation():
+    # LLM API call without any validation markers
+    content = b"""
+import os
+
+def call_model():
+    client.chat.completions.create(
+        model="gpt-4",
+        messages=[{"role": "user", "content": "Hello"}]
+    )
+"""
+    rules = [
+        {
+            "id": "MIT-003-A",
+            "rule_id": "ALN-001",
+            "name": "LLM Output Without Validation",
+            "match": {
+                "type": "regex",
+                "pattern": r"\.(create|send)\s*\("
+            },
+            "message": "LLM API call detected. Ensure output is validated.",
+            "severity": "error"
+        }
+    ]
+    engine = PolicyEngine(config={"rules": rules})
+    adapter = PythonAdapter()
+
+    results = engine.scan_file(content, "test.py", adapter)
+    assert len(results["violations"]) == 1
+    assert "MIT-003-A" in results["violations"][0]["id"]
+
+def test_aln_001_with_validation_pydantic():
+    # LLM API call accompanied by pydantic import
+    content = b"""
+import os
+from pydantic import BaseModel
+
+class OutputSchema(BaseModel):
+    response: str
+
+def call_model():
+    client.chat.completions.create(
+        model="gpt-4",
+        messages=[{"role": "user", "content": "Hello"}]
+    )
+"""
+    rules = [
+        {
+            "id": "MIT-003-A",
+            "rule_id": "ALN-001",
+            "name": "LLM Output Without Validation",
+            "match": {
+                "type": "regex",
+                "pattern": r"\.(create|send)\s*\("
+            },
+            "message": "LLM API call detected. Ensure output is validated.",
+            "severity": "error"
+        }
+    ]
+    engine = PolicyEngine(config={"rules": rules})
+    adapter = PythonAdapter()
+
+    results = engine.scan_file(content, "test.py", adapter)
+    # The evaluation pipeline should discard this candidate because of BaseModel/pydantic
+    assert len(results["violations"]) == 0
+
+def test_aln_001_with_validation_comment():
+    # LLM API call with explicit validation comment
+    content = b"""
+def call_model():
+    # anchor: validate
+    client.chat.completions.create(
+        model="gpt-4",
+        messages=[{"role": "user", "content": "Hello"}]
+    )
+"""
+    rules = [
+        {
+            "id": "MIT-003-A",
+            "rule_id": "ALN-001",
+            "name": "LLM Output Without Validation",
+            "match": {
+                "type": "regex",
+                "pattern": r"\.(create|send)\s*\("
+            },
+            "message": "LLM API call detected. Ensure output is validated.",
+            "severity": "error"
+        }
+    ]
+    engine = PolicyEngine(config={"rules": rules})
+    adapter = PythonAdapter()
+
+    results = engine.scan_file(content, "test.py", adapter)
+    assert len(results["violations"]) == 0