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)