From bda6b096532bfa7c97fb4210e4faf8ebefbdbeca Mon Sep 17 00:00:00 2001 From: dimakis Date: Thu, 2 Jul 2026 10:21:06 +0100 Subject: [PATCH 1/2] =?UTF-8?q?feat(transport):=20p0=20foundation=20?= =?UTF-8?q?=E2=80=94=20session=5Fstate=5Fchanged=20event,=20crash=20recove?= =?UTF-8?q?ry,=20SSE=20default?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Transport SSOT Phase 0: additive foundation, no behavior change. - Add ClientSessionState type and SessionStateEvent interface (types.ts) - Emit session_state_changed event from setSessionState() with 7→3 state mapping - Sync is_active from state on every transition (backwards-compatible) - Add recoverStaleSessions() to mark ACTIVE/STARTING/DETACHED/SUSPENDED → ENDED on startup - Call recoverStaleSessions() in server startup (index.ts) - Add state field to /api/sessions/:id/meta response (app.ts) - Flip SSE to default transport, WS opt-in via localStorage (client-store.ts) - Add session_state_changed case to protocol parser (no-op, ready for P1) Design doc: docs/design/transport-ssot.md (on design/transport-ssot branch) Telos: 97361f814c5f54df Co-Authored-By: Claude Opus 4.6 --- frontend/src/client-store.ts | 10 ++-- packages/client/src/protocol-parser.ts | 4 ++ packages/protocol/src/event-store.ts | 74 +++++++++++++++++++++++++- packages/protocol/src/types.ts | 16 ++++++ server/app.ts | 1 + server/index.ts | 6 +++ 6 files changed, 105 insertions(+), 6 deletions(-) diff --git a/frontend/src/client-store.ts b/frontend/src/client-store.ts index 33a59dc8..130cc3f0 100644 --- a/frontend/src/client-store.ts +++ b/frontend/src/client-store.ts @@ -17,13 +17,13 @@ import { eventBus } from './lib/event-bus-singleton'; import { getPreferredModel } from './lib/model-preference'; /** - * Transport selector — set localStorage 'mitzo:transport' to 'sse' to use - * SSE + HTTP POST instead of WebSocket. Default is 'ws'. + * Transport selector — SSE + HTTP POST is the default transport (Transport SSOT P0). + * Set localStorage 'mitzo:transport' to 'ws' to fall back to WebSocket. * - * Toggle from console: localStorage.setItem('mitzo:transport', 'sse'); location.reload(); - * Revert: localStorage.removeItem('mitzo:transport'); location.reload(); + * Force WS: localStorage.setItem('mitzo:transport', 'ws'); location.reload(); + * Revert SSE: localStorage.removeItem('mitzo:transport'); location.reload(); */ -const useSSE = typeof window !== 'undefined' && localStorage.getItem('mitzo:transport') === 'sse'; +const useSSE = typeof window !== 'undefined' && localStorage.getItem('mitzo:transport') !== 'ws'; const sseConfig: SseConnectionConfig | undefined = useSSE ? { diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index 5fab88e6..a48eed3e 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -292,6 +292,10 @@ export function parseServerMessage( callbacks.onSessionRenamed?.(msg.name as string); break; + case 'session_state_changed': + // P0: receive and log, no UI action yet (Phase 1 will bind to running state) + break; + case 'message_start': result.messagesActions.push({ type: 'MESSAGE_START', diff --git a/packages/protocol/src/event-store.ts b/packages/protocol/src/event-store.ts index 8578ddbc..ec5bd838 100644 --- a/packages/protocol/src/event-store.ts +++ b/packages/protocol/src/event-store.ts @@ -5,11 +5,38 @@ import type { SessionMeta, SessionSearchResult, SessionState, + ClientSessionState, EventStoreLogger, } from './types.js'; // Re-export types for consumer convenience -export type { StoredEvent, SessionMeta, SessionSearchResult, SessionState, EventStoreLogger }; +export type { + StoredEvent, + SessionMeta, + SessionSearchResult, + SessionState, + ClientSessionState, + EventStoreLogger, +}; + +/** Map internal 7-state lifecycle to client-facing 3-state. */ +function toClientState(state: SessionState): ClientSessionState { + switch (state) { + case 'STARTING': + case 'ACTIVE': + return 'running'; + case 'CREATED': + case 'CLOSING': + case 'ENDED': + return 'idle'; + // DETACHED/SUSPENDED: preserve last emitted state. + // At the EventStore level we don't track "last emitted", so return 'idle' + // as the safe default. The server layer can override if needed. + case 'DETACHED': + case 'SUSPENDED': + return 'idle'; + } +} const noopLogger: EventStoreLogger = { info() {} }; @@ -598,10 +625,26 @@ export class EventStore { this.stmts.setSessionState.run(newState, now, sessionId); + // Backwards-compatible: sync is_active from state (P0) + const isActive = newState !== 'ENDED' && newState !== 'CLOSING' ? 1 : 0; + this.db!.prepare( + "UPDATE sessions SET is_active = ?, updated_at = unixepoch('now', 'subsec') * 1000 WHERE session_id = ?", + ).run(isActive, sessionId); + + // Emit session_state_changed event for client consumption (P0) + const clientState = toClientState(newState); + this.append(sessionId, 'session_state_changed', { + sessionId, + state: clientState, + internalState: newState, + timestamp: now, + }); + this.log.info('session state transition', { sessionId, fromState, toState: newState, + clientState, clientId: opts?.clientId, reason: opts?.reason, }); @@ -612,6 +655,35 @@ export class EventStore { return (row?.state as SessionState) ?? null; } + /** + * Recover sessions left in incomplete states after a server crash/restart. + * Any session in ACTIVE, STARTING, DETACHED, or SUSPENDED is transitioned to ENDED. + * Returns the number of sessions recovered. + */ + recoverStaleSessions(): number { + const staleStates = ['ACTIVE', 'STARTING', 'DETACHED', 'SUSPENDED']; + const placeholders = staleStates.map(() => '?').join(', '); + const rows = this.db!.prepare( + `SELECT session_id FROM sessions WHERE state IN (${placeholders})`, + ).all(...staleStates) as Array<{ session_id: string }>; + + for (const row of rows) { + this.setSessionState(row.session_id, 'ENDED', { + reason: 'server_restart', + force: true, + }); + } + + if (rows.length > 0) { + this.log.info('recovered stale sessions on startup', { + count: rows.length, + sessionIds: rows.map((r) => r.session_id), + }); + } + + return rows.length; + } + recordUsage( sessionId: string, usage: { diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index 81410574..f7f4f031 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -208,6 +208,22 @@ export type SessionState = | 'CLOSING' | 'ENDED'; +/** + * Client-facing session state derived from the internal 7-state machine. + * Mirrors Anthropic SDK convention: idle | running | requires_action. + */ +export type ClientSessionState = 'idle' | 'running' | 'requires_action'; + +/** Server-authoritative state event emitted on every lifecycle transition. */ +export interface SessionStateEvent { + type: 'session_state_changed'; + sessionId: string; + state: ClientSessionState; + /** Internal lifecycle state for debugging (not used for UI). */ + internalState: SessionState; + timestamp: number; +} + export interface Session { id: string; summary: string; diff --git a/server/app.ts b/server/app.ts index 87984fbd..72bd0f9e 100644 --- a/server/app.ts +++ b/server/app.ts @@ -1218,6 +1218,7 @@ app.get('/api/sessions/:id/meta', (req, res) => { cwd: meta.cwd, mode: meta.mode, isActive: meta.isActive, + state: meta.state, totalTokens, totalCostUsd: meta.totalCostUsd, numTurns: meta.numTurns, diff --git a/server/index.ts b/server/index.ts index c55c33f9..1800983e 100644 --- a/server/index.ts +++ b/server/index.ts @@ -1024,6 +1024,12 @@ checkPort(PORT).then((inUse) => { // Start periodic sync for connection-level delivery guarantee connRegistry.startPeriodicSync(); + // Recover sessions left in incomplete states after crash/restart (Transport SSOT P0) + const recovered = eventStore.recoverStaleSessions(); + if (recovered > 0) { + log.info(`recovered ${recovered} stale session(s) on startup`); + } + // Eagerly reconcile sessions so the first /api/sessions request is fast and accurate. reconcileSessionsBackground(); // Clean up stale worktrees across all repos. From 58699f1403ebf1a1346281c0ee7b3652d1850a90 Mon Sep 17 00:00:00 2001 From: dimakis Date: Thu, 2 Jul 2026 10:50:33 +0100 Subject: [PATCH 2/2] =?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 | 91 +++-------- packages/client/__tests__/store.test.ts | 145 +----------------- packages/client/src/protocol-parser.ts | 56 ++----- 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, 86 insertions(+), 446 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 0ecda2da..689c8cab 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' }); }); }); 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 a48eed3e..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,7 +265,10 @@ export function parseServerMessage( break; case 'session_state_changed': - // P0: receive and log, no UI action yet (Phase 1 will bind to running state) + // 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': @@ -376,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); } @@ -434,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; @@ -442,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 1800983e..d2d1f758 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 }); }, ); }