fix(p0): Stop hook transitions to COMPLETED on explicit signal (CL-3)

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 []
 
 

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 == []