feat: add deterministic session review projection core

docs/session-review-projection.md

@@ -0,0 +1,25 @@
+# Session review projection
+
+`agent_trace.review` provides a small deterministic layer between raw trace events and a review UI.
+
+It is intentionally conservative: the projection classifies only facts that are present in the recorded event stream. It does not infer a missing tool call, retry, test, or provider event from surrounding prose.
+
+## Ordering and relationships
+
+Items are ordered by event timestamp with recorded source order as the tie breaker. A `parent_id` is marked `linked` only when that event ID is present in the supplied stream. Missing parents are kept verbatim, marked `orphaned`, and placed in the `gaps` filter.
+
+This makes an incomplete trace visibly incomplete instead of manufacturing a clean execution tree.
+
+## Filters
+
+The first schema version exposes reviewer-oriented categories for files, tools, commands, tests, failures, retries, recovery, decisions, privacy transformations, and gaps.
+
+Core event types provide categories such as files, tools, failures, and decisions. More semantic categories such as tests, commands, retries, and recovery require explicit event metadata (`is_test`, `is_command`, `retry_of`/`retry`, or `recovery_of`/`recovered`). The projector deliberately does not guess these from command strings or natural-language output.
+
+## Raw evidence
+
+Each projected item retains the source event ID, event type, timestamp, parent ID, redaction flag, source index, and a copy of the event's provider-specific `data`. A UI can therefore expose the underlying recorded fields rather than presenting derived labels as ground truth.
+
+## Scope
+
+This is a foundation for issue #242. It does not yet replace the local dashboard, calculate evidence health, render annotations, or add URL/keyboard navigation. Those surfaces can consume the projection without duplicating event-ordering and relationship rules.

src/agent_trace/review.py

