diff --git a/README.md b/README.md index 3d4cd99..6b42fdd 100644 --- a/README.md +++ b/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]") diff --git a/src/pisama/_analyze.py b/src/pisama/_analyze.py index 639b6ee..2d5a074 100644 --- a/src/pisama/_analyze.py +++ b/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]: diff --git a/src/pisama/cli/analyze_cmd.py b/src/pisama/cli/analyze_cmd.py index 0c8fcdc..53fcd3d 100644 --- a/src/pisama/cli/analyze_cmd.py +++ b/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) diff --git a/src/pisama/cli/check_cmd.py b/src/pisama/cli/check_cmd.py index ebe2965..0e92d5f 100644 --- a/src/pisama/cli/check_cmd.py +++ b/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: diff --git a/src/pisama/cli/replay.py b/src/pisama/cli/replay.py index be44364..e6d25b9 100644 --- a/src/pisama/cli/replay.py +++ b/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) diff --git a/src/pisama/cli/smoke.py b/src/pisama/cli/smoke.py index 93e9bff..fba127d 100644 --- a/src/pisama/cli/smoke.py +++ b/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) diff --git a/src/pisama/cli/watch.py b/src/pisama/cli/watch.py index cc794ed..93ee879 100644 --- a/src/pisama/cli/watch.py +++ b/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) diff --git a/src/pisama/output/terminal.py b/src/pisama/output/terminal.py index e03f5d5..313488e 100644 --- a/src/pisama/output/terminal.py +++ b/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 diff --git a/src/pisama/replay/comparator.py b/src/pisama/replay/comparator.py index b0b30bd..e8f9c6f 100644 --- a/src/pisama/replay/comparator.py +++ b/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,6 +59,8 @@ 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: @@ -62,8 +68,10 @@ def compare(cls, a: AnalyzeResult, b: AnalyzeResult) -> "ComparisonResult": 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) diff --git a/src/pisama/replay/smoke_runner.py b/src/pisama/replay/smoke_runner.py index 25af87a..d046e72 100644 --- a/src/pisama/replay/smoke_runner.py +++ b/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: diff --git a/tests/test_assessment_coverage.py b/tests/test_assessment_coverage.py new file mode 100644 index 0000000..aba4684 --- /dev/null +++ b/tests/test_assessment_coverage.py @@ -0,0 +1,112 @@ +"""Real result serialization and consumer output coverage (no mocked detectors).""" + +import json +import os +import subprocess +import sys +from dataclasses import asdict +from pathlib import Path + +import pytest +from pisama_core.detection.orchestrator import AnalysisResult +from pisama_core.detection.result import DetectionResult +from pisama_core.traces.enums import Platform + +from pisama import analyze +from pisama._analyze import _convert_assessments + + +def test_explicit_metadata_and_legacy_unknown_preserved(): + outcomes = [ + DetectionResult("a", metadata={"assessment": "abstained", "checked_contracts": 0}), + DetectionResult("b", metadata={"assessment": "contract_satisfied", "checked_contracts": 1}), + DetectionResult("c", metadata={"error": "private input must not leak"}), + DetectionResult("d"), + ] + result = _convert_assessments( + AnalysisResult(trace_id="synthetic", platform=Platform.GENERIC, detection_results=outcomes) + ) + assert [row["assessment"] for row in result] == [ + "abstained", + "contract_satisfied", + "error", + "unknown", + ] + assert "private input" not in json.dumps(result) + + +@pytest.mark.parametrize("assessment", ["contract_satisfied", "abstained"]) +def test_findings_cannot_be_classified_as_pass_or_abstention(assessment): + result = DetectionResult("synthetic", detected=True, metadata={"assessment": assessment}) + converted = _convert_assessments( + AnalysisResult(trace_id="synthetic", platform=Platform.GENERIC, detection_results=[result]) + ) + assert converted[0]["assessment"] == "finding" + + +@pytest.mark.parametrize( + "checked", [{"marker": "SYNTHETIC_PRIVATE"}, "SYNTHETIC_PRIVATE", True, -1] +) +def test_metadata_values_are_typed_and_allowlisted(checked): + result = DetectionResult( + "synthetic", + metadata={ + "assessment": "abstained", + "checked_contracts": checked, + "confidence_basis": "SYNTHETIC_PRIVATE", + }, + ) + converted = _convert_assessments( + AnalysisResult(trace_id="synthetic", platform=Platform.GENERIC, detection_results=[result]) + ) + assert "checked_contracts" not in converted[0] + assert "confidence_basis" not in converted[0] + assert "SYNTHETIC_PRIVATE" not in json.dumps(converted) + + +def test_undetected_violation_is_unknown(): + result = DetectionResult("synthetic", metadata={"assessment": "contract_violated"}) + converted = _convert_assessments( + AnalysisResult(trace_id="synthetic", platform=Platform.GENERIC, detection_results=[result]) + ) + assert converted[0]["assessment"] == "unknown" + + +def test_real_core_results_have_transparent_coverage(tmp_path): + trace = { + "spans": [ + { + "kind": "llm", + "input_data": {"prompt": "Explain JSON."}, + "output_data": {"content": "A data format."}, + } + ] + } + result = analyze(trace, detectors=["communication"]) + records = asdict(result)["detector_assessments"] + assert len(records) == 1 + # Released legacy core has no assessment; candidate core explicitly abstains. + assert records[0]["assessment"] in {"unknown", "abstained"} + path = tmp_path / "trace.json" + path.write_text(json.dumps(trace)) + env = { + **os.environ, + "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src") + + os.pathsep + + os.environ.get("PYTHONPATH", ""), + } + command = [ + sys.executable, + "-c", + "from pisama.cli.main import main; main()", + "analyze", + str(path), + ] + rendered = subprocess.run(command, capture_output=True, text=True, env=env, timeout=15) + assert rendered.returncode == 0 + assert "Coverage:" in rendered.stdout + assert "unspecified" in rendered.stdout + serialized = subprocess.run( + command + ["--json"], capture_output=True, text=True, env=env, timeout=15 + ) + assert "detector_assessments" in json.loads(serialized.stdout) diff --git a/tests/test_detector_error_exit.py b/tests/test_detector_error_exit.py new file mode 100644 index 0000000..9c24551 --- /dev/null +++ b/tests/test_detector_error_exit.py @@ -0,0 +1,110 @@ +"""Exercise a real raising detector through the real orchestrator and CLI.""" + +import asyncio +import json + +from click.testing import CliRunner +from pisama_core.detection.base import BaseDetector +from pisama_core.detection.registry import registry + +from pisama.cli.main import main + + +class RaisingDetector(BaseDetector): + name = "synthetic_error_probe" + description = "Test an actual detector exception" + platforms = [] + + async def detect(self, trace): + raise RuntimeError("synthetic detector failure") + + +def test_detector_errors_fail_analyze_and_never_check(tmp_path): + path = tmp_path / "trace.json" + path.write_text(json.dumps({"spans": [{"name": "synthetic"}]})) + registry.register(RaisingDetector()) + try: + analyzed = CliRunner().invoke(main, ["analyze", str(path), "--json"]) + assert analyzed.exit_code == 1, analyzed.output + assert any( + item["assessment"] == "error" + for item in json.loads(analyzed.output)["detector_assessments"] + ) + checked = CliRunner().invoke(main, ["check", str(path), "--json", "--fail-on", "never"]) + assert checked.exit_code == 1, checked.output + payload = json.loads(checked.output) + assert payload["summary"]["passed"] is False + assert payload["summary"]["files_clean"] == 0 + assert payload["results"][0]["status"] == "detector_error" + from pisama._loader import load_trace + from pisama.replay.smoke_runner import SmokeRunner + + smoke = asyncio.run( + SmokeRunner().run([load_trace(str(path))], detectors=[RaisingDetector.name]) + ) + assert smoke.errors + assert smoke.to_dict()["detector_assessments"][0]["assessments"][0]["assessment"] == "error" + finally: + registry.unregister(RaisingDetector.name) + + +def test_legacy_unknown_is_not_clean_but_preserves_threshold_pass(tmp_path): + path = tmp_path / "trace.json" + path.write_text(json.dumps({"spans": [{"name": "synthetic"}]})) + checked = CliRunner().invoke(main, ["check", str(path), "--json", "--detectors", "context"]) + assert checked.exit_code == 0 + payload = json.loads(checked.output) + assert payload["summary"]["passed"] is True + assert payload["summary"]["files_clean"] == 0 + assert payload["results"][0]["status"] == "unassessed" + assert payload["results"][0]["detector_assessments"][0]["assessment"] == "unknown" + + +def test_replay_disappearance_without_coverage_not_fixed(): + from pisama._analyze import AnalyzeResult, Issue + from pisama.replay.comparator import ComparisonResult + + prior = AnalyzeResult([Issue("synthetic", "failure", 75, 0.9, [], None)], "a", 1, 1) + after = AnalyzeResult( + [], "b", 1, 1, [{"detector_name": "synthetic", "assessment": "abstained"}] + ) + comparison = ComparisonResult.compare(prior, after) + assert comparison.fixed == [] + assert comparison.unassessed == ["synthetic"] + assert comparison.assessments_b == after.detector_assessments + + +def test_different_contract_pass_is_not_proof_of_prior_fix(): + from pisama._analyze import AnalyzeResult, Issue + from pisama.replay.comparator import ComparisonResult + + before = AnalyzeResult( + [Issue("communication", "Expected OK, got NO", 55, 0.9, [], None)], + "literal-run", + 1, + 1, + [ + { + "detector_name": "communication", + "assessment": "contract_violated", + "checked_contracts": 1, + } + ], + ) + after = AnalyzeResult( + [], + "different-json-run", + 1, + 1, + [ + { + "detector_name": "communication", + "assessment": "contract_satisfied", + "checked_contracts": 1, + } + ], + ) + result = ComparisonResult.compare(before, after) + assert result.fixed == [] + assert not result.has_improvements + assert result.unassessed == ["communication"] diff --git a/tests/test_public_trace_workflows.py b/tests/test_public_trace_workflows.py index 27d346c..4d8d9ce 100644 --- a/tests/test_public_trace_workflows.py +++ b/tests/test_public_trace_workflows.py @@ -177,7 +177,7 @@ def test_cli_check_and_detector_inventory_cover_the_ci_user_journey( ) assert check.exit_code == 0, check.output payload = json.loads(check.output) - assert payload["schema_version"] == 2 + assert payload["schema_version"] == 3 assert payload["summary"]["files_total"] == 1 assert payload["summary"]["files_analyzed"] == 1 assert payload["summary"]["passed"] is True @@ -207,8 +207,9 @@ async def test_batch_smoke_comparison_and_terminal_views_use_real_results( short_result = await async_analyze(short_trace) comparison = ComparisonResult.compare(full_result, short_result) - assert comparison.has_improvements - assert {"context", "communication"}.intersection(comparison.fixed) + # Removing most of a trace is not evidence that its failures were fixed. + assert not comparison.has_improvements + assert {"context", "communication"}.intersection(comparison.unassessed) reverse = ComparisonResult.compare(short_result, full_result) assert reverse.has_regressions