From 802ec4361ef3101be96fdb08a37b01cf8cff46f0 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 01:39:30 -0700 Subject: [PATCH 1/2] Add tests for chat stream JSON parse drop (CL-6837) A malformed live chat.message currently vanishes until the next poll, and the EventSource boundary uses JSON.parse instead of arktype. These tests pin an immediate toast and feed refresh instead of a silent wait. --- .../test/stream-message-parse-drop.test.tsx | 190 ++++++++++++++++++ .../test/use-workbench-stream.test.tsx | 88 +++++++- 2 files changed, 272 insertions(+), 6 deletions(-) create mode 100644 packages/chat-ui/test/stream-message-parse-drop.test.tsx diff --git a/packages/chat-ui/test/stream-message-parse-drop.test.tsx b/packages/chat-ui/test/stream-message-parse-drop.test.tsx new file mode 100644 index 000000000..9b3b83e09 --- /dev/null +++ b/packages/chat-ui/test/stream-message-parse-drop.test.tsx @@ -0,0 +1,190 @@ +// CL-6837: a chat.message the stream cannot parse must not sit silent +// until the next poll. Schema miss and JSON miss both toast and refresh +// the feed immediately. + +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { act, createElement } from "react"; +import { createRoot } from "react-dom/client"; +import type { Root } from "react-dom/client"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import * as reactUi from "@corbits/react-ui"; + +const realFetch = globalThis.fetch; +const realEventSource = globalThis.EventSource; + +class StubEventSource { + static instances: StubEventSource[] = []; + onopen: (() => void) | null = null; + onerror: (() => void) | null = null; + readyState = 1; + listeners = new Map void>(); + + constructor(readonly url: string) { + StubEventSource.instances.push(this); + } + + addEventListener( + eventType: string, + listener: (message: MessageEvent) => void, + ) { + this.listeners.set(eventType, listener); + } + + emit(eventType: string, data: unknown) { + this.listeners.get(eventType)?.({ + data: JSON.stringify(data), + } as MessageEvent); + } + + emitRaw(eventType: string, data: string) { + this.listeners.get(eventType)?.({ data } as MessageEvent); + } + + close() { + this.readyState = 2; + } +} + +afterEach(() => { + globalThis.fetch = realFetch; + globalThis.EventSource = realEventSource; + StubEventSource.instances = []; +}); + +const WORKBENCH_WIRE = { + id: "ch_1", + title: "Launch Planning", + kind: "workbench", + pinned: false, + participants: [] as { address: string; handle: string }[], +}; + +let feedRefetchCount = 0; + +function stubFetch() { + globalThis.EventSource = StubEventSource as unknown as typeof EventSource; + feedRefetchCount = 0; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const path = typeof input === "string" ? input : String(input); + const method = init?.method ?? "GET"; + if ( + method === "GET" && + /\/chat\/workbenches\/[^/]+\/(messages|threads|pins)(\?|$)/.test(path) + ) { + feedRefetchCount += 1; + } + const json = (body: unknown) => + new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + if (/\/chat\/workbenches\?kind=workbench$/.test(path)) { + return json({ items: [WORKBENCH_WIRE] }); + } + if (/\/chat\/workbenches\?kind=chat$/.test(path)) { + return json({ items: [] }); + } + if (/\/chat\/workbenches\/[^/]+\/threads$/.test(path)) { + return json({ rootThreadId: "", items: [] }); + } + if (/\/chat\/workbenches\/[^/]+\/messages/.test(path)) { + return json({ items: [] }); + } + if (/\/chat\/workbenches\/[^/]+\/pins$/.test(path)) { + return json({ items: [] }); + } + if (/\/chat\/workbenches\/[^/]+\/read-state$/.test(path)) return json({}); + if (/\/chat\/workbenches\/[^/]+\/invitable$/.test(path)) { + return json({ items: [] }); + } + if ( + /\/chat\/workbenches\/[^/]+\/presence$/.test(path) && + init?.method === "POST" + ) { + return json({}); + } + throw new Error(`unstubbed fetch: ${path}`); + }) as typeof fetch; +} + +const { ChatWorkspace } = await import("../src/chat-workspace"); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +function mount(props: Parameters[0]) { + const container = document.createElement("div"); + document.body.appendChild(container); + const root: Root = createRoot(container); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: 0, staleTime: 0 } }, + }); + act(() => { + root.render( + createElement( + QueryClientProvider, + { client: queryClient }, + createElement(ChatWorkspace, props), + ), + ); + }); + return { + container, + settle: (ms = 30) => act(() => sleep(ms)), + unmount: () => root.unmount(), + }; +} + +function firstStream(): StubEventSource { + const instance = StubEventSource.instances[0]; + if (instance === undefined) throw new Error("no stream connected"); + return instance; +} + +describe("dropped chat.message is not silent until poll (CL-6837)", () => { + test("a schema miss toasts and refreshes the feed", async () => { + stubFetch(); + const toast = spyOn(reactUi, "toast"); + const harness = mount({ + tenant: { kind: "ready", tenantId: "tnt_1" }, + workbenchId: "ch_1", + }); + await harness.settle(); + const feedRefetchCountAfterHydration = feedRefetchCount; + + act(() => { + firstStream().emit("chat.message", { not: "a message" }); + }); + await harness.settle(300); + + expect(toast).toHaveBeenCalledWith( + "Couldn't apply that message live — refreshing.", + ); + expect(feedRefetchCount).toBeGreaterThan(feedRefetchCountAfterHydration); + toast.mockRestore(); + harness.unmount(); + }); + + test("invalid JSON toasts and refreshes the feed instead of forwarding garbage", async () => { + stubFetch(); + const toast = spyOn(reactUi, "toast"); + const harness = mount({ + tenant: { kind: "ready", tenantId: "tnt_1" }, + workbenchId: "ch_1", + }); + await harness.settle(); + const feedRefetchCountAfterHydration = feedRefetchCount; + + act(() => { + firstStream().emitRaw("chat.message", "not-json{"); + }); + await harness.settle(300); + + expect(toast).toHaveBeenCalledWith( + "Couldn't apply that message live — refreshing.", + ); + expect(feedRefetchCount).toBeGreaterThan(feedRefetchCountAfterHydration); + expect(harness.container.textContent).not.toContain("not-json"); + toast.mockRestore(); + harness.unmount(); + }); +}); diff --git a/packages/chat-ui/test/use-workbench-stream.test.tsx b/packages/chat-ui/test/use-workbench-stream.test.tsx index 65649ba1f..e492caaeb 100644 --- a/packages/chat-ui/test/use-workbench-stream.test.tsx +++ b/packages/chat-ui/test/use-workbench-stream.test.tsx @@ -11,7 +11,10 @@ import { afterEach, describe, expect, test } from "bun:test"; import { act, createElement } from "react"; import { createRoot } from "react-dom/client"; -import { useWorkbenchStream } from "../src/use-workbench-stream"; +import { + parseStreamEventJson, + useWorkbenchStream, +} from "../src/use-workbench-stream"; const realEventSource = globalThis.EventSource; const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); @@ -44,6 +47,10 @@ class StubEventSource { } as MessageEvent); } + emitRaw(eventType: string, data: string) { + this.listeners.get(eventType)?.({ data } as MessageEvent); + } + close() { this.readyState = StubEventSource.CLOSED; } @@ -64,11 +71,14 @@ afterEach(() => { StubEventSource.instances = []; }); -function mount(intervals: { - pollMs: number; - baseDelayMs: number; - maxDelayMs: number; -}) { +function mount( + intervals: { + pollMs: number; + baseDelayMs: number; + maxDelayMs: number; + }, + onParseError?: (eventType: string) => void, +) { globalThis.EventSource = StubEventSource as unknown as typeof EventSource; const container = document.createElement("div"); document.body.appendChild(container); @@ -84,6 +94,7 @@ function mount(intervals: { pollCount += 1; }, intervals, + onParseError, ); return null; } @@ -195,4 +206,69 @@ describe("useWorkbenchStream (reconnect + poll wiring)", () => { ]); harness.unmount(); }); + + test("invalid JSON on chat.message is not forwarded and polls immediately (CL-6837)", async () => { + const parseErrors: string[] = []; + const harness = mount( + { + pollMs: 1000, + baseDelayMs: 5000, + maxDelayMs: 5000, + }, + (eventType) => parseErrors.push(eventType), + ); + harness.latest().open(); + const pollsAfterOpen = harness.pollCount(); + + harness.latest().emitRaw("chat.message", "not-json{"); + + expect(harness.events()).toEqual([]); + expect(harness.pollCount()).toBe(pollsAfterOpen + 1); + expect(parseErrors).toEqual(["chat.message"]); + harness.unmount(); + }); + + test("valid JSON on chat.message still forwards and does not poll (CL-6837)", async () => { + const harness = mount({ + pollMs: 1000, + baseDelayMs: 5000, + maxDelayMs: 5000, + }); + harness.latest().open(); + const pollsAfterOpen = harness.pollCount(); + + harness.latest().emit("chat.message", { id: "m1", parts: [] }); + + expect(harness.events()).toEqual([ + ["chat.message", { id: "m1", parts: [] }], + ]); + expect(harness.pollCount()).toBe(pollsAfterOpen); + harness.unmount(); + }); +}); + +describe("parseStreamEventJson (CL-6837)", () => { + test("parses a JSON object through arktype", () => { + expect(parseStreamEventJson('{"id":"m1"}')).toEqual({ + ok: true, + data: { id: "m1" }, + }); + }); + + test("undefined payload is a successful empty event", () => { + expect(parseStreamEventJson(undefined)).toEqual({ + ok: true, + data: undefined, + }); + }); + + test("invalid JSON is a parse miss, not a thrown error", () => { + expect(parseStreamEventJson("not-json{")).toEqual({ ok: false }); + }); + + test("a non-string payload is a parse miss", () => { + expect(parseStreamEventJson({ id: "already-parsed" })).toEqual({ + ok: false, + }); + }); }); From 753b54d2d49c33b4682e816af2802b1f834fd0e4 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 01:39:45 -0700 Subject: [PATCH 2/2] Parse stream JSON with arktype and surface dropped messages (CL-6837) The EventSource handler caught JSON.parse failures and forwarded the raw string, so ChatMessageEventData missed and the room stayed silent until a later poll. Parse at that boundary with arktype, toast, and refresh the feed immediately. chat.settings handling from CL-6476 is unchanged. --- packages/chat-ui/src/chat-workspace.tsx | 27 ++++++++------ packages/chat-ui/src/strings.ts | 1 + packages/chat-ui/src/use-workbench-stream.ts | 38 +++++++++++++++----- scripts/checks/report-error-baseline.txt | 1 - 4 files changed, 48 insertions(+), 19 deletions(-) diff --git a/packages/chat-ui/src/chat-workspace.tsx b/packages/chat-ui/src/chat-workspace.tsx index 255cb6e4e..4f07749c9 100644 --- a/packages/chat-ui/src/chat-workspace.tsx +++ b/packages/chat-ui/src/chat-workspace.tsx @@ -960,17 +960,18 @@ function ChatWorkspaceInner({ switch (eventType) { case "chat.message": { const parsed = ChatMessageEventData(data); - if (!(parsed instanceof type.errors)) { - applyStreamMessage(queryClient, tenantId, activeWorkbenchId, { - id: parsed.id, - createdAt: parsed.createdAt, - parts: parsed.parts, - sender: parsed.sender, - ...(parsed.threadId !== null - ? { threadId: parsed.threadId } - : {}), - }); + if (parsed instanceof type.errors) { + toast(CHAT_STRINGS.streamMessageDropped); + refreshFeed(); + break; } + applyStreamMessage(queryClient, tenantId, activeWorkbenchId, { + id: parsed.id, + createdAt: parsed.createdAt, + parts: parsed.parts, + sender: parsed.sender, + ...(parsed.threadId !== null ? { threadId: parsed.threadId } : {}), + }); break; } case "chat.reaction": { @@ -1020,6 +1021,12 @@ function ChatWorkspaceInner({ } }, refreshFeed, + {}, + (eventType) => { + if (eventType !== "chat.message") return; + toast(CHAT_STRINGS.streamMessageDropped); + refreshFeed(); + }, ); /** The one door into the workbench settings surface — the gear button and diff --git a/packages/chat-ui/src/strings.ts b/packages/chat-ui/src/strings.ts index 0722209a0..bec00ae23 100644 --- a/packages/chat-ui/src/strings.ts +++ b/packages/chat-ui/src/strings.ts @@ -156,6 +156,7 @@ export const CHAT_STRINGS = { reactionPickerLabel: "Choose a reaction", reactionPickerOptionLabel: (emoji: string) => `React with ${emoji}`, reactionToggleError: "Couldn't update that reaction — try again.", + streamMessageDropped: "Couldn't apply that message live — refreshing.", pinMessageAction: "Pin message", unpinMessageAction: "Unpin message", pinMessageError: "Couldn't pin that message — try again.", diff --git a/packages/chat-ui/src/use-workbench-stream.ts b/packages/chat-ui/src/use-workbench-stream.ts index 26bfd1013..9b6940995 100644 --- a/packages/chat-ui/src/use-workbench-stream.ts +++ b/packages/chat-ui/src/use-workbench-stream.ts @@ -6,12 +6,30 @@ // stream emits (chat.settings, chat.typing, and the platform's own run // events); the caller decides what to do with each — this hook only owns // the transport, and connection state is never something the caller should -// render as chrome. +// render as chrome. JSON at this boundary is parsed with arktype +// (`string.json.parse`); a miss is never forwarded as a raw string. +import { type } from "arktype"; import { useEffect, useRef, useState } from "react"; export type WorkbenchStreamState = "connecting" | "live" | "polling"; +const StreamEventJson = type("string.json.parse"); + +/** + * The EventSource payload is a JSON string (or absent). Parse it at this + * boundary so a SyntaxError never becomes a silently dropped `chat.message`. + */ +export function parseStreamEventJson( + raw: unknown, +): { readonly ok: true; readonly data: unknown } | { readonly ok: false } { + if (raw === undefined) return { ok: true, data: undefined }; + if (typeof raw !== "string") return { ok: false }; + const parsed = StreamEventJson(raw); + if (parsed instanceof type.errors) return { ok: false }; + return { ok: true, data: parsed }; +} + /** * The S3 fix, isolated as a pure rule: with no active workbench there is * nothing to stream, so the hook must not open an `EventSource` or start a @@ -53,12 +71,15 @@ export function useWorkbenchStream( onEvent: (eventType: string, data: unknown) => void, onPoll: () => void, intervals?: WorkbenchStreamIntervals, + onParseError?: (eventType: string) => void, ): WorkbenchStreamState { const [state, setState] = useState("connecting"); const onEventRef = useRef(onEvent); onEventRef.current = onEvent; const onPollRef = useRef(onPoll); onPollRef.current = onPoll; + const onParseErrorRef = useRef(onParseError); + onParseErrorRef.current = onParseError; const pollMs = intervals?.pollMs ?? POLL_INTERVAL_MS; const baseDelayMs = intervals?.baseDelayMs ?? BASE_DELAY_MS; const maxDelayMs = intervals?.maxDelayMs ?? MAX_DELAY_MS; @@ -120,14 +141,15 @@ export function useWorkbenchStream( const forward = (eventType: string) => (message: MessageEvent) => { if (cancelled) return; - try { - onEventRef.current( - eventType, - message.data === undefined ? undefined : JSON.parse(message.data), - ); - } catch { - onEventRef.current(eventType, message.data); + const parsed = parseStreamEventJson(message.data); + if (!parsed.ok) { + onParseErrorRef.current?.(eventType); + if (eventType === "chat.message") { + onPollRef.current(); + } + return; } + onEventRef.current(eventType, parsed.data); }; // "chat.message" is every message posted into the room; diff --git a/scripts/checks/report-error-baseline.txt b/scripts/checks/report-error-baseline.txt index 326cf34da..a840cc303 100644 --- a/scripts/checks/report-error-baseline.txt +++ b/scripts/checks/report-error-baseline.txt @@ -132,7 +132,6 @@ packages/chat-ui/src/timeline.tsx 1 toast(CHAT_STRINGS.copyTextError); packages/chat-ui/src/tool-activity.ts 1 return output; packages/chat-ui/src/use-optimistic-sends.ts 1 if (cause instanceof ChatApiError && cause.status === 403) { packages/chat-ui/src/use-thread-navigation.ts 1 toast(CHAT_STRINGS.forkThreadError); -packages/chat-ui/src/use-workbench-stream.ts 1 onEventRef.current(eventType, message.data); packages/chat/src/platform-adapter.ts 1 continue; packages/chat/src/platform-adapter.ts 1 wakeLogger.error`drift check for ${binding.roomAddress} (run ${live.run.id}) failed, leaving it as-is: ${ packages/chat/src/platform-adapter.ts 1 wakeLogger.error`inference-source reconcile for ${binding.roomAddress} (run ${run.id}) failed, leaving it as-is: ${