@@ -0,0 +1,135 @@
+"""Deterministic projection of trace events into reviewer-facing evidence.
+
+This module does not infer missing activity.  It only classifies facts present in
+TraceEvent fields and marks broken relationships explicitly so UI consumers can
+show gaps instead of inventing a complete story.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from typing import Iterable, Literal
+
+from .models import EventType, TraceEvent
+
+ReviewCategory = Literal[
+    "commands",
+    "decisions",
+    "failures",
+    "files",
+    "gaps",
+    "privacy",
+    "recovery",
+    "retry",
+    "tests",
+    "tools",
+]
+RelationshipState = Literal["none", "linked", "orphaned"]
+
+
+@dataclass(frozen=True)
+class ReviewItem:
+    event_id: str
+    event_type: str
+    timestamp: float
+    parent_id: str | None
+    relationship: RelationshipState
+    categories: tuple[ReviewCategory, ...]
+    redacted: bool
+    source_index: int
+    data: dict[str, object] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, object]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class ReviewProjection:
+    schema_version: int
+    items: tuple[ReviewItem, ...]
+    orphaned_event_ids: tuple[str, ...]
+
+    def filter(self, category: ReviewCategory) -> tuple[ReviewItem, ...]:
+        return tuple(item for item in self.items if category in item.categories)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": self.schema_version,
+            "items": [item.to_dict() for item in self.items],
+            "orphaned_event_ids": list(self.orphaned_event_ids),
+        }
+
+
+def _explicit_categories(event: TraceEvent) -> set[ReviewCategory]:
+    categories: set[ReviewCategory] = set()
+    if event.event_type in (EventType.FILE_READ, EventType.FILE_WRITE):
+        categories.add("files")
+    if event.event_type in (EventType.TOOL_CALL, EventType.TOOL_RESULT):
+        categories.add("tools")
+    if event.event_type == EventType.ERROR:
+        categories.add("failures")
+    if event.event_type == EventType.DECISION:
+        categories.add("decisions")
+    if event.redacted:
+        categories.add("privacy")
+
+    data = event.data
+    if data.get("is_test") is True or data.get("category") == "test":
+        categories.add("tests")
+    if data.get("is_command") is True or data.get("category") == "command":
+        categories.add("commands")
+    if data.get("retry_of") or data.get("retry") is True:
+        categories.add("retry")
+    if data.get("recovery_of") or data.get("recovered") is True:
+        categories.add("recovery")
+    if data.get("privacy_transformed") is True:
+        categories.add("privacy")
+    return categories
+
+
+def build_review_projection(events: Iterable[TraceEvent]) -> ReviewProjection:
+    """Build a stable review projection without guessing provider semantics.
+
+    Parent links are checked only against event IDs present in the supplied event
+    stream.  Missing parents are surfaced as ``orphaned`` and categorized as a
+    gap; no synthetic parent is created.
+    """
+    indexed = list(enumerate(events))
+    known_ids = {event.event_id for _, event in indexed if event.event_id}
+    items: list[ReviewItem] = []
+    orphaned: list[str] = []
+
+    for source_index, event in indexed:
+        categories = _explicit_categories(event)
+        parent_id = event.parent_id or None
+        if parent_id is None:
+            relationship: RelationshipState = "none"
+        elif parent_id in known_ids:
+            relationship = "linked"
+        else:
+            relationship = "orphaned"
+            categories.add("gaps")
+            orphaned.append(event.event_id)
+
+        items.append(
+            ReviewItem(
+                event_id=event.event_id,
+                event_type=event.event_type.value,
+                timestamp=event.timestamp,
+                parent_id=parent_id,
+                relationship=relationship,
+                categories=tuple(sorted(categories)),
+                redacted=event.redacted,
+                source_index=source_index,
+                data=dict(event.data),
+            )
+        )
+
+    # Timestamp is the primary execution order. source_index makes ties stable
+    # and preserves recorded order when providers emit identical timestamps.
+    items.sort(key=lambda item: (item.timestamp, item.source_index))
+    return ReviewProjection(
+        schema_version=1,
+        items=tuple(items),
+        orphaned_event_ids=tuple(orphaned),
+    )

tests/test_review.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import unittest
+
+from agent_trace.models import EventType, TraceEvent
+from agent_trace.review import build_review_projection
+
+
+class ReviewProjectionTests(unittest.TestCase):
+    def test_orders_events_and_links_tool_result(self) -> None:
+        call = TraceEvent(
+            EventType.TOOL_CALL,
+            timestamp=2.0,
+            event_id="call-1",
+            data={"name": "shell"},
+        )
+        result = TraceEvent(
+            EventType.TOOL_RESULT,
+            timestamp=3.0,
+            event_id="result-1",
+            parent_id="call-1",
+            data={"output": "ok"},
+        )
+        prompt = TraceEvent(EventType.USER_PROMPT, timestamp=1.0, event_id="prompt-1")
+
+        projection = build_review_projection([call, result, prompt])
+
+        self.assertEqual([item.event_id for item in projection.items], ["prompt-1", "call-1", "result-1"])
+        result_item = next(item for item in projection.items if item.event_id == "result-1")
+        self.assertEqual(result_item.relationship, "linked")
+        self.assertIn("tools", result_item.categories)
+
+    def test_surfaces_orphan_as_gap_without_synthetic_parent(self) -> None:
+        orphan = TraceEvent(
+            EventType.TOOL_RESULT,
+            timestamp=1.0,
+            event_id="orphan-1",
+            parent_id="missing-call",
+        )
+
+        projection = build_review_projection([orphan])
+
+        self.assertEqual(projection.orphaned_event_ids, ("orphan-1",))
+        self.assertEqual(projection.items[0].relationship, "orphaned")
+        self.assertIn("gaps", projection.items[0].categories)
+        self.assertEqual(projection.items[0].parent_id, "missing-call")
+
+    def test_filters_explicit_review_categories(self) -> None:
+        events = [
+            TraceEvent(EventType.FILE_WRITE, timestamp=1, event_id="file"),
+            TraceEvent(
+                EventType.TOOL_CALL,
+                timestamp=2,
+                event_id="test",
+                data={"is_test": True, "is_command": True},
+            ),
+            TraceEvent(
+                EventType.ERROR,
+                timestamp=3,
+                event_id="failure",
+                data={"retry": True},
+            ),
+            TraceEvent(
+                EventType.TOOL_RESULT,
+                timestamp=4,
+                event_id="recovery",
+                data={"recovered": True},
+            ),
+        ]
+
+        projection = build_review_projection(events)
+
+        self.assertEqual([item.event_id for item in projection.filter("files")], ["file"])
+        self.assertEqual([item.event_id for item in projection.filter("tests")], ["test"])
+        self.assertEqual([item.event_id for item in projection.filter("commands")], ["test"])
+        self.assertEqual([item.event_id for item in projection.filter("failures")], ["failure"])
+        self.assertEqual([item.event_id for item in projection.filter("retry")], ["failure"])
+        self.assertEqual([item.event_id for item in projection.filter("recovery")], ["recovery"])
+
+    def test_redaction_and_privacy_transform_are_visible(self) -> None:
+        redacted = TraceEvent(
+            EventType.USER_PROMPT,
+            timestamp=1,
+            event_id="redacted",
+            redacted=True,
+            data={"privacy_transformed": True},
+        )
+
+        projection = build_review_projection([redacted])
+
+        item = projection.items[0]
+        self.assertTrue(item.redacted)
+        self.assertIn("privacy", item.categories)
+        self.assertEqual(projection.to_dict()["schema_version"], 1)
+
+    def test_timestamp_ties_preserve_recorded_order(self) -> None:
+        first = TraceEvent(EventType.USER_PROMPT, timestamp=1, event_id="first")
+        second = TraceEvent(EventType.ASSISTANT_RESPONSE, timestamp=1, event_id="second")
+
+        projection = build_review_projection([first, second])
+
+        self.assertEqual([item.event_id for item in projection.items], ["first", "second"])
+
+
+if __name__ == "__main__":
+    unittest.main()