Add Sentry observability to audit and eval pipelines

Breadcrumbs on capture-layer parse failures, manual AI spans for Ollama,
per-case spans with scoring data in eval runner, malformed-case warnings.
Zero-cost when SENTRY_DSN is unset. 38 tests pass, 1 xfailed.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

agents/semantic_proposer.py

@@ -6,6 +6,8 @@
 import urllib.request
 from typing import Callable
 
+import sentry_sdk
+
 from agents.semantic_confirmer import ALLOWED_RELATION_TYPES
 
 
@@ -83,6 +85,16 @@ def _json_from_text(text: str) -> dict:
     try:
         parsed = json.loads(_strip_code_fence(text))
     except json.JSONDecodeError as error:
+        sentry_sdk.add_breadcrumb(
+            category="capture_layer",
+            message="proposer output failed JSON parse",
+            level="error",
+            data={
+                "raw_length": len(text),
+                "raw_preview": text[:200],
+                "starts_with_fence": text.strip().startswith("```"),
+            },
+        )
         raise SemanticProposerError("proposer returned non-json output") from error
     if isinstance(parsed, list):
         parsed = {"proposals": parsed}
@@ -96,6 +108,12 @@ def _normalize_proposal(raw: dict) -> dict:
         raise SemanticProposerError("proposal must be an object")
     missing = PROPOSER_SCHEMA_FIELDS - set(raw)
     if missing:
+        sentry_sdk.add_breadcrumb(
+            category="capture_layer",
+            message=f"proposal missing required fields: {sorted(missing)}",
+            level="warning",
+            data={"proposal_keys": sorted(raw.keys())},
+        )
         raise SemanticProposerError(f"proposal missing required field(s): {sorted(missing)}")
     relation_type = str(raw["type"]).strip()
     if relation_type not in ALLOWED_RELATION_TYPES:
@@ -186,33 +204,38 @@ def call_anthropic(prompt: str) -> str:
 def call_local_llama(prompt: str) -> str:
     # HTTP API instead of `ollama run` subprocess: the CLI path corrupted
     # captured output with ANSI/spinner bytes (2026-07-01 eval attempt #1).
