diff --git a/docs/design/prompt-delivery.md b/docs/design/prompt-delivery.md
new file mode 100644
index 00000000..4adb5c17
--- /dev/null
+++ b/docs/design/prompt-delivery.md
@@ -0,0 +1,84 @@
+# Prompt delivery across reconnect
+
+The first prompt after foregrounding must reach the existing agent without a second prompt. Three independent failures caused this behavior: client queueing based on stale `running`, ignored HTTP failures, and registry rekeying while the SDK query loop retained the old key.
+
+## Contract
+
+- A runtime registry key is stable for the entire SDK query. `ownerConnectionId` identifies the replaceable event connection. Reconnect, send, interrupt, suspend, and disconnect use the connection identity without renaming the runtime.
+- `/api/chat/send` accepts an authenticated, schema-validated command even without an SSE stream. `clientMsgId` identifies the whole command, including creation of a new session.
+- SQLite stores the complete command and its allocated session UUID before dispatch. A retry with the same payload returns the same receipt; reusing an ID with different content fails. The session UUID is passed to SDK startup and registered before asynchronous boot work, so rapid follow-ups find the same input queue.
+- HTTP 202 acknowledges durable acceptance, not completion of model execution. Startup failures are persisted and replayed. After a server restart, commands without evidence of delivery are surfaced as interrupted; they are not automatically re-executed across an ambiguous crash boundary.
+- The client outbox submits in order, independently of SSE readiness and agent running state. Network errors, 429, 5xx, missing acknowledgements, and 15-second request/body timeouts trigger retries with the same ID, backing off to ten seconds. Definitive rejections are shown in the UI.
+- Unacknowledged prompts survive page reload in per-tab session storage when available. Storage failure preserves in-memory retries. Switching conversations does not discard accepted user intent. Only unattempted follow-ups from the same draft inherit a newly acknowledged session ID.
+- SSE replay remains separately acknowledged, with a 15-second timeout. Any tracked session is replayed on welcome, including a command accepted before the first stream exists. Foregrounding rebuilds the stream rather than trusting a stale connected flag. Replaced streams cannot deliver late callbacks.
+
+This is one durable command dispatch per message ID, not a claim of exactly-once external tool side effects. Server crashes interrupt live model execution; recovery reports uncertainty instead of repeating possible side effects.
+
+## Salvaged work
+
+PR #445 supplies the removal of `wasRunning`, `pendingSend`, its five-second timer, and parser-side queue draining, with the corresponding tests. Unrelated changes in that branch are excluded.
+
+PR #440 correctly identifies reconnect ownership churn as redundant. This fix removes runtime rekeying while retaining transport reattachment, permission/suspend recovery, cursor replay, and periodic event sync. Removing those mechanisms wholesale is not necessary to fix prompt delivery and would widen the validation surface.
+
+## Regression evidence
+
+`reconnect-delivery.integration.test.ts` connects the actual HTTP router, durable EventStore, SseConnection/outbox, SessionRegistry, and query loop to a deterministic SDK stream. It creates a session, suspends it, reconnects, submits one prompt, loses its HTTP acknowledgement, and verifies automatic retry produces exactly one additional SDK input and a response on the new stream, with no runtime-key change.
+
+Additional tests cover receipt deduplication, changed-payload rejection, native commands, restart interruption, SDK pre-registration, rapid follow-ups, offline acceptance, reload recovery, response-body stalls, reconnect timeouts, and stale callbacks.
+
+## Review follow-up
+
+The first Centaur review identified two client correctness bugs and an unhandled
+interrupt startup rejection. Replay requests are now serialized per EventSource
+and connection ID. A receipt adding a session during replay schedules one later
+replay with the updated session set; readiness is emitted after that replay.
+Receipts for sessions already tracked do not trigger redundant replays. Navigation
+clears the previous conversation's delivery banner. Interrupt resume reports
+startup rejection through the transport, like normal send startup.
+
+The response parsing concern also exposed a real distinction: a non-retryable
+HTTP rejection with an HTML body must fail visibly and release the next queued
+command. A malformed successful response remains ambiguous and is retried with
+the same command ID, because the server may already have executed it.
+
+Other review suggestions were intentionally not adopted:
+
+- Navigation preserves submitted prompts. Clearing them would silently lose
+ acknowledged user intent. Draft scopes prevent cross-conversation reassignment.
+- Receipts require an explicit session ID or null. Missing session identity is
+ not equivalent to a successful native command and must not discard the outbox
+ entry. Persistent malformed responses keep delivery pending, visibly, rather
+ than claiming success or inviting a duplicate command.
+- Stopping delivery pauses new sends while allowing the bounded in-flight
+ request to settle. A valid receipt is saved immediately, even while stopped.
+ An unresolved request remains eligible for a deduplicated retry on restart.
+- Browser foreground recovery probes the existing stream with a nonce delivered
+ over SSE. A matching reply preserves the stream and resumes/replays suspended
+ sessions. Missing replies rebuild the stream after one second. HTTP success
+ alone never proves SSE liveness. Native force-reconnect and bfcache recovery
+ remain available; prompt delivery never waits for the probe.
+- The post-dispatch receipt read remains: legacy handlers report some synchronous
+ failures through transport events rather than throwing. The durable failure
+ check prevents those paths from returning a successful receipt. Replacing the
+ handler result contract is separate work, not a cosmetic simplification.
+
+## Second review and integration with #456
+
+Startup metadata now enters the durable event log through the HTTP command
+transport even without a stream. Already sequenced query events are not appended
+again, and user-message echoes carry the sequence of their original append.
+Delivery status is separate from delivery errors. Tests cover stopped
+acknowledgements, 429/5xx retries, invalid receipt identities, pending retry status,
+queue capacity, startup persistence, and healthy/dead foreground probes.
+
+Merge #455 before integrating #456's provider lifecycle. #456 is stacked on #453;
+#453 must land before #456, but neither is required by #455. A merge simulation
+against #456 at e4322eb found conflicts in server/chat.ts,
+server/ws-handler-v2.ts and frontend/src/styles/global.css. Reconcile those in
+the provider stack after this transport change, preserving initialSessionId,
+clientMsgId, the stable runtime registry key and account/provider selection.
+In particular, the transport receipt must identify the same session as the
+Codex conversation and its durable queue. Repeat lost-acknowledgement and
+foreground/reconnect tests through the provider queue to prove a retry creates
+one queue entry and executes one turn. Do not infer compatibility from a clean
+textual merge or activate the development-gated provider as part of this fix.
diff --git a/frontend/src/client-store.ts b/frontend/src/client-store.ts
index 8cebb204..dbc7de26 100644
--- a/frontend/src/client-store.ts
+++ b/frontend/src/client-store.ts
@@ -29,6 +29,7 @@ const useSSE = typeof window !== 'undefined' && localStorage.getItem('mitzo:tran
const sseConfig: SseConnectionConfig | undefined = useSSE
? {
baseUrl: getApiBaseUrl(),
+ outboxStorage: sessionStorage,
fetch: (url, init) => apiFetch(url, init),
suspendUrl: `${getApiBaseUrl()}/api/sessions/suspend`,
}
diff --git a/frontend/src/pages/ChatView.tsx b/frontend/src/pages/ChatView.tsx
index 65b1f04f..7407ffa0 100644
--- a/frontend/src/pages/ChatView.tsx
+++ b/frontend/src/pages/ChatView.tsx
@@ -36,6 +36,8 @@ export function ChatView() {
const messages = useMessages();
const connection = useConnection();
const tokens = useTokens();
+ const sendError = useMitzoStore((s) => s.sendError);
+ const sendStatus = useMitzoStore((s) => s.sendStatus);
const activeSessionId = useMitzoStore((s) => s.sessions.active);
// Select individual action functions — stable references
@@ -280,6 +282,16 @@ export function ChatView() {
)}
+ {(sendError || sendStatus) && (
+
+ {sendError || sendStatus}
+
+ )}
): ProtocolParserState {
return {
currentSessionId: undefined,
- pendingSend: [],
...overrides,
};
}
@@ -17,7 +16,6 @@ function makeCallbacks(overrides?: Partial): ProtocolCallback
onMessagesRestored: vi.fn(),
onSessionRenamed: vi.fn(),
setWsRunning: vi.fn(),
- sendQueued: vi.fn(),
...overrides,
};
}
@@ -157,21 +155,15 @@ describe('session lifecycle', () => {
]);
});
- it('session_end dequeues first pending send and queues it', () => {
- const state = makeState({
- pendingSend: [
- { type: 'send', prompt: 'follow-up' },
- { type: 'send', prompt: 'second' },
- ],
- });
+ it('session_end dispatches SESSION_END action', () => {
const cb = makeCallbacks();
- const r = parseServerMessage({ type: 'session_end', sessionId: 'sid' }, state, cb, POOL_KEY);
+ const r = parseServerMessage(
+ { type: 'session_end', sessionId: 'sid' },
+ makeState(),
+ cb,
+ POOL_KEY,
+ );
expect(r.messagesActions).toContainEqual({ type: 'SESSION_END', sessionId: 'sid' });
- // 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' }]);
});
});
@@ -378,10 +370,15 @@ describe('error handling', () => {
expect(r.messagesActions).toEqual([{ type: 'ERROR', error: 'Something broke' }]);
});
- it('error clears pendingSend queue', () => {
- const state = makeState({ pendingSend: [{ type: 'send', prompt: 'test' }] });
- parseServerMessage({ type: 'error', error: 'fail' }, state, makeCallbacks(), POOL_KEY);
- expect(state.pendingSend).toEqual([]);
+ it('error does not require pendingSend cleanup (removed in P2)', () => {
+ const state = makeState();
+ const r = parseServerMessage(
+ { type: 'error', error: 'fail' },
+ state,
+ makeCallbacks(),
+ POOL_KEY,
+ );
+ expect(r.messagesActions).toContainEqual({ type: 'ERROR', error: 'fail' });
});
});
diff --git a/packages/client/__tests__/store-sessions.test.ts b/packages/client/__tests__/store-sessions.test.ts
index a0623d2e..17dd8efc 100644
--- a/packages/client/__tests__/store-sessions.test.ts
+++ b/packages/client/__tests__/store-sessions.test.ts
@@ -138,3 +138,13 @@ describe('refreshSessions', () => {
expect(store.getState().sessions.list[0].name).toBe('Existing');
});
});
+
+describe('delivery status on navigation', () => {
+ it.each(['switch', 'new'])('clears the previous conversation error on %s', async (action) => {
+ const store = createMitzoStore(makeOptions());
+ store.setState({ sendError: 'Reconnecting — your message will retry automatically.' });
+ if (action === 'switch') await store.getState().switchSession('other');
+ else store.getState().newSession();
+ expect(store.getState().sendError).toBeNull();
+ });
+});
diff --git a/packages/client/__tests__/store.test.ts b/packages/client/__tests__/store.test.ts
index 2cf2a355..9fe41cab 100644
--- a/packages/client/__tests__/store.test.ts
+++ b/packages/client/__tests__/store.test.ts
@@ -402,89 +402,23 @@ describe('sendMessage', () => {
);
});
- it('queues second message while first turn is running', async () => {
+ it('sends second message immediately while first turn is running (server dedup)', async () => {
const store = createReadyStore();
store.getState().sendMessage('first');
lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-q' });
- // Turn is still running — second send should queue
+ // Turn is still running — P2: second send goes immediately (server deduplicates)
const sentBefore = lastWs.sent.length;
store.getState().sendMessage('second');
- // Should NOT have sent yet
- const newSendsImmediate = lastWs.sent.slice(sentBefore).map((s) => JSON.parse(s));
- expect(newSendsImmediate.filter((m) => m.type === 'send')).toHaveLength(0);
-
- // session_end triggers flush of queued message
- lastWs.simulateMessage({ type: 'session_end', sessionId: 'sess-q' });
-
const newSends = lastWs.sent.slice(sentBefore).map((s) => JSON.parse(s));
expect(newSends).toContainEqual(expect.objectContaining({ type: 'send', prompt: 'second' }));
- });
-
- it('queues second message as pendingSend while first turn is active', () => {
- const store = createReadyStore();
-
- store.getState().sendMessage('first');
- lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-pend' });
-
- const sentBefore = lastWs.sent.length;
- store.getState().sendMessage('second');
-
- // Second message should be queued, not immediately sent
- const immediateSends = lastWs.sent
- .slice(sentBefore)
- .filter((s) => JSON.parse(s).type === 'send');
- expect(immediateSends).toHaveLength(0);
- // But the optimistic user message should appear in the store
+ // Optimistic user message should appear in the store
const userMsgs = store.getState().messages.messages.filter((m) => m.role === 'user');
expect(userMsgs).toHaveLength(2);
});
-
- it('cancels pending timeout when session_end arrives in time', () => {
- vi.useFakeTimers();
- try {
- const store = createMitzoStore(makeOptions());
- lastWs.completeHandshake();
-
- store.getState().sendMessage('first');
- lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-ok' });
-
- store.getState().sendMessage('second');
- lastWs.simulateMessage({ type: 'session_end', sessionId: 'sess-ok' });
-
- const sentAfterEnd = lastWs.sent.length;
- vi.advanceTimersByTime(6_000);
-
- expect(lastWs.sent.length).toBe(sentAfterEnd);
- } finally {
- vi.useRealTimers();
- }
- });
-
- it('cancels pending timeout on newSession', () => {
- vi.useFakeTimers();
- try {
- const store = createMitzoStore(makeOptions());
- lastWs.completeHandshake();
-
- store.getState().sendMessage('first');
- lastWs.simulateMessage({ type: 'session_id', sessionId: 'sess-new' });
-
- store.getState().sendMessage('second');
- const sentBefore = lastWs.sent.length;
-
- store.getState().newSession();
- vi.advanceTimersByTime(6_000);
-
- const flushed = lastWs.sent.slice(sentBefore).filter((s) => JSON.parse(s).type === 'send');
- expect(flushed).toHaveLength(0);
- } finally {
- vi.useRealTimers();
- }
- });
});
describe('WS → store wiring', () => {
@@ -1343,3 +1277,26 @@ describe('account selection', () => {
});
});
});
+
+describe('delivery status', () => {
+ it('keeps pending delivery separate from an error', () => {
+ const store = createMitzoStore(makeOptions());
+ lastWs.completeHandshake();
+ store.getState().sendMessage('hello');
+ const command = lastWs.parsedSent().find((m) => m.type === 'send')!;
+ lastWs.simulateMessage({
+ type: '_send_pending',
+ clientMsgId: command.clientMsgId,
+ retrying: true,
+ });
+ expect(store.getState().sendError).toBeNull();
+ expect(store.getState().sendStatus).toContain('retry');
+ lastWs.simulateMessage({
+ type: '_send_failed',
+ clientMsgId: command.clientMsgId,
+ error: 'Rejected',
+ });
+ expect(store.getState().sendStatus).toBeNull();
+ expect(store.getState().sendError).toBe('Rejected');
+ });
+});
diff --git a/packages/client/src/__tests__/send-outbox.test.ts b/packages/client/src/__tests__/send-outbox.test.ts
new file mode 100644
index 00000000..1f831f75
--- /dev/null
+++ b/packages/client/src/__tests__/send-outbox.test.ts
@@ -0,0 +1,249 @@
+import { describe, it, expect, vi, afterEach } from 'vitest';
+import { SendOutbox } from '../send-outbox.js';
+
+const prompt = { type: 'send', sessionId: null, clientMsgId: 'one', prompt: 'hello' };
+const ack = (id = 'one') =>
+ ({
+ ok: true,
+ status: 202,
+ json: async () => ({ accepted: true, clientMsgId: id, sessionId: 'session' }),
+ }) as Response;
+afterEach(() => vi.useRealTimers());
+
+describe('send outbox', () => {
+ it('surfaces a definitive HTML rejection and lets the next prompt proceed', async () => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockResolvedValueOnce({
+ ok: false,
+ status: 403,
+ json: async () => {
+ throw new SyntaxError('HTML');
+ },
+ })
+ .mockResolvedValueOnce(ack('two'));
+ const notify = vi.fn();
+ const outbox = new SendOutbox({ fetch, notify, url: '/send' });
+ outbox.enqueue(prompt, 0);
+ outbox.enqueue({ ...prompt, clientMsgId: 'two' }, 1);
+ outbox.start();
+ await vi.advanceTimersByTimeAsync(0);
+ expect(notify).toHaveBeenCalledWith(
+ expect.objectContaining({
+ type: '_send_failed',
+ clientMsgId: 'one',
+ }),
+ );
+ expect(notify).toHaveBeenCalledWith(
+ expect.objectContaining({
+ type: '_send_accepted',
+ clientMsgId: 'two',
+ }),
+ );
+ expect(fetch).toHaveBeenCalledTimes(2);
+ outbox.stop();
+ });
+
+ it.each(['html', 'missing session'])(
+ 'retains ambiguous %s success for a safe retry',
+ async (kind) => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockResolvedValueOnce({
+ ok: true,
+ status: 202,
+ json: async () => {
+ if (kind === 'html') throw new SyntaxError('HTML');
+ return { accepted: true, clientMsgId: 'one' };
+ },
+ })
+ .mockResolvedValueOnce(ack());
+ const notify = vi.fn();
+ const outbox = new SendOutbox({ fetch, notify, url: '/send' });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(fetch).toHaveBeenCalledTimes(2);
+ expect(fetch.mock.calls[0][1].body).toBe(fetch.mock.calls[1][1].body);
+ expect(notify).not.toHaveBeenCalledWith(expect.objectContaining({ type: '_send_failed' }));
+ expect(notify).toHaveBeenCalledWith(expect.objectContaining({ type: '_send_accepted' }));
+ outbox.stop();
+ },
+ );
+
+ it('captures an in-flight acknowledgement while stopped without starting another prompt', async () => {
+ vi.useFakeTimers();
+ let resolve!: (response: Response) => void;
+ const fetch = vi
+ .fn()
+ .mockReturnValueOnce(
+ new Promise((r) => {
+ resolve = r;
+ }),
+ )
+ .mockResolvedValue(ack('two'));
+ const notify = vi.fn();
+ const outbox = new SendOutbox({ fetch, notify, url: '/send' });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ outbox.enqueue({ ...prompt, clientMsgId: 'two' }, 0);
+ outbox.stop();
+ resolve(ack());
+ await vi.advanceTimersByTimeAsync(0);
+ expect(notify).toHaveBeenCalledWith(
+ expect.objectContaining({ type: '_send_accepted', clientMsgId: 'one' }),
+ );
+ expect(fetch).toHaveBeenCalledTimes(1);
+ outbox.start();
+ await vi.advanceTimersByTimeAsync(0);
+ expect(JSON.parse(fetch.mock.calls[1][1].body).clientMsgId).toBe('two');
+ outbox.stop();
+ });
+
+ it.each([429, 500, 503])(
+ 'retries HTTP %s with the same identity and reports retry status',
+ async (status) => {
+ vi.useFakeTimers();
+ const fetch = vi.fn().mockResolvedValueOnce({ status }).mockResolvedValue(ack());
+ const notify = vi.fn();
+ const outbox = new SendOutbox({ fetch, notify, url: '/send' });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(fetch.mock.calls[0][1].body).toBe(fetch.mock.calls[1][1].body);
+ expect(notify).toHaveBeenCalledWith(
+ expect.objectContaining({ type: '_send_pending', retrying: true, clientMsgId: 'one' }),
+ );
+ outbox.stop();
+ },
+ );
+
+ it.each([
+ { accepted: 'true', clientMsgId: 'one' },
+ { accepted: true, clientMsgId: 'wrong' },
+ ])('retains a receipt with invalid identity or acceptance: %j', async (receipt) => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockResolvedValueOnce({
+ ok: true,
+ status: 202,
+ json: async () => ({ ...receipt, sessionId: 'session' }),
+ })
+ .mockResolvedValue(ack());
+ const outbox = new SendOutbox({ fetch, notify: vi.fn(), url: '/send' });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(fetch).toHaveBeenCalledTimes(2);
+ expect(fetch.mock.calls[0][1].body).toBe(fetch.mock.calls[1][1].body);
+ outbox.stop();
+ });
+
+ it('rejects a full queue without discarding an accepted entry', () => {
+ let saved = '';
+ const outbox = new SendOutbox({
+ fetch: vi.fn(),
+ notify: vi.fn(),
+ url: '/send',
+ storage: {
+ getItem: () => null,
+ setItem: (_, value) => {
+ saved = value;
+ },
+ },
+ });
+ for (let i = 0; i < 100; i++)
+ expect(outbox.enqueue({ ...prompt, clientMsgId: String(i) }, 0)).toBe(true);
+ expect(outbox.enqueue({ ...prompt, clientMsgId: 'overflow' }, 0)).toBe(false);
+ expect(JSON.parse(saved)).toHaveLength(100);
+ expect(JSON.parse(saved)[0].body.clientMsgId).toBe('0');
+ });
+
+ it('retries a lost response with the identical command ID without waiting for SSE', async () => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockRejectedValueOnce(new Error('lost response'))
+ .mockResolvedValue(ack());
+ const notify = vi.fn();
+ const outbox = new SendOutbox({ fetch, notify, url: '/send' });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(fetch).toHaveBeenCalledTimes(2);
+ expect(fetch.mock.calls[0][1].body).toBe(fetch.mock.calls[1][1].body);
+ expect(notify).toHaveBeenCalledWith(
+ expect.objectContaining({ type: '_send_accepted', clientMsgId: 'one', sessionId: 'session' }),
+ );
+ outbox.stop();
+ });
+
+ it('times out a hung POST and retries automatically', async () => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockReturnValueOnce(new Promise(() => {}))
+ .mockResolvedValue(ack());
+ const outbox = new SendOutbox({ fetch, notify: vi.fn(), url: '/send', timeoutMs: 100 });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ await vi.advanceTimersByTimeAsync(1100);
+ expect(fetch).toHaveBeenCalledTimes(2);
+ outbox.stop();
+ });
+
+ it('also times out a response whose body never arrives', async () => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockResolvedValueOnce({ ok: true, status: 202, json: () => new Promise(() => {}) })
+ .mockResolvedValue(ack());
+ const outbox = new SendOutbox({ fetch, notify: vi.fn(), url: '/send', timeoutMs: 100 });
+ outbox.start();
+ outbox.enqueue(prompt, 0);
+ await vi.advanceTimersByTimeAsync(1100);
+ expect(fetch).toHaveBeenCalledTimes(2);
+ outbox.stop();
+ });
+
+ it('binds rapid follow-ups to the first accepted session while keeping new drafts separate', async () => {
+ vi.useFakeTimers();
+ const fetch = vi
+ .fn()
+ .mockResolvedValueOnce(ack())
+ .mockResolvedValueOnce(ack('two'))
+ .mockResolvedValueOnce(ack('three'));
+ const outbox = new SendOutbox({ fetch, notify: vi.fn(), url: '/send' });
+ outbox.enqueue(prompt, 0);
+ outbox.enqueue({ ...prompt, clientMsgId: 'two' }, 0);
+ outbox.enqueue({ ...prompt, clientMsgId: 'three' }, 1);
+ outbox.start();
+ await vi.advanceTimersByTimeAsync(0);
+ expect(JSON.parse(fetch.mock.calls[1][1].body).sessionId).toBe('session');
+ expect(JSON.parse(fetch.mock.calls[2][1].body).sessionId).toBeNull();
+ outbox.stop();
+ });
+
+ it('restores unacknowledged prompts after a reload', async () => {
+ vi.useFakeTimers();
+ let value: string | null = null;
+ const storage = {
+ getItem: () => value,
+ setItem: (_: string, v: string) => {
+ value = v;
+ },
+ };
+ const first = new SendOutbox({ fetch: vi.fn(), notify: vi.fn(), url: '/send', storage });
+ first.enqueue(prompt, 0);
+ const fetch = vi.fn().mockResolvedValue(ack());
+ const second = new SendOutbox({ fetch, notify: vi.fn(), url: '/send', storage });
+ second.start();
+ await vi.advanceTimersByTimeAsync(0);
+ expect(JSON.parse(fetch.mock.calls[0][1].body)).toEqual(prompt);
+ expect(JSON.parse(value!)).toEqual([]);
+ second.stop();
+ });
+});
diff --git a/packages/client/src/__tests__/sse-connection.test.ts b/packages/client/src/__tests__/sse-connection.test.ts
index e00e62ae..97f2064d 100644
--- a/packages/client/src/__tests__/sse-connection.test.ts
+++ b/packages/client/src/__tests__/sse-connection.test.ts
@@ -86,6 +86,139 @@ describe('SseConnection', () => {
vi.useRealTimers();
});
+ it('serializes replay requests and includes sessions accepted during an in-flight replay', async () => {
+ const replays: Array<{
+ body: { sessions: Array<{ sessionId: string }> };
+ resolve: (r: Response) => void;
+ }> = [];
+ const fetch = vi.fn((url: string, init?: RequestInit) => {
+ if (url.endsWith('/reconnect'))
+ return new Promise((resolve) =>
+ replays.push({ body: JSON.parse(String(init?.body)), resolve }),
+ );
+ const body = JSON.parse(String(init?.body));
+ return Promise.resolve({
+ ok: true,
+ status: 202,
+ json: async () => ({
+ accepted: true,
+ clientMsgId: body.clientMsgId,
+ sessionId: body.clientMsgId,
+ }),
+ } as Response);
+ });
+ const conn = new SseConnection(createConfig({ fetch }));
+ const listener = vi.fn();
+ conn.onMessage(listener);
+ conn.connect();
+ lastES()._emit('welcome', { connectionId: 'c1' });
+ listener.mockClear();
+ conn.send({ type: 'send', clientMsgId: 'one', sessionId: null, prompt: 'one' });
+ await vi.advanceTimersByTimeAsync(0);
+ conn.clearPendingSends();
+ conn.send({ type: 'send', clientMsgId: 'two', sessionId: null, prompt: 'two' });
+ await vi.advanceTimersByTimeAsync(0);
+ expect(replays).toHaveLength(1);
+ replays[0].resolve({ ok: true } as Response);
+ await vi.advanceTimersByTimeAsync(0);
+ expect(replays).toHaveLength(2);
+ expect(replays[1].body.sessions.map((s) => s.sessionId)).toEqual(['one', 'two']);
+ expect(listener.mock.calls.filter(([e]) => e.type === '_open')).toHaveLength(0);
+ replays[1].resolve({ ok: true } as Response);
+ await vi.advanceTimersByTimeAsync(0);
+ expect(listener.mock.calls.filter(([e]) => e.type === '_open')).toHaveLength(1);
+ conn.disconnect();
+ });
+
+ it('rebuilds a seemingly connected stream when its foreground probe receives no reply', async () => {
+ const doc = new EventTarget();
+ Object.assign(doc, { visibilityState: 'visible' });
+ vi.stubGlobal('document', doc);
+ vi.stubGlobal('addEventListener', vi.fn());
+ vi.stubGlobal('removeEventListener', vi.fn());
+ const conn = new SseConnection(createConfig());
+ try {
+ conn.connect();
+ const old = lastES();
+ old._emit('welcome', { connectionId: 'old' });
+ doc.dispatchEvent(new Event('visibilitychange'));
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(old.readyState).toBe(2);
+ expect(MockEventSource.instances).toHaveLength(2);
+ expect(conn.isConnected()).toBe(false);
+ } finally {
+ conn.disconnect();
+ vi.unstubAllGlobals();
+ }
+ });
+
+ it('keeps a stream that delivers the foreground probe over SSE', async () => {
+ const doc = new EventTarget();
+ Object.assign(doc, { visibilityState: 'visible' });
+ vi.stubGlobal('document', doc);
+ vi.stubGlobal('addEventListener', vi.fn());
+ vi.stubGlobal('removeEventListener', vi.fn());
+ const fetch = vi.fn().mockResolvedValue({ ok: true });
+ const conn = new SseConnection(createConfig({ fetch }));
+ try {
+ conn.connect();
+ const es = lastES();
+ es._emit('welcome', { connectionId: 'healthy' });
+ conn.trackSeq('suspended-session', 12);
+ doc.dispatchEvent(new Event('visibilitychange'));
+ const call = fetch.mock.calls.find(([url]) => url.endsWith('/probe'));
+ expect(call).toBeDefined();
+ es._emit('message', { type: '_probe', nonce: JSON.parse(String(call![1].body)).nonce });
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(MockEventSource.instances).toHaveLength(1);
+ expect(fetch.mock.calls.some(([url]) => url.endsWith('/reconnect'))).toBe(true);
+ expect(conn.isConnected()).toBe(true);
+ } finally {
+ conn.disconnect();
+ vi.unstubAllGlobals();
+ }
+ });
+
+ it('submits the first prompt while SSE and reconnect are unavailable', async () => {
+ const fetch = vi.fn().mockResolvedValue({
+ ok: true,
+ status: 202,
+ json: async () => ({ accepted: true, clientMsgId: 'first', sessionId: 'session' }),
+ });
+ const listener = vi.fn();
+ const conn = new SseConnection(createConfig({ fetch }));
+ conn.onMessage(listener);
+ conn.connect();
+ conn.send({ type: 'send', prompt: 'first', clientMsgId: 'first', sessionId: null });
+ await vi.advanceTimersByTimeAsync(0);
+ expect(fetch).toHaveBeenCalledWith(
+ 'https://localhost:3100/api/chat/send',
+ expect.objectContaining({ body: expect.stringContaining('first') }),
+ );
+ expect(listener).toHaveBeenCalledWith(
+ expect.objectContaining({ type: '_send_accepted', sessionId: 'session' }),
+ );
+ expect(conn.getTrackedSessions()).toContain('session');
+ lastES()._emit('welcome', { connectionId: 'fresh' });
+ await vi.advanceTimersByTimeAsync(0);
+ expect(fetch).toHaveBeenCalledWith(
+ 'https://localhost:3100/api/chat/reconnect',
+ expect.objectContaining({ body: expect.stringContaining('session') }),
+ );
+ conn.disconnect();
+ });
+
+ it('rebuilds the stream when a reconnect POST hangs', async () => {
+ const fetch = vi.fn().mockReturnValue(new Promise(() => {}));
+ const conn = new SseConnection(createConfig({ fetch }));
+ conn.connect();
+ conn.trackSeq('session', 1);
+ lastES()._emit('welcome', { connectionId: 'old' });
+ await vi.advanceTimersByTimeAsync(15500);
+ expect(MockEventSource.instances).toHaveLength(2);
+ conn.disconnect();
+ });
+
// ─── Connection lifecycle ────────────────────────────────────────────────
it('creates EventSource on connect()', () => {
@@ -166,16 +299,16 @@ describe('SseConnection', () => {
conn.connect();
lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' });
- conn.send({ type: 'send', sessionId: 'sess-1', prompt: 'hello', clientMsgId: 'msg-1' });
+ conn.send({ type: 'watch', sessionId: 'sess-1', prompt: 'hello', clientMsgId: 'msg-1' });
- expect(mockFetch).toHaveBeenCalledWith('https://localhost:3100/api/chat/send', {
+ expect(mockFetch).toHaveBeenCalledWith('https://localhost:3100/api/chat/watch', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Connection-ID': 'conn-abc',
},
body: JSON.stringify({
- type: 'send',
+ type: 'watch',
sessionId: 'sess-1',
prompt: 'hello',
clientMsgId: 'msg-1',
@@ -190,7 +323,7 @@ describe('SseConnection', () => {
lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' });
const types: Array<[string, string]> = [
- ['send', 'send'],
+ ['watch', 'watch'],
['stop', 'stop'],
['interrupt', 'interrupt'],
['permission_response', 'permission'],
@@ -229,7 +362,7 @@ describe('SseConnection', () => {
conn.connect();
// Not yet connected — should queue
- const queued = conn.send({ type: 'send', prompt: 'queued', clientMsgId: 'q-1' });
+ const queued = conn.send({ type: 'watch', prompt: 'queued', clientMsgId: 'q-1' });
expect(queued).toBe(true);
expect(mockFetch).not.toHaveBeenCalled();
@@ -237,7 +370,7 @@ describe('SseConnection', () => {
lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' });
expect(mockFetch).toHaveBeenCalledWith(
- 'https://localhost:3100/api/chat/send',
+ 'https://localhost:3100/api/chat/watch',
expect.objectContaining({ method: 'POST' }),
);
});
@@ -247,7 +380,7 @@ describe('SseConnection', () => {
conn.connect();
for (let i = 0; i < 101; i++) {
- conn.send({ type: 'send', prompt: `msg-${i}`, clientMsgId: `id-${i}` });
+ conn.send({ type: 'watch', prompt: `msg-${i}`, clientMsgId: `id-${i}` });
}
// clearPendingSends exposes queue length indirectly
@@ -261,12 +394,12 @@ describe('SseConnection', () => {
expect(mockFetch).toHaveBeenCalledTimes(100);
});
- it('clearPendingSends() empties the queue', () => {
+ it('clearPendingSends() empties the control queue', () => {
const mockFetch = vi.fn().mockResolvedValue({ ok: true });
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
conn.connect();
- conn.send({ type: 'send', prompt: 'will be cleared', clientMsgId: 'c-1' });
+ conn.send({ type: 'watch', prompt: 'will be cleared', clientMsgId: 'c-1' });
conn.clearPendingSends();
lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' });
@@ -423,7 +556,7 @@ describe('SseConnection', () => {
// Resolve the FIRST (stale) reconnect POST
resolveFirst({ ok: true });
- await vi.runAllTimersAsync();
+ await vi.advanceTimersByTimeAsync(0);
// Must NOT set _connected — connectionId has moved on to conn-ghi
expect(conn.isConnected()).toBe(false);
@@ -432,7 +565,7 @@ describe('SseConnection', () => {
// Resolve the SECOND (current) reconnect POST
listener.mockClear();
resolveSecond({ ok: true });
- await vi.runAllTimersAsync();
+ await vi.advanceTimersByTimeAsync(0);
// Now _connected should be true
expect(conn.isConnected()).toBe(true);
@@ -459,7 +592,7 @@ describe('SseConnection', () => {
// Force reconnect — sends are now queued
conn.checkAndReconnect(true);
- conn.send({ type: 'send', prompt: 'queued msg', clientMsgId: 'q-1' });
+ conn.send({ type: 'watch', prompt: 'queued msg', clientMsgId: 'q-1' });
postEndpoints.length = 0;
// Welcome — reconnect POST fires, queued send waits
@@ -472,7 +605,7 @@ describe('SseConnection', () => {
resolveReconnect({ ok: true });
await vi.runAllTimersAsync();
- expect(postEndpoints).toEqual(['reconnect', 'send']);
+ expect(postEndpoints).toEqual(['reconnect', 'watch']);
});
it('stays disconnected when reconnect POST fails', async () => {
@@ -491,7 +624,7 @@ describe('SseConnection', () => {
// Force reconnect
conn.checkAndReconnect(true);
- conn.send({ type: 'send', prompt: 'should stay queued', clientMsgId: 'q-1' });
+ conn.send({ type: 'watch', prompt: 'should stay queued', clientMsgId: 'q-1' });
listener.mockClear();
// New welcome — reconnect POST will fail
@@ -625,7 +758,7 @@ describe('SseConnection', () => {
// Force reconnect and queue a send
conn.checkAndReconnect(true);
- conn.send({ type: 'send', prompt: 'must survive', clientMsgId: 'q-1' });
+ conn.send({ type: 'watch', prompt: 'must survive', clientMsgId: 'q-1' });
postEndpoints.length = 0;
// First welcome — reconnect fails, send stays queued
@@ -638,7 +771,7 @@ describe('SseConnection', () => {
lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-ghi' });
await vi.runAllTimersAsync();
expect(conn.isConnected()).toBe(true);
- expect(postEndpoints).toEqual(['reconnect', 'send']);
+ expect(postEndpoints).toEqual(['reconnect', 'watch']);
});
it('schedules delayed reconnect when reconnect POST fails', async () => {
@@ -778,7 +911,7 @@ describe('SseConnection', () => {
expect(listener).toHaveBeenCalledWith({ type: '_open' });
});
- it('does not send reconnect POST on first connection', () => {
+ it('restores tracked sessions on the first connection', () => {
const mockFetch = vi.fn().mockResolvedValue({ ok: true });
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
conn.connect();
@@ -786,10 +919,10 @@ describe('SseConnection', () => {
// Track a session BEFORE welcome (simulating a pre-existing session)
conn.trackSeq('sess-1', 5);
- // First welcome — _isReconnect is false, so no reconnect POST
+ // A restored/accepted session needs replay even on the first stream.
lastES()._emit('welcome', { type: 'welcome', protocolVersion: 2, connectionId: 'conn-abc' });
- expect(mockFetch).not.toHaveBeenCalledWith(
+ expect(mockFetch).toHaveBeenCalledWith(
'https://localhost:3100/api/chat/reconnect',
expect.any(Object),
);
diff --git a/packages/client/src/protocol-parser.ts b/packages/client/src/protocol-parser.ts
index be21abb9..1b22c8b1 100644
--- a/packages/client/src/protocol-parser.ts
+++ b/packages/client/src/protocol-parser.ts
@@ -47,12 +47,6 @@ export interface ProtocolCallbacks {
/** @deprecated v1 only — called to mark the WS pool entry as running/not-running. */
setWsRunning?(poolKey: string, running: boolean): void;
- /** @deprecated v1 only — called to send a queued message after session_end. */
- sendQueued?(poolKey: string, msg: unknown): void;
-
- /** v2: Called when a queued message should be sent after session_end. */
- onSendQueued?(msg: Record): void;
-
/** v2: Called with token data from session_switched response. */
onTokensHydrated?(tokens: Record): void;
@@ -65,9 +59,6 @@ export interface ProtocolCallbacks {
export interface ProtocolParserState {
/** Currently tracked session ID (used for expiry detection). */
currentSessionId: string | undefined;
-
- /** Queued messages to send after current session ends (FIFO). */
- pendingSend: Record[];
}
// ─── Parser result ───────────────────────────────────────────────────────────
@@ -364,18 +355,6 @@ export function parseServerMessage(
if (msg.sessionId && !state.currentSessionId) {
callbacks.onSessionAssigned(msg.sessionId as string);
}
- // Drain first queued message. Optimistic running=true avoids UI flicker
- // between the send and the server's session_state_changed confirmation.
- const pending = state.pendingSend.shift();
- if (pending) {
- result.messagesActions.push({ type: 'SESSION_STATE_CHANGED', state: 'running' });
- if (callbacks.onSendQueued) {
- callbacks.onSendQueued(pending);
- } else {
- callbacks.setWsRunning?.(poolKey, true);
- callbacks.sendQueued?.(poolKey, pending);
- }
- }
break;
}
@@ -450,7 +429,6 @@ export function parseServerMessage(
const errorMsg = msg.error as string;
callbacks.setWsRunning?.(poolKey, false);
- state.pendingSend = [];
result.messagesActions.push({
type: 'ERROR',
error: errorMsg || 'Unknown error',
diff --git a/packages/client/src/send-outbox.ts b/packages/client/src/send-outbox.ts
new file mode 100644
index 00000000..84032f77
--- /dev/null
+++ b/packages/client/src/send-outbox.ts
@@ -0,0 +1,159 @@
+/** Ordered, acknowledged HTTP delivery. SSE state never gates prompt submission. */
+interface Entry {
+ body: Record;
+ scope: number;
+}
+interface Config {
+ url: string;
+ fetch: (url: string, init?: RequestInit) => Promise;
+ notify: (event: Record) => void;
+ headers?: () => Record;
+ storage?: Pick;
+ timeoutMs?: number;
+}
+
+export class SendOutbox {
+ private entries: Entry[] = [];
+ private active = false;
+ private busy = false;
+ private timer?: ReturnType;
+ private failures = 0;
+ private readonly key: string;
+
+ constructor(private config: Config) {
+ this.key = `mitzo-send-outbox:${config.url}`;
+ try {
+ const saved: unknown = JSON.parse(config.storage?.getItem(this.key) ?? '[]');
+ if (Array.isArray(saved))
+ this.entries = saved.filter(
+ (entry): entry is Entry =>
+ entry &&
+ typeof entry.scope === 'number' &&
+ entry.body?.type === 'send' &&
+ typeof entry.body.clientMsgId === 'string' &&
+ typeof entry.body.prompt === 'string',
+ );
+ } catch {
+ /* Storage can be unavailable in private browsing. */
+ }
+ }
+
+ start(): void {
+ this.active = true;
+ void this.pump();
+ }
+ stop(): void {
+ this.active = false;
+ if (this.timer) clearTimeout(this.timer);
+ this.timer = undefined;
+ // Pause new work; the bounded in-flight request may still acknowledge.
+ }
+
+ enqueue(body: Record, scope: number): boolean {
+ if (this.entries.length >= 100) return false;
+ this.entries.push({ body: { ...body }, scope });
+ this.persist();
+ this.config.notify({
+ type: '_send_pending',
+ clientMsgId: body.clientMsgId,
+ sessionId: body.sessionId,
+ });
+ void this.pump();
+ return true;
+ }
+
+ private persist(): void {
+ try {
+ this.config.storage?.setItem(this.key, JSON.stringify(this.entries));
+ } catch {
+ /* In-memory retries still work if storage is full/disabled. */
+ }
+ }
+
+ private async pump(): Promise {
+ if (!this.active || this.busy || this.timer || !this.entries.length) return;
+ this.busy = true;
+ const entry = this.entries[0];
+ const abort = new AbortController();
+ let timeout: ReturnType | undefined;
+ try {
+ const { response, receipt } = await Promise.race([
+ this.config
+ .fetch(this.config.url, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', ...this.config.headers?.() },
+ body: JSON.stringify(entry.body),
+ signal: abort.signal,
+ })
+ .then(async (response) => {
+ if (response.status === 429 || response.status >= 500)
+ throw new Error('Server temporarily unavailable');
+ // A definitive HTTP rejection remains definitive even when an
+ // intermediary supplies HTML. A malformed success is ambiguous:
+ // retain its command ID and retry for the authoritative receipt.
+ const receipt = response.ok
+ ? await response.json()
+ : await response.json().catch(() => null);
+ return { response, receipt };
+ }),
+ new Promise((_, reject) => {
+ abort.signal.addEventListener('abort', () => reject(new Error('Delivery interrupted')), {
+ once: true,
+ });
+ timeout = setTimeout(() => abort.abort(), this.config.timeoutMs ?? 15000);
+ }),
+ ]);
+ if (!response.ok) {
+ this.entries.shift();
+ this.config.notify({
+ type: '_send_failed',
+ clientMsgId: entry.body.clientMsgId,
+ sessionId: entry.body.sessionId,
+ error: receipt?.error ?? `Message was not accepted (HTTP ${response.status}).`,
+ });
+ } else {
+ if (
+ receipt.accepted !== true ||
+ receipt.clientMsgId !== entry.body.clientMsgId ||
+ (typeof receipt.sessionId !== 'string' && receipt.sessionId !== null)
+ )
+ throw new Error('Missing message acknowledgement');
+ this.entries.shift();
+ // Only unattempted follow-ups in the same draft inherit its session.
+ if (entry.body.sessionId === null && receipt.sessionId) {
+ for (const queued of this.entries) {
+ if (queued.scope === entry.scope && queued.body.sessionId === null)
+ queued.body.sessionId = receipt.sessionId;
+ }
+ }
+ this.config.notify({
+ type: '_send_accepted',
+ ...receipt,
+ originalSessionId: entry.body.sessionId,
+ });
+ }
+ this.failures = 0;
+ this.persist();
+ } catch {
+ if (this.active) {
+ this.config.notify({
+ type: '_send_pending',
+ clientMsgId: entry.body.clientMsgId,
+ sessionId: entry.body.sessionId,
+ retrying: true,
+ });
+ this.timer = setTimeout(
+ () => {
+ this.timer = undefined;
+ void this.pump();
+ },
+ Math.min(1000 * 2 ** this.failures++, 10000),
+ );
+ }
+ } finally {
+ if (timeout) clearTimeout(timeout);
+ this.busy = false;
+ if (this.active && !this.timer) void this.pump();
+ }
+ }
+}
diff --git a/packages/client/src/sse-connection.ts b/packages/client/src/sse-connection.ts
index 8ca85ed8..8fdd4a59 100644
--- a/packages/client/src/sse-connection.ts
+++ b/packages/client/src/sse-connection.ts
@@ -13,6 +13,7 @@
* The server runs both transports in parallel during the migration period.
*/
+import { SendOutbox } from './send-outbox.js';
import type { ConnectionListener } from './connection.js';
import type { ChatConnection } from './chat-connection.js';
@@ -26,6 +27,7 @@ export interface SseConnectionConfig {
reconnectDelayMs?: number;
/** URL for the sendBeacon suspend fallback. */
suspendUrl?: string;
+ outboxStorage?: Pick;
}
const MAX_PENDING_SENDS = 100;
@@ -34,7 +36,15 @@ export class SseConnection implements ChatConnection {
private es: EventSource | null = null;
private _connectionId: string | null = null;
private _connected = false;
- private _isReconnect = false;
+ private sendScope = Date.now();
+ private replayRequest: {
+ es: EventSource | null;
+ connectionId: string;
+ dirty: boolean;
+ } | null = null;
+ private outbox: SendOutbox;
+ private foregroundProbe: { nonce: string; cancel: () => void } | null = null;
+ private probeCounter = 0;
private listener: ConnectionListener | null = null;
private seqBySession = new Map();
private pendingSends: Array<{ endpoint: string; body: Record }> = [];
@@ -42,7 +52,8 @@ export class SseConnection implements ChatConnection {
private boundOnVisibility: (() => void) | null = null;
private boundOnPageShow: ((e: PageTransitionEvent) => void) | null = null;
private boundOnPageHide: (() => void) | null = null;
- private config: Required;
+ private config: Required> &
+ Pick;
constructor(config: SseConnectionConfig) {
this.config = {
@@ -51,14 +62,43 @@ export class SseConnection implements ChatConnection {
suspendUrl: '',
...config,
};
+ this.outbox = new SendOutbox({
+ url: `${config.baseUrl}/api/chat/send`,
+ fetch: config.fetch,
+ storage: config.outboxStorage,
+ headers: (): Record =>
+ this._connectionId ? { 'X-Connection-ID': this._connectionId } : {},
+ notify: (event) => {
+ this.listener?.(event);
+ if (event.type === '_send_accepted' && typeof event.sessionId === 'string') {
+ const sessionId = event.sessionId as string;
+ if (!this.seqBySession.has(sessionId)) {
+ this.seqBySession.set(sessionId, 0);
+ // Include acknowledgements arriving during the welcome replay too.
+ if (
+ this.es &&
+ this._connectionId &&
+ (this._connected ||
+ (this.replayRequest?.es === this.es &&
+ this.replayRequest.connectionId === this._connectionId))
+ )
+ void this.doReconnectPost(this._connectionId, this.es);
+ }
+ }
+ },
+ });
}
connect(): void {
+ this.outbox.start();
this.doConnect();
this.addBrowserListeners();
}
disconnect(): void {
+ this.foregroundProbe?.cancel();
+ this.outbox.stop();
+ this.clearPendingSends();
this.removeBrowserListeners();
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
@@ -80,10 +120,12 @@ export class SseConnection implements ChatConnection {
* { type: 'interrupt', ... } → POST /api/chat/interrupt
* etc.
*
- * Returns true if the message was sent or queued, false if not connected
- * and not reconnecting.
+ * Prompts enter the acknowledged outbox regardless of SSE readiness.
+ * Control messages wait for replay readiness. False means the request
+ * could not be queued; true is local acceptance, not server delivery.
*/
send(msg: Record): boolean {
+ if (msg.type === 'send') return this.outbox.enqueue(msg, this.sendScope);
const endpoint = this.messageTypeToEndpoint(msg.type as string);
if (!endpoint) return false;
@@ -128,7 +170,10 @@ export class SseConnection implements ChatConnection {
this.seqBySession.delete(sessionId);
}
+ // Navigation discards stale controls, not submitted prompts. Scope prevents
+ // pending prompts in a different draft from inheriting an earlier receipt.
clearPendingSends(): void {
+ this.sendScope++;
this.pendingSends = [];
}
@@ -175,6 +220,7 @@ export class SseConnection implements ChatConnection {
*/
checkAndReconnect(force = false): void {
if (!force && this._connected) return;
+ this.foregroundProbe?.cancel();
if (this.reconnectTimer) return;
if (this.es) {
this.es.close();
@@ -209,6 +255,7 @@ export class SseConnection implements ChatConnection {
// Welcome event — server sends connectionId
es.addEventListener('welcome', (e: MessageEvent) => {
+ if (this.es !== es) return;
let msg: Record;
try {
msg = JSON.parse(e.data);
@@ -217,24 +264,24 @@ export class SseConnection implements ChatConnection {
}
this._connectionId = msg.connectionId as string;
- // _connected deferred until doReconnectPost succeeds — prevents
- // external send() from bypassing the pending queue mid-reconnect.
+ // Control messages wait for replay readiness. Prompt delivery uses
+ // its independent HTTP outbox and never waits for this handshake.
// Capture both connectionId and ES instance for the staleness guard.
const welcomeConnectionId = this._connectionId;
const welcomeEs = this.es;
- if (this._isReconnect && this.seqBySession.size > 0) {
+ if (this.seqBySession.size > 0) {
this.doReconnectPost(welcomeConnectionId, welcomeEs);
} else {
this._connected = true;
this.flushPendingSends();
this.listener?.({ type: '_open' });
}
- this._isReconnect = true;
});
// Catch-all for session events. Server sends all non-welcome events as
// `event: message`, so es.onmessage handles everything — no allowlist needed.
es.onmessage = (e: MessageEvent) => {
+ if (this.es !== es) return;
let msg: Record;
try {
msg = JSON.parse(e.data);
@@ -242,6 +289,16 @@ export class SseConnection implements ChatConnection {
return;
}
+ if (msg.type === '_probe') {
+ if (this.foregroundProbe && msg.nonce === this.foregroundProbe.nonce) {
+ this.foregroundProbe.cancel();
+ // Backgrounding suspended the sessions even if the stream survived.
+ if (this._connectionId && this.seqBySession.size)
+ void this.doReconnectPost(this._connectionId, es);
+ }
+ return;
+ }
+
if (typeof msg.seq === 'number' && typeof msg.sessionId === 'string') {
this.seqBySession.set(msg.sessionId as string, msg.seq as number);
}
@@ -250,6 +307,7 @@ export class SseConnection implements ChatConnection {
};
es.onerror = () => {
+ if (this.es !== es) return;
// EventSource auto-reconnects on error. We only need to update
// our state and notify the listener.
if (this._connected) {
@@ -267,34 +325,55 @@ export class SseConnection implements ChatConnection {
*
* On failure the client stays disconnected — the next EventSource
* auto-reconnect will trigger a fresh welcome + retry. This prevents
- * flushing pending sends into the void when the server never ran
+ * flushing control messages before the server has run
* handleReconnect (no watch, no reattach, no replay).
*/
private async doReconnectPost(
welcomeConnectionId: string,
welcomeEs: EventSource | null,
): Promise {
+ if (
+ this.replayRequest?.es === welcomeEs &&
+ this.replayRequest.connectionId === welcomeConnectionId
+ ) {
+ this.replayRequest.dirty = true;
+ return;
+ }
+ const request = { es: welcomeEs, connectionId: welcomeConnectionId, dirty: false };
+ this.replayRequest = request;
+ const abort = new AbortController();
+ let timeout: ReturnType | undefined;
try {
- const res = await this.config.fetch(`${this.config.baseUrl}/api/chat/reconnect`, {
- method: 'POST',
- headers: {
- 'Content-Type': 'application/json',
- 'X-Connection-ID': welcomeConnectionId,
- },
- body: JSON.stringify({
- type: 'reconnect',
- sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({
- sessionId,
- lastSeq,
- })),
+ const res = await Promise.race([
+ this.config.fetch(`${this.config.baseUrl}/api/chat/reconnect`, {
+ method: 'POST',
+ signal: abort.signal,
+ headers: {
+ 'Content-Type': 'application/json',
+ 'X-Connection-ID': welcomeConnectionId,
+ },
+ body: JSON.stringify({
+ type: 'reconnect',
+ sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({
+ sessionId,
+ lastSeq,
+ })),
+ }),
}),
- });
+ new Promise((_, reject) => {
+ timeout = setTimeout(() => {
+ abort.abort();
+ reject(new Error('Reconnect timed out'));
+ }, 15000);
+ }),
+ ]);
// Guard: bail if disconnect() was called, a newer welcome arrived,
// or checkAndReconnect replaced the EventSource while in-flight.
if (!this.es || this.es !== welcomeEs || this._connectionId !== welcomeConnectionId) return;
if (res.ok) {
+ if (request.dirty) return;
this._connected = true;
this.flushPendingSends();
this.listener?.({ type: '_open' });
@@ -306,12 +385,28 @@ export class SseConnection implements ChatConnection {
if (!this.es || this.es !== welcomeEs || this._connectionId !== welcomeConnectionId) return;
console.warn('[SseConnection] reconnect POST failed', err);
this.scheduleReconnect();
+ } finally {
+ if (timeout) clearTimeout(timeout);
+ if (this.replayRequest === request) {
+ this.replayRequest = null;
+ if (
+ request.dirty &&
+ this.es &&
+ this.es === welcomeEs &&
+ this._connectionId === welcomeConnectionId
+ )
+ void this.doReconnectPost(welcomeConnectionId, welcomeEs);
+ }
}
}
/** Tear down and reconnect after a delay to avoid tight retry loops. */
private scheduleReconnect(): void {
if (this.reconnectTimer) return;
+ if (this._connected) {
+ this._connected = false;
+ this.listener?.({ type: '_close' });
+ }
if (this.es) {
this.es.close();
this.es = null;
@@ -325,17 +420,18 @@ export class SseConnection implements ChatConnection {
private async doPost(endpoint: string, body: Record): Promise {
if (!this._connectionId) return;
try {
- await this.config.fetch(`${this.config.baseUrl}/api/chat/${endpoint}`, {
+ const res = await this.config.fetch(`${this.config.baseUrl}/api/chat/${endpoint}`, {
method: 'POST',
- headers: {
- 'Content-Type': 'application/json',
- 'X-Connection-ID': this._connectionId,
- },
+ headers: { 'Content-Type': 'application/json', 'X-Connection-ID': this._connectionId },
body: JSON.stringify(body),
});
+ if (!res.ok)
+ this.listener?.({
+ type: 'error',
+ error: `Could not ${endpoint} (${res.status}). Please retry.`,
+ });
} catch {
- // POST failures are non-fatal — the server may be temporarily
- // unreachable. The SSE stream will reconnect and replay missed events.
+ this.listener?.({ type: 'error', error: `Could not ${endpoint}. Please retry.` });
}
}
@@ -383,12 +479,46 @@ export class SseConnection implements ChatConnection {
// ─── Browser lifecycle ─────────────────────────────────────────────────────
+ private probeForeground(): void {
+ if (!this._connected || !this.es || !this._connectionId) {
+ this.checkAndReconnect();
+ return;
+ }
+ if (this.foregroundProbe) return;
+ const es = this.es;
+ const connectionId = this._connectionId;
+ const nonce = String(++this.probeCounter);
+ const abort = new AbortController();
+ const cancel = () => {
+ clearTimeout(timer);
+ abort.abort();
+ if (this.foregroundProbe?.nonce === nonce) this.foregroundProbe = null;
+ };
+ const fail = () => {
+ if (this.foregroundProbe?.nonce !== nonce) return;
+ cancel();
+ if (this.es === es && this._connectionId === connectionId) this.checkAndReconnect(true);
+ };
+ const timer = setTimeout(fail, 1000);
+ this.foregroundProbe = { nonce, cancel };
+ void this.config
+ .fetch(`${this.config.baseUrl}/api/chat/probe`, {
+ method: 'POST',
+ signal: abort.signal,
+ headers: { 'Content-Type': 'application/json', 'X-Connection-ID': connectionId },
+ body: JSON.stringify({ nonce }),
+ })
+ .then((res) => {
+ if (!res.ok) fail();
+ }, fail);
+ }
+
private addBrowserListeners(): void {
if (typeof globalThis.document === 'undefined') return;
this.boundOnVisibility = () => {
if (document.visibilityState === 'visible') {
- this.checkAndReconnect();
+ this.probeForeground();
this.listener?.({ type: '_foreground' });
} else if (document.visibilityState === 'hidden') {
this.sendSuspend();
@@ -396,7 +526,7 @@ export class SseConnection implements ChatConnection {
};
this.boundOnPageShow = (e: PageTransitionEvent) => {
- if (e.persisted) this.checkAndReconnect();
+ if (e.persisted) this.checkAndReconnect(true);
};
this.boundOnPageHide = () => {
diff --git a/packages/client/src/store.ts b/packages/client/src/store.ts
index 9adb7767..09a9cf85 100644
--- a/packages/client/src/store.ts
+++ b/packages/client/src/store.ts
@@ -87,6 +87,7 @@ export interface MitzoStoreState {
// Error state
sendError: string | null;
+ sendStatus: string | null;
// Pending session (for "Start Session" from inbox/todo)
pendingSession: PendingSession | null;
@@ -174,8 +175,6 @@ function removeTaskFromTree(tasks: Task[], id: string): Task[] {
});
}
-const PENDING_SEND_TIMEOUT_MS = 5_000;
-
// ─── Factory ─────────────────────────────────────────────────────────────────
export function createMitzoStore(options: MitzoStoreOptions): StoreApi {
@@ -184,12 +183,7 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi;
- } = {
- currentSessionId: undefined,
- pendingSend: [],
- };
+ const parserState: ProtocolParserState = { currentSessionId: undefined };
let recoveryInFlight = false;
@@ -221,13 +215,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi((set, get) => ({
// ── Initial state ────────────────────────────────────────────────────
@@ -244,6 +231,7 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({
sessions: { ...s.sessions, active: id },
messages: INITIAL_MESSAGES_STATE,
+ sendError: null,
+ sendStatus: null,
permissions: INITIAL_PERMISSIONS_STATE,
tokens: INITIAL_TOKENS_STATE,
progress: INITIAL_PROGRESS_STATE,
@@ -294,13 +282,13 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi => {
const msg: Record = {
@@ -349,42 +336,13 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({
- messages: messagesReducer(s.messages, {
- type: 'SESSION_STATE_CHANGED',
- state: 'running',
- }),
- }));
- connection.send(pending);
- // Reschedule for remaining queued messages
- if (parserState.pendingSend.length > 0) {
- parserState.pendingSendTimer = setTimeout(drainOne, PENDING_SEND_TIMEOUT_MS);
- } else {
- parserState.pendingSendTimer = undefined;
- }
- }, PENDING_SEND_TIMEOUT_MS);
- } else {
- const sent = connection.send(msg);
- if (!sent) {
- set({ sendError: 'Not connected. Message will be sent when reconnected.' });
- }
- }
+ const sent = connection.send(msg);
+ if (!sent) set({ sendError: 'Message could not be queued. Please retry.' });
},
interruptMessage(text: string, opts?: SendMessageOptions) {
@@ -694,10 +652,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi) {
- connection.send(msg);
- },
-
onReconnected() {
const activeId = parserState.currentSessionId;
if (activeId) fetchAndRestoreMessages(activeId);
@@ -718,6 +672,33 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi) {
+ if (
+ msg.type === '_send_pending' ||
+ msg.type === '_send_failed' ||
+ msg.type === '_send_accepted'
+ ) {
+ const visible = store
+ .getState()
+ .messages.messages.some((m) => m.messageId === msg.clientMsgId);
+ if (visible) {
+ store.setState({
+ sendError: msg.type === '_send_failed' ? String(msg.error) : null,
+ sendStatus:
+ msg.type === '_send_pending'
+ ? msg.retrying
+ ? 'Reconnecting — your message will retry automatically.'
+ : 'Sending…'
+ : null,
+ });
+ if (
+ msg.type === '_send_accepted' &&
+ typeof msg.sessionId === 'string' &&
+ !parserState.currentSessionId
+ )
+ callbacks.onSessionAssigned(msg.sessionId as string);
+ }
+ return;
+ }
// Foreground recovery: when the page becomes visible again (iOS may have
// evicted it from memory, losing in-memory state), re-fetch messages from
// the REST API if we have an active session but no messages in the store.
@@ -762,12 +743,6 @@ export function createMitzoStore(options: MitzoStoreOptions): StoreApi ({
messages: messagesReducer(s.messages, action),
diff --git a/packages/harness/src/session-registry.ts b/packages/harness/src/session-registry.ts
index 8f8678ac..8052ceda 100644
--- a/packages/harness/src/session-registry.ts
+++ b/packages/harness/src/session-registry.ts
@@ -26,6 +26,8 @@ import type {
} from '@mitzo/protocol';
export interface ManagedSession {
+ /** Current event connection; the registry key remains stable for the query lifetime. */
+ ownerConnectionId?: string;
transport: SessionTransport;
abortController: AbortController;
sessionId?: string;
diff --git a/packages/protocol/src/event-store.ts b/packages/protocol/src/event-store.ts
index 92fefd59..0ed371d9 100644
--- a/packages/protocol/src/event-store.ts
+++ b/packages/protocol/src/event-store.ts
@@ -87,7 +87,22 @@ interface SessionRow {
updated_at: number;
}
+export interface SendCommandReceipt {
+ clientMsgId: string;
+ sessionId: string | null;
+ payload: Record;
+ error: string | null;
+}
+
const SCHEMA = `
+ CREATE TABLE IF NOT EXISTS send_commands (
+ client_msg_id TEXT PRIMARY KEY,
+ session_id TEXT NOT NULL,
+ payload TEXT NOT NULL,
+ error TEXT,
+ created_at INTEGER NOT NULL DEFAULT (unixepoch('now', 'subsec') * 1000)
+ );
+
CREATE TABLE IF NOT EXISTS events (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
@@ -144,6 +159,74 @@ export class EventStore {
getSessionState: Database.Statement;
};
+ getSendCommand(clientMsgId: string): SendCommandReceipt | undefined {
+ const row = this.db!.prepare('SELECT * FROM send_commands WHERE client_msg_id = ?').get(
+ clientMsgId,
+ ) as
+ | { client_msg_id: string; session_id: string; payload: string; error: string | null }
+ | undefined;
+ return (
+ row && {
+ clientMsgId: row.client_msg_id,
+ sessionId: row.session_id || null,
+ payload: JSON.parse(row.payload),
+ error: row.error,
+ }
+ );
+ }
+
+ /** Synchronous insert before dispatch: retries can never allocate another session. */
+ insertSendCommand(
+ clientMsgId: string,
+ sessionId: string,
+ payload: Record,
+ ): void {
+ this.db!.prepare(
+ 'INSERT INTO send_commands (client_msg_id, session_id, payload) VALUES (?, ?, ?)',
+ ).run(clientMsgId, sessionId, JSON.stringify(payload));
+ }
+
+ completeNativeSendCommand(clientMsgId: string): void {
+ this.db!.prepare("UPDATE send_commands SET session_id = '' WHERE client_msg_id = ?").run(
+ clientMsgId,
+ );
+ }
+
+ /** A crash may happen between acceptance and dispatch. Never silently discard
+ * that receipt or re-execute a possibly side-effecting command after restart. */
+ recoverPendingSendCommands(): void {
+ const rows = this.db!.prepare(
+ `SELECT client_msg_id, session_id, payload FROM send_commands c
+ WHERE error IS NULL AND session_id != '' AND NOT EXISTS (
+ SELECT 1 FROM events e WHERE e.session_id = c.session_id AND e.type = 'user_message'
+ AND json_extract(e.payload, '$.messageId') = c.client_msg_id
+ )`,
+ ).all() as Array<{ client_msg_id: string; session_id: string; payload: string }>;
+ for (const row of rows) {
+ const error =
+ 'Server restarted before message execution was confirmed. Please check the conversation and retry.';
+ this.failSendCommand(row.client_msg_id, error);
+ if (!this.getSession(row.session_id)) {
+ const payload = JSON.parse(row.payload);
+ this.upsertSession({ sessionId: row.session_id, initialPrompt: payload.prompt });
+ }
+ this.append(row.session_id, 'error', {
+ type: 'error',
+ v: 2,
+ sessionId: row.session_id,
+ error,
+ });
+ this.setSessionState(row.session_id, 'ENDED', { force: true, reason: 'server_restart' });
+ }
+ }
+
+ failSendCommand(clientMsgId: string, error: string): void {
+ this.db!.prepare('UPDATE send_commands SET error = ? WHERE client_msg_id = ?').run(
+ error,
+ clientMsgId,
+ );
+ }
+
constructor(dbPath: string, logger?: EventStoreLogger) {
this.log = logger ?? noopLogger;
const db = new Database(dbPath);
diff --git a/server/__tests__/chat-rest-handler.test.ts b/server/__tests__/chat-rest-handler.test.ts
index 66251fe3..72fa5675 100644
--- a/server/__tests__/chat-rest-handler.test.ts
+++ b/server/__tests__/chat-rest-handler.test.ts
@@ -1,5 +1,6 @@
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import express from 'express';
+import { EventStore } from '../event-store.js';
import request from 'supertest';
import { SessionSseRegistry } from '../session-sse-registry.js';
import { SseTransport } from '../sse-transport.js';
@@ -55,7 +56,7 @@ function buildApp(sseRegistry: SessionSseRegistry, connRegistry: ConnectionRegis
const ctx: V2HandlerContext = {
connRegistry,
sessionRegistry: {} as V2HandlerContext['sessionRegistry'],
- eventStore: {} as V2HandlerContext['eventStore'],
+ eventStore: new EventStore(':memory:'),
nativeCommands: {} as V2HandlerContext['nativeCommands'],
};
@@ -71,6 +72,7 @@ describe('chat-rest-handler', () => {
let sseRegistry: SessionSseRegistry;
let connRegistry: ConnectionRegistry;
let testApp: express.Express;
+ let eventStore: EventStore;
const CONNECTION_ID = 'conn-test-123';
beforeEach(() => {
@@ -84,33 +86,69 @@ describe('chat-rest-handler', () => {
const transport = new SseTransport(CONNECTION_ID, sseRegistry);
connRegistry.register(CONNECTION_ID, transport);
- const { app } = buildApp(sseRegistry, connRegistry);
+ const { app, ctx } = buildApp(sseRegistry, connRegistry);
testApp = app;
+ eventStore = ctx.eventStore;
});
afterEach(() => {
+ eventStore.close();
sseRegistry.destroy();
connRegistry.dispose();
});
// ─── Header validation ──────────────────────────────────────────────────
- it('rejects requests without X-Connection-ID', async () => {
- const res = await request(testApp)
- .post('/api/chat/send')
- .send({ prompt: 'hello', clientMsgId: 'msg-1' });
+ it('returns a liveness nonce through the existing SSE connection', async () => {
+ const transport = connRegistry.get(CONNECTION_ID)!.transport;
+ const send = vi.spyOn(transport, 'send');
+ const response = await request(testApp)
+ .post('/api/chat/probe')
+ .set('X-Connection-ID', CONNECTION_ID)
+ .send({ nonce: 'probe-1' });
+ expect(response.status).toBe(202);
+ expect(send).toHaveBeenCalledWith({ type: '_probe', nonce: 'probe-1' });
+ });
- expect(res.status).toBe(400);
- expect(res.body.error).toContain('X-Connection-ID');
+ it('persists startup events before a stream exists without duplicating sequenced events', async () => {
+ vi.mocked(handleSendV2).mockImplementationOnce((_id, transport, _msg, _ctx, delivery) => {
+ if (transport.isOpen()) transport.send({ type: 'session_info', branch: 'main' });
+ const sid = delivery!.initialSessionId!;
+ const seq = eventStore.append(sid, 'message_start', {
+ v: 2,
+ type: 'message_start',
+ messageId: 'm',
+ });
+ transport.send({ v: 2, type: 'message_start', messageId: 'm', seq });
+ });
+ const response = await request(testApp).post('/api/chat/send').send({
+ type: 'send',
+ sessionId: null,
+ prompt: 'hello',
+ clientMsgId: 'early',
+ });
+ expect(response.status).toBe(202);
+ const events = eventStore.getEventsAfter(response.body.sessionId, 0);
+ expect(events.map((e) => e.type)).toEqual(['session_info', 'message_start']);
});
- it('rejects requests with unknown connection (getTransport path)', async () => {
- const res = await request(testApp)
+ it('accepts the first prompt with no SSE stream and deduplicates a retry', async () => {
+ const message = { type: 'send', sessionId: null, prompt: 'hello', clientMsgId: 'msg-1' };
+ const first = await request(testApp).post('/api/chat/send').send(message);
+ const second = await request(testApp)
.post('/api/chat/send')
- .set('X-Connection-ID', 'conn-nonexistent')
- .send({ prompt: 'hello', clientMsgId: 'msg-1' });
-
- expect(res.status).toBe(404);
+ .set('X-Connection-ID', 'expired')
+ .send(message);
+ expect(first.status).toBe(202);
+ expect(second.body).toEqual(first.body);
+ expect(first.body).toEqual(
+ expect.objectContaining({
+ accepted: true,
+ clientMsgId: 'msg-1',
+ sessionId: expect.any(String),
+ }),
+ );
+ expect(handleSendV2).toHaveBeenCalledTimes(1);
});
it('rejects requests with unknown connection (requireConnection path)', async () => {
@@ -140,9 +178,10 @@ describe('chat-rest-handler', () => {
expect(handleSendV2).toHaveBeenCalledOnce();
expect(handleSendV2).toHaveBeenCalledWith(
CONNECTION_ID,
- expect.any(SseTransport),
+ expect.objectContaining({ send: expect.any(Function), isOpen: expect.any(Function) }),
expect.objectContaining({ prompt: 'hello world' }),
expect.any(Object),
+ expect.objectContaining({ initialSessionId: res.body.sessionId }),
);
});
diff --git a/server/__tests__/chat-send-startup.test.ts b/server/__tests__/chat-send-startup.test.ts
new file mode 100644
index 00000000..1116d665
--- /dev/null
+++ b/server/__tests__/chat-send-startup.test.ts
@@ -0,0 +1,82 @@
+import { afterEach, expect, it, vi } from 'vitest';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { query } from '@anthropic-ai/claude-agent-sdk';
+vi.mock('@anthropic-ai/claude-agent-sdk', async (original) => ({
+ ...(await original