diff --git a/agent/main.py b/agent/main.py index 6f29c06b..67e04933 100644 --- a/agent/main.py +++ b/agent/main.py @@ -846,7 +846,22 @@ def _cancel_event(): ) await submission_queue.put(approval_submission) console.print() # spacing after approval - # Silently ignore other events + elif event.event_type == "session_terminated": + shimmer.stop() + stream_buf.discard() + data = event.data or {} + user_message = data.get( + "user_message", "Session terminated unexpectedly." + ) + reason = data.get("reason", "unknown") + print_error(f"{user_message} (reason: {reason})") + turn_complete_event.set() + else: + logger.warning( + "Unhandled event type %r with data: %s", + event.event_type, + event.data, + ) except asyncio.CancelledError: break diff --git a/frontend/src/hooks/useAgentChat.ts b/frontend/src/hooks/useAgentChat.ts index 1fbd17b8..05758273 100644 --- a/frontend/src/hooks/useAgentChat.ts +++ b/frontend/src/hooks/useAgentChat.ts @@ -416,6 +416,9 @@ export function useAgentChat({ sessionId, isActive, isProcessing = false, onRead } }, onInterrupted: () => { /* no-op — handled by stop() caller */ }, + onSessionTerminated: (reason: string, userMessage: string) => { + logger.warn(`Session terminated (reason: ${reason}): ${userMessage}`); + }, onRecoverMessages: async ({ submittedText, currentMessageCount, @@ -777,6 +780,25 @@ export function useAgentChat({ sessionId, isActive, isProcessing = false, onRead remaining_usd?: number | null; }); } + } else if (et === 'session_terminated') { + const userMessage = (event.data?.user_message as string) || 'Session terminated unexpectedly.'; + sideChannel.onError(userMessage); + sideChannel.onProcessingDone(); + stopReconnect(); + const result = await hydrateMessages(); + if (result) { + const uiMsgs = llmMessagesToUIMessages( + result.data, + result.pendingIds, + chatActionsRef.current.messages, + result.pendingItems, + ); + if (uiMsgs.length > 0) { + chat.setMessages(uiMsgs); + saveMessages(sessionId, uiMsgs); + } + } + return true; } else if (et === 'turn_complete' || et === 'error' || et === 'interrupted') { sideChannel.onProcessingDone(); stopReconnect(); diff --git a/frontend/src/lib/sse-chat-transport.ts b/frontend/src/lib/sse-chat-transport.ts index d19d303c..1e821977 100644 --- a/frontend/src/lib/sse-chat-transport.ts +++ b/frontend/src/lib/sse-chat-transport.ts @@ -42,6 +42,7 @@ export interface SideChannelCallbacks { onUsageEvent: (eventType: 'llm_call' | 'hf_job_complete' | 'sandbox_destroy', data: Record) => void; onSessionUpdate: (data: Record) => void; onInterrupted: () => void; + onSessionTerminated: (reason: string, userMessage: string) => void; onRecoverMessages: (context: MessageRecoveryContext) => Promise; } @@ -458,6 +459,18 @@ function createEventToChunkStream(sideChannel: SideChannelCallbacks): TransformS break; } + case 'session_terminated': { + const reason = (event.data?.reason as string) || 'unknown'; + const userMessage = (event.data?.user_message as string) || 'Session terminated unexpectedly.'; + endTextPart(controller); + controller.enqueue({ type: 'finish-step' }); + controller.enqueue({ type: 'finish', finishReason: 'error' }); + sideChannel.onSessionTerminated(reason, userMessage); + sideChannel.onError(userMessage); + sideChannel.onProcessingDone(); + break; + } + default: logger.log('SSE transport: unknown event', event); } diff --git a/frontend/src/types/events.ts b/frontend/src/types/events.ts index 1bef1e4d..0fe38b43 100644 --- a/frontend/src/types/events.ts +++ b/frontend/src/types/events.ts @@ -24,6 +24,7 @@ export type EventType = | 'shutdown' | 'interrupted' | 'undo_complete' + | 'session_terminated' | 'plan_update'; export interface AgentEvent { diff --git a/pyproject.toml b/pyproject.toml index c52b1ebb..11ca3106 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,3 +80,4 @@ package = true [tool.pytest.ini_options] asyncio_mode = "auto" +pythonpath = "." diff --git a/tests/unit/test_session_terminated_dispatch.py b/tests/unit/test_session_terminated_dispatch.py new file mode 100644 index 00000000..e16f4976 --- /dev/null +++ b/tests/unit/test_session_terminated_dispatch.py @@ -0,0 +1,131 @@ +"""Tests for session_terminated event dispatch in the CLI event listener. + +Covers two requirements from issue #345: +1. ``session_terminated`` events are explicitly handled (error printed, + turn_complete_event set) instead of silently dropped. +2. Unknown/unhandled event types now produce a ``logger.warning`` instead + of silently passing — a regression guard for this entire bug class. +""" + +from __future__ import annotations + +import asyncio +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from agent.core.session import Event + + +# ── helpers ──────────────────────────────────────────────────────────── + + +def _make_queues(): + return asyncio.Queue(), asyncio.Queue() + + +def _make_events(): + return asyncio.Event(), asyncio.Event() + + +async def _run_listener_with_events( + events: list[Event], + *, + config=None, + timeout: float = 2.0, +): + """Feed *events* into the event_listener, then send a shutdown to stop it. + + Returns the turn_complete_event so callers can assert on it. + """ + from agent.main import event_listener + + event_queue, submission_queue = _make_queues() + turn_complete_event, ready_event = _make_events() + + # Minimal prompt_session stub (not exercised for these event types) + prompt_session = SimpleNamespace(prompt_async=AsyncMock(return_value="n")) + + if config is None: + config = SimpleNamespace(yolo_mode=False) + + for ev in events: + await event_queue.put(ev) + + # Sentinel to break the listener's while-True loop + await event_queue.put(Event(event_type="shutdown")) + + await asyncio.wait_for( + event_listener( + event_queue, + submission_queue, + turn_complete_event, + ready_event, + prompt_session, + config, + session_holder=[None], + ), + timeout=timeout, + ) + return turn_complete_event + + +# ── session_terminated handling ──────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_event_listener_handles_session_terminated(capsys): + """session_terminated must print an error and set turn_complete_event.""" + ev = Event( + event_type="session_terminated", + data={ + "reason": "compaction_failed", + "user_message": "Conversation too large to continue.", + }, + ) + + turn_complete = await _run_listener_with_events([ev]) + + assert turn_complete.is_set(), ( + "turn_complete_event must be set so the CLI prompt loop doesn't hang" + ) + + captured = capsys.readouterr() + assert "compaction_failed" in captured.out or "compaction_failed" in captured.err + + +@pytest.mark.asyncio +async def test_session_terminated_uses_default_message_when_data_missing(capsys): + """If data is None the handler must still work with defaults.""" + ev = Event(event_type="session_terminated", data=None) + + turn_complete = await _run_listener_with_events([ev]) + + assert turn_complete.is_set() + captured = capsys.readouterr() + # The default reason should appear + assert "unknown" in captured.out or "unknown" in captured.err + + +# ── unknown event type warning ───────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_event_listener_warns_on_unknown_event_type(caplog): + """An event type with no explicit handler must produce a warning log.""" + ev = Event( + event_type="completely_novel_event_type", + data={"some": "payload"}, + ) + + with caplog.at_level(logging.WARNING, logger="agent.main"): + await _run_listener_with_events([ev]) + + assert any( + "completely_novel_event_type" in record.message for record in caplog.records + ), ( + "Unknown event types must trigger a logger.warning with the event type name. " + f"Got log records: {[r.message for r in caplog.records]}" + )