-    request = urllib.request.Request(
-        OLLAMA_API_URL,
-        data=json.dumps(
-            {
-                "model": OLLAMA_MODEL,
-                "prompt": prompt,
-                "stream": False,
-            }
-        ).encode("utf-8"),
-        headers={"content-type": "application/json"},
-        method="POST",
-    )
-    try:
-        with urllib.request.urlopen(request, timeout=120) as response:
-            payload = json.loads(response.read().decode("utf-8"))
-    except urllib.error.URLError as error:
-        raise SemanticProposerError(
-            f"local llama proposer call failed: network error: {str(error.reason)[:240]}"
-        ) from error
-    except TimeoutError as error:
-        raise SemanticProposerError("local llama proposer call failed: timeout") from error
-    except json.JSONDecodeError as error:
-        raise SemanticProposerError("local llama proposer call failed: invalid json response") from error
-    text = str(payload.get("response", "")).strip()
-    if not text:
-        raise SemanticProposerError("local llama proposer returned no text")
-    return text
+    with sentry_sdk.start_span(op="ai.chat_completions.create", name="ollama.generate") as span:
+        span.set_data("ai.model_id", OLLAMA_MODEL)
+        span.set_data("ai.provider", "ollama")
+        span.set_data("ai.input_messages", [{"role": "user", "content": prompt[:200] + "..."}])
+        request = urllib.request.Request(
+            OLLAMA_API_URL,
+            data=json.dumps(
+                {
+                    "model": OLLAMA_MODEL,
+                    "prompt": prompt,
+                    "stream": False,
+                }
+            ).encode("utf-8"),
+            headers={"content-type": "application/json"},
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=120) as response:
+                payload = json.loads(response.read().decode("utf-8"))
+        except urllib.error.URLError as error:
+            raise SemanticProposerError(
+                f"local llama proposer call failed: network error: {str(error.reason)[:240]}"
+            ) from error
+        except TimeoutError as error:
+            raise SemanticProposerError("local llama proposer call failed: timeout") from error
+        except json.JSONDecodeError as error:
+            raise SemanticProposerError("local llama proposer call failed: invalid json response") from error
+        text = str(payload.get("response", "")).strip()
+        if not text:
+            raise SemanticProposerError("local llama proposer returned no text")
+        span.set_data("ai.response_chars", len(text))
+        return text
 
 
 def propose_authority_changes(

audit_cli.py

@@ -1,19 +1,40 @@
 from __future__ import annotations
 
 import json
+import os
 import sys
 from pathlib import Path
 
+import sentry_sdk
+
 from audit_pipeline import run_audit
 
 
+def _init_sentry() -> None:
+    dsn = os.environ.get("SENTRY_DSN")
+    if not dsn:
+        return
+    sentry_sdk.init(
+        dsn=dsn,
+        traces_sample_rate=1.0,
+        send_default_pii=True,
+        enable_logs=True,
+        release=os.environ.get("SENTRY_RELEASE", "memory-authority-auditor@0.1.0"),
+        environment=os.environ.get("SENTRY_ENVIRONMENT", "development"),
+    )
+
+
 def main() -> int:
     if len(sys.argv) != 2:
         print("Usage: python3 audit_cli.py <memory-file>", file=sys.stderr)
         return 2
 
+    _init_sentry()
+
     path = Path(sys.argv[1])
-    result = run_audit(path.read_text(encoding="utf-8"))
+    with sentry_sdk.start_transaction(op="audit.run", name="audit pipeline") as txn:
+        txn.set_data("input_file", str(path))
+        result = run_audit(path.read_text(encoding="utf-8"))
     print(json.dumps(result, indent=2))
     return 0
 

audit_pipeline.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import sentry_sdk
+
 from agents.memory_extractor import extract_memories
 from agents.authority_classifier import classify_items
 from agents.conflict_detector import detect_conflicts
@@ -8,6 +10,7 @@
 from agents.authority_mapper import authority_map
 
 
+@sentry_sdk.trace
 def run_audit(text: str) -> dict:
     items = extract_memories(text)
     item_dicts = [item.to_dict() for item in items]

path_a_eval_runner.py

@@ -2,10 +2,13 @@
 
 import argparse
 import json
+import os
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Callable
 
+import sentry_sdk
+
 from agents.memory_extractor import extract_memories
 from agents.semantic_confirmer import ALLOWED_RELATION_TYPES, CONFIDENCE_THRESHOLD, confirm_authority_changes
 from agents.semantic_proposer import (
@@ -233,51 +236,72 @@ def run_engine(engine: str, cases: list[dict]) -> dict:
     remove_citation_false_fires = 0
 
     for case in cases:
-        items = _items(case)
-        prompt = build_authority_change_prompt(items)
-        raw_output = ""
-        malformed_reasons: list[str] = []
-        try:
-            raw_output = provider(prompt)
-            proposals, malformed_reasons = _parse_case_output(raw_output)
-        except Exception as error:  # per-case malformed/degraded bucket, never abort the run
-            proposals = []
-            malformed_reasons = [f"{type(error).__name__}: {error}"]
-
-        if malformed_reasons:
-            malformed_count += 1
-
-        confirmed = confirm_authority_changes(proposals, items)
-        score = _score_confirmed(case, confirmed)
-        no_confirmer = _score_remove_confirmer(case, proposals)
-        no_citation = _score_confirmed(case, _confirm_without_citation(proposals, items))
-        lexical = _lexical_result(case)
-
-        if score["expected_positive"]:
-            positives += 1
-            positives_caught += int(score["caught"])
-            positives_caught_direction += int(score["direction_caught"])
-        else:
-            negatives += 1
-            negatives_passed += int(score["negative_passed"])
-            remove_confirmer_false_fires += int(no_confirmer["would_false_fire_without_confirmation"])
-            remove_citation_false_fires += int(not no_citation["negative_passed"])
-
-        case_results.append({
-            "case_id": case["id"],
-            "class": case["class"],
-            "malformed": bool(malformed_reasons),
-            "malformed_reasons": malformed_reasons,
-            "proposal_count": len(proposals),
-            "proposals": proposals,
-            "confirmed_findings": confirmed["findings"],
-            "needs_human_judgment": confirmed["needs_human_judgment"],
-            "score": score,
-            "ablation_remove_confirmer": no_confirmer,
-            "ablation_remove_citation_requirement": no_citation,
-            "lexical_baseline_current_detector": lexical,
-            "raw_output": raw_output,
-        })
+        with sentry_sdk.start_span(op="eval.case", name=f"case:{case['id']}") as case_span:
+            case_span.set_data("case_id", case["id"])
+            case_span.set_data("case_class", case["class"])
+            case_span.set_data("engine", engine)
+            items = _items(case)
+            prompt = build_authority_change_prompt(items)
+            raw_output = ""
+            malformed_reasons: list[str] = []
+            try:
+                raw_output = provider(prompt)
+                proposals, malformed_reasons = _parse_case_output(raw_output)
+            except Exception as error:  # per-case malformed/degraded bucket, never abort the run
+                proposals = []
+                malformed_reasons = [f"{type(error).__name__}: {error}"]
+                sentry_sdk.capture_exception(error)
+
+            if malformed_reasons:
+                malformed_count += 1
+                case_span.set_data("malformed", True)
+                case_span.set_data("malformed_reasons", malformed_reasons)
+                sentry_sdk.add_breadcrumb(
+                    category="eval",
+                    message=f"proposer output malformed: {case['id']}",
+                    level="warning",
+                    data={
+                        "case_id": case["id"],
+                        "engine": engine,
+                        "reasons": malformed_reasons,
+                    },
+                )
+
+            confirmed = confirm_authority_changes(proposals, items)
+            score = _score_confirmed(case, confirmed)
+            no_confirmer = _score_remove_confirmer(case, proposals)
+            no_citation = _score_confirmed(case, _confirm_without_citation(proposals, items))
+            lexical = _lexical_result(case)
+
+            if score["expected_positive"]:
+                positives += 1
+                positives_caught += int(score["caught"])
+                positives_caught_direction += int(score["direction_caught"])
+            else:
+                negatives += 1
+                negatives_passed += int(score["negative_passed"])
+                remove_confirmer_false_fires += int(no_confirmer["would_false_fire_without_confirmation"])
+                remove_citation_false_fires += int(not no_citation["negative_passed"])
+
+            case_span.set_data("score", score)
+            case_span.set_data("proposal_count", len(proposals))
+            case_span.set_data("confirmed_count", len(confirmed["findings"]))
+
+            case_results.append({
+                "case_id": case["id"],
+                "class": case["class"],
+                "malformed": bool(malformed_reasons),
+                "malformed_reasons": malformed_reasons,
+                "proposal_count": len(proposals),
+                "proposals": proposals,
+                "confirmed_findings": confirmed["findings"],
+                "needs_human_judgment": confirmed["needs_human_judgment"],
+                "score": score,
+                "ablation_remove_confirmer": no_confirmer,
+                "ablation_remove_citation_requirement": no_citation,
+                "lexical_baseline_current_detector": lexical,
+                "raw_output": raw_output,
+            })
 
     return {
         "engine": engine,
@@ -347,27 +371,65 @@ def _write_markdown(result: dict, path: Path) -> None:
     path.write_text("\n".join(lines), encoding="utf-8")
 
 
+def _init_sentry() -> None:
+    dsn = os.environ.get("SENTRY_DSN")
+    if not dsn:
+        return
+    sentry_sdk.init(
+        dsn=dsn,
+        traces_sample_rate=1.0,
+        send_default_pii=True,
+        enable_logs=True,
+        release=os.environ.get("SENTRY_RELEASE", "memory-authority-auditor@0.1.0"),
+        environment=os.environ.get("SENTRY_ENVIRONMENT", "development"),
+    )
+
+
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--engines", default="anthropic,local_llama3.2")
     parser.add_argument("--output-dir", default=str(ARTIFACT_DIR))
     parser.add_argument("--fixture", default=str(FIXTURE))
     args = parser.parse_args()
 
+    _init_sentry()
+
     run_id = _utc_slug()
     output_dir = Path(args.output_dir)
     output_dir.mkdir(parents=True, exist_ok=True)
     fixture_path = Path(args.fixture)
     cases = _cases(fixture_path)
     engines = [engine.strip() for engine in args.engines.split(",") if engine.strip()]
-    result = {
-        "run_id": run_id,
-        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
-        "fixture": str(fixture_path),
-        "scoring_rules": SCORING_RULES,
-        "engines": [run_engine(engine, cases) for engine in engines],
-        "boundary": "No public claim from this artifact until Ka'el and Fable re-verify.",
-    }
+
+    with sentry_sdk.start_transaction(op="eval.run", name="path_a_eval") as txn:
+        txn.set_data("run_id", run_id)
+        txn.set_data("fixture", str(fixture_path))
+        txn.set_data("engines", engines)
+        txn.set_data("case_count", len(cases))
+
+        engine_results = []
+        for engine in engines:
+            with sentry_sdk.start_span(op="eval.engine", name=f"engine:{engine}") as eng_span:
+                eng_span.set_data("engine", engine)
+                eng_result = run_engine(engine, cases)
+                eng_span.set_data("summary", eng_result["summary"])
+                malformed = eng_result["summary"]["malformed_cases"]
+                if malformed > 0:
+                    sentry_sdk.capture_message(
+                        f"engine {engine} completed with {malformed}/{len(cases)} malformed cases",
+                        level="warning",
+                    )
+                engine_results.append(eng_result)
+
+        result = {
+            "run_id": run_id,
+            "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+            "fixture": str(fixture_path),
+            "scoring_rules": SCORING_RULES,
+            "engines": engine_results,
+            "boundary": "No public claim from this artifact until Ka'el and Fable re-verify.",
+        }
+
     json_path = output_dir / f"path_a_eval_{run_id}.json"
     md_path = output_dir / f"path_a_eval_{run_id}.md"
     json_path.write_text(json.dumps(result, indent=2), encoding="utf-8")

requirements.txt

@@ -0,0 +1 @@
+sentry-sdk[anthropic]>=2.0