diff --git a/docs/evidence-health.md b/docs/evidence-health.md new file mode 100644 index 0000000..b4fac4d --- /dev/null +++ b/docs/evidence-health.md @@ -0,0 +1,57 @@ +# Evidence health + +Evidence health answers a narrow review question: **is this captured session structurally usable as evidence, and what limitations should the reviewer know about?** + +It does not score agent quality, infer misconduct, or claim that a provider exposed everything that happened. + +## Versioned result + +`agent_trace.evidence_health.assess_evidence_health()` returns a versioned result with one of four statuses: + +- `healthy` — the observed stream has the expected boundaries and relationships and no declared capture limitation was supplied. +- `partial` — the stream is reviewable but has an observed gap or a declared provider limitation. +- `unknown` — there is not enough completed evidence to judge yet, for example an active session without an end marker. +- `invalid` — the stream is internally contradictory, for example duplicate boundaries, duplicate event IDs, mixed session IDs, or a non-finite timestamp. + +Every non-healthy result includes machine-readable reason codes and human explanations. Reasons also distinguish observed defects from provider limitations. + +## Observed checks + +The first version checks: + +- session start/end boundaries; +- events before start or after end; +- duplicate event IDs; +- mixed session IDs; +- non-finite timestamps and timestamp regressions; +- unpaired tool calls/results; +- unpaired LLM requests/responses; and +- export/drop failures explicitly reported by the caller. + +The calculation is deterministic and local. It performs no network calls. + +## Provider limitations + +A missing event alone is not enough to conclude that capture failed: some providers never expose certain categories of activity. Callers should therefore pass known blind spots from a versioned provider/capture matrix: + +```python +from agent_trace.evidence_health import assess_evidence_health + +health = assess_evidence_health( + events, + provider="codex", + capture_method="hooks", + provider_blind_spots=[ + "example limitation supplied by the capture matrix", + ], + session_finalized=True, +) +``` + +A declared blind spot makes the result `partial` even when the observed event stream is structurally clean. This prevents `healthy` from becoming a false claim about activity a provider cannot expose. + +## Integration boundary + +The core assessment module intentionally does not read configuration files, provider matrices, session metadata, or UI state. Those surfaces should adapt their existing data into the same function so CLI, JSON/API, replay, and the local dashboard can expose one consistent result. + +This keeps the health rules independently testable and gives later integration work one stable source of reason codes rather than several slightly different implementations. diff --git a/src/agent_trace/evidence_health.py b/src/agent_trace/evidence_health.py new file mode 100644 index 0000000..2e4325e --- /dev/null +++ b/src/agent_trace/evidence_health.py @@ -0,0 +1,391 @@ +"""Deterministic evidence-health assessment for captured agent sessions. + +The goal of evidence health is narrower than judging whether an agent behaved +correctly. It answers whether the *recorded evidence* is structurally usable for +review and which limitations a reviewer must keep in mind. + +The calculation is deliberately local and dependency-free. Provider blind +spots are supplied explicitly by capture adapters/matrices instead of inferred +from missing events, so a clean event stream is never presented as proof that a +provider exposed everything that happened. +""" + +from __future__ import annotations + +import math +from dataclasses import asdict, dataclass, field +from enum import Enum +from typing import Iterable, Sequence + +from .models import EventType, TraceEvent + +EVIDENCE_HEALTH_SCHEMA_VERSION = 1 + + +class EvidenceHealthStatus(str, Enum): + """Reviewability state for one captured session.""" + + HEALTHY = "healthy" + PARTIAL = "partial" + UNKNOWN = "unknown" + INVALID = "invalid" + + +class EvidenceHealthReasonKind(str, Enum): + """Whether a reason comes from observed evidence or declared capture limits.""" + + OBSERVED = "observed" + PROVIDER_LIMITATION = "provider_limitation" + + +@dataclass(frozen=True) +class EvidenceHealthReason: + """Machine-readable reason explaining an evidence-health result.""" + + code: str + message: str + kind: EvidenceHealthReasonKind = EvidenceHealthReasonKind.OBSERVED + event_id: str = "" + + +@dataclass(frozen=True) +class EvidenceHealthResult: + """Versioned result returned by :func:`assess_evidence_health`.""" + + status: EvidenceHealthStatus + reasons: tuple[EvidenceHealthReason, ...] = field(default_factory=tuple) + schema_version: int = EVIDENCE_HEALTH_SCHEMA_VERSION + provider: str = "" + capture_method: str = "" + observed_start: bool = False + observed_end: bool = False + event_count: int = 0 + + def to_dict(self) -> dict: + """Return a stable JSON-compatible representation.""" + payload = asdict(self) + payload["status"] = self.status.value + for reason in payload["reasons"]: + reason["kind"] = reason["kind"].value + return payload + + +@dataclass(frozen=True) +class _RelationshipSpec: + request_type: EventType + result_type: EventType + missing_result_code: str + orphan_result_code: str + duplicate_outcome_code: str + out_of_order_outcome_code: str + label: str + error_is_terminal: bool = False + + +_RELATIONSHIPS = ( + _RelationshipSpec( + request_type=EventType.TOOL_CALL, + result_type=EventType.TOOL_RESULT, + missing_result_code="unpaired_tool_call", + orphan_result_code="orphan_tool_result", + duplicate_outcome_code="duplicate_tool_outcome", + out_of_order_outcome_code="out_of_order_tool_outcome", + label="tool", + error_is_terminal=True, + ), + _RelationshipSpec( + request_type=EventType.LLM_REQUEST, + result_type=EventType.LLM_RESPONSE, + missing_result_code="unpaired_llm_request", + orphan_result_code="orphan_llm_response", + duplicate_outcome_code="duplicate_llm_outcome", + out_of_order_outcome_code="out_of_order_llm_outcome", + label="LLM", + error_is_terminal=True, + ), +) + + +def _reason( + code: str, + message: str, + *, + kind: EvidenceHealthReasonKind = EvidenceHealthReasonKind.OBSERVED, + event: TraceEvent | None = None, +) -> EvidenceHealthReason: + return EvidenceHealthReason( + code=code, + message=message, + kind=kind, + event_id=event.event_id if event is not None else "", + ) + + +def _relationship_reasons( + events: Sequence[TraceEvent], + spec: _RelationshipSpec, +) -> list[EvidenceHealthReason]: + request_positions = { + event.event_id: index + for index, event in enumerate(events) + if event.event_type == spec.request_type + } + requests = { + event.event_id: event + for event in events + if event.event_type == spec.request_type + } + outcomes_by_request: dict[str, list[tuple[int, TraceEvent]]] = {} + reasons: list[EvidenceHealthReason] = [] + + for index, event in enumerate(events): + is_result = event.event_type == spec.result_type + is_terminal_error = ( + spec.error_is_terminal + and event.event_type == EventType.ERROR + and event.parent_id.strip() in requests + ) + if not is_result and not is_terminal_error: + continue + + parent_id = event.parent_id.strip() + if parent_id and parent_id in requests: + outcomes_by_request.setdefault(parent_id, []).append((index, event)) + continue + + if is_result: + detail = ( + f" references unknown parent {parent_id!r}" + if parent_id + else " has no parent_id" + ) + reasons.append( + _reason( + spec.orphan_result_code, + f"{spec.label} result {event.event_id!r}{detail}", + event=event, + ) + ) + + for request_id, request in requests.items(): + outcomes = outcomes_by_request.get(request_id, []) + if not outcomes: + reasons.append( + _reason( + spec.missing_result_code, + f"{spec.label} request {request_id!r} has no recorded terminal outcome", + event=request, + ) + ) + continue + + if len(outcomes) > 1: + reasons.append( + _reason( + spec.duplicate_outcome_code, + f"{spec.label} request {request_id!r} has {len(outcomes)} recorded terminal outcomes", + event=outcomes[1][1], + ) + ) + + request_position = request_positions[request_id] + for outcome_position, outcome in outcomes: + if outcome_position < request_position: + reasons.append( + _reason( + spec.out_of_order_outcome_code, + f"{spec.label} outcome {outcome.event_id!r} appears before request {request_id!r}", + event=outcome, + ) + ) + + return reasons + + +def _structural_reasons( + events: Sequence[TraceEvent], + *, + session_finalized: bool | None, +) -> tuple[list[EvidenceHealthReason], bool]: + reasons: list[EvidenceHealthReason] = [] + invalid = False + + starts = [event for event in events if event.event_type == EventType.SESSION_START] + ends = [event for event in events if event.event_type == EventType.SESSION_END] + + if not starts: + reasons.append(_reason("missing_session_start", "session start marker was not captured")) + elif len(starts) > 1: + reasons.append( + _reason( + "duplicate_session_start", + f"captured {len(starts)} session start markers", + event=starts[1], + ) + ) + invalid = True + + if not ends: + if session_finalized is False: + reasons.append(_reason("session_active", "session is still active; no end marker expected yet")) + else: + reasons.append(_reason("missing_session_end", "session end marker was not captured")) + elif len(ends) > 1: + reasons.append( + _reason( + "duplicate_session_end", + f"captured {len(ends)} session end markers", + event=ends[1], + ) + ) + invalid = True + + if starts and events[0] is not starts[0]: + reasons.append( + _reason( + "events_before_session_start", + "events were recorded before the first session start marker", + event=events[0], + ) + ) + invalid = True + + if ends and events[-1] is not ends[-1]: + reasons.append( + _reason( + "events_after_session_end", + "events were recorded after the last session end marker", + event=events[-1], + ) + ) + invalid = True + + session_ids = {event.session_id for event in events if event.session_id} + if len(session_ids) > 1: + reasons.append( + _reason( + "mixed_session_ids", + f"event stream contains {len(session_ids)} different session IDs", + ) + ) + invalid = True + + previous_timestamp: float | None = None + for event in events: + timestamp = event.timestamp + if not isinstance(timestamp, (int, float)) or not math.isfinite(float(timestamp)): + reasons.append( + _reason( + "invalid_timestamp", + f"event {event.event_id!r} has a non-finite timestamp", + event=event, + ) + ) + invalid = True + continue + if previous_timestamp is not None and timestamp < previous_timestamp: + reasons.append( + _reason( + "timestamp_regression", + f"event {event.event_id!r} is earlier than the preceding event", + event=event, + ) + ) + previous_timestamp = timestamp + + event_ids: set[str] = set() + for event in events: + if not event.event_id: + reasons.append(_reason("missing_event_id", "captured event has no event_id", event=event)) + invalid = True + continue + if event.event_id in event_ids: + reasons.append( + _reason( + "duplicate_event_id", + f"event_id {event.event_id!r} appears more than once", + event=event, + ) + ) + invalid = True + event_ids.add(event.event_id) + + for spec in _RELATIONSHIPS: + reasons.extend(_relationship_reasons(events, spec)) + + return reasons, invalid + + +def assess_evidence_health( + events: Iterable[TraceEvent], + *, + provider: str = "", + capture_method: str = "", + provider_blind_spots: Iterable[str] = (), + session_finalized: bool | None = None, + export_failures: Iterable[str] = (), +) -> EvidenceHealthResult: + """Assess whether a captured session is structurally reviewable. + + ``provider_blind_spots`` must come from declared provider/capture metadata. + This function never invents a provider limitation from an absent event. + + ``session_finalized=False`` distinguishes an active session from a completed + session whose end marker is unexpectedly absent. + """ + event_list = list(events) + no_events = not event_list + if no_events: + reasons = [_reason("no_events", "no captured events are available for review")] + invalid = False + else: + reasons, invalid = _structural_reasons( + event_list, + session_finalized=session_finalized, + ) + + for failure in export_failures: + failure_text = str(failure).strip() + if failure_text: + reasons.append( + _reason( + "export_failure", + f"capture/export pipeline reported a failure: {failure_text}", + ) + ) + + for limitation in provider_blind_spots: + limitation_text = str(limitation).strip() + if limitation_text: + reasons.append( + _reason( + "provider_blind_spot", + limitation_text, + kind=EvidenceHealthReasonKind.PROVIDER_LIMITATION, + ) + ) + + active_only = ( + session_finalized is False + and reasons + and all(reason.code == "session_active" for reason in reasons) + ) + + if invalid: + status = EvidenceHealthStatus.INVALID + elif no_events or active_only: + status = EvidenceHealthStatus.UNKNOWN + elif reasons: + status = EvidenceHealthStatus.PARTIAL + else: + status = EvidenceHealthStatus.HEALTHY + + return EvidenceHealthResult( + status=status, + reasons=tuple(reasons), + provider=provider, + capture_method=capture_method, + observed_start=any(event.event_type == EventType.SESSION_START for event in event_list), + observed_end=any(event.event_type == EventType.SESSION_END for event in event_list), + event_count=len(event_list), + ) diff --git a/tests/test_evidence_health.py b/tests/test_evidence_health.py new file mode 100644 index 0000000..5b00d04 --- /dev/null +++ b/tests/test_evidence_health.py @@ -0,0 +1,278 @@ +import unittest + +from agent_trace.evidence_health import ( + EVIDENCE_HEALTH_SCHEMA_VERSION, + EvidenceHealthReasonKind, + EvidenceHealthStatus, + assess_evidence_health, +) +from agent_trace.models import EventType, TraceEvent + + +SESSION_ID = "review-fixture" + + +def event( + event_type: EventType, + timestamp: float, + event_id: str, + *, + parent_id: str = "", + session_id: str = SESSION_ID, +) -> TraceEvent: + return TraceEvent( + event_type=event_type, + timestamp=timestamp, + event_id=event_id, + session_id=session_id, + parent_id=parent_id, + ) + + +class TestEvidenceHealth(unittest.TestCase): + def test_complete_paired_session_is_healthy(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.LLM_REQUEST, 2.0, "llm-request"), + event(EventType.LLM_RESPONSE, 3.0, "llm-response", parent_id="llm-request"), + event(EventType.TOOL_CALL, 4.0, "tool-call"), + event(EventType.TOOL_RESULT, 5.0, "tool-result", parent_id="tool-call"), + event(EventType.SESSION_END, 6.0, "end"), + ] + + result = assess_evidence_health( + events, + provider="codex", + capture_method="hooks", + session_finalized=True, + ) + + self.assertEqual(result.status, EvidenceHealthStatus.HEALTHY) + self.assertEqual(result.reasons, ()) + self.assertTrue(result.observed_start) + self.assertTrue(result.observed_end) + self.assertEqual(result.event_count, 6) + self.assertEqual(result.provider, "codex") + self.assertEqual(result.capture_method, "hooks") + + def test_failed_tool_call_error_is_a_terminal_outcome(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.TOOL_CALL, 2.0, "tool-call"), + event(EventType.ERROR, 3.0, "tool-error", parent_id="tool-call"), + event(EventType.SESSION_END, 4.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + + self.assertEqual(result.status, EvidenceHealthStatus.HEALTHY) + self.assertNotIn("unpaired_tool_call", [reason.code for reason in result.reasons]) + + def test_failed_llm_call_error_is_a_terminal_outcome(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.LLM_REQUEST, 2.0, "llm-request"), + event(EventType.ERROR, 3.0, "llm-error", parent_id="llm-request"), + event(EventType.SESSION_END, 4.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + + self.assertEqual(result.status, EvidenceHealthStatus.HEALTHY) + self.assertNotIn("unpaired_llm_request", [reason.code for reason in result.reasons]) + + def test_out_of_order_tool_result_is_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.TOOL_RESULT, 2.0, "tool-result", parent_id="tool-call"), + event(EventType.TOOL_CALL, 3.0, "tool-call"), + event(EventType.SESSION_END, 4.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + codes = [reason.code for reason in result.reasons] + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertIn("out_of_order_tool_outcome", codes) + self.assertNotIn("unpaired_tool_call", codes) + + def test_duplicate_tool_terminal_outcomes_are_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.TOOL_CALL, 2.0, "tool-call"), + event(EventType.TOOL_RESULT, 3.0, "tool-result", parent_id="tool-call"), + event(EventType.ERROR, 4.0, "tool-error", parent_id="tool-call"), + event(EventType.SESSION_END, 5.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + codes = [reason.code for reason in result.reasons] + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertIn("duplicate_tool_outcome", codes) + self.assertNotIn("unpaired_tool_call", codes) + + def test_duplicate_llm_responses_are_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.LLM_REQUEST, 2.0, "llm-request"), + event(EventType.LLM_RESPONSE, 3.0, "llm-response-1", parent_id="llm-request"), + event(EventType.LLM_RESPONSE, 4.0, "llm-response-2", parent_id="llm-request"), + event(EventType.SESSION_END, 5.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertIn("duplicate_llm_outcome", [reason.code for reason in result.reasons]) + + def test_empty_evidence_is_unknown(self): + result = assess_evidence_health([]) + + self.assertEqual(result.status, EvidenceHealthStatus.UNKNOWN) + self.assertEqual([reason.code for reason in result.reasons], ["no_events"]) + + def test_empty_evidence_keeps_observable_failures_and_declared_limits(self): + result = assess_evidence_health( + [], + provider_blind_spots=["tool execution is not exposed by this capture adapter"], + export_failures=["collector rejected batch 1"], + ) + + self.assertEqual(result.status, EvidenceHealthStatus.UNKNOWN) + self.assertEqual( + [reason.code for reason in result.reasons], + ["no_events", "export_failure", "provider_blind_spot"], + ) + self.assertEqual( + result.reasons[-1].kind, + EvidenceHealthReasonKind.PROVIDER_LIMITATION, + ) + + def test_active_session_without_end_is_unknown_not_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.USER_PROMPT, 2.0, "prompt"), + ] + + result = assess_evidence_health(events, session_finalized=False) + + self.assertEqual(result.status, EvidenceHealthStatus.UNKNOWN) + self.assertEqual([reason.code for reason in result.reasons], ["session_active"]) + + def test_finalized_session_without_end_is_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.USER_PROMPT, 2.0, "prompt"), + ] + + result = assess_evidence_health(events, session_finalized=True) + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertIn("missing_session_end", [reason.code for reason in result.reasons]) + + def test_unpaired_tool_relationships_are_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.TOOL_CALL, 2.0, "call-without-result"), + event(EventType.TOOL_RESULT, 3.0, "orphan-result", parent_id="unknown-call"), + event(EventType.SESSION_END, 4.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + codes = {reason.code for reason in result.reasons} + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertEqual(codes, {"unpaired_tool_call", "orphan_tool_result"}) + + def test_timestamp_regression_is_partial(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.USER_PROMPT, 3.0, "prompt"), + event(EventType.ASSISTANT_RESPONSE, 2.0, "response"), + event(EventType.SESSION_END, 4.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertIn("timestamp_regression", [reason.code for reason in result.reasons]) + + def test_provider_blind_spots_are_declared_limitations(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.SESSION_END, 2.0, "end"), + ] + + result = assess_evidence_health( + events, + provider="example-provider", + provider_blind_spots=["file reads are not exposed by this capture adapter"], + session_finalized=True, + ) + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertEqual(len(result.reasons), 1) + self.assertEqual(result.reasons[0].code, "provider_blind_spot") + self.assertEqual( + result.reasons[0].kind, + EvidenceHealthReasonKind.PROVIDER_LIMITATION, + ) + + def test_mixed_session_ids_are_invalid(self): + events = [ + event(EventType.SESSION_START, 1.0, "start", session_id="one"), + event(EventType.USER_PROMPT, 2.0, "prompt", session_id="two"), + event(EventType.SESSION_END, 3.0, "end", session_id="one"), + ] + + result = assess_evidence_health(events, session_finalized=True) + + self.assertEqual(result.status, EvidenceHealthStatus.INVALID) + self.assertIn("mixed_session_ids", [reason.code for reason in result.reasons]) + + def test_duplicate_boundaries_and_event_ids_are_invalid(self): + events = [ + event(EventType.SESSION_START, 1.0, "same"), + event(EventType.SESSION_START, 2.0, "same"), + event(EventType.SESSION_END, 3.0, "end"), + ] + + result = assess_evidence_health(events, session_finalized=True) + codes = {reason.code for reason in result.reasons} + + self.assertEqual(result.status, EvidenceHealthStatus.INVALID) + self.assertIn("duplicate_session_start", codes) + self.assertIn("duplicate_event_id", codes) + + def test_export_failures_are_observed_partial_evidence(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.SESSION_END, 2.0, "end"), + ] + + result = assess_evidence_health( + events, + export_failures=["collector rejected batch 4"], + session_finalized=True, + ) + + self.assertEqual(result.status, EvidenceHealthStatus.PARTIAL) + self.assertEqual(result.reasons[0].code, "export_failure") + self.assertEqual(result.reasons[0].kind, EvidenceHealthReasonKind.OBSERVED) + + def test_result_serializes_with_versioned_machine_readable_fields(self): + events = [ + event(EventType.SESSION_START, 1.0, "start"), + event(EventType.SESSION_END, 2.0, "end"), + ] + + payload = assess_evidence_health(events, session_finalized=True).to_dict() + + self.assertEqual(payload["schema_version"], EVIDENCE_HEALTH_SCHEMA_VERSION) + self.assertEqual(payload["status"], "healthy") + self.assertEqual(payload["reasons"], ()) + + +if __name__ == "__main__": + unittest.main()