Skip to content

Commit 4b38750

Browse files
dimakisclaude
dimakis
and
claude
authored
feat(transport): stable client connectionId + POST failure resilience (#401)
* feat(transport): stable client connectionId + POST failure resilience Client generates a UUID per tab, sends it via ?cid= on SSE GET. Server reuses it across reconnects, eliminating the ownership race in handleReconnect that caused the "send twice to wake agent" bug. - SseConnection: stable cid in URL, welcome validation, backward compat - ConnectionRegistry: reconnect() swaps transport preserving state, markInactive() with 60s TTL + onExpired callback for deferred cleanup - Server: accept ?cid=, validate format, connRegistry.reconnect() on existing cid, deferred close with onExpired → detachChat() - Frontend: sessionStorage-persisted connectionId per tab - Store: optimistic SET_RUNNING after send, _send_failed handler - doPost: re-queue failed sends (404/429/5xx) with _send_failed emission - Tests: 17 new tests for stable cid, reconnect, TTL, POST failures Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address Centaur review — ownership check, observer leak, retry limit - onExpired: verify session still owned by expired connectionId before detaching (prevents cross-tab detach race) - Stable-cid close: call removeObserver(transport) to prevent stale transport accumulation in SessionRegistry observer sets - CID regex: case-insensitive match for crypto.randomUUID() compat - doPost: add retry counter (max 3), emit willRetry:false after exhaustion - Export SendFailedEvent interface for type-safe event handling - Test: verify message dropped after max retries with willRetry:false Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address second Centaur review — spinner stuck, 404 handling, regex - store: clear SET_RUNNING when _send_failed fires with willRetry:false (prevents stuck "thinking" spinner on permanent failure) - sse-connection: treat 404 as permanent failure (no retry), separate from transient 429/5xx retries - server: strict UUID regex, import getOwnerConnection() from ws-handler-v2 instead of reimplementing - tests: 404 permanent failure, 4xx silently swallowed (400/403) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address third Centaur review — sendError cleanup, CID logging - Clear sendError on _open (reconnect success) to prevent stale error banners - Remove redundant type cast on msg.willRetry in _send_failed handler - Log warning when client sends invalid cid format on SSE connect Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address fourth Centaur review — SSE close race, retry timer - Fix critical race: stale close handler removing new SSE stream. Added removeIfCurrent() to SessionSseRegistry — only removes if the response being closed is still the current entry for that connectionId - Add timer-based retry flush (3s) for failed POSTs when SSE stays healthy, so messages don't sit in the queue waiting for a reconnect that never comes - Update ChatConnection jsdoc — connectionId is now client-generated - Clean up retry timer on disconnect Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address fifth Centaur review — guard markInactive race - Make removeIfCurrent() return boolean so close handler knows if stream was actually removed vs already replaced by reconnect - Only call markInactive() when the closing stream was the current one, preventing stale close handlers from starting TTL on fresh connections - Add SessionSseRegistry unit tests for removeIfCurrent() Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address sixth Centaur review — surface all 4xx failures - Emit _send_failed for all non-OK send responses, not just 404. Covers 400/401/403 which previously left the UI stuck in 'thinking' state - Fix error message: "retrying..." instead of "retrying on reconnect..." since scheduleRetryFlush now retries without waiting for reconnect Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address seventh Centaur review — clear stale sendError - Clear sendError on any incoming server event, confirming message delivery even when retry succeeded without SSE reconnect - Add SessionSseRegistry.add() stream-close test for reconnect safety Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(test): fix type assertion in session-sse-registry test Use proper Response intersection type instead of `as never` to satisfy tsc --noEmit in CI. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): scope sendError clearing to session events only Global events (task_state, inbox_updated) no longer clear the retry banner — only session-scoped events confirm message delivery. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(transport): address eighth review — scope sendError clearing - Remove premature sendError clear on _open (flushPendingSends hasn't resolved yet — let session events confirm delivery instead) - Scope sendError clear to active session events only, preventing background session activity from hiding retry state Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent b7631c9 commit 4b38750

10 files changed

Lines changed: 860 additions & 60 deletions

File tree

‎frontend/src/client-store.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,23 @@ import { getPreferredModel } from './lib/model-preference';
2525
*/
2626
const useSSE = typeof window !== 'undefined' && localStorage.getItem('mitzo:transport') === 'sse';
2727

28+
/** Stable connectionId per browser tab — survives SSE reconnects. */
29+
function getOrCreateConnectionId(): string {
30+
const KEY = 'mitzo:connectionId';
31+
let cid = sessionStorage.getItem(KEY);
32+
if (!cid) {
33+
cid = `cid-${crypto.randomUUID()}`;
34+
sessionStorage.setItem(KEY, cid);
35+
}
36+
return cid;
37+
}
38+
2839
const sseConfig: SseConnectionConfig | undefined = useSSE
2940
? {
3041
baseUrl: getApiBaseUrl(),
3142
fetch: (url, init) => apiFetch(url, init),
3243
suspendUrl: `${getApiBaseUrl()}/api/sessions/suspend`,
44+
connectionId: getOrCreateConnectionId(),
3345
}
3446
: undefined;
3547

‎packages/client/src/__tests__/sse-connection.test.ts‎

Lines changed: 321 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ function createConfig(overrides?: Partial<SseConnectionConfig>): SseConnectionCo
6666
baseUrl: 'https://localhost:3100',
6767
fetch: vi.fn().mockResolvedValue({ ok: true }),
6868
createEventSource: (url: string) => new MockEventSource(url) as unknown as EventSource,
69+
connectionId: 'cid-test',
6970
...overrides,
7071
};
7172
}
@@ -93,7 +94,7 @@ describe('SseConnection', () => {
9394
conn.connect();
9495

9596
expect(MockEventSource.instances).toHaveLength(1);
96-
expect(lastES().url).toBe('https://localhost:3100/api/chat/events');
97+
expect(lastES().url).toBe('https://localhost:3100/api/chat/events?cid=cid-test');
9798
});
9899

99100
it('becomes connected after welcome event', () => {
@@ -804,9 +805,10 @@ describe('SseConnection', () => {
804805
// Force reconnect
805806
conn.checkAndReconnect(true);
806807

807-
// URL should be clean — reconnect is handled via POST, not query param
808+
// URL should include cid but NOT sessions (reconnect is handled via POST)
808809
const newES = lastES();
809-
expect(newES.url).toBe('https://localhost:3100/api/chat/events');
810+
expect(newES.url).toContain('/api/chat/events?cid=');
811+
expect(newES.url).not.toContain('sessions=');
810812
});
811813

812814
it('checkAndReconnect(false) is no-op when connected', () => {
@@ -821,6 +823,80 @@ describe('SseConnection', () => {
821823
expect(MockEventSource.instances.length).toBe(count);
822824
});
823825

826+
// ─── Stable connectionId ────────────────────────────────────────────────
827+
828+
describe('stable connectionId', () => {
829+
it('includes client connectionId in EventSource URL', () => {
830+
const conn = new SseConnection(createConfig({ connectionId: 'cid-my-tab' }));
831+
conn.connect();
832+
expect(lastES().url).toBe('https://localhost:3100/api/chat/events?cid=cid-my-tab');
833+
});
834+
835+
it('auto-generates connectionId when not provided', () => {
836+
const conn = new SseConnection(createConfig({ connectionId: undefined }));
837+
conn.connect();
838+
expect(lastES().url).toMatch(/\?cid=cid-[0-9a-f-]{36}$/);
839+
});
840+
841+
it('preserves connectionId when server echoes it back', () => {
842+
const conn = new SseConnection(createConfig({ connectionId: 'cid-stable' }));
843+
conn.connect();
844+
lastES()._emit('welcome', {
845+
type: 'welcome',
846+
protocolVersion: 2,
847+
connectionId: 'cid-stable',
848+
});
849+
expect(conn.getConnectionId()).toBe('cid-stable');
850+
});
851+
852+
it('falls back to server connectionId when it differs (old server)', () => {
853+
const conn = new SseConnection(createConfig({ connectionId: 'cid-client' }));
854+
conn.connect();
855+
lastES()._emit('welcome', {
856+
type: 'welcome',
857+
protocolVersion: 2,
858+
connectionId: 'conn-server-assigned',
859+
});
860+
expect(conn.getConnectionId()).toBe('conn-server-assigned');
861+
});
862+
863+
it('uses same connectionId across reconnects', () => {
864+
const conn = new SseConnection(createConfig({ connectionId: 'cid-stable' }));
865+
conn.connect();
866+
lastES()._emit('welcome', {
867+
type: 'welcome',
868+
protocolVersion: 2,
869+
connectionId: 'cid-stable',
870+
});
871+
872+
conn.checkAndReconnect(true);
873+
874+
expect(lastES().url).toBe('https://localhost:3100/api/chat/events?cid=cid-stable');
875+
});
876+
877+
it('sends connectionId in X-Connection-ID header on POST', () => {
878+
const mockFetch = vi.fn().mockResolvedValue({ ok: true });
879+
const conn = new SseConnection(createConfig({ fetch: mockFetch, connectionId: 'cid-hdr' }));
880+
conn.connect();
881+
lastES()._emit('welcome', {
882+
type: 'welcome',
883+
protocolVersion: 2,
884+
connectionId: 'cid-hdr',
885+
});
886+
887+
conn.send({ type: 'send', prompt: 'hi' });
888+
889+
expect(mockFetch).toHaveBeenCalledWith(
890+
expect.any(String),
891+
expect.objectContaining({
892+
headers: expect.objectContaining({
893+
'X-Connection-ID': 'cid-hdr',
894+
}),
895+
}),
896+
);
897+
});
898+
});
899+
824900
// ─── Session tracking ──────────────────────────────────────────────────
825901

826902
it('trackSeq / getLastSeq / clearSession', () => {
@@ -872,4 +948,246 @@ describe('SseConnection', () => {
872948

873949
expect(mockFetch).not.toHaveBeenCalled();
874950
});
951+
952+
// ─── POST failure surfacing ───────────────────────────────────────────
953+
954+
describe('POST failure surfacing', () => {
955+
it('emits _send_failed on 500 response for send endpoint', async () => {
956+
const mockFetch = vi.fn().mockResolvedValue({ ok: false, status: 500 });
957+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
958+
const listener = vi.fn();
959+
conn.onMessage(listener);
960+
conn.connect();
961+
lastES()._emit('welcome', {
962+
type: 'welcome',
963+
protocolVersion: 2,
964+
connectionId: 'cid-test',
965+
});
966+
967+
conn.send({ type: 'send', prompt: 'hello', clientMsgId: 'msg-1' });
968+
await vi.runAllTimersAsync();
969+
970+
expect(listener).toHaveBeenCalledWith(
971+
expect.objectContaining({
972+
type: '_send_failed',
973+
clientMsgId: 'msg-1',
974+
willRetry: true,
975+
}),
976+
);
977+
});
978+
979+
it('emits _send_failed on network error for send endpoint', async () => {
980+
const mockFetch = vi.fn().mockRejectedValue(new Error('network error'));
981+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
982+
const listener = vi.fn();
983+
conn.onMessage(listener);
984+
conn.connect();
985+
lastES()._emit('welcome', {
986+
type: 'welcome',
987+
protocolVersion: 2,
988+
connectionId: 'cid-test',
989+
});
990+
991+
conn.send({ type: 'send', prompt: 'hello', clientMsgId: 'msg-2' });
992+
await vi.runAllTimersAsync();
993+
994+
expect(listener).toHaveBeenCalledWith(
995+
expect.objectContaining({
996+
type: '_send_failed',
997+
clientMsgId: 'msg-2',
998+
willRetry: true,
999+
}),
1000+
);
1001+
});
1002+
1003+
it('re-queues failed send for retry on reconnect', async () => {
1004+
let callCount = 0;
1005+
const mockFetch = vi.fn().mockImplementation(() => {
1006+
callCount++;
1007+
if (callCount === 1) return Promise.resolve({ ok: false, status: 500 });
1008+
return Promise.resolve({ ok: true });
1009+
});
1010+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1011+
conn.connect();
1012+
lastES()._emit('welcome', {
1013+
type: 'welcome',
1014+
protocolVersion: 2,
1015+
connectionId: 'cid-test',
1016+
});
1017+
1018+
conn.send({ type: 'send', prompt: 'retry me', clientMsgId: 'r-1' });
1019+
await vi.runAllTimersAsync();
1020+
1021+
conn.checkAndReconnect(true);
1022+
lastES()._emit('welcome', {
1023+
type: 'welcome',
1024+
protocolVersion: 2,
1025+
connectionId: 'cid-test',
1026+
});
1027+
await vi.runAllTimersAsync();
1028+
1029+
expect(mockFetch).toHaveBeenCalledTimes(2);
1030+
});
1031+
1032+
it('does not emit _send_failed for non-send endpoints', async () => {
1033+
const mockFetch = vi.fn().mockResolvedValue({ ok: false, status: 500 });
1034+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1035+
const listener = vi.fn();
1036+
conn.onMessage(listener);
1037+
conn.connect();
1038+
lastES()._emit('welcome', {
1039+
type: 'welcome',
1040+
protocolVersion: 2,
1041+
connectionId: 'cid-test',
1042+
});
1043+
1044+
conn.send({ type: 'stop', sessionId: 'sess-1' });
1045+
await vi.runAllTimersAsync();
1046+
1047+
expect(listener).not.toHaveBeenCalledWith(expect.objectContaining({ type: '_send_failed' }));
1048+
});
1049+
1050+
it('drops message after max retries and emits willRetry: false', async () => {
1051+
const mockFetch = vi.fn().mockResolvedValue({ ok: false, status: 500 });
1052+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1053+
const listener = vi.fn();
1054+
conn.onMessage(listener);
1055+
conn.connect();
1056+
lastES()._emit('welcome', {
1057+
type: 'welcome',
1058+
protocolVersion: 2,
1059+
connectionId: 'cid-test',
1060+
});
1061+
1062+
// Initial send fails → queued with retries:1
1063+
conn.send({ type: 'send', prompt: 'persistent fail', clientMsgId: 'pf-1' });
1064+
await vi.runAllTimersAsync();
1065+
1066+
// 3 reconnect cycles — each flush retries and fails again
1067+
for (let i = 0; i < 3; i++) {
1068+
conn.checkAndReconnect(true);
1069+
lastES()._emit('welcome', {
1070+
type: 'welcome',
1071+
protocolVersion: 2,
1072+
connectionId: 'cid-test',
1073+
});
1074+
await vi.runAllTimersAsync();
1075+
}
1076+
1077+
// 4 _send_failed events: 3 willRetry:true + 1 willRetry:false
1078+
const failedEvents = listener.mock.calls
1079+
.map((c: unknown[]) => c[0] as Record<string, unknown>)
1080+
.filter((m) => m.type === '_send_failed');
1081+
1082+
expect(failedEvents).toHaveLength(4);
1083+
expect(failedEvents[3]).toEqual(expect.objectContaining({ willRetry: false }));
1084+
1085+
// One more reconnect — message should NOT be retried (was dropped)
1086+
mockFetch.mockClear();
1087+
conn.checkAndReconnect(true);
1088+
lastES()._emit('welcome', {
1089+
type: 'welcome',
1090+
protocolVersion: 2,
1091+
connectionId: 'cid-test',
1092+
});
1093+
await vi.runAllTimersAsync();
1094+
1095+
expect(mockFetch).not.toHaveBeenCalled();
1096+
});
1097+
1098+
it('emits _send_failed with willRetry:false for 404 (permanent failure)', async () => {
1099+
const mockFetch = vi.fn().mockResolvedValue({ ok: false, status: 404 });
1100+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1101+
const listener = vi.fn();
1102+
conn.onMessage(listener);
1103+
conn.connect();
1104+
lastES()._emit('welcome', {
1105+
type: 'welcome',
1106+
protocolVersion: 2,
1107+
connectionId: 'cid-test',
1108+
});
1109+
1110+
conn.send({ type: 'send', prompt: 'hello', clientMsgId: 'msg-404' });
1111+
await vi.runAllTimersAsync();
1112+
1113+
expect(listener).toHaveBeenCalledWith(
1114+
expect.objectContaining({
1115+
type: '_send_failed',
1116+
clientMsgId: 'msg-404',
1117+
willRetry: false,
1118+
}),
1119+
);
1120+
});
1121+
1122+
it('emits _send_failed with willRetry:false for 400 (permanent failure)', async () => {
1123+
const mockFetch = vi.fn().mockResolvedValue({ ok: false, status: 400 });
1124+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1125+
const listener = vi.fn();
1126+
conn.onMessage(listener);
1127+
conn.connect();
1128+
lastES()._emit('welcome', {
1129+
type: 'welcome',
1130+
protocolVersion: 2,
1131+
connectionId: 'cid-test',
1132+
});
1133+
1134+
conn.send({ type: 'send', prompt: 'bad request', clientMsgId: 'msg-400' });
1135+
await vi.runAllTimersAsync();
1136+
1137+
expect(listener).toHaveBeenCalledWith(
1138+
expect.objectContaining({
1139+
type: '_send_failed',
1140+
clientMsgId: 'msg-400',
1141+
willRetry: false,
1142+
}),
1143+
);
1144+
});
1145+
1146+
it('emits _send_failed on 429 rate limit', async () => {
1147+
const mockFetch = vi.fn().mockResolvedValue({ ok: false, status: 429 });
1148+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1149+
const listener = vi.fn();
1150+
conn.onMessage(listener);
1151+
conn.connect();
1152+
lastES()._emit('welcome', {
1153+
type: 'welcome',
1154+
protocolVersion: 2,
1155+
connectionId: 'cid-test',
1156+
});
1157+
1158+
conn.send({ type: 'send', prompt: 'hello', clientMsgId: 'msg-3' });
1159+
await vi.runAllTimersAsync();
1160+
1161+
expect(listener).toHaveBeenCalledWith(
1162+
expect.objectContaining({
1163+
type: '_send_failed',
1164+
willRetry: true,
1165+
}),
1166+
);
1167+
});
1168+
1169+
it('retries failed send after delay even without reconnect', async () => {
1170+
let callCount = 0;
1171+
const mockFetch = vi.fn().mockImplementation(() => {
1172+
callCount++;
1173+
if (callCount === 1) return Promise.resolve({ ok: false, status: 500 });
1174+
return Promise.resolve({ ok: true });
1175+
});
1176+
const conn = new SseConnection(createConfig({ fetch: mockFetch }));
1177+
conn.connect();
1178+
lastES()._emit('welcome', {
1179+
type: 'welcome',
1180+
protocolVersion: 2,
1181+
connectionId: 'cid-test',
1182+
});
1183+
1184+
// First send fails with 500 — gets queued for retry
1185+
conn.send({ type: 'send', prompt: 'retry me', clientMsgId: 'r-1' });
1186+
await vi.runAllTimersAsync();
1187+
1188+
// The retry flush timer fires after 3s and retries the queued message
1189+
// (no reconnect needed — SSE stream stayed connected)
1190+
expect(mockFetch).toHaveBeenCalledTimes(2);
1191+
});
1192+
});
8751193
});

‎packages/client/src/chat-connection.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ export interface ChatConnection {
1717
onMessage(listener: ConnectionListener): void;
1818
/** Whether the transport is connected and ready to send. */
1919
isConnected(): boolean;
20-
/** Server-assigned connection ID (null before welcome). */
20+
/** Client-generated stable connection ID. Available immediately after construction. */
2121
getConnectionId(): string | null;
2222
/** Track the latest received seq for a session (for reconnect replay). */
2323
trackSeq(sessionId: string, seq: number): void;

0 commit comments

Comments
 (0)