diff --git a/bridge/hook_listener.py b/bridge/hook_listener.py index 10ef640..c4444ee 100644 --- a/bridge/hook_listener.py +++ b/bridge/hook_listener.py @@ -215,11 +215,32 @@ def _handle_user_prompt_submit(listener: HookListener, event: Dict[str, Any], def _handle_stop(listener: HookListener, event: Dict[str, Any], payload: Dict[str, Any], session_id: str) -> List[StateChange]: + """Per PRD §CL-3: Stop events transition the session to COMPLETED + when an explicit completion reason is provided; otherwise they + arm the idle-WAITING timer. + + The Claude Code Stop hook payload may include `reason`, `final`, or + `session_ended` flags signaling the session is concluding. Without + such a signal, we treat Stop as "agent yielded control, awaiting + next prompt" — that's the WAITING semantics handled by the idle + timer. With an explicit signal, we transition to COMPLETED. + """ session = listener._find_session(session_id) if session is None: return [] - # Mark the stop time; tick_idle_check will flip to WAITING after the threshold. + # Always record the stop time so tick_idle_check can fire if no + # subsequent productive event arrives. listener._last_stop_time[session.session_id] = time.time() + + # Inspect payload for explicit-completion signals. + is_final = bool( + payload.get("final") + or payload.get("session_ended") + or payload.get("reason") == "session_complete" + or payload.get("reason") == "completed" + ) + if is_final and session.status != SessionStatus.COMPLETED: + return [listener._set_status(session, SessionStatus.COMPLETED, note="stop_final")] return [] diff --git a/bridge/tests/test_stop_to_completed.py b/bridge/tests/test_stop_to_completed.py new file mode 100644 index 0000000..1152539 --- /dev/null +++ b/bridge/tests/test_stop_to_completed.py @@ -0,0 +1,119 @@ +"""Tests for the Stop → COMPLETED transition (P0 #3 from CTO audit). + +PRD §CL-3: 'Detect Stop events to transition a session to Completed.' +The handler accepts several explicit-completion signals from the +Stop hook payload; absent any signal, Stop only arms the idle-WAITING +timer (preserving the existing semantics). +""" + +import time + +import pytest + +from bridge.hook_listener import HookListener +from bridge.session_state import SessionState, SessionStatus + + +def _state_with_active(sid: str = "s1") -> SessionState: + state = SessionState(kingdom_name="Camelot") + s = state.add_session("Royal Court") + s.session_id = sid + s.status = SessionStatus.ACTIVE + return state + + +def test_stop_with_final_flag_transitions_to_completed(): + state = _state_with_active() + listener = HookListener(state=state) + changes = listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1", "final": True}, + }) + assert state.sessions[0].status == SessionStatus.COMPLETED + assert any(c.note == "stop_final" for c in changes) + + +def test_stop_with_session_ended_flag_transitions_to_completed(): + state = _state_with_active() + listener = HookListener(state=state) + listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1", "session_ended": True}, + }) + assert state.sessions[0].status == SessionStatus.COMPLETED + + +def test_stop_with_reason_session_complete_transitions_to_completed(): + state = _state_with_active() + listener = HookListener(state=state) + listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1", "reason": "session_complete"}, + }) + assert state.sessions[0].status == SessionStatus.COMPLETED + + +def test_stop_with_reason_completed_transitions_to_completed(): + state = _state_with_active() + listener = HookListener(state=state) + listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1", "reason": "completed"}, + }) + assert state.sessions[0].status == SessionStatus.COMPLETED + + +def test_stop_without_signal_only_arms_idle_timer(): + """Backward-compat: a vanilla Stop with no completion signal still + just arms the idle timer so tick_idle_check eventually flips + ACTIVE → WAITING. Status stays ACTIVE in the immediate response. + """ + state = _state_with_active() + listener = HookListener(state=state, idle_threshold_sec=30) + changes = listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1"}, + }) + assert state.sessions[0].status == SessionStatus.ACTIVE + assert changes == [] + # Stop time recorded so the next tick_idle_check (after threshold) + # fires the WAITING transition. + fake_now = time.time() + 31 + later_changes = listener.tick_idle_check(now=fake_now) + assert state.sessions[0].status == SessionStatus.WAITING + assert any(c.note == "idle_threshold_crossed" for c in later_changes) + + +def test_stop_with_unrelated_reason_does_not_complete(): + state = _state_with_active() + listener = HookListener(state=state) + changes = listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1", "reason": "interrupted"}, + }) + assert state.sessions[0].status == SessionStatus.ACTIVE + assert changes == [] + + +def test_stop_idempotent_when_already_completed(): + state = _state_with_active() + state.sessions[0].status = SessionStatus.COMPLETED + listener = HookListener(state=state) + changes = listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "s1", "final": True}, + }) + # Already COMPLETED — no spurious status change emitted. + assert state.sessions[0].status == SessionStatus.COMPLETED + assert changes == [] + + +def test_stop_for_unknown_session_is_noop(): + state = _state_with_active() + listener = HookListener(state=state) + changes = listener.feed({ + "event_type": "Stop", + "payload": {"session_id": "sess-other", "final": True}, + }) + assert state.sessions[0].status == SessionStatus.ACTIVE + assert changes == []