合并远程主分支客户端收口

src/stores/opencode.ts

@@ -274,6 +274,7 @@ function removeEventListenerIfSupported(
 let projectEventReconnectTimer: ReturnType<typeof setTimeout> | null = null;
 let projectEventReconnectAttempt = 0;
 const activeSessionRunTokens = new Map<string, number>();
+const pendingSessionRunStreamBatches = new Map<string, PendingSessionRunStreamBatch>();
 const drainingAbortedSessionRunIds = new Map<string, number>();
 const ignoredSuppressedPolledAbortKeys = new Map<string, { runId: number; keys: Set<string> }>();
 const IMAGE_DATA_URL_PATTERN = /^data:image\/[^,]+;base64,/i;
@@ -282,6 +283,27 @@ let commandLoadSequence = 0;
 let runtimeStatusSequence = 0;
 let sessionDiffLoadSequence = 0;
 
+const STREAM_UPDATE_BATCH_MS = 32;
+const SESSION_STATUS_FALLBACK_POLL_MS = POLL_INTERVAL_MS;
+const SESSION_MESSAGE_FALLBACK_POLL_MS = 1_500;
+
+type SessionRunStreamEventType =
+  | 'message.updated'
+  | 'message.part.updated'
+  | 'message.part.delta'
+  | 'message.part.removed';
+
+type SessionRunStreamEvent = {
+  type: SessionRunStreamEventType;
+  payload: Record<string, unknown>;
+};
+
+type PendingSessionRunStreamBatch = {
+  runToken: number;
+  events: SessionRunStreamEvent[];
+  timer: ReturnType<typeof setTimeout> | null;
+};
+
 function beginSessionRun(sessionId: string): number {
   sessionRunNonce += 1;
   const nextToken = sessionRunNonce;
@@ -300,6 +322,9 @@ function isSessionRunCurrent(sessionId: string, token: number): boolean {
 
 function closeActiveSessionEventSource(): void {
   activeSessionRunTokens.clear();
+  for (const sessionId of pendingSessionRunStreamBatches.keys()) {
+    cancelSessionRunStreamBatch(sessionId);
+  }
   drainingAbortedSessionRunIds.clear();
   ignoredSuppressedPolledAbortKeys.clear();
   for (const cleanup of activeSessionRunListenerCleanups.values()) cleanup();
@@ -321,6 +346,7 @@ function closeActiveSessionEventSource(): void {
 }
 
 function closeSessionEventSource(sessionId: string): void {
+  cancelSessionRunStreamBatch(sessionId);
   activeSessionRunListenerCleanups.get(sessionId)?.();
   activeSessionRunListenerCleanups.delete(sessionId);
   activeSessionRunListenerRegistrations.delete(sessionId);
@@ -350,6 +376,97 @@ function registerSessionRunEventListener(
   });
 }
 
+function cancelSessionRunStreamBatch(sessionId: string): void {
+  const pending = pendingSessionRunStreamBatches.get(sessionId);
+  if (!pending) return;
+  if (pending.timer) clearTimeout(pending.timer);
+  pendingSessionRunStreamBatches.delete(sessionId);
+}
+
+function hasSessionRunEventListener(sessionId: string): boolean {
+  return activeSessionRunTokens.has(sessionId)
+    && (activeSessionRunListenerRegistrations.get(sessionId)?.length ?? 0) > 0;
+}
+
+function getStreamEventIdentity(payload: Record<string, unknown>): string | null {
+  const eventId = typeof payload.eventID === 'string' && payload.eventID.trim()
+    ? payload.eventID.trim()
+    : typeof payload.eventId === 'string' && payload.eventId.trim()
+      ? payload.eventId.trim()
+      : typeof payload.sequence === 'string' && payload.sequence.trim()
+        ? payload.sequence.trim()
+        : typeof payload.sequence === 'number' && Number.isFinite(payload.sequence)
+          ? String(payload.sequence)
+          : null;
+  return eventId;
+}
+
+function getStreamDeltaKey(payload: Record<string, unknown>): string | null {
+  const messageId = typeof payload.messageID === 'string'
+    ? payload.messageID
+    : typeof payload.messageId === 'string'
+      ? payload.messageId
+      : '';
+  const partId = typeof payload.partID === 'string'
+    ? payload.partID
+    : typeof payload.partId === 'string'
+      ? payload.partId
+      : '';
+  const field = typeof payload.field === 'string' ? payload.field : 'text';
+  return messageId && partId ? `${messageId}:${partId}:${field}` : null;
+}
+
+function getStringStreamDelta(payload: Record<string, unknown>): string | null {
+  return typeof payload.delta === 'string' ? payload.delta : null;
+}
+
+function coalesceSessionRunStreamEvents(events: SessionRunStreamEvent[]): SessionRunStreamEvent[] {
+  const coalesced: SessionRunStreamEvent[] = [];
+
+  for (const event of events) {
+    if (event.type !== 'message.part.delta') {
+      coalesced.push(event);
+      continue;
+    }
+
+    const delta = getStringStreamDelta(event.payload);
+    const deltaKey = getStreamDeltaKey(event.payload);
+    const previous = coalesced.at(-1);
+    const previousDelta = previous && previous.type === 'message.part.delta'
+      ? getStringStreamDelta(previous.payload)
+      : null;
+    const previousKey = previous && previous.type === 'message.part.delta'
+      ? getStreamDeltaKey(previous.payload)
+      : null;
+
+    // Only merge anonymous, adjacent chunks for the same part. Events with an
+    // explicit identity remain separate so reducer-level replay protection is
+    // preserved, and interleaved parts retain their original order.
+    if (
+      delta
+      && deltaKey
+      && !getStreamEventIdentity(event.payload)
+      && previousDelta
+      && previousKey === deltaKey
+      && previous
+      && !getStreamEventIdentity(previous.payload)
+    ) {
+      coalesced[coalesced.length - 1] = {
+        ...previous,
+        payload: {
+          ...previous.payload,
+          delta: `${previousDelta}${delta}`,
+        },
+      };
+      continue;
+    }
+
+    coalesced.push(event);
+  }
+
+  return coalesced;
+}
+
 export function resetOpencodeStoreEphemeralStateForTests(): void {
   closeActiveSessionEventSource();
   sessionRunNonce = 0;
@@ -1750,7 +1867,7 @@ function getSessionTranscriptPatch(
     state.sessionTranscriptBySessionId[sessionId],
     { type, payload } satisfies OpenCodeEventEnvelope,
   );
-  if (!next) return {};
+  if (!next || next === state.sessionTranscriptBySessionId[sessionId]) return {};
   return {
     sessionTranscriptBySessionId: {
       ...state.sessionTranscriptBySessionId,
@@ -1759,6 +1876,129 @@ function getSessionTranscriptPatch(
   };
 }
 
+function getSessionStreamingEventPatch(
+  state: OpencodeState,
+  sessionId: string,
+  type: SessionRunStreamEventType,
+  payload: Record<string, unknown>,
+): Partial<OpencodeState> {
+  const transcriptPatch = getSessionTranscriptPatch(state, type, payload);
+
+  if (type === 'message.updated') {
+    const nextMessage = createStreamingMessageFromEvent(payload);
+    if (!nextMessage || nextMessage.role !== 'assistant') return transcriptPatch;
+    const currentMessage = state.streamingMessagesBySessionId[sessionId];
+    return {
+      ...transcriptPatch,
+      ...getSessionStreamingPatch(state, sessionId, {
+        ...nextMessage,
+        content: currentMessage?.content ?? nextMessage.content,
+        _attachedFiles: currentMessage?._attachedFiles,
+      }),
+    };
+  }
+
+  if (type === 'message.part.updated') {
+    const nextToolStatus = getStreamingToolStatus(payload);
+    const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
+    const currentTools = state.streamingToolsBySessionId[sessionId] ?? [];
+    const streamingMessage = applyStreamingPartToMessage(currentMessage, payload);
+    const streamingTools = nextToolStatus
+      ? mergeStreamingToolStatuses(currentTools, nextToolStatus)
+      : currentTools;
+    const part = payload.part;
+    const compacting = Boolean(
+      part
+      && typeof part === 'object'
+      && !Array.isArray(part)
+      && (part as Record<string, unknown>).type === 'compaction',
+    );
+    return {
+      ...transcriptPatch,
+      ...getSessionStreamingPatch(state, sessionId, streamingMessage, streamingTools),
+      compactingSessionIds: compacting
+        ? { ...state.compactingSessionIds, [sessionId]: true }
+        : state.compactingSessionIds,
+    };
+  }
+
+  if (type === 'message.part.delta') {
+    const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
+    return {
+      ...transcriptPatch,
+      ...getSessionStreamingPatch(
+        state,
+        sessionId,
+        applyStreamingPartDeltaToMessage(currentMessage, payload),
+      ),
+    };
+  }
+
+  const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
+  return {
+    ...transcriptPatch,
+    ...getSessionStreamingPatch(
+      state,
+      sessionId,
+      removeStreamingPartFromMessage(currentMessage, payload),
+    ),
+  };
+}
+
+function flushSessionRunStreamBatch(
+  set: OpencodeSet,
+  sessionId: string,
+  runToken: number,
+): void {
+  const pending = pendingSessionRunStreamBatches.get(sessionId);
+  if (!pending || pending.runToken !== runToken) return;
+  if (pending.timer) clearTimeout(pending.timer);
+  pendingSessionRunStreamBatches.delete(sessionId);
+  if (!isSessionRunCurrent(sessionId, runToken)) return;
+
+  const events = coalesceSessionRunStreamEvents(pending.events);
+  if (events.length === 0) return;
+
+  set((state) => {
+    let workingState = state;
+    let patch: Partial<OpencodeState> = {};
+    for (const event of events) {
+      const eventPatch = getSessionStreamingEventPatch(
+        workingState,
+        sessionId,
+        event.type,
+        event.payload,
+      );
+      patch = { ...patch, ...eventPatch };
+      workingState = { ...workingState, ...eventPatch };
+    }
+    return patch;
+  });
+}
+
+function queueSessionRunStreamEvent(
+  set: OpencodeSet,
+  sessionId: string,
+  runToken: number,
+  type: SessionRunStreamEventType,
+  payload: Record<string, unknown>,
+): void {
+  if (!isSessionRunCurrent(sessionId, runToken)) return;
+
+  let pending = pendingSessionRunStreamBatches.get(sessionId);
+  if (!pending || pending.runToken !== runToken) {
+    cancelSessionRunStreamBatch(sessionId);
+    pending = { runToken, events: [], timer: null };
+    pendingSessionRunStreamBatches.set(sessionId, pending);
+  }
+
+  pending.events.push({ type, payload });
+  if (pending.timer) return;
+  pending.timer = setTimeout(() => {
+    flushSessionRunStreamBatch(set, sessionId, runToken);
+  }, STREAM_UPDATE_BATCH_MS);
+}
+
 function getHydratedSessionTranscriptPatch(
   state: OpencodeState,
   sessionId: string,
@@ -1849,6 +2089,11 @@ function applyProjectEventToStore(
 
   const sessionId = getOpencodeEventSessionId(payload);
   if (!sessionId) return;
+  // A locally submitted run has a run-scoped listener on the same source.
+  // Let that listener be the sole owner of lifecycle, transcript, and live
+  // message updates; otherwise every stream chunk is reduced and published
+  // twice (once here and once by the run listener).
+  if (hasSessionRunEventListener(sessionId)) return;
   const active = isSessionEventActive(get(), sessionId);
   set((state) => {
     const transcriptPatch = getSessionTranscriptPatch(state, type, payload);
@@ -1905,49 +2150,18 @@ function applyProjectEventToStore(
     // but must not append the same delta a second time.
     if (active) return transcriptPatch;
 
-    if (type === 'message.updated') {
-      const nextMessage = createStreamingMessageFromEvent(payload);
-      return nextMessage && nextMessage.role === 'assistant'
-        ? {
-            ...transcriptPatch,
-            ...getSessionStreamingPatch(state, sessionId, nextMessage),
-          }
-        : transcriptPatch;
-    }
-    if (type === 'message.part.updated') {
-      const nextToolStatus = getStreamingToolStatus(payload);
-      const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
-      const currentTools = state.streamingToolsBySessionId[sessionId] ?? [];
-      const streamingMessage = applyStreamingPartToMessage(currentMessage, payload);
-      const streamingTools = nextToolStatus
-        ? mergeStreamingToolStatuses(currentTools, nextToolStatus)
-        : currentTools;
-      return {
-        ...transcriptPatch,
-        ...getSessionStreamingPatch(state, sessionId, streamingMessage, streamingTools),
-      };
-    }
-    if (type === 'message.part.delta') {
-      const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
-      return {
-        ...transcriptPatch,
-        ...getSessionStreamingPatch(
-          state,
-          sessionId,
-          applyStreamingPartDeltaToMessage(currentMessage, payload),
-        ),
-      };
-    }
-    if (type === 'message.part.removed') {
-      const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
-      return {
-        ...transcriptPatch,
-        ...getSessionStreamingPatch(
-          state,
-          sessionId,
-          removeStreamingPartFromMessage(currentMessage, payload),
-        ),
-      };
+    if (
+      type === 'message.updated'
+      || type === 'message.part.updated'
+      || type === 'message.part.delta'
+      || type === 'message.part.removed'
+    ) {
+      return getSessionStreamingEventPatch(
+        state,
+        sessionId,
+        type,
+        payload,
+      );
     }
     return transcriptPatch;
   });
@@ -2096,6 +2310,7 @@ function finishSessionRunSuccessfully(
   statuses: Record<string, OpencodeSessionStatus> = {},
 ): boolean {
   if (!isSessionRunCurrent(sessionId, runToken)) return false;
+  flushSessionRunStreamBatch(set, sessionId, runToken);
   closeSessionEventSource(sessionId);
   set((state) => {
     const sendingSessionIds = withoutKey(state.sendingSessionIds, sessionId);
@@ -2249,6 +2464,9 @@ async function runSessionSubmission(
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
         const status = normalizeSessionStatus(payload.status);
+        if (status.type === 'idle') {
+          flushSessionRunStreamBatch(set, sessionId, runToken);
+        }
         set((state) => ({
           ...getSessionTranscriptPatch(state, 'session.status', payload),
           ...dispatchSessionRunEvent(state, sessionId, {
@@ -2266,6 +2484,7 @@ async function runSessionSubmission(
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
+        flushSessionRunStreamBatch(set, sessionId, runToken);
         set((state) => ({
           ...getSessionTranscriptPatch(state, 'session.idle', payload),
           ...dispatchSessionRunEvent(state, sessionId, {
@@ -2284,6 +2503,7 @@ async function runSessionSubmission(
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
+        flushSessionRunStreamBatch(set, sessionId, runToken);
         set((state) => ({
           ...getSessionTranscriptPatch(state, 'session.compacted', payload),
           compactingSessionIds: withoutKey(state.compactingSessionIds, sessionId),
@@ -2294,6 +2514,7 @@ async function runSessionSubmission(
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
+        flushSessionRunStreamBatch(set, sessionId, runToken);
         const message = getSessionErrorMessage(payload);
         if (isAbortSessionErrorMessage(message)) {
           const drainingAbortedRunId = drainingAbortedSessionRunIds.get(sessionId);
@@ -2430,82 +2651,28 @@ async function runSessionSubmission(
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
-        const nextMessage = createStreamingMessageFromEvent(payload);
-        if (!nextMessage || nextMessage.role !== 'assistant') return;
-        set((state) => {
-          const currentMessage = state.streamingMessagesBySessionId[sessionId];
-          return {
-            ...getSessionTranscriptPatch(state, 'message.updated', payload),
-            ...getSessionStreamingPatch(state, sessionId, {
-              ...nextMessage,
-              content: currentMessage?.content ?? nextMessage.content,
-              _attachedFiles: currentMessage?._attachedFiles,
-            }),
-          };
-        });
+        queueSessionRunStreamEvent(set, sessionId, runToken, 'message.updated', payload);
       });
 
       registerSessionRunEventListener(sessionId, source, 'message.part.updated', (event) => {
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
-        const nextToolStatus = getStreamingToolStatus(payload);
-        const part = payload.part;
-        const compacting = Boolean(
-          part
-          && typeof part === 'object'
-          && !Array.isArray(part)
-          && (part as Record<string, unknown>).type === 'compaction',
-        );
-        set((state) => {
-          const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
-          const currentTools = state.streamingToolsBySessionId[sessionId] ?? [];
-          const streamingMessage = applyStreamingPartToMessage(currentMessage, payload);
-          const streamingTools = nextToolStatus
-            ? mergeStreamingToolStatuses(currentTools, nextToolStatus)
-            : currentTools;
-          return {
-            ...getSessionTranscriptPatch(state, 'message.part.updated', payload),
-            ...getSessionStreamingPatch(state, sessionId, streamingMessage, streamingTools),
-            compactingSessionIds: compacting
-              ? { ...state.compactingSessionIds, [sessionId]: true }
-              : state.compactingSessionIds,
-          };
-        });
+        queueSessionRunStreamEvent(set, sessionId, runToken, 'message.part.updated', payload);
       });
 
       registerSessionRunEventListener(sessionId, source, 'message.part.delta', (event) => {
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
-        set((state) => {
-          const currentMessage = state.streamingMessagesBySessionId[sessionId] ?? null;
-          return {
-            ...getSessionTranscriptPatch(state, 'message.part.delta', payload),
-            ...getSessionStreamingPatch(
-              state,
-              sessionId,
-              applyStreamingPartDeltaToMessage(currentMessage, payload),
-            ),
-          };
-        });
+        queueSessionRunStreamEvent(set, sessionId, runToken, 'message.part.delta', payload);
       });
 
       registerSessionRunEventListener(sessionId, source, 'message.part.removed', (event) => {
         const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
         if (!isSessionRunCurrent(sessionId, runToken)) return;
         if (getOpencodeEventSessionId(payload) !== sessionId) return;
-        set((state) => ({
-          ...getSessionTranscriptPatch(state, 'message.part.removed', payload),
-          ...getSessionStreamingPatch(
-            state,
-            sessionId,
-            removeStreamingPartFromMessage(
-              state.streamingMessagesBySessionId[sessionId] ?? null,
-              payload,
-            ),
-          ),
-        }));
+        queueSessionRunStreamEvent(set, sessionId, runToken, 'message.part.removed', payload);
       });
       }
     } catch {
@@ -2561,6 +2728,8 @@ async function runSessionSubmission(
     let messages: RawMessage[] = getSessionMessagesForState(get(), sessionId);
     let status: OpencodeSessionStatus = { type: 'busy' };
     const useBaselineInference = submission.kind !== 'prompt';
+    let nextStatusFallbackPollAt = 0;
+    let nextMessageFallbackPollAt = 0;
 
     while (true) {
       if (!isSessionRunCurrent(sessionId, runToken)) {
@@ -2571,9 +2740,30 @@ async function runSessionSubmission(
         throw liveSessionError;
       }
 
-      const statusResponse = await hostApiFetch<OpencodeSessionStatusActionResponse>('/api/opencode/sessions/status');
-      const statuses = normalizeSessionStatusMap(statusResponse.statuses);
-      const refreshedMessages = await fetchSessionMessages(sessionId);
+      const now = Date.now();
+      const runListenerOwnsStream = hasSessionRunEventListener(sessionId);
+      const eventStatus = runListenerOwnsStream
+        ? get().sessionStatuses[sessionId]
+        : undefined;
+      const shouldPollStatus = now >= nextStatusFallbackPollAt
+        || eventStatus?.type === 'idle';
+      let statuses: OpencodeSessionStatusMap = {};
+      if (shouldPollStatus) {
+        const statusResponse = await hostApiFetch<OpencodeSessionStatusActionResponse>('/api/opencode/sessions/status');
+        statuses = normalizeSessionStatusMap(statusResponse.statuses);
+        nextStatusFallbackPollAt = Date.now() + SESSION_STATUS_FALLBACK_POLL_MS;
+      }
+      const httpStatus = statuses[sessionId];
+      const observedStatus = shouldPollStatus
+        ? httpStatus ?? eventStatus
+        : eventStatus ?? httpStatus ?? status;
+      const shouldRefreshMessages = Date.now() >= nextMessageFallbackPollAt
+        || observedStatus?.type === 'idle';
+      let refreshedMessages = messages;
+      if (shouldRefreshMessages) {
+        refreshedMessages = await fetchSessionMessages(sessionId);
+        nextMessageFallbackPollAt = Date.now() + SESSION_MESSAGE_FALLBACK_POLL_MS;
+      }
       if (liveSessionError) {
         throw liveSessionError;
       }
@@ -2597,12 +2787,17 @@ async function runSessionSubmission(
         : sendingUserMessage
           ? getAssistantErrorAfterPrompt(refreshedMessages, sendingUserMessage)
           : null;
-      const polledStatus = statuses[sessionId]
-        ?? (useBaselineInference
-          ? inferMissingPolledStatusAfterBaseline(refreshedMessages, baseline)
-          : sendingUserMessage
-            ? inferMissingPolledStatus(refreshedMessages, sendingUserMessage)
-            : IDLE_SESSION_STATUS);
+      const inferredStatus = useBaselineInference
+        ? inferMissingPolledStatusAfterBaseline(refreshedMessages, baseline)
+        : sendingUserMessage
+          ? inferMissingPolledStatus(refreshedMessages, sendingUserMessage)
+          : IDLE_SESSION_STATUS;
+      // A missing HTTP status is common while the runtime is transitioning.
+      // Prefer a newly observed assistant response over the optimistic/event
+      // busy marker, otherwise a completed run can remain busy forever.
+      const polledStatus = shouldPollStatus
+        ? httpStatus ?? inferredStatus
+        : eventStatus ?? status ?? inferredStatus;
       status = assistantError || assistantAborted ? IDLE_SESSION_STATUS : polledStatus;
       const currentMessages = getSessionMessagesForState(get(), sessionId);
       messages = assistantAborted || status.type === 'idle'
@@ -2670,22 +2865,24 @@ async function runSessionSubmission(
         return messages;
       }
 
-      set((state) => ({
-        ...getSessionMessagePatch(state, sessionId, messages),
-        ...(assistantError
-          ? dispatchSessionRunEvent(state, sessionId, {
-              type: 'remote_failed',
-              runId: runToken,
-              error: assistantError,
-            })
-          : {}),
-        sessionStatuses: {
-          ...state.sessionStatuses,
-          ...statuses,
-          [sessionId]: status,
-        },
-        ...errorState(assistantError ?? null),
-      }));
+      if (shouldRefreshMessages || shouldPollStatus || assistantError) {
+        set((state) => ({
+          ...getSessionMessagePatch(state, sessionId, messages),
+          ...(assistantError
+            ? dispatchSessionRunEvent(state, sessionId, {
+                type: 'remote_failed',
+                runId: runToken,
+                error: assistantError,
+              })
+            : {}),
+          sessionStatuses: {
+            ...state.sessionStatuses,
+            ...statuses,
+            [sessionId]: status,
+          },
+          ...errorState(assistantError ?? null),
+        }));
+      }
 
       if (assistantError) {
         throw new Error(assistantError);