From 9e2fcbc5787a7df7c9436c047bd6859cfd58232e Mon Sep 17 00:00:00 2001 From: dimakis Date: Thu, 2 Jul 2026 23:07:27 +0100 Subject: [PATCH] =?UTF-8?q?feat(transport):=20p0=20foundation=20=E2=80=94?= =?UTF-8?q?=20session=20state=20machine,=20crash=20recovery,=20SSE=20defau?= =?UTF-8?q?lt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Transport SSOT Phase 0: server-authoritative session state foundation. - Add ClientSessionState type (idle/running/requires_action) with exhaustive toClientState() mapping from 7-state SessionState - Emit session_state_changed events on every setSessionState() call - Sync is_active column from state in the same UPDATE (backwards-compatible) - Add recoverStaleSessions() for crash recovery (ACTIVE/STARTING/ DETACHED/SUSPENDED/CLOSING → ENDED) - Re-export ClientSessionState and SessionStateEvent from package index - Flip default transport from WS to SSE (Transport SSOT P0) - Add protocol-parser handler for session_state_changed (P0: observability) - 151 tests covering state mapping, is_active sync, crash recovery, and protocol-parser event handling Co-Authored-By: Claude Opus 4.6 --- frontend/src/client-store.ts | 11 +- .../client/__tests__/protocol-parser.test.ts | 42 ++++ packages/client/src/protocol-parser.ts | 9 + .../protocol/__tests__/event-store.test.ts | 209 ++++++++++++++++++ packages/protocol/src/event-store.ts | 82 ++++++- packages/protocol/src/index.ts | 2 + packages/protocol/src/types.ts | 16 ++ server/app.ts | 1 + server/index.ts | 5 + 9 files changed, 370 insertions(+), 7 deletions(-) diff --git a/frontend/src/client-store.ts b/frontend/src/client-store.ts index 33a59dc8..8cebb204 100644 --- a/frontend/src/client-store.ts +++ b/frontend/src/client-store.ts @@ -17,13 +17,14 @@ 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). + * This is an intentional flip from WS-default per the transport-ssot design doc. + * 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/__tests__/protocol-parser.test.ts b/packages/client/__tests__/protocol-parser.test.ts index 0ecda2da..109afa42 100644 --- a/packages/client/__tests__/protocol-parser.test.ts +++ b/packages/client/__tests__/protocol-parser.test.ts @@ -965,6 +965,48 @@ describe('boot_context', () => { }); }); +// ─── Session state (Transport SSOT P0) ────────────────────────────────────── + +describe('session_state_changed', () => { + it('produces no message actions (P0: observability only)', () => { + const r = parseServerMessage( + { + type: 'session_state_changed', + sessionId: 'sid-1', + state: 'running', + internalState: 'ACTIVE', + timestamp: 1234567890, + }, + makeState(), + makeCallbacks(), + POOL_KEY, + ); + expect(r.messagesActions).toHaveLength(0); + }); + + it('logs via console.debug', () => { + const spy = vi.spyOn(console, 'debug').mockImplementation(() => {}); + parseServerMessage( + { + type: 'session_state_changed', + sessionId: 'sid-1', + state: 'idle', + internalState: 'ENDED', + timestamp: 1234567890, + }, + makeState(), + makeCallbacks(), + POOL_KEY, + ); + expect(spy).toHaveBeenCalledWith('[mitzo] session_state_changed', { + sessionId: 'sid-1', + state: 'idle', + internalState: 'ENDED', + }); + spy.mockRestore(); + }); +}); + // ─── Subagent cancellation ─────────────────────────────────────────────────── describe('subagent_cancelled', () => { diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts index 5fab88e6..1028dbad 100644 --- a/packages/client/src/protocol-parser.ts +++ b/packages/client/src/protocol-parser.ts @@ -292,6 +292,15 @@ export function parseServerMessage( callbacks.onSessionRenamed?.(msg.name as string); 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, + }); + break; + case 'message_start': result.messagesActions.push({ type: 'MESSAGE_START', diff --git a/packages/protocol/__tests__/event-store.test.ts b/packages/protocol/__tests__/event-store.test.ts index 7e742784..8630b286 100644 --- a/packages/protocol/__tests__/event-store.test.ts +++ b/packages/protocol/__tests__/event-store.test.ts @@ -642,6 +642,215 @@ describe('EventStore', () => { }); }); + describe('toClientState mapping (via session_state_changed events)', () => { + const sid = 'client-state-test'; + + beforeEach(() => { + store.upsertSession({ sessionId: sid }); + }); + + it('maps CREATED to idle', () => { + store.setSessionState(sid, 'CREATED', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('idle'); + expect(stateEvent?.payload.internalState).toBe('CREATED'); + }); + + it('maps STARTING to running', () => { + store.setSessionState(sid, 'STARTING', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('running'); + }); + + it('maps ACTIVE to running', () => { + store.setSessionState(sid, 'ACTIVE', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('running'); + }); + + it('maps DETACHED to idle', () => { + store.setSessionState(sid, 'DETACHED', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('idle'); + }); + + it('maps SUSPENDED to idle', () => { + store.setSessionState(sid, 'SUSPENDED', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('idle'); + }); + + it('maps CLOSING to idle', () => { + store.setSessionState(sid, 'CLOSING', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('idle'); + }); + + it('maps ENDED to idle', () => { + store.setSessionState(sid, 'ENDED', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.state).toBe('idle'); + }); + + it('includes timestamp in event payload', () => { + const before = Date.now(); + store.setSessionState(sid, 'ACTIVE', { force: true }); + const events = store.getSessionEvents(sid); + const stateEvent = events.find((e) => e.type === 'session_state_changed'); + expect(stateEvent?.payload.timestamp).toBeGreaterThanOrEqual(before); + }); + }); + + describe('setSessionState syncs is_active', () => { + const sid = 'is-active-sync-test'; + + beforeEach(() => { + store.upsertSession({ sessionId: sid }); + }); + + it('sets is_active=true for ACTIVE', () => { + store.setSessionState(sid, 'ACTIVE', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(true); + }); + + it('sets is_active=true for DETACHED', () => { + store.setSessionState(sid, 'DETACHED', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(true); + }); + + it('sets is_active=true for SUSPENDED', () => { + store.setSessionState(sid, 'SUSPENDED', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(true); + }); + + it('sets is_active=false for ENDED', () => { + store.setSessionState(sid, 'ENDED', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(false); + }); + + it('sets is_active=false for CLOSING', () => { + store.setSessionState(sid, 'CLOSING', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(false); + }); + + it('sets is_active=true for CREATED', () => { + store.setSessionState(sid, 'CREATED', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(true); + }); + + it('sets is_active=true for STARTING', () => { + store.setSessionState(sid, 'STARTING', { force: true }); + expect(store.getSession(sid)!.isActive).toBe(true); + }); + }); + + describe('recoverStaleSessions', () => { + it('transitions ACTIVE sessions to ENDED', () => { + store.upsertSession({ sessionId: 'active-1' }); + store.setSessionState('active-1', 'ACTIVE', { force: true }); + + const count = store.recoverStaleSessions(); + + expect(count).toBe(1); + expect(store.getSessionState('active-1')).toBe('ENDED'); + }); + + it('transitions STARTING sessions to ENDED', () => { + store.upsertSession({ sessionId: 'starting-1' }); + store.setSessionState('starting-1', 'STARTING', { force: true }); + + store.recoverStaleSessions(); + + expect(store.getSessionState('starting-1')).toBe('ENDED'); + }); + + it('transitions DETACHED sessions to ENDED', () => { + store.upsertSession({ sessionId: 'detached-1' }); + store.setSessionState('detached-1', 'DETACHED', { force: true }); + + store.recoverStaleSessions(); + + expect(store.getSessionState('detached-1')).toBe('ENDED'); + }); + + it('transitions SUSPENDED sessions to ENDED', () => { + store.upsertSession({ sessionId: 'suspended-1' }); + store.setSessionState('suspended-1', 'SUSPENDED', { force: true }); + + store.recoverStaleSessions(); + + expect(store.getSessionState('suspended-1')).toBe('ENDED'); + }); + + it('transitions CLOSING sessions to ENDED', () => { + store.upsertSession({ sessionId: 'closing-1' }); + store.setSessionState('closing-1', 'CLOSING', { force: true }); + + store.recoverStaleSessions(); + + expect(store.getSessionState('closing-1')).toBe('ENDED'); + }); + + it('does not touch ENDED sessions', () => { + store.upsertSession({ sessionId: 'ended-1' }); + store.setSessionState('ended-1', 'ENDED', { force: true }); + + const count = store.recoverStaleSessions(); + + expect(count).toBe(0); + expect(store.getSessionState('ended-1')).toBe('ENDED'); + }); + + it('does not touch CREATED sessions', () => { + store.upsertSession({ sessionId: 'created-1' }); + store.setSessionState('created-1', 'CREATED', { force: true }); + + const count = store.recoverStaleSessions(); + + expect(count).toBe(0); + expect(store.getSessionState('created-1')).toBe('CREATED'); + }); + + it('returns correct count for multiple stale sessions', () => { + store.upsertSession({ sessionId: 'stale-1' }); + store.upsertSession({ sessionId: 'stale-2' }); + store.upsertSession({ sessionId: 'ok-1' }); + store.setSessionState('stale-1', 'ACTIVE', { force: true }); + store.setSessionState('stale-2', 'DETACHED', { force: true }); + store.setSessionState('ok-1', 'ENDED', { force: true }); + + const count = store.recoverStaleSessions(); + + expect(count).toBe(2); + }); + + it('emits session_state_changed events for recovered sessions', () => { + store.upsertSession({ sessionId: 'recover-1' }); + store.setSessionState('recover-1', 'ACTIVE', { force: true }); + + // Clear events from setup + const beforeCount = store.getSessionEvents('recover-1').length; + + store.recoverStaleSessions(); + + const events = store.getSessionEvents('recover-1'); + // Should have new session_state_changed event from recovery + const recoveryEvent = events + .slice(beforeCount) + .find((e) => e.type === 'session_state_changed'); + expect(recoveryEvent).toBeDefined(); + expect(recoveryEvent?.payload.state).toBe('idle'); + expect(recoveryEvent?.payload.internalState).toBe('ENDED'); + }); + }); + describe('close', () => { it('is safe to call multiple times', () => { store.close(); diff --git a/packages/protocol/src/event-store.ts b/packages/protocol/src/event-store.ts index 8578ddbc..5aa63428 100644 --- a/packages/protocol/src/event-store.ts +++ b/packages/protocol/src/event-store.ts @@ -5,11 +5,42 @@ 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. + * 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 { + switch (state) { + case 'STARTING': + case 'ACTIVE': + return 'running'; + case 'CREATED': + case 'CLOSING': + case 'ENDED': + case 'DETACHED': + case 'SUSPENDED': + return 'idle'; + default: { + const _exhaustive: never = state; + return _exhaustive; + } + } +} const noopLogger: EventStoreLogger = { info() {} }; @@ -191,6 +222,7 @@ export class EventStore { `UPDATE sessions SET state = ?, last_state_change = ?, + is_active = ?, updated_at = unixepoch('now', 'subsec') * 1000 WHERE session_id = ?`, ), @@ -596,12 +628,25 @@ export class EventStore { } } - this.stmts.setSessionState.run(newState, now, sessionId); + // Sync is_active from state (backwards-compatible, P0). + // Only ENDED/CLOSING are inactive. CREATED is transient — see recoverStaleSessions(). + const isActive = newState !== 'ENDED' && newState !== 'CLOSING' ? 1 : 0; + this.stmts.setSessionState.run(newState, now, 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 +657,39 @@ 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, SUSPENDED, or CLOSING is transitioned to ENDED. + * CLOSING is included because the process performing graceful shutdown is gone after a crash. + * Returns the number of sessions recovered. + */ + recoverStaleSessions(): number { + // CREATED excluded: transient state, moves to STARTING synchronously in startChat(). + // The crash window between CREATED and STARTING is negligible. + const staleStates = ['ACTIVE', 'STARTING', 'DETACHED', 'SUSPENDED', 'CLOSING']; + const placeholders = staleStates.map(() => '?').join(', '); + // Inline prepare is intentional — this runs once at startup, not worth caching. + 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/index.ts b/packages/protocol/src/index.ts index b0abd6da..7c1040f7 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -19,6 +19,8 @@ export type { Session, SessionClosedBy, SessionState, + ClientSessionState, + SessionStateEvent, StoredEvent, SessionMeta, SessionSearchResult, 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..2bd65a80 100644 --- a/server/index.ts +++ b/server/index.ts @@ -1024,6 +1024,11 @@ 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). + // Must run before reconcileSessionsBackground() so reconciliation sees ENDED states. + // recoverStaleSessions() logs internally — no need to log here. + eventStore.recoverStaleSessions(); + // Eagerly reconcile sessions so the first /api/sessions request is fast and accurate. reconcileSessionsBackground(); // Clean up stale worktrees across all repos.