diff --git a/web/src/api/chatStream.test.ts b/web/src/api/chatStream.test.ts index 1320dce3..5d1e2fd5 100644 --- a/web/src/api/chatStream.test.ts +++ b/web/src/api/chatStream.test.ts @@ -33,3 +33,13 @@ test('parseSSEChunk: ignores heartbeat comments and keeps last id', () => { assert.equal(events.length, 2); assert.equal(lastSeq, 2); }); + +test('parseSSEChunk: skips a single malformed frame and keeps parsing the next valid frame', () => { + const raw = + 'id: 1\nevent: delta\ndata: {not valid json\n\n' + + 'id: 2\nevent: delta\ndata: {"type":"delta","delta":"ok"}\n\n'; + const { events, lastSeq } = parseSSEChunk(raw); + assert.equal(events.length, 1); + assert.equal(events[0].delta, 'ok'); + assert.equal(lastSeq, 2); +}); diff --git a/web/src/api/taskRuns.test.ts b/web/src/api/taskRuns.test.ts index a3ef0ad4..4b8e70cd 100644 --- a/web/src/api/taskRuns.test.ts +++ b/web/src/api/taskRuns.test.ts @@ -1,6 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { createTaskRun, readTaskRunSSEOnce, subscribeTaskRunEvents } from './taskRuns'; +import { createTaskRun, readTaskRunSSEOnce, subscribeTaskRunEvents, TaskRunHttpError } from './taskRuns'; import { parseSSEChunk } from './chatStream'; test('createTaskRun exposes DOC_BUSY detail without a retryable run', async () => { @@ -100,3 +100,59 @@ test('readTaskRunSSEOnce marks non-terminal EOF for reconnect', async () => { globalThis.fetch = originalFetch; } }); + +test('subscribeTaskRunEvents fails fast on 401 without retrying', async () => { + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = async () => { + calls += 1; + return new Response('', { status: 401 }); + }; + try { + await assert.rejects( + () => + subscribeTaskRunEvents('run_401', { + onEvent: () => undefined, + maxReconnects: 5, + reconnectDelayMs: 1, + sleep: async () => undefined, + }), + (err: unknown) => { + assert.ok(err instanceof TaskRunHttpError); + assert.equal(err.status, 401); + return true; + }, + ); + assert.equal(calls, 1, 'must not retry on 401'); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test('subscribeTaskRunEvents fails fast on 404 without retrying', async () => { + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = async () => { + calls += 1; + return new Response('', { status: 404 }); + }; + try { + await assert.rejects( + () => + subscribeTaskRunEvents('run_404', { + onEvent: () => undefined, + maxReconnects: 5, + reconnectDelayMs: 1, + sleep: async () => undefined, + }), + (err: unknown) => { + assert.ok(err instanceof TaskRunHttpError); + assert.equal(err.status, 404); + return true; + }, + ); + assert.equal(calls, 1, 'must not retry on 404'); + } finally { + globalThis.fetch = originalFetch; + } +}); diff --git a/web/src/hooks/useTurnRuntime.test.ts b/web/src/hooks/useTurnRuntime.test.ts index efb11b11..26b49643 100644 --- a/web/src/hooks/useTurnRuntime.test.ts +++ b/web/src/hooks/useTurnRuntime.test.ts @@ -44,3 +44,42 @@ test('turn runtime: session A stream survives session B switch', () => { assert.equal(isSessionThinking('session-A'), true); assert.equal(isSessionThinking('session-B'), true); }); + +test('abortAllSessions aborts controllers across all sessions', () => { + type TurnRuntime = { + clientTurnId: string; + sessionId: string; + status: string; + controller: { aborted: boolean; abort: () => void }; + }; + const store = new Map>(); + + const makeController = () => { + const c = { aborted: false, abort() { c.aborted = true; } }; + return c; + }; + + const register = (rt: TurnRuntime) => { + if (!store.has(rt.sessionId)) store.set(rt.sessionId, new Map()); + store.get(rt.sessionId)!.set(rt.clientTurnId, rt); + }; + + const abortAllSessions = () => { + for (const sessionMap of store.values()) { + for (const rt of sessionMap.values()) { + if (rt.status === 'pending' || rt.status === 'streaming') { + rt.controller.abort(); + rt.status = 'cancelled'; + } + } + } + }; + + const a = makeController(); + const b = makeController(); + register({ clientTurnId: 't1', sessionId: 's1', status: 'streaming', controller: a }); + register({ clientTurnId: 't2', sessionId: 's2', status: 'pending', controller: b }); + abortAllSessions(); + assert.equal(a.aborted, true); + assert.equal(b.aborted, true); +});