fix(transport): deliver first prompts reliably across reconnect

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.

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`,
     }

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() {
           </>
         )}
       </header>
+      {(sendError || sendStatus) && (
+        <div
+          role={sendError ? 'alert' : 'status'}
+          className={
+            sendError ? 'chat-delivery-status chat-delivery-error' : 'chat-delivery-status'
+          }
+        >
+          {sendError || sendStatus}
+        </div>
+      )}
       <div className="chat-account-bar">
         <AccountModelPicker
           disabled={messages.running}

frontend/src/pages/__tests__/DesktopChatView.test.tsx

@@ -118,6 +118,7 @@ function createMockStore() {
     },
     progress: { blocks: {}, toolIndex: {} },
     sendError: null,
+    sendStatus: null,
     dispatchMessages: vi.fn(),
     switchSession: vi.fn().mockResolvedValue(undefined),
     newSession: vi.fn(),

frontend/src/styles/global.css

@@ -6280,3 +6280,14 @@ textarea:focus {
   overflow-wrap: anywhere;
   line-height: 1.5;
 }
+
+.chat-delivery-status {
+  padding: 0.5rem 1rem;
+  font-size: 0.85rem;
+  color: var(--text-secondary);
+  border-bottom: 1px solid var(--border);
+}
+
+.chat-delivery-error {
+  color: var(--color-danger, #ef4444);
+}

packages/client/__tests__/protocol-parser.test.ts

@@ -5,7 +5,6 @@ import type { ProtocolCallbacks, ProtocolParserState } from '../src/protocol-par
 function makeState(overrides?: Partial<ProtocolParserState>): ProtocolParserState {
   return {
     currentSessionId: undefined,
-    pendingSend: [],
     ...overrides,
   };
 }
@@ -17,7 +16,6 @@ function makeCallbacks(overrides?: Partial<ProtocolCallbacks>): 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' });
   });
 });
 

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();
+  });
+});

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');
+  });
+});