fix: audit turn journal terminal collisions

api/session_recovery.py

@@ -499,7 +499,7 @@ def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None)
 
     for session_id in iter_turn_journal_session_ids(session_dir):
         journal = read_turn_journal(session_id, session_dir=session_dir)
-        states = derive_turn_journal_states(journal.get('events') or [])
+        states, _ = derive_turn_journal_states(journal.get('events') or [])
         live_path = session_dir / f"{session_id}.json"
         live_messages = _msg_count(live_path)
         existing_user_messages: set[str] = set()

api/turn_journal.py

@@ -102,20 +102,43 @@ def read_turn_journal(session_id: str, *, session_dir: Path | None = None) -> di
     return {"session_id": str(session_id), "events": events, "malformed": malformed}
 
 
-def derive_turn_journal_states(events: Iterable[dict]) -> dict[str, dict]:
-    """Return the latest event per ``turn_id``."""
+def derive_turn_journal_states(events: Iterable[dict]) -> tuple[dict[str, dict], list[dict]]:
+    '''Return the latest event per ``turn_id`` and any terminal-collision entries.
+
+    The first element is the latest event per turn_id (same overwrite-by-timestamp
+    behaviour as before).  The second element is a list of collision records, one
+    per turn_id that had more than one terminal event.  Each collision record
+    contains ``turn_id`` and the ``events`` list (in ascending created_at order).
+
+    A collision means the same logical turn recorded both ``completed`` and
+    ``interrupted`` terminal events -- the derived state still picks the latest
+    by timestamp, but callers can now detect and audit the double-terminal
+    situation explicitly rather than having it silently collapse.
+    '''
     states: dict[str, dict] = {}
+    # Collect all terminal events per turn_id to detect collisions
+    terminal_events: dict[str, list[dict]] = {}
     for event in events:
         if not isinstance(event, dict):
             continue
-        turn_id = str(event.get("turn_id") or "").strip()
+        turn_id = str(event.get('turn_id') or '').strip()
         if not turn_id:
             continue
+        # Track terminal events for collision detection
+        if is_terminal_turn_event(event):
+            terminal_events.setdefault(turn_id, []).append(event)
+        # Existing latest-by-timestamp derivation
         previous = states.get(turn_id)
-        if previous is None or float(event.get("created_at") or 0) >= float(previous.get("created_at") or 0):
+        if previous is None or float(event.get('created_at') or 0) >= float(previous.get('created_at') or 0):
             states[turn_id] = event
-    return states
 
+    # Build collision list: turn_ids with more than one terminal event
+    collisions = [
+        {'turn_id': tid, 'events': sorted(evts, key=lambda e: float(e.get('created_at') or 0))}
+        for tid, evts in terminal_events.items()
+        if len(evts) > 1
+    ]
+    return states, collisions
 
 def _latest_turn_id_for_stream(events: Iterable[dict], stream_id: str) -> str | None:
     stream = str(stream_id or "").strip()

tests/test_turn_journal.py

@@ -58,7 +58,7 @@ def test_read_turn_journal_tolerates_malformed_lines(tmp_path):
 
 
 def test_derive_turn_journal_states_keeps_latest_event_per_turn():
-    states = derive_turn_journal_states([
+    states, _ = derive_turn_journal_states([
         {"event": "submitted", "turn_id": "turn-1", "created_at": 1},
         {"event": "worker_started", "turn_id": "turn-1", "created_at": 2},
         {"event": "submitted", "turn_id": "turn-2", "created_at": 3},
@@ -70,7 +70,7 @@ def test_derive_turn_journal_states_keeps_latest_event_per_turn():
 
 
 def test_derive_turn_journal_states_uses_created_at_not_file_order():
-    states = derive_turn_journal_states([
+    states, _ = derive_turn_journal_states([
         {"event": "completed", "turn_id": "turn-1", "created_at": 20},
         {"event": "submitted", "turn_id": "turn-1", "created_at": 10},
     ])
@@ -133,3 +133,35 @@ def test_audit_ignores_completed_or_already_materialized_turn_journal_entry(tmp_
 
     assert report["status"] == "ok"
     assert report["items"] == []
+
+
+def test_derive_turn_journal_states_reports_terminal_collision_when_both_completed_and_interrupted():
+    # A turn that recorded both completed and interrupted terminal events should
+    # not silently collapse to one winner — the collision must be reported.
+    events = [
+        {'event': 'submitted', 'turn_id': 'turn-double-terminal', 'created_at': 1},
+        {'event': 'worker_started', 'turn_id': 'turn-double-terminal', 'created_at': 2},
+        {'event': 'completed', 'turn_id': 'turn-double-terminal', 'created_at': 3},
+        {'event': 'interrupted', 'turn_id': 'turn-double-terminal', 'created_at': 4, 'reason': 'server_restart'},
+    ]
+    states, collisions = derive_turn_journal_states(events)
+
+    # Derived state still picks the latest by timestamp (interrupted)
+    assert states['turn-double-terminal']['event'] == 'interrupted'
+    # But the collision is explicitly reported so callers can audit it
+    assert len(collisions) == 1
+    assert collisions[0]['turn_id'] == 'turn-double-terminal'
+    assert [e['event'] for e in collisions[0]['events']] == ['completed', 'interrupted']
+
+
+def test_derive_turn_journal_states_no_collision_when_single_terminal():
+    # A normal turn with only one terminal event must not produce a collision.
+    events = [
+        {'event': 'submitted', 'turn_id': 'turn-normal', 'created_at': 1},
+        {'event': 'worker_started', 'turn_id': 'turn-normal', 'created_at': 2},
+        {'event': 'completed', 'turn_id': 'turn-normal', 'created_at': 3},
+    ]
+    states, collisions = derive_turn_journal_states(events)
+
+    assert states['turn-normal']['event'] == 'completed'
+    assert collisions == []

tests/test_turn_journal_lifecycle.py

@@ -21,7 +21,7 @@ def test_append_turn_journal_event_for_stream_reuses_submitted_turn_id(tmp_path)
 
     assert submitted["turn_id"] == "turn-1"
     assert worker["turn_id"] == "turn-1"
-    states = derive_turn_journal_states([submitted, worker])
+    states, _ = derive_turn_journal_states([submitted, worker])
     assert states["turn-1"]["event"] == "worker_started"