diff --git a/api/session_recovery.py b/api/session_recovery.py index 10f9d394..d00c6d68 100644 --- a/api/session_recovery.py +++ b/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() diff --git a/api/turn_journal.py b/api/turn_journal.py index f25268a4..628d2175 100644 --- a/api/turn_journal.py +++ b/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() diff --git a/tests/test_turn_journal.py b/tests/test_turn_journal.py index c5626358..bfefc407 100644 --- a/tests/test_turn_journal.py +++ b/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 == [] diff --git a/tests/test_turn_journal_lifecycle.py b/tests/test_turn_journal_lifecycle.py index ea7ae704..3f5a317a 100644 --- a/tests/test_turn_journal_lifecycle.py +++ b/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"