From e22744686f31ecf794bac202b5213cd089917bc5 Mon Sep 17 00:00:00 2001 From: dimakis Date: Thu, 2 Jul 2026 10:50:33 +0100 Subject: [PATCH 1/5] =?UTF-8?q?feat(transport):=20p1=20server-authoritativ?= =?UTF-8?q?e=20state=20=E2=80=94=20replace=20SET=5FRUNNING=20with=20sessio?= =?UTF-8?q?n=5Fstate=5Fchanged?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Transport SSOT Phase 1: client reads running state from server events only. Server: - handleReconnect uses getSessionState() instead of getSession().isActive - Remove redundant markSessionInactive() in query-loop (setSessionState syncs is_active) - Remove running field from reconnected and session_switched messages - connection-registry isSessionActive uses state instead of is_active column Client: - Activate session_state_changed handler to dispatch SESSION_STATE_CHANGED action - Replace SET_RUNNING action with SESSION_STATE_CHANGED (derives running from state) - Remove syncRunningState() REST polling and foreground call site - Remove running field handling from reconnected/session_switched/takeover/subscribed parsers - handleStop uses SESSION_STATE_CHANGED instead of SET_RUNNING Dead code removed: SET_RUNNING action type, syncRunningState(), runningSyncInFlight, running field in reconnected summaries and session_switched messages. Design doc: docs/design/transport-ssot.md (on design/transport-ssot branch) Depends on: PR #431 (P0 foundation) Co-Authored-By: Claude Opus 4.6 --- frontend/src/pages/ChatView.tsx | 2 +- frontend/src/pages/DesktopChatView.tsx | 2 +- .../client/__tests__/protocol-parser.test.ts | 109 +++---------- packages/client/__tests__/store.test.ts | 145 +----------------- packages/client/src/protocol-parser.ts | 61 ++------ packages/client/src/slices/messages.ts | 7 +- packages/client/src/store.ts | 34 +--- server/__tests__/ws-handler-v2.test.ts | 145 ++---------------- server/index.ts | 7 +- server/query-loop.ts | 3 +- server/ws-handler-v2.ts | 40 ++--- 11 files changed, 92 insertions(+), 463 deletions(-) diff --git a/frontend/src/pages/ChatView.tsx b/frontend/src/pages/ChatView.tsx index fbbc9702..7e7d4618 100644 --- a/frontend/src/pages/ChatView.tsx +++ b/frontend/src/pages/ChatView.tsx @@ -185,7 +185,7 @@ export function ChatView() { const handleStop = useCallback(() => { storeStopGeneration(); - storeDispatchMessages({ type: 'SET_RUNNING', running: false }); + storeDispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); }, [storeStopGeneration, storeDispatchMessages]); function handlePermission( diff --git a/frontend/src/pages/DesktopChatView.tsx b/frontend/src/pages/DesktopChatView.tsx index 2d897d60..d4815f44 100644 --- a/frontend/src/pages/DesktopChatView.tsx +++ b/frontend/src/pages/DesktopChatView.tsx @@ -148,7 +148,7 @@ export function DesktopChatView() { const handleStop = useCallback(() => { storeStopGeneration(); - storeDispatchMessages({ type: 'SET_RUNNING', running: false }); + storeDispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); }, [storeStopGeneration, storeDispatchMessages]); function handlePermission( diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index 109afa42..4ff68a5e 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -50,7 +50,7 @@ describe('pool lifecycle events', () => { // ─── Reattach ───────────────────────────────────────────────────────────────── describe('reattach', () => { - it('reattached sets running and calls onSessionAssigned', () => { + it('reattached calls onSessionAssigned and setWsRunning (P1: no SET_RUNNING)', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'reattached', clientId: 'c1', sessionId: 'sid-1', running: true }, @@ -58,13 +58,14 @@ describe('reattach', () => { cb, POOL_KEY, ); - expect(r.messagesActions).toEqual([{ type: 'SET_RUNNING', running: true }]); + // P1: running state comes from session_state_changed events, not reattached + expect(r.messagesActions).toHaveLength(0); expect(cb.onSessionAssigned).toHaveBeenCalledWith('sid-1'); expect(cb.setWsRunning).toHaveBeenCalledWith(POOL_KEY, true); expect(r.connectionUpdate).toEqual({ status: 'connected' }); }); - it('reattach_failed sets running=false and marks connection connected', () => { + it('reattach_failed marks connection connected (P1: no SET_RUNNING)', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'reattach_failed', clientId: 'old' }, @@ -72,7 +73,8 @@ describe('reattach', () => { cb, POOL_KEY, ); - expect(r.messagesActions).toEqual([{ type: 'SET_RUNNING', running: false }]); + // P1: running state comes from session_state_changed events + expect(r.messagesActions).toHaveLength(0); expect(cb.setWsRunning).toHaveBeenCalledWith(POOL_KEY, false); expect(r.connectionUpdate).toEqual({ status: 'connected' }); }); @@ -165,7 +167,8 @@ describe('session lifecycle', () => { const cb = makeCallbacks(); const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, state, cb, POOL_KEY); expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' }); - expect(r.messagesActions).toContainEqual({ type: 'SET_RUNNING', running: true }); + // P1: no SET_RUNNING — running state from server's session_state_changed + expect(r.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); expect(cb.sendQueued).toHaveBeenCalledWith(POOL_KEY, { type: 'send', prompt: 'follow-up' }); // Second message stays queued expect(state.pendingSend).toEqual([{ type: 'send', prompt: 'second' }]); @@ -385,7 +388,7 @@ describe('error handling', () => { // ─── Subscribed ────────────────────────────────────────────────────────────── describe('subscribed', () => { - it('subscribed with running=true sets running', () => { + it('subscribed with running=true calls setWsRunning (P1: no SET_RUNNING)', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'subscribed', sessionId: 'sid', running: true }, @@ -393,7 +396,8 @@ describe('subscribed', () => { cb, POOL_KEY, ); - expect(r.messagesActions).toEqual([{ type: 'SET_RUNNING', running: true }]); + // P1: running state from session_state_changed events, not subscribed + expect(r.messagesActions).toHaveLength(0); expect(cb.setWsRunning).toHaveBeenCalledWith(POOL_KEY, true); }); @@ -591,7 +595,7 @@ describe('misc', () => { // ─── session_takeover ──────────────────────────────────────────────────────── describe('session_takeover', () => { - it('produces SET_RUNNING false and ERROR message', () => { + it('produces ERROR message (P1: no SET_RUNNING, state from server event)', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'session_takeover', sessionId: 'sess-1' }, @@ -599,7 +603,8 @@ describe('session_takeover', () => { cb, POOL_KEY, ); - expect(r.messagesActions).toContainEqual({ type: 'SET_RUNNING', running: false }); + // P1: running=false comes from server's session_state_changed event + expect(r.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); expect(r.messagesActions).toContainEqual( expect.objectContaining({ type: 'ERROR', error: expect.stringContaining('another device') }), ); @@ -616,75 +621,17 @@ describe('reconnected', () => { expect(onReconnected).toHaveBeenCalled(); }); - it('dispatches SET_RUNNING false when active session reports not running', () => { - const state = makeState({ currentSessionId: 'sid-1' }); - const r = parseServerMessage( - { type: 'reconnected', sessions: [{ sessionId: 'sid-1', replayed: 0, running: false }] }, - state, - makeCallbacks(), - POOL_KEY, - ); - expect(r.messagesActions).toContainEqual({ type: 'SET_RUNNING', running: false }); - }); - - it('dispatches SET_RUNNING true when active session is still running', () => { - const setWsRunning = vi.fn(); + it('P1: reconnected does not dispatch SET_RUNNING — state from replayed events', () => { const state = makeState({ currentSessionId: 'sid-1' }); const r = parseServerMessage( - { type: 'reconnected', sessions: [{ sessionId: 'sid-1', replayed: 0, running: true }] }, - state, - makeCallbacks({ setWsRunning }), - POOL_KEY, - ); - expect(r.messagesActions).toContainEqual({ type: 'SET_RUNNING', running: true }); - expect(setWsRunning).toHaveBeenCalledWith(POOL_KEY, true); - }); - - it('no-ops when no currentSessionId', () => { - const state = makeState({ currentSessionId: undefined }); - const r = parseServerMessage( - { type: 'reconnected', sessions: [{ sessionId: 'sid-1', replayed: 0, running: false }] }, + { type: 'reconnected', sessions: [{ sessionId: 'sid-1', replayed: 3 }] }, state, makeCallbacks(), POOL_KEY, ); - expect(r.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); - }); - - it('no-ops when sessions field is undefined (backward compat)', () => { - const state = makeState({ currentSessionId: 'sid-1' }); - const r = parseServerMessage({ type: 'reconnected' }, state, makeCallbacks(), POOL_KEY); - expect(r.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); - }); - - it('no-ops when sessions has invalid shape (runtime validation)', () => { - const state = makeState({ currentSessionId: 'sid-1' }); - // Invalid: sessions is not an array - const r1 = parseServerMessage( - { type: 'reconnected', sessions: { invalid: 'shape' } }, - state, - makeCallbacks(), - POOL_KEY, - ); - expect(r1.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); - - // Invalid: array contains entries missing required fields - const r2 = parseServerMessage( - { type: 'reconnected', sessions: [{ sessionId: 'sid-1' }] }, // missing running field - state, - makeCallbacks(), - POOL_KEY, - ); - expect(r2.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); - - // Invalid: running field has wrong type - const r3 = parseServerMessage( - { type: 'reconnected', sessions: [{ sessionId: 'sid-1', running: 'yes' }] }, - state, - makeCallbacks(), - POOL_KEY, - ); - expect(r3.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); + // Running state comes from replayed session_state_changed events, not reconnected payload + expect(r.messagesActions).toHaveLength(0); + expect(r.connectionUpdate).toEqual({ status: 'connected' }); }); }); @@ -965,10 +912,10 @@ describe('boot_context', () => { }); }); -// ─── Session state (Transport SSOT P0) ────────────────────────────────────── +// ─── Session state (Transport SSOT P1) ────────────────────────────────────── describe('session_state_changed', () => { - it('produces no message actions (P0: observability only)', () => { + it('dispatches SESSION_STATE_CHANGED action with state', () => { const r = parseServerMessage( { type: 'session_state_changed', @@ -981,12 +928,11 @@ describe('session_state_changed', () => { makeCallbacks(), POOL_KEY, ); - expect(r.messagesActions).toHaveLength(0); + expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'running' }); }); - it('logs via console.debug', () => { - const spy = vi.spyOn(console, 'debug').mockImplementation(() => {}); - parseServerMessage( + it('dispatches idle state on ENDED', () => { + const r = parseServerMessage( { type: 'session_state_changed', sessionId: 'sid-1', @@ -998,12 +944,7 @@ describe('session_state_changed', () => { makeCallbacks(), POOL_KEY, ); - expect(spy).toHaveBeenCalledWith('[mitzo] session_state_changed', { - sessionId: 'sid-1', - state: 'idle', - internalState: 'ENDED', - }); - spy.mockRestore(); + expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); }); }); diff --git a/packages/client/__tests__/store.test.ts b/packages/client/__tests__/store.test.ts index 2362b0f2..b81cf40a 100644 --- a/packages/client/__tests__/store.test.ts +++ b/packages/client/__tests__/store.test.ts @@ -548,7 +548,7 @@ describe('WS → store wiring', () => { }); it('dispatches session_end and clears running', () => { - store.getState().dispatchMessages({ type: 'SET_RUNNING', running: true }); + store.getState().dispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'running' }); expect(store.getState().messages.running).toBe(true); lastWs.simulateMessage({ type: 'session_end', sessionId: 'test-session' }); @@ -859,7 +859,7 @@ describe('session isolation via sessionId filtering', () => { it('rejects session_end from a foreign session when active session is set', async () => { const store = createReadyStore(); await store.getState().switchSession('session-b'); - store.getState().dispatchMessages({ type: 'SET_RUNNING', running: true }); + store.getState().dispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'running' }); lastWs.simulateMessage({ type: 'session_end', @@ -909,7 +909,7 @@ describe('session isolation via sessionId filtering', () => { it('drops session_end when no active session (null-session filter)', async () => { const store = createReadyStore(); await store.getState().switchSession('old-session'); - store.getState().dispatchMessages({ type: 'SET_RUNNING', running: true }); + store.getState().dispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'running' }); store.getState().newSession(); @@ -967,7 +967,7 @@ describe('setModel', () => { describe('dispatchMessages', () => { it('applies messages reducer action directly', () => { const store = createReadyStore(); - store.getState().dispatchMessages({ type: 'SET_RUNNING', running: true }); + store.getState().dispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'running' }); expect(store.getState().messages.running).toBe(true); }); }); @@ -1299,139 +1299,6 @@ describe('foreground recovery', () => { expect(state.messages[0].messageId).toBe('user-1'); }); - it('syncs running state from session meta on foreground when session_end was missed', async () => { - const transport = mockTransport(); - const store = createReadyStore(transport); - - // Set up active session with running=true (agent was processing) - store.setState((s) => ({ - sessions: { ...s.sessions, active: 'sess-1' }, - messages: { ...s.messages, running: true }, - })); - - // Mock fetch to return messages for /messages and isActive=false for /meta - (transport.fetch as ReturnType).mockImplementation((url: string) => { - if (typeof url === 'string' && url.includes('/meta')) { - return Promise.resolve({ - ok: true, - json: () => Promise.resolve({ sessionId: 'sess-1', isActive: false }), - }); - } - // /messages endpoint - return Promise.resolve({ - ok: true, - json: () => - Promise.resolve([ - { messageId: 'msg-1', role: 'assistant', blocks: [], isStreaming: false }, - ]), - }); - }); - - // Simulate foreground return (iOS Safari coming back from background) - lastWs.simulateMessage({ type: '_foreground' }); - - // Wait for both async fetches to resolve - await vi.waitFor(() => { - expect(store.getState().messages.running).toBe(false); - }); - }); - - it('does not change running state when session is still active on foreground', async () => { - const transport = mockTransport(); - const store = createReadyStore(transport); - - store.setState((s) => ({ - sessions: { ...s.sessions, active: 'sess-1' }, - messages: { ...s.messages, running: true }, - })); - - (transport.fetch as ReturnType).mockImplementation((url: string) => { - if (typeof url === 'string' && url.includes('/meta')) { - return Promise.resolve({ - ok: true, - json: () => Promise.resolve({ sessionId: 'sess-1', isActive: true }), - }); - } - return Promise.resolve({ - ok: true, - json: () => Promise.resolve([]), - }); - }); - - lastWs.simulateMessage({ type: '_foreground' }); - await new Promise((r) => setTimeout(r, 50)); - - expect(store.getState().messages.running).toBe(true); - }); - - it('does not clear running when session switches before meta fetch resolves', async () => { - const transport = mockTransport(); - const store = createReadyStore(transport); - - store.setState((s) => ({ - sessions: { ...s.sessions, active: 'sess-old' }, - messages: { ...s.messages, running: true }, - })); - - // Meta fetch for sess-old returns isActive=false, but with a delay - (transport.fetch as ReturnType).mockImplementation((url: string) => { - if (typeof url === 'string' && url.includes('/meta')) { - return new Promise((resolve) => - setTimeout( - () => - resolve({ - ok: true, - json: () => Promise.resolve({ sessionId: 'sess-old', isActive: false }), - }), - 30, - ), - ); - } - return Promise.resolve({ - ok: true, - json: () => Promise.resolve([]), - }); - }); - - // Trigger foreground — meta fetch starts for sess-old - lastWs.simulateMessage({ type: '_foreground' }); - - // User switches to a new session before meta fetch resolves - await store.getState().switchSession('sess-new'); - store.setState((s) => ({ - messages: { ...s.messages, running: true }, - })); - - // Wait for the delayed meta fetch to resolve - await new Promise((r) => setTimeout(r, 60)); - - // running should still be true — stale meta response was discarded - expect(store.getState().messages.running).toBe(true); - }); - - it('preserves running state when meta fetch fails (network error)', async () => { - const transport = mockTransport(); - const store = createReadyStore(transport); - - store.setState((s) => ({ - sessions: { ...s.sessions, active: 'sess-1' }, - messages: { ...s.messages, running: true }, - })); - - (transport.fetch as ReturnType).mockImplementation((url: string) => { - if (typeof url === 'string' && url.includes('/meta')) { - return Promise.reject(new Error('network error')); - } - return Promise.resolve({ - ok: true, - json: () => Promise.resolve([]), - }); - }); - - lastWs.simulateMessage({ type: '_foreground' }); - await new Promise((r) => setTimeout(r, 50)); - - // running unchanged — fetch failure is non-fatal - expect(store.getState().messages.running).toBe(true); - }); + // P1: syncRunningState tests removed — running state is now server-authoritative + // via session_state_changed events. No polling on foreground needed. }); diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index 1028dbad..6a04fe8a 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -121,36 +121,15 @@ export function parseServerMessage( // ── v2 handshake events ──────────────────────────────────────────────── - case 'reconnected': { + case 'reconnected': + // P1: running state is derived from replayed session_state_changed events, + // not from the reconnected message payload. result.connectionUpdate = { status: 'connected' }; - // Apply authoritative running state from the server for the active session. - // Validate runtime shape: sessions must be an array, and each entry must have - // sessionId (string) and running (boolean). Explicit running === false check - // guards against undefined/missing field. - const sessions = msg.sessions as unknown; - if ( - Array.isArray(sessions) && - state.currentSessionId && - sessions.every( - (s): s is { sessionId: string; running: boolean } => - typeof s === 'object' && - s !== null && - typeof s.sessionId === 'string' && - typeof s.running === 'boolean', - ) - ) { - const active = sessions.find((s) => s.sessionId === state.currentSessionId); - if (active) { - result.messagesActions.push({ type: 'SET_RUNNING', running: active.running }); - if (active.running) callbacks.setWsRunning?.(poolKey, true); - } - } callbacks.onReconnected?.(); break; - } case 'session_takeover': - result.messagesActions.push({ type: 'SET_RUNNING', running: false }); + // P1: running=false comes from server's session_state_changed event result.messagesActions.push({ type: 'ERROR', error: 'Session resumed on another device.', @@ -162,14 +141,7 @@ export function parseServerMessage( if (tokens) { callbacks.onTokensHydrated?.(tokens); } - // Restore running state from the server so the client UI matches - // the actual session state on reattach. Without this, the client - // defaults to running=false after switchSession resets state, - // causing the first send to go through the normal path even when - // the session is actively generating. - if (typeof msg.running === 'boolean') { - result.messagesActions.push({ type: 'SET_RUNNING', running: msg.running }); - } + // P1: running state restored via replayed session_state_changed events break; } @@ -179,14 +151,14 @@ export function parseServerMessage( // ── v1 handshake events (kept for backward compat) ───────────────────── case 'reattached': - result.messagesActions.push({ type: 'SET_RUNNING', running: true }); + // P1: running state from server's session_state_changed events callbacks.setWsRunning?.(poolKey, true); result.connectionUpdate = { status: 'connected' }; if (msg.sessionId) callbacks.onSessionAssigned(msg.sessionId as string); break; case 'reattach_failed': - result.messagesActions.push({ type: 'SET_RUNNING', running: false }); + // P1: running=false from server's session_state_changed events callbacks.setWsRunning?.(poolKey, false); result.connectionUpdate = { status: 'connected' }; if (state.currentSessionId && callbacks.fetchMessages) { @@ -293,12 +265,10 @@ export function parseServerMessage( break; case 'session_state_changed': - // P0: log for observability, no UI action yet (Phase 1 will bind to running state) - console.debug('[mitzo] session_state_changed', { - sessionId: msg.sessionId, - state: msg.state, - internalState: msg.internalState, - }); + // P1: server-authoritative state — derive running from this event only + if (typeof msg.state === 'string') { + result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: msg.state }); + } break; case 'message_start': @@ -381,14 +351,13 @@ export function parseServerMessage( if (msg.sessionId && !state.currentSessionId) { callbacks.onSessionAssigned(msg.sessionId as string); } + // P1: pending send still drains queued messages, but running state + // comes from server's session_state_changed event (P2 removes pendingSend entirely) const pending = state.pendingSend.shift(); if (pending) { - result.messagesActions.push({ type: 'SET_RUNNING', running: true }); - // v2 path: use onSendQueued callback (no pool key needed) if (callbacks.onSendQueued) { callbacks.onSendQueued(pending); } else { - // v1 fallback callbacks.setWsRunning?.(poolKey, true); callbacks.sendQueued?.(poolKey, pending); } @@ -439,7 +408,7 @@ export function parseServerMessage( command: 'close', content: 'Session closing... The agent will commit work and write a summary.', }); - result.messagesActions.push({ type: 'SET_RUNNING', running: false }); + // P1: running=false from server's session_state_changed (CLOSING → idle) } break; @@ -447,8 +416,8 @@ export function parseServerMessage( break; case 'subscribed': + // P1: running state from session_state_changed events, not subscribed payload if (msg.running) { - result.messagesActions.push({ type: 'SET_RUNNING', running: true }); callbacks.setWsRunning?.(poolKey, true); } break; diff --git a/packages/client/src/slices/messages.ts b/packages/client/src/slices/messages.ts index ace8e2de..e5666884 100644 --- a/packages/client/src/slices/messages.ts +++ b/packages/client/src/slices/messages.ts @@ -16,6 +16,7 @@ import type { StreamingSubagentState, FinishedSubagentState, ToolResultImage, + ClientSessionState, } from '@mitzo/protocol'; // ─── State ─────────────────────────────────────────────────────────────────── @@ -156,7 +157,7 @@ export type MessagesAction = images?: string[]; contextBlocks?: string[]; } - | { type: 'SET_RUNNING'; running: boolean } + | { type: 'SESSION_STATE_CHANGED'; state: ClientSessionState } | { type: 'CONNECTION_LOST' } | { type: 'PERMISSION_REQUEST'; payload: PermissionRequest } | { type: 'PERMISSION_TIMEOUT'; permId: string } @@ -573,8 +574,8 @@ export function messagesReducer(state: MessagesState, action: MessagesAction): M running: true, }; - case 'SET_RUNNING': - return { ...state, running: action.running }; + case 'SESSION_STATE_CHANGED': + return { ...state, running: action.state === 'running' }; case 'CONNECTION_LOST': return { diff --git a/packages/client/src/store.ts b/packages/client/src/store.ts index 3a5efdc2..153021c1 100644 --- a/packages/client/src/store.ts +++ b/packages/client/src/store.ts @@ -190,7 +190,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi { - if (!meta) return; - // Guard: if the user switched sessions while the fetch was in flight, - // don't apply stale isActive=false to the new session's running state. - const { sessions, messages: msgState } = store.getState(); - if (sessions.active !== sessionId) return; - if (msgState.running && !meta.isActive) { - store.setState((s) => ({ - messages: messagesReducer(s.messages, { type: 'SET_RUNNING', running: false }), - })); - } - }) - .catch(() => { - // Non-fatal — running state will correct on next event or user action - }) - .finally(() => { - runningSyncInFlight = false; - }); - } + // P1: syncRunningState removed — running state is server-authoritative + // via session_state_changed events. No polling needed. function clearPendingSendTimer() { if (parserState.pendingSendTimer) { @@ -385,9 +361,7 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({ - messages: messagesReducer(s.messages, { type: 'SET_RUNNING', running: true }), - })); + // P1: running state from server's session_state_changed event connection.send(pending); // Reschedule for remaining queued messages if (parserState.pendingSend.length > 0) { @@ -730,7 +704,7 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi { expect(ctx.connRegistry.hasOpenWatchers('sess-1')).toBe(true); }); - it('returns running: true when session has active query loop', async () => { + it('P1: session_switched does not include running field', async () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1', session: {} }); sessionReg.isActive.mockReturnValue(true); @@ -648,40 +648,9 @@ describe('handleSwitchSession', () => { await handleSwitchSession('c1', { type: 'switch_session', sessionId: 'sess-1' }, ctx); const resp = transport.sent[0]; - expect(resp).toHaveProperty('running', true); - }); - - it('returns running: false when store state is ENDED (zombie)', async () => { - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1', session: {} }); - sessionReg.isActive.mockReturnValue(true); - - const eventStore = mockEventStore(); - eventStore.getSession.mockReturnValue({ - sessionId: 'sess-1', - mode: 'agent', - cwd: '/test', - branch: 'main', - wtId: null, - inputTokens: 0, - outputTokens: 0, - cacheReadTokens: 0, - cacheCreationTokens: 0, - totalCostUsd: 0, - }); - eventStore.getSessionState.mockReturnValue('ENDED'); - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - eventStore: eventStore as unknown as V2HandlerContext['eventStore'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - await handleSwitchSession('c1', { type: 'switch_session', sessionId: 'sess-1' }, ctx); - - const resp = transport.sent[0]; - expect(resp).toHaveProperty('running', false); + // P1: running state from session_state_changed events, not session_switched + expect(resp).not.toHaveProperty('running'); + expect(resp).toHaveProperty('type', 'session_switched'); }); }); @@ -1048,10 +1017,10 @@ describe('handlePermissionResponseV2', () => { }); }); -// ─── handleReconnect — running status ─────────────────────────────────────── +// ─── handleReconnect — P1: no running field in summary ────────────────────── -describe('handleReconnect running status', () => { - it('reports running=true when session has an active driver', () => { +describe('handleReconnect reconnected summary (P1)', () => { + it('reconnected summary does not include running field', () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'driver-1' }); sessionReg.isActive.mockReturnValue(true); @@ -1069,59 +1038,21 @@ describe('handleReconnect running status', () => { ); const summary = transport.sent.find((m) => m.type === 'reconnected') as { - sessions: Array<{ sessionId: string; running: boolean }>; + sessions: Array<{ sessionId: string; replayed: number }>; }; - expect(summary.sessions[0].running).toBe(true); + expect(summary.sessions[0]).not.toHaveProperty('running'); + expect(summary.sessions[0]).toHaveProperty('sessionId', 'sess-1'); + expect(summary.sessions[0]).toHaveProperty('replayed'); }); - it('reports running=false when session has no active driver', () => { - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId.mockReturnValue({ clientId: 'driver-1' }); - sessionReg.isActive.mockReturnValue(false); - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'sess-1', lastSeq: 0 }] }, - ctx, - ); - - const summary = transport.sent.find((m) => m.type === 'reconnected') as { - sessions: Array<{ sessionId: string; running: boolean }>; - }; - expect(summary.sessions[0].running).toBe(false); - }); - - it('reports running=false when session is not in registry', () => { - const ctx = createContext(); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - handleReconnect( - 'c1', - { type: 'reconnect', sessions: [{ sessionId: 'unknown-sess', lastSeq: 0 }] }, - ctx, - ); - - const summary = transport.sent.find((m) => m.type === 'reconnected') as { - sessions: Array<{ sessionId: string; running: boolean }>; - }; - expect(summary.sessions[0].running).toBe(false); - }); - - it('reports running=false when registry says active but EventStore says inactive (zombie session)', () => { + it('removes stale session from registry when store state is ENDED (zombie)', () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'driver-1' }); sessionReg.isActive.mockReturnValue(true); sessionReg.isAttached.mockReturnValue(false); const eventStore = mockEventStore(); - eventStore.getSession.mockReturnValue({ isActive: false }); + eventStore.getSessionState.mockReturnValue('ENDED'); (reattachChat as ReturnType).mockClear(); @@ -1138,55 +1069,10 @@ describe('handleReconnect running status', () => { ctx, ); - const summary = transport.sent.find((m) => m.type === 'reconnected') as { - sessions: Array<{ sessionId: string; running: boolean }>; - }; - expect(summary.sessions[0].running).toBe(false); + expect(sessionReg.remove).toHaveBeenCalledWith('driver-1'); expect(reattachChat).not.toHaveBeenCalled(); }); - it('handles mixed running states across multiple sessions', () => { - const sessionReg = mockSessionRegistry(); - sessionReg.findBySessionId - .mockReturnValueOnce({ clientId: 'driver-1' }) - .mockReturnValueOnce(null) - .mockReturnValueOnce({ clientId: 'driver-3' }); - sessionReg.isActive.mockReturnValueOnce(true).mockReturnValueOnce(false); - - const ctx = createContext({ - sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], - }); - const transport = mockTransport(); - ctx.connRegistry.register('c1', transport); - - handleReconnect( - 'c1', - { - type: 'reconnect', - sessions: [ - { sessionId: 'sess-1', lastSeq: 0 }, - { sessionId: 'sess-2', lastSeq: 0 }, - { sessionId: 'sess-3', lastSeq: 0 }, - ], - }, - ctx, - ); - - const summary = transport.sent.find((m) => m.type === 'reconnected') as { - sessions: Array<{ sessionId: string; running: boolean }>; - }; - expect(summary.sessions).toHaveLength(3); - expect(summary.sessions[0]).toEqual( - expect.objectContaining({ sessionId: 'sess-1', running: true }), - ); - expect(summary.sessions[1]).toEqual( - expect.objectContaining({ sessionId: 'sess-2', running: false }), - ); - expect(summary.sessions[2]).toEqual( - expect.objectContaining({ sessionId: 'sess-3', running: false }), - ); - }); - it('replays multiple events in sequence order', () => { const eventStore = mockEventStore(); eventStore.getEventsAfter.mockReturnValue([ @@ -2850,7 +2736,8 @@ describe('stale session cleanup removes registry entry', () => { sessionReg.isActive.mockReturnValue(true); const eventStore = mockEventStore(); - eventStore.getSession.mockReturnValue({ sessionId: 'sess-1', isActive: false }); + // P1: handleReconnect uses getSessionState() instead of getSession().isActive + eventStore.getSessionState.mockReturnValue('ENDED'); const ctx = createContext({ sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], diff --git a/server/index.ts b/server/index.ts index 2bd65a80..8cb4640d 100644 --- a/server/index.ts +++ b/server/index.ts @@ -101,11 +101,14 @@ const connRegistry = new ConnectionRegistry(); setConnectionRegistry(connRegistry); // Wire up EventStore for periodic sync (enables delivery guarantee). -// Provide isSessionActive so periodic sync skips ended sessions. +// Provide isSessionActive so periodic sync skips ended sessions (P1: use state, not is_active). connRegistry.setEventStore({ getEventsAfter: (sessionId, afterSeq, limit) => eventStore.getEventsAfter(sessionId, afterSeq, limit), - isSessionActive: (sessionId) => eventStore.getSession(sessionId)?.isActive ?? false, + isSessionActive: (sessionId) => { + const state = eventStore.getSessionState(sessionId); + return state !== null && state !== 'ENDED' && state !== 'CLOSING'; + }, }); // Resolve cert paths relative to the project root (where package.json lives) diff --git a/server/query-loop.ts b/server/query-loop.ts index 4fc0e953..3b2f21ba 100644 --- a/server/query-loop.ts +++ b/server/query-loop.ts @@ -1396,9 +1396,8 @@ async function _runQueryLoopInner( durationApiMs: 0, // only available from SDK result }); } - // Mark session as inactive in durable store + // Mark session as ended in durable store (P1: setSessionState syncs is_active) if (store && resolvedSessionId) { - store.markSessionInactive(resolvedSessionId); store.setSessionState(resolvedSessionId, 'ENDED', { clientId, reason: caughtError ? 'error' : 'completed', diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index 14e3ce1b..f5e7588a 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -229,7 +229,7 @@ export function handleReconnect( 'ws.reconnect', { 'ws.connectionId': connectionId, 'ws.sessionCount': msg.sessions.length }, () => { - const summaries: Array<{ sessionId: string; replayed: number; running: boolean }> = []; + const summaries: Array<{ sessionId: string; replayed: number }> = []; for (const entry of msg.sessions) { ctx.connRegistry.watch(connectionId, entry.sessionId); @@ -251,22 +251,20 @@ export function handleReconnect( const newCursor = events.length > 0 ? events[events.length - 1].seq : entry.lastSeq; ctx.connRegistry.resetCursor(connectionId, entry.sessionId, newCursor); - // Cross-reference with the durable EventStore: markSessionInactive() is - // called in the query loop's finally block, so isActive=false in the - // store is ground truth that the loop has ended. + // Cross-reference with the durable EventStore: state=ENDED in the store + // is ground truth that the query loop has finished (P1: use state, not is_active). const found = ctx.sessionRegistry.findBySessionId(entry.sessionId); + const storeState = ctx.eventStore.getSessionState(entry.sessionId); let running = found ? ctx.sessionRegistry.isActive(found.clientId) : false; - if (running) { - const storeMeta = ctx.eventStore.getSession(entry.sessionId); - if (storeMeta && !storeMeta.isActive) { - running = false; - log.info('removing stale session from registry', { - connectionId, - sessionId: entry.sessionId, - clientId: found!.clientId, - }); - ctx.sessionRegistry.remove(found!.clientId); - } + if (running && (storeState === 'ENDED' || storeState === 'CLOSING')) { + running = false; + log.info('removing stale session from registry (state-based)', { + connectionId, + sessionId: entry.sessionId, + clientId: found!.clientId, + storeState, + }); + ctx.sessionRegistry.remove(found!.clientId); } if (found && running && !ctx.sessionRegistry.isAttached(found.clientId)) { const ownerConnection = getOwnerConnection(found.clientId); @@ -322,7 +320,6 @@ export function handleReconnect( summaries.push({ sessionId: entry.sessionId, replayed: events.length + suspendReplayed, - running, }); // Re-send boot_context so pills reappear after reconnect. @@ -410,14 +407,6 @@ export async function handleSwitchSession( ctx.connRegistry.setActive(connectionId, msg.sessionId); // Cross-reference registry with durable state to avoid reporting - // a zombie query loop as running. - const found = ctx.sessionRegistry.findBySessionId(msg.sessionId); - const storeState = ctx.eventStore.getSessionState(msg.sessionId); - const running = - found && storeState !== 'ENDED' && storeState !== 'CLOSING' && storeState !== null - ? ctx.sessionRegistry.isActive(found.clientId) - : false; - ctx.connRegistry.get(connectionId)?.transport.send({ type: 'session_switched', sessionId: msg.sessionId, @@ -425,7 +414,6 @@ export async function handleSwitchSession( cwd: sessionMeta.cwd, branch: sessionMeta.branch, wtId: sessionMeta.wtId, - running, tokens: { input: sessionMeta.inputTokens, output: sessionMeta.outputTokens, @@ -439,7 +427,7 @@ export async function handleSwitchSession( // Uses shared helper with hot (in-memory) + cold (EventStore) paths. sendBootContext(connectionId, msg.sessionId, ctx); - log.info('switch_session', { connectionId, sessionId: msg.sessionId, running }); + log.info('switch_session', { connectionId, sessionId: msg.sessionId }); }, ); } From e770d0364a356a68a8e96d7c986bec22fbdaad75 Mon Sep 17 00:00:00 2001 From: dimakis Date: Thu, 2 Jul 2026 23:51:22 +0100 Subject: [PATCH 2/5] fix(transport): address Centaur review findings on transport SSOT p1 - Emit session_state_changed on session switch to close running-state gap - Validate msg.state against ClientSessionState union before dispatch - Add requires_action + unknown state tests - Add optimistic dispatch comments in handleStop - Export toClientState for server-side state mapping Co-Authored-By: Claude Opus 4.6 --- frontend/src/pages/ChatView.tsx | 2 ++ frontend/src/pages/DesktopChatView.tsx | 2 ++ .../client/__tests__/protocol-parser.test.ts | 35 +++++++++++++++++++ packages/client/src/protocol-parser.ts | 11 ++++-- packages/protocol/src/event-store.ts | 2 +- server/event-store.ts | 2 +- server/ws-handler-v2.ts | 14 ++++++++ 7 files changed, 63 insertions(+), 5 deletions(-) diff --git a/frontend/src/pages/ChatView.tsx b/frontend/src/pages/ChatView.tsx index 7e7d4618..3d191e42 100644 --- a/frontend/src/pages/ChatView.tsx +++ b/frontend/src/pages/ChatView.tsx @@ -185,6 +185,8 @@ export function ChatView() { const handleStop = useCallback(() => { storeStopGeneration(); + // Optimistic update — server confirms via session_state_changed event, + // but we set idle immediately for responsive UI on the stop button. storeDispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); }, [storeStopGeneration, storeDispatchMessages]); diff --git a/frontend/src/pages/DesktopChatView.tsx b/frontend/src/pages/DesktopChatView.tsx index d4815f44..1e85533b 100644 --- a/frontend/src/pages/DesktopChatView.tsx +++ b/frontend/src/pages/DesktopChatView.tsx @@ -148,6 +148,8 @@ export function DesktopChatView() { const handleStop = useCallback(() => { storeStopGeneration(); + // Optimistic update — server confirms via session_state_changed event, + // but we set idle immediately for responsive UI on the stop button. storeDispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); }, [storeStopGeneration, storeDispatchMessages]); diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index 4ff68a5e..f3cfbbc1 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -946,6 +946,41 @@ describe('session_state_changed', () => { ); expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); }); + + it('dispatches requires_action state', () => { + const r = parseServerMessage( + { + type: 'session_state_changed', + sessionId: 'sid-1', + state: 'requires_action', + internalState: 'ACTIVE', + timestamp: 1234567890, + }, + makeState(), + makeCallbacks(), + POOL_KEY, + ); + expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'requires_action' }); + }); + + it('warns on unknown state', () => { + const spy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const r = parseServerMessage( + { + type: 'session_state_changed', + sessionId: 'sid-1', + state: 'bogus', + internalState: 'ACTIVE', + timestamp: 1234567890, + }, + makeState(), + makeCallbacks(), + POOL_KEY, + ); + expect(r.messagesActions).toHaveLength(0); + expect(spy).toHaveBeenCalledWith('[mitzo] unknown session state:', 'bogus'); + spy.mockRestore(); + }); }); // ─── Subagent cancellation ─────────────────────────────────────────────────── diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index 6a04fe8a..87ef98b9 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -16,6 +16,7 @@ import type { ToolTier, RawToolInput, ToolResultImage, + ClientSessionState, } from '@mitzo/protocol'; import type { MessagesAction } from './slices/messages.js'; import type { WsMsg } from './server-messages.js'; @@ -264,12 +265,16 @@ export function parseServerMessage( callbacks.onSessionRenamed?.(msg.name as string); break; - case 'session_state_changed': + case 'session_state_changed': { // P1: server-authoritative state — derive running from this event only - if (typeof msg.state === 'string') { - result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: msg.state }); + const validStates: ReadonlySet = new Set(['idle', 'running', 'requires_action']); + if (typeof msg.state === 'string' && validStates.has(msg.state)) { + result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: msg.state as ClientSessionState }); + } else if (typeof msg.state === 'string') { + console.warn('[mitzo] unknown session state:', msg.state); } break; + } case 'message_start': result.messagesActions.push({ diff --git a/packages/protocol/src/event-store.ts b/packages/protocol/src/event-store.ts index 5aa63428..74271737 100644 --- a/packages/protocol/src/event-store.ts +++ b/packages/protocol/src/event-store.ts @@ -24,7 +24,7 @@ export type { * Note: 'requires_action' is never returned here — it is emitted separately * by the permission_request handler (Phase 1), not from lifecycle transitions. */ -function toClientState(state: SessionState): ClientSessionState { +export function toClientState(state: SessionState): ClientSessionState { switch (state) { case 'STARTING': case 'ACTIVE': diff --git a/server/event-store.ts b/server/event-store.ts index ec2c2dae..09f215a5 100644 --- a/server/event-store.ts +++ b/server/event-store.ts @@ -1,2 +1,2 @@ -export { EventStore } from '@mitzo/protocol/event-store'; +export { EventStore, toClientState } from '@mitzo/protocol/event-store'; export type { StoredEvent, SessionMeta, SessionSearchResult } from '@mitzo/protocol'; diff --git a/server/ws-handler-v2.ts b/server/ws-handler-v2.ts index f5e7588a..32d9d5a1 100644 --- a/server/ws-handler-v2.ts +++ b/server/ws-handler-v2.ts @@ -11,6 +11,7 @@ import type { SessionTransport, ConnectionRegistry } from '@mitzo/harness'; import type { SessionRegistry } from './session-registry.js'; import type { EventStore } from './event-store.js'; +import { toClientState } from './event-store.js'; import { IncomingWsMessageV2, WatchMessage, @@ -423,6 +424,19 @@ export async function handleSwitchSession( }, }); + // Immediately emit current session state so the client derives running + // status without waiting for the next live event or periodic sync. + const currentState = ctx.eventStore.getSessionState(msg.sessionId); + if (currentState) { + ctx.connRegistry.get(connectionId)?.transport.send({ + type: 'session_state_changed', + sessionId: msg.sessionId, + state: toClientState(currentState), + internalState: currentState, + timestamp: Date.now(), + }); + } + // Re-send boot_context so pills appear on session switch. // Uses shared helper with hot (in-memory) + cold (EventStore) paths. sendBootContext(connectionId, msg.sessionId, ctx); From ad7dee11fd850ea24fb2cf1a60da997228948f65 Mon Sep 17 00:00:00 2001 From: dimakis Date: Fri, 3 Jul 2026 00:00:41 +0100 Subject: [PATCH 3/5] fix(transport): address second Centaur review findings on SSOT p1 - Fix requires_action mapping: action.state !== 'idle' (keeps stop button during permission prompts) - Add optimistic running=true on pending send drain (session_end + timer) - Add session_state_changed e2e test in store foreground recovery - Fix prettier formatting Co-Authored-By: Claude Opus 4.6 --- .../client/__tests__/protocol-parser.test.ts | 9 +++-- packages/client/__tests__/store.test.ts | 34 +++++++++++++++++-- packages/client/src/protocol-parser.ts | 13 +++++-- packages/client/src/slices/messages.ts | 2 +- packages/client/src/store.ts | 8 ++++- 5 files changed, 56 insertions(+), 10 deletions(-) diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index f3cfbbc1..3be7b553 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -167,8 +167,8 @@ describe('session lifecycle', () => { const cb = makeCallbacks(); const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, state, cb, POOL_KEY); expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' }); - // P1: no SET_RUNNING — running state from server's session_state_changed - expect(r.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); + // P1: optimistic running=true when draining pending send + expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'running' }); expect(cb.sendQueued).toHaveBeenCalledWith(POOL_KEY, { type: 'send', prompt: 'follow-up' }); // Second message stays queued expect(state.pendingSend).toEqual([{ type: 'send', prompt: 'second' }]); @@ -960,7 +960,10 @@ describe('session_state_changed', () => { makeCallbacks(), POOL_KEY, ); - expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'requires_action' }); + expect(r.messagesActions).toContainEqual({ + type: 'SESSION_STATE_CHANGED', + state: 'requires_action', + }); }); it('warns on unknown state', () => { diff --git a/packages/client/__tests__/store.test.ts b/packages/client/__tests__/store.test.ts index b81cf40a..e3424707 100644 --- a/packages/client/__tests__/store.test.ts +++ b/packages/client/__tests__/store.test.ts @@ -1299,6 +1299,36 @@ describe('foreground recovery', () => { expect(state.messages[0].messageId).toBe('user-1'); }); - // P1: syncRunningState tests removed — running state is now server-authoritative - // via session_state_changed events. No polling on foreground needed. + // P1: syncRunningState removed — running state is server-authoritative. + // Verify session_state_changed events correctly update running after foreground. + it('session_state_changed updates running after foreground restore', async () => { + const transport = mockTransport(); + const store = createReadyStore(transport); + + // Establish session so session-scoped event filter accepts sess-1 events + lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-1' }); + + // Session is idle + expect(store.getState().messages.running).toBe(false); + + // Server sends session_state_changed: running + lastWs.simulateMessage({ + type: 'session_state_changed', + sessionId: 'sess-1', + state: 'running', + internalState: 'ACTIVE', + timestamp: Date.now(), + }); + expect(store.getState().messages.running).toBe(true); + + // Server sends session_state_changed: idle + lastWs.simulateMessage({ + type: 'session_state_changed', + sessionId: 'sess-1', + state: 'idle', + internalState: 'ENDED', + timestamp: Date.now(), + }); + expect(store.getState().messages.running).toBe(false); + }); }); diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index 87ef98b9..21d30687 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -269,7 +269,10 @@ export function parseServerMessage( // P1: server-authoritative state — derive running from this event only const validStates: ReadonlySet = new Set(['idle', 'running', 'requires_action']); if (typeof msg.state === 'string' && validStates.has(msg.state)) { - result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: msg.state as ClientSessionState }); + result.messagesActions.push({ + type: 'SESSION_STATE_CHANGED', + state: msg.state as ClientSessionState, + }); } else if (typeof msg.state === 'string') { console.warn('[mitzo] unknown session state:', msg.state); } @@ -356,10 +359,14 @@ export function parseServerMessage( if (msg.sessionId && !state.currentSessionId) { callbacks.onSessionAssigned(msg.sessionId as string); } - // P1: pending send still drains queued messages, but running state - // comes from server's session_state_changed event (P2 removes pendingSend entirely) + // Drain first queued message. Optimistic running=true avoids UI flicker + // between the send and the server's session_state_changed confirmation. const pending = state.pendingSend.shift(); if (pending) { + result.messagesActions.push({ + type: 'SESSION_STATE_CHANGED', + state: 'running' as ClientSessionState, + }); if (callbacks.onSendQueued) { callbacks.onSendQueued(pending); } else { diff --git a/packages/client/src/slices/messages.ts b/packages/client/src/slices/messages.ts index e5666884..87ca76bf 100644 --- a/packages/client/src/slices/messages.ts +++ b/packages/client/src/slices/messages.ts @@ -575,7 +575,7 @@ export function messagesReducer(state: MessagesState, action: MessagesAction): M }; case 'SESSION_STATE_CHANGED': - return { ...state, running: action.state === 'running' }; + return { ...state, running: action.state !== 'idle' }; case 'CONNECTION_LOST': return { diff --git a/packages/client/src/store.ts b/packages/client/src/store.ts index 153021c1..cd07e6a4 100644 --- a/packages/client/src/store.ts +++ b/packages/client/src/store.ts @@ -361,7 +361,13 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({ + messages: messagesReducer(s.messages, { + type: 'SESSION_STATE_CHANGED', + state: 'running', + }), + })); connection.send(pending); // Reschedule for remaining queued messages if (parserState.pendingSend.length > 0) { From 339c79180bf0ae1aeb0febd777329815c9fb20bf Mon Sep 17 00:00:00 2001 From: dimakis Date: Fri, 3 Jul 2026 00:08:27 +0100 Subject: [PATCH 4/5] fix(transport): address third Centaur review findings on SSOT p1 - Fix session_takeover regression: inline idle dispatch (server unwatches old client) - Fix session_close_ack: inline idle dispatch for 'no active agent' path - Hoist VALID_CLIENT_STATES Set to module scope - Remove unnecessary 'as ClientSessionState' cast on string literal - Add test: session_state_changed emission on session switch - Add test: no emission when getSessionState returns null Co-Authored-By: Claude Opus 4.6 --- .../client/__tests__/protocol-parser.test.ts | 5 +-- packages/client/src/protocol-parser.ts | 20 +++++---- server/__tests__/ws-handler-v2.test.ts | 41 +++++++++++++++++++ 3 files changed, 55 insertions(+), 11 deletions(-) diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index 3be7b553..a4801449 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -595,7 +595,7 @@ describe('misc', () => { // ─── session_takeover ──────────────────────────────────────────────────────── describe('session_takeover', () => { - it('produces ERROR message (P1: no SET_RUNNING, state from server event)', () => { + it('clears running and produces ERROR (server unwatches after takeover)', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'session_takeover', sessionId: 'sess-1' }, @@ -603,8 +603,7 @@ describe('session_takeover', () => { cb, POOL_KEY, ); - // P1: running=false comes from server's session_state_changed event - expect(r.messagesActions).not.toContainEqual(expect.objectContaining({ type: 'SET_RUNNING' })); + expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); expect(r.messagesActions).toContainEqual( expect.objectContaining({ type: 'ERROR', error: expect.stringContaining('another device') }), ); diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index 21d30687..a98264ee 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -102,6 +102,10 @@ export interface ParseResult { | { type: 'workload_batch_updated'; items: WorkloadItem[]; created: number }; } +// ─── Constants ────────────────────────────────────────────────────────────── + +const VALID_CLIENT_STATES: ReadonlySet = new Set(['idle', 'running', 'requires_action']); + // ─── Parser ────────────────────────────────────────────────────────────────── export function parseServerMessage( @@ -130,7 +134,9 @@ export function parseServerMessage( break; case 'session_takeover': - // P1: running=false comes from server's session_state_changed event + // Server unwatches the old client after takeover, so no subsequent + // session_state_changed event will arrive — clear running inline. + result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); result.messagesActions.push({ type: 'ERROR', error: 'Session resumed on another device.', @@ -267,8 +273,7 @@ export function parseServerMessage( case 'session_state_changed': { // P1: server-authoritative state — derive running from this event only - const validStates: ReadonlySet = new Set(['idle', 'running', 'requires_action']); - if (typeof msg.state === 'string' && validStates.has(msg.state)) { + if (typeof msg.state === 'string' && VALID_CLIENT_STATES.has(msg.state)) { result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: msg.state as ClientSessionState, @@ -363,10 +368,7 @@ export function parseServerMessage( // between the send and the server's session_state_changed confirmation. const pending = state.pendingSend.shift(); if (pending) { - result.messagesActions.push({ - type: 'SESSION_STATE_CHANGED', - state: 'running' as ClientSessionState, - }); + result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: 'running' }); if (callbacks.onSendQueued) { callbacks.onSendQueued(pending); } else { @@ -420,7 +422,9 @@ export function parseServerMessage( command: 'close', content: 'Session closing... The agent will commit work and write a summary.', }); - // P1: running=false from server's session_state_changed (CLOSING → idle) + // Safety net: server's 'no active agent' path may not emit + // session_state_changed, so clear running inline. + result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); } break; diff --git a/server/__tests__/ws-handler-v2.test.ts b/server/__tests__/ws-handler-v2.test.ts index 1371b75c..91669f68 100644 --- a/server/__tests__/ws-handler-v2.test.ts +++ b/server/__tests__/ws-handler-v2.test.ts @@ -651,6 +651,47 @@ describe('handleSwitchSession', () => { // P1: running state from session_state_changed events, not session_switched expect(resp).not.toHaveProperty('running'); expect(resp).toHaveProperty('type', 'session_switched'); + + // P1: session_state_changed emitted immediately after session_switched + const stateMsg = transport.sent[1]; + expect(stateMsg).toMatchObject({ + type: 'session_state_changed', + sessionId: 'sess-1', + state: 'running', + internalState: 'ACTIVE', + }); + expect(stateMsg).toHaveProperty('timestamp'); + }); + + it('P1: no session_state_changed when getSessionState returns null', async () => { + const sessionReg = mockSessionRegistry(); + const eventStore = mockEventStore(); + eventStore.getSession.mockReturnValue({ + sessionId: 'sess-1', + mode: 'agent', + cwd: '/test', + branch: 'main', + wtId: null, + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalCostUsd: 0, + }); + eventStore.getSessionState.mockReturnValue(null); + + const ctx = createContext({ + sessionRegistry: sessionReg as unknown as V2HandlerContext['sessionRegistry'], + eventStore: eventStore as unknown as V2HandlerContext['eventStore'], + }); + const transport = mockTransport(); + ctx.connRegistry.register('c1', transport); + + await handleSwitchSession('c1', { type: 'switch_session', sessionId: 'sess-1' }, ctx); + + // Only session_switched should be sent, no session_state_changed + expect(transport.sent).toHaveLength(1); + expect(transport.sent[0]).toHaveProperty('type', 'session_switched'); }); }); From 0268afb214fdc1043864d8fbc87e0e7187f44018 Mon Sep 17 00:00:00 2001 From: dimakis Date: Fri, 3 Jul 2026 00:23:55 +0100 Subject: [PATCH 5/5] fix(transport): address fourth Centaur review findings on SSOT p1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix session_takeover regression: inline idle dispatch (server unwatches old client) - Fix session_close_ack: inline idle dispatch for 'no active agent' path - Add session_close_ack test coverage - Remove temporal 'P1:' comment prefixes — describe rationale, not phase - Note periodic sync as iOS foreground gap fallback Critical finding (is_active stale) was a false positive — setSessionState already syncs is_active at event-store.ts:631-634. Co-Authored-By: Claude Opus 4.6 --- .../client/__tests__/protocol-parser.test.ts | 43 +++++++++++++++---- packages/client/__tests__/store.test.ts | 2 +- packages/client/src/protocol-parser.ts | 12 +++--- packages/client/src/store.ts | 6 +-- server/__tests__/ws-handler-v2.test.ts | 12 +++--- 5 files changed, 51 insertions(+), 24 deletions(-) diff --git a/packages/client/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index a4801449..e26cdebb 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -50,7 +50,7 @@ describe('pool lifecycle events', () => { // ─── Reattach ───────────────────────────────────────────────────────────────── describe('reattach', () => { - it('reattached calls onSessionAssigned and setWsRunning (P1: no SET_RUNNING)', () => { + it('reattached calls onSessionAssigned and setWsRunning', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'reattached', clientId: 'c1', sessionId: 'sid-1', running: true }, @@ -58,14 +58,14 @@ describe('reattach', () => { cb, POOL_KEY, ); - // P1: running state comes from session_state_changed events, not reattached + // Running state comes from session_state_changed events, not reattached expect(r.messagesActions).toHaveLength(0); expect(cb.onSessionAssigned).toHaveBeenCalledWith('sid-1'); expect(cb.setWsRunning).toHaveBeenCalledWith(POOL_KEY, true); expect(r.connectionUpdate).toEqual({ status: 'connected' }); }); - it('reattach_failed marks connection connected (P1: no SET_RUNNING)', () => { + it('reattach_failed marks connection connected', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'reattach_failed', clientId: 'old' }, @@ -73,7 +73,7 @@ describe('reattach', () => { cb, POOL_KEY, ); - // P1: running state comes from session_state_changed events + // Running state comes from session_state_changed events expect(r.messagesActions).toHaveLength(0); expect(cb.setWsRunning).toHaveBeenCalledWith(POOL_KEY, false); expect(r.connectionUpdate).toEqual({ status: 'connected' }); @@ -167,7 +167,7 @@ describe('session lifecycle', () => { const cb = makeCallbacks(); const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, state, cb, POOL_KEY); expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' }); - // P1: optimistic running=true when draining pending send + // Optimistic running=true when draining pending send expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'running' }); expect(cb.sendQueued).toHaveBeenCalledWith(POOL_KEY, { type: 'send', prompt: 'follow-up' }); // Second message stays queued @@ -388,7 +388,7 @@ describe('error handling', () => { // ─── Subscribed ────────────────────────────────────────────────────────────── describe('subscribed', () => { - it('subscribed with running=true calls setWsRunning (P1: no SET_RUNNING)', () => { + it('subscribed with running=true calls setWsRunning', () => { const cb = makeCallbacks(); const r = parseServerMessage( { type: 'subscribed', sessionId: 'sid', running: true }, @@ -396,7 +396,7 @@ describe('subscribed', () => { cb, POOL_KEY, ); - // P1: running state from session_state_changed events, not subscribed + // Running state from session_state_changed events, not subscribed expect(r.messagesActions).toHaveLength(0); expect(cb.setWsRunning).toHaveBeenCalledWith(POOL_KEY, true); }); @@ -620,7 +620,7 @@ describe('reconnected', () => { expect(onReconnected).toHaveBeenCalled(); }); - it('P1: reconnected does not dispatch SET_RUNNING — state from replayed events', () => { + it('reconnected does not dispatch SET_RUNNING — state from replayed events', () => { const state = makeState({ currentSessionId: 'sid-1' }); const r = parseServerMessage( { type: 'reconnected', sessions: [{ sessionId: 'sid-1', replayed: 3 }] }, @@ -985,6 +985,33 @@ describe('session_state_changed', () => { }); }); +// ─── session_close_ack ────────────────────────────────────────────────────── + +describe('session_close_ack', () => { + it('dispatches idle state and close result when accepted', () => { + const r = parseServerMessage( + { type: 'session_close_ack', accepted: true }, + makeState(), + makeCallbacks(), + POOL_KEY, + ); + expect(r.messagesActions).toContainEqual( + expect.objectContaining({ type: 'NATIVE_COMMAND_RESULT', command: 'close' }), + ); + expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'idle' }); + }); + + it('produces no actions when not accepted', () => { + const r = parseServerMessage( + { type: 'session_close_ack', accepted: false }, + makeState(), + makeCallbacks(), + POOL_KEY, + ); + expect(r.messagesActions).toHaveLength(0); + }); +}); + // ─── Subagent cancellation ─────────────────────────────────────────────────── describe('subagent_cancelled', () => { diff --git a/packages/client/__tests__/store.test.ts b/packages/client/__tests__/store.test.ts index e3424707..2fba7283 100644 --- a/packages/client/__tests__/store.test.ts +++ b/packages/client/__tests__/store.test.ts @@ -1299,7 +1299,7 @@ describe('foreground recovery', () => { expect(state.messages[0].messageId).toBe('user-1'); }); - // P1: syncRunningState removed — running state is server-authoritative. + // syncRunningState removed — running state is server-authoritative. // Verify session_state_changed events correctly update running after foreground. it('session_state_changed updates running after foreground restore', async () => { const transport = mockTransport(); diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index a98264ee..bd06d755 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -127,7 +127,7 @@ export function parseServerMessage( // ── v2 handshake events ──────────────────────────────────────────────── case 'reconnected': - // P1: running state is derived from replayed session_state_changed events, + // Running state derived from replayed session_state_changed events, // not from the reconnected message payload. result.connectionUpdate = { status: 'connected' }; callbacks.onReconnected?.(); @@ -148,7 +148,7 @@ export function parseServerMessage( if (tokens) { callbacks.onTokensHydrated?.(tokens); } - // P1: running state restored via replayed session_state_changed events + // Running state restored via replayed session_state_changed events break; } @@ -158,14 +158,14 @@ export function parseServerMessage( // ── v1 handshake events (kept for backward compat) ───────────────────── case 'reattached': - // P1: running state from server's session_state_changed events + // Running state from server's session_state_changed events callbacks.setWsRunning?.(poolKey, true); result.connectionUpdate = { status: 'connected' }; if (msg.sessionId) callbacks.onSessionAssigned(msg.sessionId as string); break; case 'reattach_failed': - // P1: running=false from server's session_state_changed events + // Running state from server's session_state_changed events callbacks.setWsRunning?.(poolKey, false); result.connectionUpdate = { status: 'connected' }; if (state.currentSessionId && callbacks.fetchMessages) { @@ -272,7 +272,7 @@ export function parseServerMessage( break; case 'session_state_changed': { - // P1: server-authoritative state — derive running from this event only + // Server-authoritative state — derive running from this event only if (typeof msg.state === 'string' && VALID_CLIENT_STATES.has(msg.state)) { result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', @@ -432,7 +432,7 @@ export function parseServerMessage( break; case 'subscribed': - // P1: running state from session_state_changed events, not subscribed payload + // Running state from session_state_changed events, not subscribed payload if (msg.running) { callbacks.setWsRunning?.(poolKey, true); } diff --git a/packages/client/src/store.ts b/packages/client/src/store.ts index cd07e6a4..acc99340 100644 --- a/packages/client/src/store.ts +++ b/packages/client/src/store.ts @@ -216,8 +216,8 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi { expect(ctx.connRegistry.hasOpenWatchers('sess-1')).toBe(true); }); - it('P1: session_switched does not include running field', async () => { + it('session_switched does not include running field', async () => { const sessionReg = mockSessionRegistry(); sessionReg.findBySessionId.mockReturnValue({ clientId: 'c1:sess-1', session: {} }); sessionReg.isActive.mockReturnValue(true); @@ -648,11 +648,11 @@ describe('handleSwitchSession', () => { await handleSwitchSession('c1', { type: 'switch_session', sessionId: 'sess-1' }, ctx); const resp = transport.sent[0]; - // P1: running state from session_state_changed events, not session_switched + // Running state from session_state_changed events, not session_switched expect(resp).not.toHaveProperty('running'); expect(resp).toHaveProperty('type', 'session_switched'); - // P1: session_state_changed emitted immediately after session_switched + // session_state_changed emitted immediately after session_switched const stateMsg = transport.sent[1]; expect(stateMsg).toMatchObject({ type: 'session_state_changed', @@ -663,7 +663,7 @@ describe('handleSwitchSession', () => { expect(stateMsg).toHaveProperty('timestamp'); }); - it('P1: no session_state_changed when getSessionState returns null', async () => { + it('no session_state_changed when getSessionState returns null', async () => { const sessionReg = mockSessionRegistry(); const eventStore = mockEventStore(); eventStore.getSession.mockReturnValue({ @@ -1058,7 +1058,7 @@ describe('handlePermissionResponseV2', () => { }); }); -// ─── handleReconnect — P1: no running field in summary ────────────────────── +// ─── handleReconnect — no running field in summary ────────────────────────── describe('handleReconnect reconnected summary (P1)', () => { it('reconnected summary does not include running field', () => { @@ -2777,7 +2777,7 @@ describe('stale session cleanup removes registry entry', () => { sessionReg.isActive.mockReturnValue(true); const eventStore = mockEventStore(); - // P1: handleReconnect uses getSessionState() instead of getSession().isActive + // handleReconnect uses getSessionState() instead of getSession().isActive eventStore.getSessionState.mockReturnValue('ENDED'); const ctx = createContext({