improvement(mothership): streaming state transitions

[icecrasher321]

Summary

Improve streaming state transitions and fix issues like switching tabs leading to odd stream UIs.

Type of Change

• Bug fix
• Other: Code quality

Testing

Tested manually

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		May 4, 2026 11:37pm

[cursor]

PR Summary

Medium Risk
Touches chat streaming stop/reconnect, queued-message handoff, and cache invalidation logic; regressions could cause stuck UI, dropped queued messages, or missed task refreshes under race conditions. Changes are bounded to mothership/copilot streaming paths but are behaviorally complex and timing-sensitive.

Overview
Propagates streamId through task status pubsub/SSE (taskPubSub, mothership events, copilot stop/post) so clients can correlate started/completed events to a specific stream.

Hardens use-chat streaming state transitions: adds explicit timeouts/abort-signal utilities, prevents duplicate cursor processing, supports resolving chatId from stream batches, recovers active streams on pageshow/visibility/online, and makes queued-message handoff robust with TTL/claims/backoff and a dedicated queued-handoff stop mode.

Updates task-event handling to invalidate/remove the right React Query caches (including detail on relevant events) while avoiding detail invalidation when a status event is stale relative to an optimistic/newer active stream; expands tests to cover these race cases.

^{Reviewed by Cursor Bugbot for commit b8d76f9. Configure here.}

[greptile-apps]

Greptile Summary

This PR improves streaming state transitions in the mothership chat UI by: (1) adding streamId correlation to all task status events so clients can make smarter cache-invalidation decisions without spurious refetches, and (2) introducing a visibility/online recovery mechanism that resumes an active SSE stream when the user returns to a tab, fixing odd UI state caused by the stream stalling while the tab was hidden.

Supporting infrastructure includes combineAbortSignals, sleepWithAbort, resolveChatIdForStream, owner-tagged handoff claims, a queued-handoff stop mode, and targetChatId propagation through all finalize/reconnect code paths.

Confidence Score: 5/5

Safe to merge — no P0/P1 issues found; two P2 observations on abort-signal hygiene.

All findings are P2. The resolveChatIdForStream inside stopBarrier is bounded by its own internal 10 s timeout so it cannot hang indefinitely; the combineAbortSignals cleanup-listener ordering issue is a minor GC concern with no observable data-loss path. Core logic for stream recovery, cache invalidation, and cross-tab handoff is correct and well-tested.

use-chat.ts — the stopBarrier in stopGeneration (queued-handoff branch) and combineAbortSignals helper are worth a second read for abort-hygiene completeness.

Important Files Changed

Filename	Overview
apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts	Major refactor: adds stream recovery on tab-return (visibility/online events), owner-tagged handoff claims, queued-handoff stop mode, combineAbortSignals/sleepWithAbort primitives, resolveChatIdForStream, adoptResolvedChatId, and propagates targetChatId through all finalize paths. Core logic is sound; one noteworthy gap in the queued-handoff stop barrier.
apps/sim/hooks/use-task-events.ts	Adds smart per-detail cache invalidation for started/completed events keyed by streamId; removes detail cache for deleted events; uses getLiveAssistantMessageId to detect locally-optimistic active streams and avoid spurious refetches.
apps/sim/hooks/use-task-events.test.ts	Comprehensive test coverage for all new invalidation decision branches; covers deleted, started, completed, renamed, and unknown event types with various cached-state permutations.
apps/sim/lib/mothership/inbox/executor.ts	Moves generateId() for userMessageId earlier so streamId can be included in both started and completed status events; no semantic change to execution logic.
apps/sim/lib/copilot/chat/post.ts	Adds streamId: userMessageId to started, completed, and error-path completed events so clients can correlate status events to specific streams.
apps/sim/app/api/mothership/events/route.ts	Conditionally forwards streamId from the task status event to the SSE payload so browser clients receive it.
apps/sim/app/api/copilot/chat/stop/route.ts	Passes streamId through to the completed status event on successful stop; straightforward one-line addition.
apps/sim/app/api/copilot/chat/stream/route.ts	Includes chatId in the resume response body when the run has an associated chatId, so clients can resolve the chatId from a batch poll without an extra request.
apps/sim/lib/copilot/tasks.ts	Adds optional streamId field to TaskStatusEvent interface to carry per-stream correlation data in pubsub events.

Sequence Diagram

sequenceDiagram
    participant UI as Browser Tab
    participant useChat as useChat hook
    participant SSE as SSE Stream
    participant Server as Server (API)
    participant Redis as Redis/Pub-Sub
    participant OtherTab as Other Tab (sessionStorage)

    Note over UI,OtherTab: Normal streaming flow
    UI->>useChat: sendMessage()
    useChat->>Server: POST /stream
    Server-->>SSE: stream events
    SSE-->>useChat: processSSEStream (live)
    Server->>Redis: publishStatus(started, streamId)
    Redis-->>useChat: task_status SSE (streamId)
    useChat->>useChat: shouldSkipDetailInvalidation?

    Note over UI,OtherTab: Tab switch / visibility recovery
    UI->>useChat: visibilitychange / pageshow / online
    useChat->>useChat: recoverActiveStreamFromRedis()
    useChat->>useChat: cancelActiveStreamReader()
    useChat->>Server: fetchStreamBatch(afterCursor)
    Server-->>useChat: batch events + chatId
    useChat->>SSE: attachToExistingStream (resume)
    SSE-->>useChat: processSSEStream (resume)

    Note over UI,OtherTab: Queued message during active stream
    UI->>useChat: sendMessage() [while streaming]
    useChat->>useChat: createQueuedMessage(+handoff)
    useChat->>useChat: stopGeneration(queued-handoff)
    useChat->>Server: POST /abort (streamId)
    alt chatId unknown
        useChat->>Server: resolveChatIdForStream (poll)
        Server-->>useChat: chatId resolved
    end
    Server->>Redis: publishStatus(completed, streamId)
    Redis-->>useChat: task_status SSE
    useChat->>useChat: shouldSkipDetailInvalidation?
    useChat->>Server: POST /stream (new message)

    Note over UI,OtherTab: Cross-tab handoff recovery
    UI->>OtherTab: writeQueuedSendHandoffState (sessionStorage)
    OtherTab->>OtherTab: writeQueuedSendHandoffClaim (ownerId)
    OtherTab->>useChat: resolveChatIdForStream
    OtherTab->>Server: POST /stream (recovered message)

Loading

_{Reviews (2): Last reviewed commit: "address comments" | Re-trigger Greptile}

[icecrasher321]

bugbot run

[icecrasher321]

@greptile

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit b8d76f9. Configure here.}