Expose detector assessment and abstention coverage

README.md

@@ -125,6 +125,19 @@ for issue in result.issues:
 
 ## CLI
 
+`AnalyzeResult.detector_assessments` and JSON CLI output preserve explicit
+detector coverage. `abstained` is not a passed check; older core versions that
+provide no assessment are marked `unknown`. Terminal output separates explicit
+contract passes, abstentions, errors and unspecified coverage. Detector execution
+counts are not counts of validated requirements. No findings is not proof of task
+success. Coverage metadata does not turn heuristic confidence into calibration.
+`check --json` schema 3 includes per-file assessments and coverage completeness.
+`passed` remains a severity-threshold gate, not complete validation; explicit
+detector errors fail even with `--fail-on never`. Unknown coverage and abstention
+are not detector errors but are never counted as checked-clean files. Replay
+comparisons do not label disappeared findings fixed: current assessment metadata
+lacks comparable contract/input provenance, even when a later check passes.
+
 ```bash
 pisama analyze trace.json          # Analyze a trace
 pisama watch python my_agent.py    # Watch a live agent (pip install "pisama[auto]")

src/pisama/_analyze.py

@@ -5,7 +5,7 @@
 import asyncio
 import threading
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Optional, Sequence, Union
 
 from pisama_core.traces.models import Trace
@@ -53,6 +53,22 @@ class AnalyzeResult:
     trace_id: str
     detectors_run: int
     execution_time_ms: float
+    detector_assessments: list[dict[str, Any]] = field(default_factory=list)
+
+    @property
+    def has_detector_errors(self) -> bool:
+        return any(item.get("assessment") == "error" for item in self.detector_assessments)
+
+    @property
+    def coverage_complete(self) -> bool:
+        return (
+            self.detectors_run > 0
+            and len(self.detector_assessments) == self.detectors_run
+            and all(
+                item.get("assessment") in {"contract_satisfied", "contract_violated", "finding"}
+                for item in self.detector_assessments
+            )
+        )
 
     @property
     def has_issues(self) -> bool:
@@ -159,9 +175,49 @@ async def async_analyze(
         trace_id=trace.trace_id,
         detectors_run=analysis.total_detectors_run,
         execution_time_ms=elapsed_ms,
+        detector_assessments=_convert_assessments(analysis),
     )
 
 
+def _convert_assessments(analysis: Any) -> list[dict[str, Any]]:
+    """Preserve explicit coverage without treating legacy silence as success."""
+    assessments = []
+    for result in analysis.detection_results:
+        metadata = result.metadata if isinstance(result.metadata, dict) else {}
+        assessment = metadata.get("assessment")
+        if "error" in metadata:
+            assessment = "error"
+        elif not isinstance(assessment, str) or assessment not in {
+            "abstained",
+            "contract_satisfied",
+            "contract_violated",
+        }:
+            assessment = "finding" if result.detected else "unknown"
+        elif result.detected and assessment != "contract_violated":
+            assessment = "finding"
+        elif not result.detected and assessment == "contract_violated":
+            assessment = "unknown"
+        checked = metadata.get("checked_contracts")
+        if assessment == "contract_satisfied" and (type(checked) is not int or checked < 1):
+            assessment = "unknown"
+        elif (
+            assessment == "abstained"
+            and checked is not None
+            and (type(checked) is not int or checked != 0)
+        ):
+            assessment = "unknown"
+        item = {"detector_name": result.detector_name, "assessment": assessment}
+        # Expose coverage provenance, not arbitrary metadata/error strings that
+        # could contain captured input or credentials.
+        if type(checked) is int and checked >= 0:
+            item["checked_contracts"] = checked
+        basis = metadata.get("confidence_basis")
+        if isinstance(basis, str) and basis == "uncalibrated contract heuristic":
+            item["confidence_basis"] = basis
+        assessments.append(item)
+    return assessments
+
+
 def _convert_issues(
     analysis: Any,
 ) -> list[Issue]:

src/pisama/cli/analyze_cmd.py

@@ -79,8 +79,8 @@ def analyze_cmd(path: str, min_severity: int, output_json: bool, scrub: bool) ->
     else:
         display_analysis_result(result)
 
-    # Exit code: 1 if critical issues found, 0 otherwise
-    if result.critical_issues:
+    # Explicit detector errors fail independently of finding severity.
+    if result.critical_issues or result.has_detector_errors:
         sys.exit(1)
 
 

src/pisama/cli/check_cmd.py

@@ -26,7 +26,7 @@
 
 #: Version of the --json payload shape. Bump when the structure changes so
 #: CI consumers can pin against it.
-JSON_SCHEMA_VERSION = 2
+JSON_SCHEMA_VERSION = 3
 
 SKIP_DIRS = {".git", ".venv", "venv", "node_modules", "__pycache__", ".next", "dist", "build"}
 TRACE_SUFFIXES = {".json", ".jsonl"}
@@ -50,6 +50,8 @@ class _CheckReport:
     results: list[dict[str, object]] = field(default_factory=list)
     files_analyzed: int = 0
     files_clean: int = 0
+    files_no_findings: int = 0
+    files_incomplete: int = 0
     files_with_issues: int = 0
     files_failed: int = 0
     analysis_errors: int = 0
@@ -101,23 +103,36 @@ def record_discovery_error(self, path: Path, exc: ValueError) -> None:
     def record_result(self, path: Path, result: AnalyzeResult) -> None:
         """Record one successful analysis, including a clean result."""
         triggered = [issue for issue in result.issues if issue.severity >= self.severity_threshold]
-        failed_threshold = bool(triggered)
+        failed_threshold = bool(triggered) or result.has_detector_errors
         self.files_analyzed += 1
         self.issues_total += len(result.issues)
         self.issues_at_or_above_threshold += len(triggered)
         self.files_with_issues += int(result.has_issues)
-        self.files_clean += int(not result.has_issues)
+        self.files_clean += int(not result.has_issues and result.coverage_complete)
+        self.files_no_findings += int(not result.has_issues)
+        self.files_incomplete += int(not result.coverage_complete)
+        self.analysis_errors += int(result.has_detector_errors)
         self.files_failed += int(failed_threshold)
         self.failed = self.failed or failed_threshold
         self.results.append(
             {
                 "file": str(path),
-                "status": "issues" if result.has_issues else "clean",
+                "status": (
+                    "detector_error"
+                    if result.has_detector_errors
+                    else "issues"
+                    if result.has_issues
+                    else "clean"
+                    if result.coverage_complete
+                    else "unassessed"
+                ),
                 "failed": failed_threshold,
                 "trace_id": result.trace_id,
                 "detectors_run": result.detectors_run,
                 "execution_time_ms": result.execution_time_ms,
                 "issues": [asdict(issue) for issue in result.issues],
+                "detector_assessments": result.detector_assessments,
+                "coverage_complete": result.coverage_complete,
                 "error": None,
             }
         )
@@ -128,6 +143,8 @@ def payload(self) -> dict[str, object]:
             "files_total": self.files_total,
             "files_analyzed": self.files_analyzed,
             "files_clean": self.files_clean,
+            "files_no_findings": self.files_no_findings,
+            "files_incomplete": self.files_incomplete,
             "files_with_issues": self.files_with_issues,
             "files_failed": self.files_failed,
             "analysis_errors": self.analysis_errors,
@@ -138,6 +155,7 @@ def payload(self) -> dict[str, object]:
             "fail_on": self.fail_on,
             "severity_threshold": self.severity_threshold,
             "passed": not self.failed,
+            "pass_basis": "severity threshold and no detector errors, not complete task validation",
         }
         return {
             "schema_version": JSON_SCHEMA_VERSION,
@@ -416,10 +434,15 @@ def _render_file_result(
     """Render one human-readable result when JSON or quiet mode is not active."""
     if quiet or output_json:
         return
+    if result.has_detector_errors:
+        console.print(f"[red]Incomplete[/red] {trace_path}: detector error")
     if result.has_issues:
         _print_file_report(trace_path, result, threshold)
     else:
-        console.print(f"[green]OK[/green] {trace_path.name}: clean")
+        console.print(
+            f"{trace_path.name}: no findings; "
+            f"coverage {'complete' if result.coverage_complete else 'incomplete/unspecified'}"
+        )
 
 
 def _print_file_report(trace_path: Path, result: AnalyzeResult, threshold: int) -> None:

src/pisama/cli/replay.py

@@ -100,7 +100,7 @@ async def _replay_async(
 
             display_analysis_result(result_a)
 
-        if result_a.critical_issues:
+        if result_a.critical_issues or result_a.has_detector_errors:
             sys.exit(1)
         return
 
@@ -138,6 +138,11 @@ async def _replay_async(
             regressed=comparison.regressed,
             unchanged=comparison.unchanged,
         )
+        if comparison.unassessed:
+            console.print(
+                "Disappeared findings without checked-pass evidence: "
+                + ", ".join(comparison.unassessed)
+            )
 
-    if comparison.has_regressions:
+    if comparison.has_regressions or result_a.has_detector_errors or result_b.has_detector_errors:
         sys.exit(1)

src/pisama/cli/smoke.py

@@ -120,5 +120,9 @@ async def _smoke_async(
         for err in result.errors[:5]:
             console.print(f"  [dim]{err}[/dim]")
 
-    if fail_on_regression and result.critical_traces:
+    console.print(
+        "Detector execution does not imply complete validation; "
+        "per-trace assessment coverage is available in --json."
+    )
+    if result.errors or (fail_on_regression and result.critical_traces):
         sys.exit(1)

src/pisama/cli/watch.py

@@ -133,14 +133,20 @@ def on_trace(trace: object) -> None:
     traces = collector.get_traces()
     collector.stop()
 
+    detection_failed = False
     if traces:
         console.print(f"\n[dim]Running detection on {len(traces)} trace(s)...[/dim]")
         for trace in traces:
             try:
                 result = asyncio.run(async_analyze(trace))
+                detection_failed = detection_failed or result.has_detector_errors
+                from pisama.output.terminal import display_analysis_result
+
+                display_analysis_result(result)
                 for issue in result.issues:
                     display.add_issue(issue)
             except Exception as exc:
+                detection_failed = True
                 console.print(
                     f"[yellow]Warning:[/yellow] Detection failed for "
                     f"trace {trace.trace_id[:12]}: {exc}"
@@ -152,6 +158,6 @@ def on_trace(trace: object) -> None:
     # Exit with code from subprocess (or 1 if critical issues)
     exit_code = proc.returncode if proc.returncode else 0
     critical = [i for i in display._issues if i.severity >= 60]
-    if critical and exit_code == 0:
+    if (critical or detection_failed) and exit_code == 0:
         exit_code = 1
     sys.exit(exit_code)

src/pisama/output/terminal.py

@@ -53,8 +53,8 @@ def display_analysis_result(result: AnalyzeResult) -> None:
         if critical_count:
             header_text += f" ({critical_count} critical)"
     else:
-        header_style = "green bold"
-        header_text = "No issues detected"
+        header_style = "bold"
+        header_text = "No findings reported (not proof of task success)"
 
     console.print()
     console.print(
@@ -67,6 +67,20 @@ def display_analysis_result(result: AnalyzeResult) -> None:
         )
     )
 
+    counts: dict[str, int] = {}
+    for assessment in result.detector_assessments:
+        status = assessment.get("assessment", "unknown")
+        counts[status] = counts.get(status, 0) + 1
+    unknown = counts.get("unknown", 0) + max(
+        0, result.detectors_run - len(result.detector_assessments)
+    )
+    console.print(
+        "Coverage: "
+        f"{counts.get('contract_satisfied', 0)} explicit contract passes; "
+        f"{counts.get('abstained', 0)} abstained; "
+        f"{counts.get('error', 0)} errors; {unknown} unspecified."
+    )
+
     if not result.issues:
         return
 

src/pisama/replay/comparator.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from typing import Any
 
 from pisama._analyze import AnalyzeResult
 
@@ -24,6 +25,9 @@ class ComparisonResult:
     improved: list[str] = field(default_factory=list)
     regressed: list[str] = field(default_factory=list)
     unchanged: list[str] = field(default_factory=list)
+    unassessed: list[str] = field(default_factory=list)
+    assessments_a: list[dict[str, Any]] = field(default_factory=list)
+    assessments_b: list[dict[str, Any]] = field(default_factory=list)
 
     @property
     def has_regressions(self) -> bool:
@@ -55,15 +59,19 @@ def compare(cls, a: AnalyzeResult, b: AnalyzeResult) -> "ComparisonResult":
         result = cls(
             trace_a_id=a.trace_id,
             trace_b_id=b.trace_id,
+            assessments_a=a.detector_assessments,
+            assessments_b=b.detector_assessments,
         )
 
         for det in all_detectors:
             a_sev = a_map.get(det, 0)
             b_sev = b_map.get(det, 0)
 
             if a_sev > 0 and b_sev == 0:
-                # Was detected, now clear
-                result.fixed.append(det)
+                # Current assessments lack contract/input identity. A pass on
+                # a different request cannot establish that the prior failure
+                # was fixed, even if the detector name matches.
+                result.unassessed.append(det)
             elif a_sev > b_sev and b_sev > 0:
                 # Severity decreased
                 result.improved.append(det)

src/pisama/replay/smoke_runner.py

@@ -43,6 +43,7 @@ class SmokeTestResult:
     per_detector_stats: dict[str, DetectorStats] = field(default_factory=dict)
     critical_traces: list[str] = field(default_factory=list)
     errors: list[str] = field(default_factory=list)
+    detector_assessments: list[dict[str, Any]] = field(default_factory=list)
 
     def to_dict(self) -> dict[str, Any]:
         """Convert to dict for JSON serialization."""
@@ -52,6 +53,7 @@ def to_dict(self) -> dict[str, Any]:
             "per_detector_stats": {k: v.to_dict() for k, v in self.per_detector_stats.items()},
             "critical_traces": self.critical_traces,
             "errors": self.errors,
+            "detector_assessments": self.detector_assessments,
         }
 
 
@@ -76,11 +78,17 @@ async def run(
 
         for trace in traces:
             try:
-                analysis = await async_analyze(trace)
+                analysis = await async_analyze(trace, detectors=detectors)
             except Exception as exc:
                 result.errors.append(f"trace {trace.trace_id[:12]}: {exc}")
                 continue
 
+            result.detector_assessments.append(
+                {"trace_id": trace.trace_id, "assessments": analysis.detector_assessments}
+            )
+            if analysis.has_detector_errors:
+                result.errors.append(f"trace {trace.trace_id[:12]}: detector execution failed")
+
             # Filter by requested detectors if specified
             issues = analysis.issues
             if detectors: