fix(mothership): stream management

[Sg312]

Summary

Fix mothership stream management

Type of Change

• Bug fix

Testing

Tested with @icecrasher321 @TheodoreSpeaks

Checklist

• Code follows project style guidelines
• Self-reviewed my changes
• Tests added/updated and passing
• No new warnings introduced
• I confirm that I have read and agree to the terms outlined in the Contributor License Agreement (CLA)

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

1 Skipped Deployment

Project	Deployment	Actions	Updated (UTC)
docs	Skipped		Mar 17, 2026 5:32pm

[greptile-apps]

Greptile Summary

This PR fixes a streaming regression in the Mothership copilot chat by introducing server-side stream serialization (pendingChatStreams) and client-side stop-generation improvements, plus a ResizeObserver-driven embedded canvas fit that replaces the fragile one-shot onInit approach.

Key changes:

• chat-streaming.ts: Adds a module-level pendingChatStreams map so a new request for the same chatId force-aborts the prior stream and awaits its finally cleanup before forwarding to Go, preventing message-ordering races.
• mothership/chat/route.ts: Calls waitForPendingChatStream(actualChatId) before constructing the new stream — the correct integration point for the new serialization mechanism.
• use-chat.ts: Adds a 3-second busy-wait in stopGeneration to allow chatIdRef to be populated before the abort call is issued, handling the race where the user stops very quickly after sending.
• tool-execution.ts: Replaces plain setTimeout in the tool-decision polling loops with abortAwareSleep, enabling immediate loop exit when the stream is aborted rather than waiting out the current poll interval.
• workflow.tsx: Replaces the one-shot embedded fitView in onInit with a ResizeObserver-backed scheduleEmbeddedFit callback (debounced via requestAnimationFrame), fixing embedded canvas layout on container resize; lowers minZoom from 0.35 to 0.1.

Issues found:

• The pendingChatStreams map has a cross-resolution race: if registerPendingChatStream is called for a chatId that already has an entry (concurrent requests, or a timed-out waitForPendingChatStream followed by a new registration), the old stream's finally block resolves and deletes the new entry, meaning a future waiter could proceed before the new stream has actually settled. Adding a streamId guard to resolvePendingChatStream would fix this.
• The client-side busy-wait in stopGeneration can exit because sendingRef.current flips to false rather than because chatIdRef.current was set, leaving the abort call potentially ineffective with a null chat ID.

Confidence Score: 3/5

• Safe to merge with low risk — the regression fix is directionally correct, but the pendingChatStreams cross-resolution race could reintroduce ordering issues under concurrent or timeout-triggered scenarios.
• The majority of the changes are well-reasoned and solve real problems (abort-aware polling, embedded canvas resize, stream serialization). However, resolvePendingChatStream lacks a streamId guard, meaning a superseded stream can resolve a newer stream's tracking entry, which could allow a subsequent request to start before the current stream has fully settled — the very problem this PR aims to fix. The client-side busy-wait exit path with a null chatId is a secondary, lower-severity concern. Neither issue is a regression relative to the pre-PR state, but they leave the fix incomplete.
• apps/sim/lib/copilot/chat-streaming.ts — specifically the registerPendingChatStream and resolvePendingChatStream pair needs a streamId guard to prevent cross-resolution.

Important Files Changed

Filename	Overview
apps/sim/lib/copilot/chat-streaming.ts	Introduces module-level pendingChatStreams map to serialize successive requests on the same chatId. Core intent is solid, but the stream-id cross-resolution race (one stream's finally resolving another stream's promise when a timeout fires or concurrent requests collide) could undermine the serialization guarantee in edge cases.
apps/sim/app/api/mothership/chat/route.ts	Adds await waitForPendingChatStream(actualChatId) before creating a new stream — correctly serializes requests. Straightforward integration of the new helper; the main risk comes from the helper itself rather than this call-site.
apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts	Adds a client-side busy-wait in stopGeneration to allow time for chatIdRef to be populated before aborting. Functional, but the loop can exit without a chatId set, which may leave the stop action partially effective.
apps/sim/app/workspace/[workspaceId]/w/[workflowId]/workflow.tsx	Replaces the one-shot onInit fitView for embedded mode with a ResizeObserver-driven scheduleEmbeddedFit callback that debounces via requestAnimationFrame. Cleanup of both the observer and any pending animation frame is correctly handled. Lowers minZoom to 0.1 to accommodate very small containers.
apps/sim/lib/copilot/orchestrator/sse/handlers/tool-execution.ts	Replaces plain setTimeout in polling loops with abortAwareSleep, which correctly resolves (not rejects) on abort and clears the timer. Callers re-check abortSignal?.aborted at loop start, so abort is detected on the very next iteration. Clean and correct implementation.

Sequence Diagram

sequenceDiagram
    participant Client
    participant Route as mothership/chat/route.ts
    participant Streaming as chat-streaming.ts
    participant Go as Go Backend

    Note over Client,Go: Happy path — no prior stream
    Client->>Route: POST /api/mothership/chat (chatId)
    Route->>Streaming: waitForPendingChatStream(chatId)
    Streaming-->>Route: (no entry — returns immediately)
    Route->>Streaming: createSSEStream(chatId, streamId)
    Streaming->>Streaming: registerPendingChatStream(chatId, streamId)
    Streaming->>Go: orchestrateCopilotStream(...)
    Go-->>Client: SSE events
    Streaming->>Streaming: finally → resolvePendingChatStream(chatId)

    Note over Client,Go: Stop & retry — previous stream still in-flight
    Client->>Route: POST /api/mothership/chat (same chatId)
    Route->>Streaming: waitForPendingChatStream(chatId)
    Streaming->>Streaming: abortActiveStream(prevStreamId)
    Streaming-->>Route: await promise (previous stream settles)
    Note over Streaming: previous finally block runs → resolve()
    Route->>Streaming: createSSEStream(chatId, newStreamId)
    Streaming->>Streaming: registerPendingChatStream(chatId, newStreamId)
    Streaming->>Go: orchestrateCopilotStream(...)
    Go-->>Client: SSE events

    Note over Client,Streaming: Client-side stop
    Client->>Client: stopGeneration()
    Client->>Client: busy-wait until chatIdRef set (≤3 s)
    Client->>Route: POST /api/copilot/chat/abort {streamId}
    Route->>Streaming: abortActiveStream(streamId)

Loading

_{Last reviewed commit: c0a4471}