Correct assessment scope and validate positional contract accounting

README.md

@@ -125,13 +125,26 @@ 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
 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

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
 
 
@@ -60,12 +61,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
             )
         )
@@ -175,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:
@@ -214,6 +229,38 @@ 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"])
+            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
         assessments.append(item)
     return assessments
 

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,
+    }

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:

src/pisama/output/terminal.py

@@ -75,11 +75,31 @@ 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."
+        )
+    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

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

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]