Skip to content

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

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

Merged
dimakis merged 5 commits into from
Jul 4, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e227446
feat(transport): p1 server-authoritative state — replace SET_RUNNING …
dimakis Jul 2, 2026
e770d03
fix(transport): address Centaur review findings on transport SSOT p1
dimakis Jul 2, 2026
ad7dee1
fix(transport): address second Centaur review findings on SSOT p1
dimakis Jul 2, 2026
339c791
fix(transport): address third Centaur review findings on SSOT p1
dimakis Jul 2, 2026
0268afb
fix(transport): address fourth Centaur review findings on SSOT p1
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
4 changes: 3 additions & 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,9 @@ export function ChatView() {

const handleStop = useCallback(() => {
storeStopGeneration();
storeDispatchMessages({ type: 'SET_RUNNING', running: false });
// Optimistic update — server confirms via session_state_changed event,

@dimakis dimakis Jul 4, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 style: The handleStop callback (with its optimistic update comment) is duplicated identically in both ChatView.tsx and DesktopChatView.tsx. This is pre-existing duplication, but since both files are modified in this PR, it may be worth extracting to a shared hook or utility. [fixable]

// but we set idle immediately for responsive UI on the stop button.
storeDispatchMessages({ type: 'SESSION_STATE_CHANGED', state: 'idle' });
}, [storeStopGeneration, storeDispatchMessages]);

function handlePermission(
Expand Down
4 changes: 3 additions & 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,9 @@ export function DesktopChatView() {

const handleStop = useCallback(() => {
storeStopGeneration();
storeDispatchMessages({ type: 'SET_RUNNING', running: false });
// 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]);

function handlePermission(
Expand Down
169 changes: 87 additions & 82 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', () => {
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 }]);
// 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', () => {
const cb = makeCallbacks();
const r = parseServerMessage(
{ type: 'reattach_failed', clientId: 'old' },
makeState(),
cb,
POOL_KEY,
);
expect(r.messagesActions).toEqual([{ type: 'SET_RUNNING', running: false }]);
// 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 });
// 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' }]);
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', () => {
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 }]);
// 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,15 @@ describe('misc', () => {
// ─── session_takeover ────────────────────────────────────────────────────────

describe('session_takeover', () => {
it('produces SET_RUNNING false and ERROR message', () => {
it('clears running and produces ERROR (server unwatches after takeover)', () => {
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 });
expect(r.messagesActions).toContainEqual({ type: 'SESSION_STATE_CHANGED', state: 'idle' });
expect(r.messagesActions).toContainEqual(
expect.objectContaining({ type: 'ERROR', error: expect.stringContaining('another device') }),
);
Expand All @@ -616,75 +620,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('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 }] },
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' }] },
{ type: 'reconnected', sessions: [{ sessionId: 'sid-1', replayed: 3 }] },
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 Expand Up @@ -965,10 +911,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',
Expand All @@ -981,12 +927,11 @@ describe('session_state_changed', () => {
makeCallbacks(),

@dimakis dimakis Jul 2, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 missing_tests: Tests cover state: 'running' and state: 'idle' (mapped from ENDED) but there is no test for state: 'requires_action' to verify the expected behavior (currently maps to running=false). Adding a test would document the intended semantics and catch regressions if the mapping changes. [fixable]

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',
Expand All @@ -998,15 +943,75 @@ describe('session_state_changed', () => {
makeCallbacks(),
POOL_KEY,
);
expect(spy).toHaveBeenCalledWith('[mitzo] session_state_changed', {
sessionId: 'sid-1',
state: 'idle',
internalState: 'ENDED',
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();
});
});

// ─── 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', () => {
Expand Down
Loading
Loading