From 39d51388d4cc247fb9e41d5000087f525178d2a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pascal=20Andr=C3=A9?= Date: Sat, 15 Aug 2026 18:27:51 +0200 Subject: [PATCH] fix(v2): render native response deltas progressively Apply OpenCode V2 text and reasoning delta events directly to the normalized message store so assistant responses appear as they are generated instead of waiting for periodic HTTP snapshots. Deduplicate replayed events, preserve ordinal part ordering, reconcile stale or superseded snapshots, settle terminal messages, retry failed terminal refreshes, and clear streaming state when sessions or workspaces are removed. This keeps snapshot refreshes as authoritative recovery without erasing visible streamed content or recreating removed stores. Add focused streaming regression coverage and extend native event tests for immediate text, reasoning, periodic refresh behavior, replay handling, deletion cleanup, and terminal reconciliation. Validated with 223 standard UI tests, 64 browser/integration tests, full workspace typechecks, a production Tauri build, and a live CDP smoke covering progressive rendering and reload restoration. --- .github/workflows/pr-build.yml | 1 + packages/ui/src/stores/instances.ts | 2 + .../stores/native-session-streaming.test.ts | 148 ++++++++++++++ .../ui/src/stores/native-session-streaming.ts | 185 ++++++++++++++++++ packages/ui/src/stores/session-api.ts | 21 +- packages/ui/src/stores/session-events.ts | 48 ++++- .../src/stores/session-native-events.test.ts | 20 +- 7 files changed, 415 insertions(+), 10 deletions(-) create mode 100644 packages/ui/src/stores/native-session-streaming.test.ts create mode 100644 packages/ui/src/stores/native-session-streaming.ts diff --git a/.github/workflows/pr-build.yml b/.github/workflows/pr-build.yml index a4c67859..544bd09b 100644 --- a/.github/workflows/pr-build.yml +++ b/.github/workflows/pr-build.yml @@ -148,6 +148,7 @@ jobs: packages/ui/src/lib/hooks/use-active-session-message-load.test.ts packages/ui/src/stores/forms.test.ts packages/ui/src/stores/instances-restore-ownership.test.ts + packages/ui/src/stores/native-session-streaming.test.ts packages/ui/src/stores/permission-lifecycle.test.ts packages/ui/src/stores/pty-store-reactivity.test.ts packages/ui/src/stores/session-actions.test.ts diff --git a/packages/ui/src/stores/instances.ts b/packages/ui/src/stores/instances.ts index 5810a52f..df10b515 100644 --- a/packages/ui/src/stores/instances.ts +++ b/packages/ui/src/stores/instances.ts @@ -41,6 +41,7 @@ import { } from "./session-state" import { setHasInstances } from "./ui" import { messageStoreBus } from "./message-v2/bus" +import { clearNativeContentDeltaState } from "./native-session-streaming" import { upsertPermissionV2, removePermissionV2, upsertQuestionV2, removeQuestionV2 } from "./message-v2/bridge" import { clearRepliedPermissions, @@ -1118,6 +1119,7 @@ function removeInstance(id: string, options: { authoritative?: boolean } = {}) { // Clean up session indexes and drafts for removed instance clearCacheForInstance(id) + clearNativeContentDeltaState(id) messageStoreBus.unregisterInstance(id) clearInstanceDraftPrompts(id) clearSessionListRequestState(id) diff --git a/packages/ui/src/stores/native-session-streaming.test.ts b/packages/ui/src/stores/native-session-streaming.test.ts new file mode 100644 index 00000000..d0359b6e --- /dev/null +++ b/packages/ui/src/stores/native-session-streaming.test.ts @@ -0,0 +1,148 @@ +import assert from "node:assert/strict" +import { describe, it } from "node:test" +import { messageStoreBus } from "./message-v2/bus.ts" +import { + applyNativeContentDelta, + clearNativeContentDeltaState, + reconcileNativeContentAfterSnapshot, + reapplyNativeContentDeltas, + settleNativeContentDeltas, +} from "./native-session-streaming.ts" + +describe("native session streaming", () => { + it("creates assistant content and applies text and reasoning deltas immediately", () => { + const instanceId = "native-streaming" + const base = { id: "event", created: 10, data: { sessionID: "session", assistantMessageID: "assistant" } } + + try { + assert.equal(applyNativeContentDelta(instanceId, { + ...base, + type: "session.text.delta", + data: { ...base.data, ordinal: 0, delta: "hello" }, + }), true) + applyNativeContentDelta(instanceId, { + ...base, + id: "event-2", + type: "session.text.delta", + data: { ...base.data, ordinal: 0, delta: " world" }, + }) + applyNativeContentDelta(instanceId, { + ...base, + id: "event-3", + type: "session.reasoning.delta", + data: { ...base.data, ordinal: 1, delta: "thinking" }, + }) + + const message = messageStoreBus.getOrCreate(instanceId).getMessage("assistant") + assert.equal(message?.status, "streaming") + assert.equal((message?.parts["assistant-text-0"]?.data as any)?.text, "hello world") + assert.equal((message?.parts["assistant-reasoning-1"]?.data as any)?.text, "thinking") + assert.equal(messageStoreBus.getOrCreate(instanceId).getMessageInfo("assistant")?.sessionID, "session") + } finally { + clearNativeContentDeltaState(instanceId) + messageStoreBus.unregisterInstance(instanceId) + } + }) + + it("rejects malformed deltas without creating state", () => { + const instanceId = "invalid-native-streaming" + try { + assert.equal(applyNativeContentDelta(instanceId, { + type: "session.text.delta", + data: { sessionID: "session", assistantMessageID: "", ordinal: 0, delta: "ignored" }, + } as any), false) + assert.equal(messageStoreBus.getOrCreate(instanceId).getSessionMessageIds("session").length, 0) + } finally { + clearNativeContentDeltaState(instanceId) + messageStoreBus.unregisterInstance(instanceId) + } + }) + + it("deduplicates replayed events and restores direct content after a stale snapshot", () => { + const instanceId = "replayed-native-streaming" + const event = { + id: "event-1", + created: 10, + type: "session.text.delta" as const, + data: { sessionID: "session", assistantMessageID: "assistant", ordinal: 0, delta: "hello" }, + } + try { + applyNativeContentDelta(instanceId, event) + applyNativeContentDelta(instanceId, event) + const store = messageStoreBus.getOrCreate(instanceId) + assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello") + + store.upsertMessage({ + id: "assistant", sessionId: "session", role: "assistant", status: "streaming", + parts: [{ id: "assistant-text-0", type: "text", text: "", sessionID: "session", messageID: "assistant" }], + }) + reapplyNativeContentDeltas(instanceId, "session") + assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello") + + settleNativeContentDeltas(instanceId, "session") + applyNativeContentDelta(instanceId, { ...event, id: "late-event", data: { ...event.data, delta: " late" } }) + assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello") + reconcileNativeContentAfterSnapshot(instanceId, "session") + assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello") + assert.equal(store.getMessage("assistant")?.status, "complete") + store.upsertMessage({ id: "assistant", sessionId: "session", role: "assistant", status: "streaming" }) + reconcileNativeContentAfterSnapshot(instanceId, "session") + assert.equal(store.getMessage("assistant")?.status, "complete") + } finally { + clearNativeContentDeltaState(instanceId) + messageStoreBus.unregisterInstance(instanceId) + } + }) + + it("renders parts in ordinal order even when events arrive out of order", () => { + const instanceId = "ordered-native-streaming" + const data = { sessionID: "session", assistantMessageID: "assistant" } + try { + applyNativeContentDelta(instanceId, { + id: "second", created: 2, type: "session.reasoning.delta", + data: { ...data, ordinal: 1, delta: "second" }, + }) + applyNativeContentDelta(instanceId, { + id: "first", created: 1, type: "session.text.delta", + data: { ...data, ordinal: 0, delta: "first" }, + }) + assert.deepEqual(messageStoreBus.getOrCreate(instanceId).getMessage("assistant")?.partIds, [ + "assistant-text-0", + "assistant-reasoning-1", + ]) + } finally { + clearNativeContentDeltaState(instanceId) + messageStoreBus.unregisterInstance(instanceId) + } + }) + + it("rejects negative ordinals", () => { + const instanceId = "negative-native-streaming" + try { + assert.equal(applyNativeContentDelta(instanceId, { + id: "event", created: 1, type: "session.text.delta", + data: { sessionID: "session", assistantMessageID: "assistant", ordinal: -1, delta: "ignored" }, + }), false) + } finally { + clearNativeContentDeltaState(instanceId) + messageStoreBus.unregisterInstance(instanceId) + } + }) + + it("does not recreate cleared session state during late reconciliation", () => { + const instanceId = "cleared-native-streaming" + try { + applyNativeContentDelta(instanceId, { + id: "event", created: 1, type: "session.text.delta", + data: { sessionID: "session", assistantMessageID: "assistant", ordinal: 0, delta: "text" }, + }) + messageStoreBus.unregisterInstance(instanceId) + clearNativeContentDeltaState(instanceId, "session") + reapplyNativeContentDeltas(instanceId, "session") + assert.equal(messageStoreBus.getInstance(instanceId), undefined) + } finally { + clearNativeContentDeltaState(instanceId) + if (messageStoreBus.getInstance(instanceId)) messageStoreBus.unregisterInstance(instanceId) + } + }) +}) diff --git a/packages/ui/src/stores/native-session-streaming.ts b/packages/ui/src/stores/native-session-streaming.ts new file mode 100644 index 00000000..85f51866 --- /dev/null +++ b/packages/ui/src/stores/native-session-streaming.ts @@ -0,0 +1,185 @@ +import type { SessionReasoningDelta, SessionTextDelta } from "@opencode-ai/client" +import type { ClientPart, MessageInfo } from "../types/message" +import { messageStoreBus } from "./message-v2/bus" + +type NativeContentDelta = SessionTextDelta | SessionReasoningDelta +type TrackedPart = { + messageId: string + partId: string + type: "text" | "reasoning" + ordinal: number + text: string + createdAt: number +} +type SessionStreamingState = { + seenEventIds: Set + eventOrder: string[] + parts: Map + settledMessageIds: Set + settledMessageOrder: string[] +} + +const MAX_TRACKED_EVENT_IDS = 20_000 +const MAX_SETTLED_MESSAGE_IDS = 100 +const streamingState = new Map>() + +function getSessionState(instanceId: string, sessionId: string): SessionStreamingState { + let instance = streamingState.get(instanceId) + if (!instance) { + instance = new Map() + streamingState.set(instanceId, instance) + } + let session = instance.get(sessionId) + if (!session) { + session = { + seenEventIds: new Set(), eventOrder: [], parts: new Map(), + settledMessageIds: new Set(), settledMessageOrder: [], + } + instance.set(sessionId, session) + } + return session +} + +function markEventSeen(state: SessionStreamingState, eventId: string): boolean { + if (state.seenEventIds.has(eventId)) return false + state.seenEventIds.add(eventId) + state.eventOrder.push(eventId) + while (state.eventOrder.length > MAX_TRACKED_EVENT_IDS) { + const expired = state.eventOrder.shift() + if (expired) state.seenEventIds.delete(expired) + } + return true +} + +function renderTrackedPart(instanceId: string, sessionId: string, tracked: TrackedPart): void { + const store = messageStoreBus.getOrCreate(instanceId) + const current = store.getMessage(tracked.messageId) + const existingPart = current?.parts[tracked.partId]?.data as ClientPart | undefined + const existingText = existingPart && "text" in existingPart ? existingPart.text : undefined + if (typeof existingText === "string") { + const terminalSnapshot = current?.status === "complete" || current?.status === "error" + if (terminalSnapshot) { + if (!tracked.text.startsWith(existingText) || existingText.length >= tracked.text.length) tracked.text = existingText + } else if (existingText.length > tracked.text.length && existingText.startsWith(tracked.text)) { + tracked.text = existingText + } + } + + const part: ClientPart = { + ...(existingPart ?? {}), + id: tracked.partId, + type: tracked.type, + text: tracked.text, + sessionID: sessionId, + messageID: tracked.messageId, + } as ClientPart + const parts = (current?.partIds ?? []) + .filter((partId) => partId !== tracked.partId) + .map((partId) => current?.parts[partId]?.data) + .filter((candidate): candidate is ClientPart => Boolean(candidate)) + parts.splice(Math.min(tracked.ordinal, parts.length), 0, part) + + store.addOrUpdateSession({ id: sessionId }) + store.upsertMessage({ + id: tracked.messageId, + sessionId, + role: "assistant", + status: "streaming", + createdAt: current?.createdAt ?? tracked.createdAt, + updatedAt: Date.now(), + parts, + }) + if (!store.getMessageInfo(tracked.messageId)) { + const info: MessageInfo = { + id: tracked.messageId, + sessionID: sessionId, + role: "assistant", + time: { created: tracked.createdAt }, + } + store.setMessageInfo(tracked.messageId, info) + } +} + +export function applyNativeContentDelta(instanceId: string, event: NativeContentDelta): boolean { + const { sessionID, assistantMessageID, ordinal, delta } = event.data + if (!instanceId || !event.id || !sessionID || !assistantMessageID || !Number.isInteger(ordinal) || ordinal < 0 || typeof delta !== "string") { + return false + } + + const state = getSessionState(instanceId, sessionID) + if (!markEventSeen(state, event.id) || state.settledMessageIds.has(assistantMessageID)) return true + const createdAt = typeof event.created === "number" ? event.created : Date.now() + const partType = event.type === "session.text.delta" ? "text" : "reasoning" + const partId = `${assistantMessageID}-${partType}-${ordinal}` + const tracked = state.parts.get(partId) ?? { + messageId: assistantMessageID, + partId, + type: partType, + ordinal, + text: "", + createdAt, + } + tracked.text += delta + state.parts.set(partId, tracked) + renderTrackedPart(instanceId, sessionID, tracked) + return true +} + +export function reapplyNativeContentDeltas(instanceId: string, sessionId: string): void { + const state = streamingState.get(instanceId)?.get(sessionId) + if (!state) return + for (const tracked of [...state.parts.values()].sort((a, b) => a.ordinal - b.ordinal)) { + renderTrackedPart(instanceId, sessionId, tracked) + } +} + +function markMessageComplete(instanceId: string, sessionId: string, messageId: string): void { + const store = messageStoreBus.getOrCreate(instanceId) + const message = store.getMessage(messageId) + if (!message || message.status === "complete" || message.status === "error") return + store.upsertMessage({ + id: message.id, + sessionId, + role: "assistant", + status: "complete", + createdAt: message.createdAt, + updatedAt: Date.now(), + }) +} + +export function settleNativeContentDeltas(instanceId: string, sessionId: string): void { + const state = streamingState.get(instanceId)?.get(sessionId) + if (!state) return + for (const tracked of state.parts.values()) { + if (!state.settledMessageIds.has(tracked.messageId)) { + state.settledMessageIds.add(tracked.messageId) + state.settledMessageOrder.push(tracked.messageId) + } + markMessageComplete(instanceId, sessionId, tracked.messageId) + } + while (state.settledMessageOrder.length > MAX_SETTLED_MESSAGE_IDS) { + const expired = state.settledMessageOrder.shift() + if (!expired) continue + state.settledMessageIds.delete(expired) + for (const [partId, tracked] of state.parts) { + if (tracked.messageId === expired) state.parts.delete(partId) + } + } +} + +export function reconcileNativeContentAfterSnapshot(instanceId: string, sessionId: string): void { + const state = streamingState.get(instanceId)?.get(sessionId) + if (!state) return + reapplyNativeContentDeltas(instanceId, sessionId) + for (const messageId of state.settledMessageIds) markMessageComplete(instanceId, sessionId, messageId) +} + +export function clearNativeContentDeltaState(instanceId: string, sessionId?: string): void { + if (!sessionId) { + streamingState.delete(instanceId) + return + } + const instance = streamingState.get(instanceId) + instance?.delete(sessionId) + if (instance?.size === 0) streamingState.delete(instanceId) +} diff --git a/packages/ui/src/stores/session-api.ts b/packages/ui/src/stores/session-api.ts index 14c018d1..c6f693ac 100644 --- a/packages/ui/src/stores/session-api.ts +++ b/packages/ui/src/stores/session-api.ts @@ -57,6 +57,7 @@ import { normalizeSessionMessage } from "./message-v2/normalizers" import { updateSessionInfo } from "./message-v2/session-info" import { seedSessionMessagesV2, reconcilePendingPermissionsV2, reconcilePendingQuestionsV2 } from "./message-v2/bridge" import { messageStoreBus } from "./message-v2/bus" +import { clearNativeContentDeltaState, reconcileNativeContentAfterSnapshot } from "./native-session-streaming" import { clearCacheForSession } from "../lib/global-cache" import { getLogger } from "../lib/logger" import { getOpencodeErrorMessage } from "../lib/opencode-api" @@ -733,6 +734,7 @@ async function deleteSession(instanceId: string, sessionId: string): Promise { - if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return prev + if (!instances().has(instanceId) || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return prev const next = new Map(prev) const nextInstanceSessions = next.get(instanceId) if (!nextInstanceSessions) return next @@ -1059,7 +1065,7 @@ async function loadMessages( const sessionForV2 = sessions().get(instanceId)?.get(sessionId) ?? { id: sessionId, title: session?.title, parentId: session?.parentId ?? null, revert: session?.revert, } - if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return + if (!instances().has(instanceId) || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return if (!seedSessionMessagesV2(instanceId, sessionForV2, messages, messagesInfo, messageRevision)) { retryAfterRevisionConflict = true } else { @@ -1092,6 +1098,9 @@ async function loadMessages( return next }) } + if (instances().has(instanceId) && sessions().get(instanceId)?.has(sessionId)) { + reconcileNativeContentAfterSnapshot(instanceId, sessionId) + } } if (retryAfterRevisionConflict && sessions().get(instanceId)?.has(sessionId)) { @@ -1104,7 +1113,9 @@ async function loadMessages( }) } - if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch) || !sessions().get(instanceId)?.has(sessionId)) return + if (!instances().has(instanceId) + || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch) + || !sessions().get(instanceId)?.has(sessionId)) return updateSessionInfo(instanceId, sessionId) if (!skipChildren && session.parentId === null) { diff --git a/packages/ui/src/stores/session-events.ts b/packages/ui/src/stores/session-events.ts index ad4fe314..4e3e129d 100644 --- a/packages/ui/src/stores/session-events.ts +++ b/packages/ui/src/stores/session-events.ts @@ -68,6 +68,11 @@ import { } from "./message-v2/bridge" import { messageStoreBus } from "./message-v2/bus" import { handleConversationAssistantPartUpdated } from "./conversation-speech" +import { + applyNativeContentDelta, + clearNativeContentDeltaState, + settleNativeContentDeltas, +} from "./native-session-streaming" const log = getLogger("sse") const pendingSessionFetches = new Map() const NATIVE_REFRESH_DELAY_MS = 75 +const NATIVE_TERMINAL_RETRY_DELAY_MS = 500 +const MAX_NATIVE_TERMINAL_REFRESH_ATTEMPTS = 3 const nativeRefreshes = new Map running?: Promise + terminalAttempts?: number }>() let activeRetryToast: ToastHandle | null = null @@ -115,21 +123,47 @@ function requestNativeSessionRefresh(instanceId: string, sessionId: string, fina if (refresh.running) return refresh.running refresh.pending = false refresh.running = (async () => { + const terminal = refresh.speakAfter + if (!instances().has(refresh.instanceId) + || !sessions().get(refresh.instanceId)?.has(refresh.sessionId) + || getAuthoritativelyDeletedSessionIdsForInstance(refresh.instanceId).has(refresh.sessionId)) { + clearNativeContentDeltaState(refresh.instanceId, refresh.sessionId) + return + } try { await loadMessages(refresh.instanceId, refresh.sessionId, { force: true, skipChildren: true }) + if (!instances().has(refresh.instanceId) + || !sessions().get(refresh.instanceId)?.has(refresh.sessionId) + || getAuthoritativelyDeletedSessionIdsForInstance(refresh.instanceId).has(refresh.sessionId)) { + clearNativeContentDeltaState(refresh.instanceId, refresh.sessionId) + return + } + if (terminal) { + settleNativeContentDeltas(refresh.instanceId, refresh.sessionId) + refresh.terminalAttempts = 0 + } } catch (error) { log.error("Failed to refresh native session messages", { instanceId, sessionId, error }) + if (terminal + && (refresh.terminalAttempts ?? 0) + 1 < MAX_NATIVE_TERMINAL_REFRESH_ATTEMPTS + && instances().has(refresh.instanceId) + && sessions().get(refresh.instanceId)?.has(refresh.sessionId)) { + refresh.terminalAttempts = (refresh.terminalAttempts ?? 0) + 1 + refresh.pending = true + } } })().finally(() => { refresh.running = undefined if (refresh.pending) { - if (refresh.speakAfter) void run() + if (refresh.speakAfter) schedule(NATIVE_TERMINAL_RETRY_DELAY_MS) else schedule(NATIVE_REFRESH_DELAY_MS) return } if (refresh.speakAfter) { refresh.speakAfter = false - speakCompletedAssistantText(refresh.instanceId, refresh.sessionId) + if (instances().has(refresh.instanceId) && sessions().get(refresh.instanceId)?.has(refresh.sessionId)) { + speakCompletedAssistantText(refresh.instanceId, refresh.sessionId) + } } nativeRefreshes.delete(key) }) @@ -158,6 +192,13 @@ function clearNativeSessionRefresh(instanceId: string, sessionId: string): void } function handleNativeSessionEvent(instanceId: string, event: NativeSessionEvent): void { + if (event.type === "session.text.delta" || event.type === "session.reasoning.delta") { + const sessionId = event.data.sessionID + if (!instances().has(instanceId) || getAuthoritativelyDeletedSessionIdsForInstance(instanceId).has(sessionId)) return + ensureSessionStatus(instanceId, sessionId, "working", event.location?.directory) + applyNativeContentDelta(instanceId, event) + return + } switch (event.type) { case "form.created": addPendingForm(instanceId, event.data.form) @@ -225,6 +266,7 @@ function setTerminalNativeSessionStatus(instanceId: string, sessionId: string, f if (existing) setSessionStatus(instanceId, sessionId, "idle", { force: true }) else ensureSessionStatus(instanceId, sessionId, "idle", directory) if (failed) messageStoreBus.getOrCreate(instanceId).failPendingSends(sessionId) + settleNativeContentDeltas(instanceId, sessionId) requestNativeSessionRefresh(instanceId, sessionId, true) } @@ -535,6 +577,7 @@ function handleSessionDeleted(instanceId: string, event: EventSessionDeleted): v log.info(`[SSE] Session deleted: ${sessionId}`) clearNativeSessionRefresh(instanceId, sessionId) + clearNativeContentDeltaState(instanceId, sessionId) removeSessionRuntimeState(instanceId, sessionId) } @@ -550,6 +593,7 @@ function handleSessionIdle(instanceId: string, event: SessionIdle): void { } ensureSessionStatus(instanceId, sessionId, "idle", event.location?.directory) + settleNativeContentDeltas(instanceId, sessionId) requestNativeSessionRefresh(instanceId, sessionId, true) log.info(`[SSE] Session idle: ${sessionId}`) } diff --git a/packages/ui/src/stores/session-native-events.test.ts b/packages/ui/src/stores/session-native-events.test.ts index d4a28157..f51e5aa6 100644 --- a/packages/ui/src/stores/session-native-events.test.ts +++ b/packages/ui/src/stores/session-native-events.test.ts @@ -5,6 +5,7 @@ import { sdkManager } from "../lib/sdk-manager.ts" import type { Session } from "../types/session.ts" import { addInstance, removeInstance } from "./instances.ts" import { messageStoreBus } from "./message-v2/bus.ts" +import { clearNativeContentDeltaState } from "./native-session-streaming.ts" import { handleNativeSessionEvent, handleSessionIdle, handleSessionStatus } from "./session-events.ts" import { clearInstanceDeletedSessionAuthority, sessions, setSessions } from "./session-state.ts" @@ -100,13 +101,14 @@ describe("native session event reducer", () => { id: "text", created: 1, type: "session.text.delta", data: { sessionID: sessionId, assistantMessageID: "assistant", ordinal: 0, delta: "streaming text" }, }) + const store = messageStoreBus.getOrCreate(instanceId) + assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "streaming text") handleNativeSessionEvent(instanceId, { id: "tool", created: 2, type: "session.tool.progress", data: { sessionID: sessionId, assistantMessageID: "assistant", id: "tool", metadata: {} }, }) await delay(120) - const store = messageStoreBus.getOrCreate(instanceId) assert.equal(calls, 1) assert.equal(sessions().get(instanceId)?.get(sessionId)?.status, "working") assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "streaming text") @@ -142,9 +144,17 @@ describe("native session event reducer", () => { const eventTypes = ["session.text.delta", "session.reasoning.delta", "session.tool.progress"] as const let eventIndex = 0 const stream = setInterval(() => { + const type = eventTypes[eventIndex % eventTypes.length] + const id = `event-${eventIndex++}` handleNativeSessionEvent(instanceId, { - type: eventTypes[eventIndex++ % eventTypes.length], - data: { sessionID: sessionId }, + id, + created: eventIndex, + type, + data: type === "session.text.delta" + ? { sessionID: sessionId, assistantMessageID: "assistant", ordinal: 0, delta: "text " } + : type === "session.reasoning.delta" + ? { sessionID: sessionId, assistantMessageID: "assistant", ordinal: 1, delta: "reason " } + : { sessionID: sessionId, assistantMessageID: "assistant", id: "tool", metadata: {} }, } as any) }, 10) @@ -155,8 +165,12 @@ describe("native session event reducer", () => { assert.ok(calls >= 2, `expected periodic refreshes, received ${calls}`) assert.ok(calls <= 5, `expected refreshes to stay bounded, received ${calls}`) + const streamed = messageStoreBus.getOrCreate(instanceId).getMessage("assistant") + assert.match((streamed?.parts["assistant-text-0"]?.data as any)?.text ?? "", /text/) + assert.match((streamed?.parts["assistant-reasoning-1"]?.data as any)?.text ?? "", /reason/) } finally { clearInterval(stream) + clearNativeContentDeltaState(instanceId) messageStoreBus.unregisterInstance(instanceId) setSessions((prev) => { const next = new Map(prev); next.delete(instanceId); return next }) clearInstanceDeletedSessionAuthority(instanceId)