From bf7b0f138d5d27694d2a119acd53641190ccf87a Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:02:51 -0700 Subject: [PATCH 1/4] feat: expose detector abstention and assessment coverage --- README.md | 7 +++ src/pisama/_analyze.py | 24 +++++++++- src/pisama/output/terminal.py | 18 +++++++- tests/test_assessment_coverage.py | 74 +++++++++++++++++++++++++++++++ 4 files changed, 120 insertions(+), 3 deletions(-) create mode 100644 tests/test_assessment_coverage.py diff --git a/README.md b/README.md index 3d4cd99..e4151ee 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,13 @@ 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. + ```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..9a97b9e 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,7 @@ 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_issues(self) -> bool: @@ -159,9 +160,30 @@ 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 or {} + assessment = metadata.get("assessment") + if "error" in metadata: + assessment = "error" + elif assessment not in {"abstained", "contract_satisfied", "contract_violated"}: + assessment = "finding" if result.detected else "unknown" + item = {"detector_name": result.detector_name, "assessment": assessment} + # Expose coverage provenance, not arbitrary metadata/error strings that + # could contain captured input or credentials. + for key in ("checked_contracts", "confidence_basis"): + if key in metadata: + item[key] = metadata[key] + assessments.append(item) + return assessments + + def _convert_issues( analysis: Any, ) -> list[Issue]: 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/tests/test_assessment_coverage.py b/tests/test_assessment_coverage.py new file mode 100644 index 0000000..a06a21e --- /dev/null +++ b/tests/test_assessment_coverage.py @@ -0,0 +1,74 @@ +"""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 + +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) + + +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) From e61bbd9780c4ae98719cf9533b578bd21d08809f Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:05:27 -0700 Subject: [PATCH 2/4] fix: validate assessment metadata and reject contradictory coverage --- src/pisama/_analyze.py | 29 +++++++++++++++++++---- tests/test_assessment_coverage.py | 38 +++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 5 deletions(-) diff --git a/src/pisama/_analyze.py b/src/pisama/_analyze.py index 9a97b9e..ff09166 100644 --- a/src/pisama/_analyze.py +++ b/src/pisama/_analyze.py @@ -168,18 +168,37 @@ 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 or {} + metadata = result.metadata if isinstance(result.metadata, dict) else {} assessment = metadata.get("assessment") if "error" in metadata: assessment = "error" - elif assessment not in {"abstained", "contract_satisfied", "contract_violated"}: + 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. - for key in ("checked_contracts", "confidence_basis"): - if key in metadata: - item[key] = metadata[key] + 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 diff --git a/tests/test_assessment_coverage.py b/tests/test_assessment_coverage.py index a06a21e..aba4684 100644 --- a/tests/test_assessment_coverage.py +++ b/tests/test_assessment_coverage.py @@ -7,6 +7,7 @@ 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 @@ -34,6 +35,43 @@ def test_explicit_metadata_and_legacy_unknown_preserved(): 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": [ From 43e4198edc86ac01d3b2b7ac030e05232876fe1b Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:13:23 -0700 Subject: [PATCH 3/4] fix: propagate detector errors and coverage through local CLI workflows --- README.md | 5 ++ src/pisama/_analyze.py | 15 ++++++ src/pisama/cli/analyze_cmd.py | 2 +- src/pisama/cli/check_cmd.py | 33 +++++++++++-- src/pisama/cli/replay.py | 9 +++- src/pisama/cli/smoke.py | 6 ++- src/pisama/cli/watch.py | 8 ++- src/pisama/replay/comparator.py | 17 ++++++- src/pisama/replay/smoke_runner.py | 10 +++- tests/test_detector_error_exit.py | 74 ++++++++++++++++++++++++++++ tests/test_public_trace_workflows.py | 7 +-- 11 files changed, 170 insertions(+), 16 deletions(-) create mode 100644 tests/test_detector_error_exit.py diff --git a/README.md b/README.md index e4151ee..ef7f594 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,11 @@ 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 without explicit pass evidence. ```bash pisama analyze trace.json # Analyze a trace diff --git a/src/pisama/_analyze.py b/src/pisama/_analyze.py index ff09166..2d5a074 100644 --- a/src/pisama/_analyze.py +++ b/src/pisama/_analyze.py @@ -55,6 +55,21 @@ class AnalyzeResult: 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: """Whether any issues were detected.""" diff --git a/src/pisama/cli/analyze_cmd.py b/src/pisama/cli/analyze_cmd.py index 0c8fcdc..ef16c9a 100644 --- a/src/pisama/cli/analyze_cmd.py +++ b/src/pisama/cli/analyze_cmd.py @@ -80,7 +80,7 @@ def analyze_cmd(path: str, min_severity: int, output_json: bool, scrub: bool) -> display_analysis_result(result) # Exit code: 1 if critical issues found, 0 otherwise - if result.critical_issues: + 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/replay/comparator.py b/src/pisama/replay/comparator.py index b0b30bd..65aaa84 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,15 @@ 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) + checked_pass = any( + item.get("detector_name") == det + and item.get("assessment") == "contract_satisfied" + for item in b.detector_assessments + ) + if checked_pass: + result.fixed.append(det) + else: + 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_detector_error_exit.py b/tests/test_detector_error_exit.py new file mode 100644 index 0000000..46b8da3 --- /dev/null +++ b/tests/test_detector_error_exit.py @@ -0,0 +1,74 @@ +"""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 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 From f526bb332749d2c54c40cabc38ed0ff4f27d6596 Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:17:09 -0700 Subject: [PATCH 4/4] fix: avoid cross-contract repair claims in replay comparisons --- README.md | 3 ++- src/pisama/cli/analyze_cmd.py | 2 +- src/pisama/replay/comparator.py | 13 ++++------- tests/test_detector_error_exit.py | 36 +++++++++++++++++++++++++++++++ 4 files changed, 43 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index ef7f594..6b42fdd 100644 --- a/README.md +++ b/README.md @@ -135,7 +135,8 @@ success. Coverage metadata does not turn heuristic confidence into calibration. `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 without explicit pass evidence. +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 diff --git a/src/pisama/cli/analyze_cmd.py b/src/pisama/cli/analyze_cmd.py index ef16c9a..53fcd3d 100644 --- a/src/pisama/cli/analyze_cmd.py +++ b/src/pisama/cli/analyze_cmd.py @@ -79,7 +79,7 @@ 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 + # 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/replay/comparator.py b/src/pisama/replay/comparator.py index 65aaa84..e8f9c6f 100644 --- a/src/pisama/replay/comparator.py +++ b/src/pisama/replay/comparator.py @@ -68,15 +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: - checked_pass = any( - item.get("detector_name") == det - and item.get("assessment") == "contract_satisfied" - for item in b.detector_assessments - ) - if checked_pass: - result.fixed.append(det) - else: - result.unassessed.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/tests/test_detector_error_exit.py b/tests/test_detector_error_exit.py index 46b8da3..9c24551 100644 --- a/tests/test_detector_error_exit.py +++ b/tests/test_detector_error_exit.py @@ -72,3 +72,39 @@ def test_replay_disappearance_without_coverage_not_fixed(): 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"]