From 2b9a56adf9a07753ceede0d221f5e5a9aa085f7a Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:28:24 -0700 Subject: [PATCH 1/3] fix: distinguish detector reporting from whole-trace coverage --- README.md | 8 ++++- src/pisama/_analyze.py | 16 +++++++-- src/pisama/cli/check_cmd.py | 15 ++++----- src/pisama/output/terminal.py | 12 +++++-- tests/test_reporting_scope.py | 61 +++++++++++++++++++++++++++++++++++ 5 files changed, 98 insertions(+), 14 deletions(-) create mode 100644 tests/test_reporting_scope.py diff --git a/README.md b/README.md index 6b42fdd..0f8d2c5 100644 --- a/README.md +++ b/README.md @@ -131,7 +131,13 @@ 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. +`check --json` schema 3 includes per-file assessments and +`assessment_reporting_complete`: one identified report per executed detector. +That flag includes explicitly unknown/abstained reports and does not describe +trace coverage. Duplicate/missing detector identities make reporting incomplete. +`trace_coverage` is currently `unassessed`; `files_clean` remains zero because +core does not provide whole-trace/span/contract coverage accounting. Even a +detector reporting a satisfied contract may have skipped other requests. `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 diff --git a/src/pisama/_analyze.py b/src/pisama/_analyze.py index 2d5a074..78fbcac 100644 --- a/src/pisama/_analyze.py +++ b/src/pisama/_analyze.py @@ -60,12 +60,24 @@ def has_detector_errors(self) -> bool: return any(item.get("assessment") == "error" for item in self.detector_assessments) @property - def coverage_complete(self) -> bool: + def assessment_reporting_complete(self) -> bool: + """Each executed detector has one identified report, not complete trace coverage.""" + names = [item.get("detector_name") for item in self.detector_assessments] return ( self.detectors_run > 0 and len(self.detector_assessments) == self.detectors_run + and all(isinstance(name, str) and bool(name) for name in names) + and len(set(names)) == self.detectors_run and all( - item.get("assessment") in {"contract_satisfied", "contract_violated", "finding"} + item.get("assessment") + in { + "contract_satisfied", + "contract_violated", + "finding", + "unknown", + "abstained", + "error", + } for item in self.detector_assessments ) ) diff --git a/src/pisama/cli/check_cmd.py b/src/pisama/cli/check_cmd.py index 0e92d5f..6e4e7ca 100644 --- a/src/pisama/cli/check_cmd.py +++ b/src/pisama/cli/check_cmd.py @@ -108,9 +108,10 @@ def record_result(self, path: Path, result: AnalyzeResult) -> None: 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 and result.coverage_complete) + # No current core result proves whole-trace/span coverage. Keep the + # legacy aggregate conservatively zero rather than count partial passes. self.files_no_findings += int(not result.has_issues) - self.files_incomplete += int(not result.coverage_complete) + self.files_incomplete += 1 self.analysis_errors += int(result.has_detector_errors) self.files_failed += int(failed_threshold) self.failed = self.failed or failed_threshold @@ -122,8 +123,6 @@ def record_result(self, path: Path, result: AnalyzeResult) -> None: if result.has_detector_errors else "issues" if result.has_issues - else "clean" - if result.coverage_complete else "unassessed" ), "failed": failed_threshold, @@ -132,7 +131,8 @@ def record_result(self, path: Path, result: AnalyzeResult) -> None: "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, + "assessment_reporting_complete": result.assessment_reporting_complete, + "trace_coverage": "unassessed", "error": None, } ) @@ -439,10 +439,7 @@ def _render_file_result( if result.has_issues: _print_file_report(trace_path, result, threshold) else: - console.print( - f"{trace_path.name}: no findings; " - f"coverage {'complete' if result.coverage_complete else 'incomplete/unspecified'}" - ) + console.print(f"{trace_path.name}: no findings; whole-trace coverage unassessed") def _print_file_report(trace_path: Path, result: AnalyzeResult, threshold: int) -> None: diff --git a/src/pisama/output/terminal.py b/src/pisama/output/terminal.py index 313488e..a8e19b7 100644 --- a/src/pisama/output/terminal.py +++ b/src/pisama/output/terminal.py @@ -75,11 +75,19 @@ def display_analysis_result(result: AnalyzeResult) -> None: 0, result.detectors_run - len(result.detector_assessments) ) console.print( - "Coverage: " - f"{counts.get('contract_satisfied', 0)} explicit contract passes; " + "Coverage: selected detector reports only; " + f"{counts.get('contract_satisfied', 0)} detectors report contract passes; " f"{counts.get('abstained', 0)} abstained; " f"{counts.get('error', 0)} errors; {unknown} unspecified." ) + console.print( + "Whole-trace/span coverage is unassessed; partial contract checks are not full coverage." + ) + if not result.assessment_reporting_complete: + console.print( + "Assessment reporting incomplete: " + "detector identities/counts are missing or inconsistent." + ) if not result.issues: return diff --git a/tests/test_reporting_scope.py b/tests/test_reporting_scope.py new file mode 100644 index 0000000..e90cdc7 --- /dev/null +++ b/tests/test_reporting_scope.py @@ -0,0 +1,61 @@ +"""Reporting completeness must not imply whole-trace semantic coverage.""" + +import json + +from click.testing import CliRunner + +from pisama import analyze +from pisama._analyze import AnalyzeResult +from pisama.cli.main import main + + +def test_duplicate_detector_reports_are_incomplete(): + result = AnalyzeResult( + [], + "synthetic", + 2, + 0, + [ + { + "detector_name": "communication", + "assessment": "contract_satisfied", + "checked_contracts": 1, + }, + { + "detector_name": "communication", + "assessment": "contract_satisfied", + "checked_contracts": 1, + }, + ], + ) + assert not result.assessment_reporting_complete + + +def test_real_two_span_partial_check_does_not_make_trace_clean(tmp_path): + trace = { + "spans": [ + { + "kind": "llm", + "input_data": {"content": "Return JSON only."}, + "output_data": {"content": "{}"}, + }, + { + "kind": "llm", + "input_data": {"content": "Explain the failure cause."}, + "output_data": {"content": "unrelated response"}, + }, + ] + } + result = analyze(trace, detectors=["communication"]) + assert result.assessment_reporting_complete + path = tmp_path / "mixed.json" + path.write_text(json.dumps(trace)) + output = CliRunner().invoke( + main, ["check", str(path), "--json", "--fail-on", "never", "--detectors", "communication"] + ) + assert output.exit_code == 0 + payload = json.loads(output.output) + assert payload["summary"]["files_clean"] == 0 + assert payload["results"][0]["trace_coverage"] == "unassessed" + assert payload["results"][0]["assessment_reporting_complete"] is True + assert "coverage_complete" not in payload["results"][0] From d73277f9f0aaa0906aaafa458efde96ae090d1ae Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:38:17 -0700 Subject: [PATCH 2/3] feat: preserve validated positional response-contract accounting --- README.md | 7 ++ src/pisama/_analyze.py | 8 ++ src/pisama/_coverage.py | 87 ++++++++++++++++ src/pisama/output/terminal.py | 12 +++ tests/test_public_trace_workflows.py | 24 ++++- tests/test_response_coverage_projection.py | 114 +++++++++++++++++++++ 6 files changed, 248 insertions(+), 4 deletions(-) create mode 100644 src/pisama/_coverage.py create mode 100644 tests/test_response_coverage_projection.py diff --git a/README.md b/README.md index 0f8d2c5..f125593 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,13 @@ for issue in result.issues: ## CLI +Validated response-contract accounting, when the core supplies it, is preserved +under each detector assessment. Version 1 includes positional span records and +checked/unsupported/outside-scope counts for eligible response pairs only. +Raw span IDs and arbitrary metadata are omitted. Invalid schemas, indices, +counts or enum values are rejected with `response_coverage_status="invalid"`; +they never establish whole-trace coverage. Business semantics remain unassessed. + `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 diff --git a/src/pisama/_analyze.py b/src/pisama/_analyze.py index 78fbcac..573041c 100644 --- a/src/pisama/_analyze.py +++ b/src/pisama/_analyze.py @@ -10,6 +10,7 @@ from pisama_core.traces.models import Trace +from pisama._coverage import validate_response_coverage from pisama._loader import load_trace @@ -226,6 +227,13 @@ def _convert_assessments(analysis: Any) -> list[dict[str, Any]]: basis = metadata.get("confidence_basis") if isinstance(basis, str) and basis == "uncalibrated contract heuristic": item["confidence_basis"] = basis + if "response_contract_coverage" in metadata: + coverage = validate_response_coverage(metadata["response_contract_coverage"]) + if coverage is None: + item["response_coverage_status"] = "invalid" + else: + item["response_coverage_status"] = "valid" + item["response_contract_coverage"] = coverage assessments.append(item) return assessments diff --git a/src/pisama/_coverage.py b/src/pisama/_coverage.py new file mode 100644 index 0000000..36c59ca --- /dev/null +++ b/src/pisama/_coverage.py @@ -0,0 +1,87 @@ +"""Strict, content-free projection of explicit response-contract accounting.""" + +from typing import Any + + +def validate_response_coverage(value: Any) -> dict[str, Any] | None: + if not isinstance(value, dict): + return None + if type(value.get("version")) is not int or value["version"] != 1: + return None + if ( + value.get("scope") != "eligible_captured_response_pairs" + or value.get("business_semantics_assessed") is not False + ): + return None + counts = ( + "trace_span_count", + "considered_count", + "checked_count", + "unsupported_count", + "outside_scope_count", + ) + if any(type(value.get(key)) is not int or value[key] < 0 for key in counts): + return None + rows = value.get("records") + if not isinstance(rows, list) or len(rows) != value["trace_span_count"]: + return None + reasons = { + "satisfied": {"explicit_contract_checked"}, + "violated": {"explicit_contract_checked"}, + "unsupported": { + "missing_attributable_pair", + "ambiguous_parent_identity", + "unsupported_or_ambiguous_contract", + }, + "outside_scope": {"span_kind_outside_response_scope"}, + } + safe = [] + seen: set[int] = set() + totals = dict.fromkeys(reasons, 0) + for row in rows: + if not isinstance(row, dict): + return None + index, status = row.get("span_index"), row.get("status") + if type(index) is not int or index < 0 or index >= len(rows) or index in seen: + return None + if not isinstance(status, str) or status not in reasons: + return None + reason, kind = row.get("reason"), row.get("contract_kind") + if not isinstance(reason, str) or reason not in reasons[status]: + return None + if status in {"satisfied", "violated"}: + if not isinstance(kind, str) or kind not in {"literal", "json"}: + return None + if row.get("relationship") not in ("captured_input_output", "explicit_message_parent"): + return None + elif kind is not None: + return None + if type(row.get("identity_ambiguous")) is not bool: + return None + seen.add(index) + totals[status] += 1 + safe.append( + { + "span_index": index, + "status": status, + "contract_kind": kind, + "reason": reason, + "identity_ambiguous": row["identity_ambiguous"], + } + ) + expected = { + "trace_span_count": len(rows), + "considered_count": len(rows) - totals["outside_scope"], + "checked_count": totals["satisfied"] + totals["violated"], + "unsupported_count": totals["unsupported"], + "outside_scope_count": totals["outside_scope"], + } + if any(value[key] != count for key, count in expected.items()): + return None + return { + "version": 1, + "scope": "eligible_captured_response_pairs", + **expected, + "records": sorted(safe, key=lambda item: item["span_index"]), + "business_semantics_assessed": False, + } diff --git a/src/pisama/output/terminal.py b/src/pisama/output/terminal.py index a8e19b7..a8b4c9b 100644 --- a/src/pisama/output/terminal.py +++ b/src/pisama/output/terminal.py @@ -88,6 +88,18 @@ def display_analysis_result(result: AnalyzeResult) -> None: "Assessment reporting incomplete: " "detector identities/counts are missing or inconsistent." ) + for assessment in result.detector_assessments: + if assessment.get("response_coverage_status") == "invalid": + console.print( + "Response-contract accounting rejected: invalid metadata; coverage unknown." + ) + coverage = assessment.get("response_contract_coverage") + if isinstance(coverage, dict): + console.print( + f"Response-contract scope: {coverage['checked_count']} checked; " + f"{coverage['unsupported_count']} unsupported; " + f"{coverage['outside_scope_count']} outside scope." + ) if not result.issues: return diff --git a/tests/test_public_trace_workflows.py b/tests/test_public_trace_workflows.py index 4d8d9ce..3a0060b 100644 --- a/tests/test_public_trace_workflows.py +++ b/tests/test_public_trace_workflows.py @@ -61,9 +61,7 @@ def test_captured_atif_loads_consistently_from_all_public_inputs( same_object = load_trace(expected) assert same_object is expected - assert {from_dict.trace_id, from_json.trace_id, from_file.trace_id} == { - expected.trace_id - } + assert {from_dict.trace_id, from_json.trace_id, from_file.trace_id} == {expected.trace_id} assert len(expected.spans) == 20 assert len({span.span_id for span in expected.spans}) == 20 assert sum(span.kind.value == "tool" for span in expected.spans) == 8 @@ -119,7 +117,25 @@ async def test_real_detector_pipeline_analyzes_captured_multi_agent_trace( assert result.execution_time_ms > 0 assert result.has_issues assert result.critical_issues - assert {"context", "communication"}.issubset(issue_types) + assert "context" in issue_types + from pisama_core.detection.detectors.communication import CommunicationDetector + + if CommunicationDetector.version.startswith("1."): + # Legacy core inferred intent failure from verb overlap. Preserve its + # compatibility expectation, not that inference as a correctness gate. + assert "communication" in issue_types + else: + assert "communication" not in issue_types + assessment = next( + item for item in result.detector_assessments if item["detector_name"] == "communication" + ) + assert assessment["assessment"] == "abstained" + assert assessment["checked_contracts"] == 0 + coverage = assessment.get("response_contract_coverage") + if coverage is not None: + assert coverage["checked_count"] == 0 + assert coverage["unsupported_count"] == 6 + assert coverage["outside_scope_count"] == 14 context_only = analyze(captured_omnigent_trajectory, detectors=["context"]) assert context_only.detectors_run == 1 diff --git a/tests/test_response_coverage_projection.py b/tests/test_response_coverage_projection.py new file mode 100644 index 0000000..7fe044a --- /dev/null +++ b/tests/test_response_coverage_projection.py @@ -0,0 +1,114 @@ +"""Validate accounting metadata without copying arbitrary trace content.""" + +import asyncio +import copy +import json + +import pytest +from pisama_core.detection.detectors.communication import CommunicationDetector +from pisama_core.traces.models import Trace + +from pisama import analyze +from pisama._coverage import validate_response_coverage + + +def valid(): + return { + "version": 1, + "scope": "eligible_captured_response_pairs", + "trace_span_count": 1, + "considered_count": 1, + "checked_count": 1, + "unsupported_count": 0, + "outside_scope_count": 0, + "business_semantics_assessed": False, + "records": [ + { + "span_index": 0, + "span_id": "PRIVATE_SYNTHETIC_MARKER", + "status": "satisfied", + "contract_kind": "json", + "reason": "explicit_contract_checked", + "identity_ambiguous": False, + "relationship": "captured_input_output", + } + ], + } + + +def test_projection_strips_identifiers_and_arbitrary_fields(): + data = valid() + data["request"] = "PRIVATE_SYNTHETIC_MARKER" + output = validate_response_coverage(data) + assert output is not None + assert "PRIVATE_SYNTHETIC_MARKER" not in json.dumps(output) + + +@pytest.mark.parametrize( + "field,value", + [ + ("version", True), + ("checked_count", True), + ("checked_count", 2), + ("trace_span_count", -1), + ("business_semantics_assessed", True), + ("scope", "whole_business_task"), + ], +) +def test_invalid_counts_or_scope_rejected(field, value): + data = valid() + data[field] = value + assert validate_response_coverage(data) is None + + +def test_duplicate_indices_rejected(): + data = valid() + data["records"].append(copy.deepcopy(data["records"][0])) + data.update(trace_span_count=2, considered_count=2, checked_count=2) + assert validate_response_coverage(data) is None + + +@pytest.mark.parametrize( + "field,value", + [ + ("span_index", True), + ("status", "PRIVATE_SYNTHETIC_MARKER"), + ("reason", "PRIVATE_SYNTHETIC_MARKER"), + ("contract_kind", "arbitrary_schema"), + ("identity_ambiguous", "no"), + ("relationship", "adjacency"), + ], +) +def test_invalid_records_rejected(field, value): + data = valid() + data["records"][0][field] = value + assert validate_response_coverage(data) is None + + +def test_real_paired_core_accounting_survives_projection(): + trace = { + "spans": [ + { + "kind": "llm", + "span_id": "PRIVATE_SYNTHETIC_MARKER", + "input_data": {"content": "Reply with OK."}, + "output_data": {"content": "OK"}, + }, + { + "kind": "llm", + "input_data": {"content": "If ready, reply with OK."}, + "output_data": {"content": "not ready"}, + }, + ] + } + raw = asyncio.run(CommunicationDetector().detect(Trace.from_dict(trace))) + result = analyze(trace, detectors=["communication"]) + assessment = result.detector_assessments[0] + if "response_contract_coverage" not in raw.metadata: + assert "response_contract_coverage" not in assessment + else: + coverage = assessment["response_contract_coverage"] + assert coverage["checked_count"] == 1 + assert coverage["unsupported_count"] == 1 + assert coverage["business_semantics_assessed"] is False + assert "PRIVATE_SYNTHETIC_MARKER" not in json.dumps(coverage) From 8df5038f0288387242f6ab7f7cb2a148853b63ee Mon Sep 17 00:00:00 2001 From: tn-pisama Date: Thu, 10 Sep 2026 00:42:08 -0700 Subject: [PATCH 3/3] fix: reconcile accounting with trace size and detector outcome --- src/pisama/_analyze.py | 33 ++++++++++++++++++++-- tests/test_response_coverage_projection.py | 32 +++++++++++++++++++++ 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/src/pisama/_analyze.py b/src/pisama/_analyze.py index 573041c..4867ee3 100644 --- a/src/pisama/_analyze.py +++ b/src/pisama/_analyze.py @@ -188,11 +188,13 @@ 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), + detector_assessments=_convert_assessments(analysis, len(trace.spans)), ) -def _convert_assessments(analysis: Any) -> list[dict[str, Any]]: +def _convert_assessments( + analysis: Any, trace_span_count: int | None = None +) -> list[dict[str, Any]]: """Preserve explicit coverage without treating legacy silence as success.""" assessments = [] for result in analysis.detection_results: @@ -229,8 +231,33 @@ def _convert_assessments(analysis: Any) -> list[dict[str, Any]]: item["confidence_basis"] = basis if "response_contract_coverage" in metadata: coverage = validate_response_coverage(metadata["response_contract_coverage"]) - if coverage is None: + consistent = coverage is not None + if coverage is not None: + violated = any(row["status"] == "violated" for row in coverage["records"]) + expected_assessment = ( + "contract_violated" + if violated + else "contract_satisfied" + if coverage["checked_count"] + else "abstained" + ) + consistent = ( + "error" not in metadata + and trace_span_count is not None + and coverage["trace_span_count"] == trace_span_count + and type(checked) is int + and checked == coverage["checked_count"] + and metadata.get("assessment") == expected_assessment + and result.detected == violated + ) + if not consistent: item["response_coverage_status"] = "invalid" + item.pop("checked_contracts", None) + item["assessment"] = ( + "error" + if "error" in metadata + else ("finding" if result.detected else "unknown") + ) else: item["response_coverage_status"] = "valid" item["response_contract_coverage"] = coverage diff --git a/tests/test_response_coverage_projection.py b/tests/test_response_coverage_projection.py index 7fe044a..d3d4a33 100644 --- a/tests/test_response_coverage_projection.py +++ b/tests/test_response_coverage_projection.py @@ -6,9 +6,13 @@ import pytest from pisama_core.detection.detectors.communication import CommunicationDetector +from pisama_core.detection.orchestrator import AnalysisResult +from pisama_core.detection.result import DetectionResult +from pisama_core.traces.enums import Platform from pisama_core.traces.models import Trace from pisama import analyze +from pisama._analyze import _convert_assessments from pisama._coverage import validate_response_coverage @@ -112,3 +116,31 @@ def test_real_paired_core_accounting_survives_projection(): assert coverage["unsupported_count"] == 1 assert coverage["business_semantics_assessed"] is False assert "PRIVATE_SYNTHETIC_MARKER" not in json.dumps(coverage) + + +@pytest.mark.parametrize("variation", ["violated", "count", "empty", "trace_count", "detected"]) +def test_cross_metadata_contradictions_are_not_passes(variation): + coverage = valid() + metadata = { + "assessment": "contract_satisfied", + "checked_contracts": 1, + "response_contract_coverage": coverage, + } + detected = False + if variation == "violated": + coverage["records"][0]["status"] = "violated" + elif variation == "count": + metadata["checked_contracts"] = 99 + elif variation == "empty": + coverage.update(records=[], trace_span_count=0, considered_count=0, checked_count=0) + elif variation == "detected": + detected = True + result = DetectionResult("communication", detected=detected, metadata=metadata) + output = _convert_assessments( + AnalysisResult(trace_id="synthetic", platform=Platform.GENERIC, detection_results=[result]), + trace_span_count=2 if variation == "trace_count" else 1, + )[0] + assert output["assessment"] == ("finding" if detected else "unknown") + assert output["response_coverage_status"] == "invalid" + assert "response_contract_coverage" not in output + assert "checked_contracts" not in output