Skip to content

feat(transport): transport SSOT phase 1 — server-authoritative state #432

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dimakis wants to merge 2 commits into from
Closed

feat(transport): transport SSOT phase 1 — server-authoritative state #432

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bda6b09
feat(transport): p0 foundation — session_state_changed event, crash r…
dimakis Jul 2, 2026
58699f1
feat(transport): p1 server-authoritative state — replace SET_RUNNING …
dimakis Jul 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions frontend/src/client-store.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
? {
Expand Down
2 changes: 1 addition & 1 deletion frontend/src/pages/ChatView.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
2 changes: 1 addition & 1 deletion frontend/src/pages/DesktopChatView.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
91 changes: 19 additions & 72 deletions packages/client/__tests__/protocol-parser.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,29 +50,31 @@ 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 },
makeState(),
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' },
makeState(),
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' });
});
Expand Down Expand Up @@ -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' }]);
Expand Down Expand Up @@ -385,15 +388,16 @@ 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 },
makeState(),
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);
});

Expand Down Expand Up @@ -591,15 +595,16 @@ 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' },
makeState({ currentSessionId: 'sess-1' }),
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') }),
);
Expand All @@ -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' });
});
});

Expand Down
145 changes: 6 additions & 139 deletions packages/client/__tests__/store.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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' });
Expand Down Expand Up @@ -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',
Expand Down Expand Up @@ -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();

Expand Down Expand Up @@ -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);
});
});
Expand Down Expand Up @@ -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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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.
});
Loading