Skip to content

fix(stream) handle task switching #3609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheodoreSpeaks merged 3 commits into from
Mar 16, 2026
Merged

fix(stream) handle task switching #3609

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6af820b
Fix task switching causing stream to abort
Mar 16, 2026
170eb79
Process all task streams all the time
Mar 16, 2026
cc1ae03
Process task streams that are in the background
Mar 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Process task streams that are in the background
  • Loading branch information
Theodore Li committed Mar 16, 2026
commit cc1ae036358bb51929fbf6e80c52ebc1df736fd4
14 changes: 6 additions & 8 deletions apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -697,14 +697,17 @@ export function useChat(
if (DEPLOY_TOOL_NAMES.has(tc.name) && tc.status === 'success') {
const output = tc.result?.output as Record<string, unknown> | undefined
const deployedWorkflowId = (output?.workflowId as string) ?? undefined
if (deployedWorkflowId) {
const isDeployed = output?.isDeployed !== false
if (deployedWorkflowId && typeof output?.isDeployed === 'boolean') {
const isDeployed = output.isDeployed as boolean
const serverDeployedAt = output.deployedAt
? new Date(output.deployedAt as string)
: undefined
useWorkflowRegistry
.getState()
.setDeploymentStatus(
deployedWorkflowId,
isDeployed,
isDeployed ? new Date() : undefined
isDeployed ? (serverDeployedAt ?? new Date()) : undefined
)
queryClient.invalidateQueries({
queryKey: deploymentKeys.info(deployedWorkflowId),
Expand Down Expand Up @@ -1147,11 +1150,6 @@ export function useChat(
useEffect(() => {
return () => {
streamGenRef.current++
// Only drop the browser→Sim read; the Sim→Go stream stays open
// so the backend can finish persisting. Explicit abort is only
// triggered by the stop button via /api/copilot/chat/abort.
abortControllerRef.current?.abort()
abortControllerRef.current = null
sendingRef.current = false
}
}, [])
Comment on lines 1150 to 1155
Copy link
Copy Markdown
Contributor

@greptile-apps greptile-apps Bot Mar 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed abort means stream continues processing after unmount

The previous cleanup explicitly called abortControllerRef.current?.abort() to cancel the in-flight SSE fetch when the component unmounts. That line was intentionally removed here to allow the backend stream to stay alive across task switches, but it has a side effect: the ReadableStreamDefaultReader inside processSSEStream's while (true) { reader.read() } loop keeps running after unmount.

streamGenRef.current++ prevents finalizeRef.current() from being called with a stale generation, but the inner flush() calls — which call setMessages, setIsSending, setIsReconnecting etc. — are not gated by the generation counter. If the stream is still in flight when the user switches tasks, state setters are called on the now-stale React fiber, which can cause React warnings and ghost state.

Consider aborting the reader without aborting the server-side stream — i.e., abort the client fetch connection while keeping the Go-side stream alive. The server-side stream can remain open without the client holding a reader:

Expand Down
40 changes: 24 additions & 16 deletions apps/sim/lib/copilot/client-sse/handlers.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -567,7 +567,6 @@ export const sseHandlers: Record<string, SSEHandler> = {
}
}

// Deploy tools: update deployment status in workflow registry
if (
targetState === ClientToolCallState.success &&
(current.name === 'deploy_api' ||
Expand All @@ -579,21 +578,30 @@ export const sseHandlers: Record<string, SSEHandler> = {
const resultPayload = asRecord(
data?.result || eventData.result || eventData.data || data?.data
)
const input = asRecord(current.params)
const workflowId =
(resultPayload?.workflowId as string) ||
(input?.workflowId as string) ||
useWorkflowRegistry.getState().activeWorkflowId
const isDeployed = resultPayload?.isDeployed !== false
if (workflowId) {
useWorkflowRegistry
.getState()
.setDeploymentStatus(workflowId, isDeployed, isDeployed ? new Date() : undefined)
logger.info('[SSE] Updated deployment status from tool result', {
toolName: current.name,
workflowId,
isDeployed,
})
if (typeof resultPayload?.isDeployed === 'boolean') {
const input = asRecord(current.params)
const workflowId =
(resultPayload?.workflowId as string) ||
(input?.workflowId as string) ||
useWorkflowRegistry.getState().activeWorkflowId
const isDeployed = resultPayload.isDeployed as boolean
const serverDeployedAt = resultPayload.deployedAt
? new Date(resultPayload.deployedAt as string)
: undefined
if (workflowId) {
useWorkflowRegistry
.getState()
.setDeploymentStatus(
workflowId,
isDeployed,
isDeployed ? (serverDeployedAt ?? new Date()) : undefined
)
logger.info('[SSE] Updated deployment status from tool result', {
toolName: current.name,
workflowId,
isDeployed,
})
}
}
} catch (err) {
logger.warn('[SSE] Failed to hydrate deployment status', {
Expand Down
8 changes: 7 additions & 1 deletion apps/sim/lib/copilot/orchestrator/tool-executor/deployment-tools/deploy.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,10 @@ export async function executeDeployChat(
return { success: false, error: 'Unauthorized chat access' }
}
await db.delete(chat).where(eq(chat.id, existing[0].id))
return { success: true, output: { success: true, action: 'undeploy', isDeployed: false } }
return {
success: true,
output: { workflowId, success: true, action: 'undeploy', isChatDeployed: false },
}
Comment on lines 88 to +93
Copy link
Copy Markdown
Contributor

@greptile-apps greptile-apps Bot Mar 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

deploy_chat undeploy response is missing isDeployed for symmetry

The deploy path returns isDeployed: true, but the undeploy path now returns isChatDeployed: false with no isDeployed field. The client-side handlers in both handlers.ts and use-chat.ts gate their registry update on typeof output?.isDeployed === 'boolean', so they silently skip the undeploy case.

This is arguably correct behaviour (undeploying a chat does not remove the workflow's API deployment), but the asymmetry between the deploy and undeploy output shapes makes intent harder to follow. A comment explaining why isDeployed is intentionally absent from the undeploy response would help future maintainers:

Suggested change
}
await db.delete(chat).where(eq(chat.id, existing[0].id))
return { success: true, output: { success: true, action: 'undeploy', isDeployed: false } }
return {
success: true,
output: { workflowId, success: true, action: 'undeploy', isChatDeployed: false },
}
return {
success: true,
// isDeployed is intentionally omitted: undeploying a chat does not
// affect the workflow's API deployment status.
output: { workflowId, success: true, action: 'undeploy', isChatDeployed: false },
}

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

}

const { hasAccess } = await checkWorkflowAccessForChatCreation(workflowId, context.userId)
Expand Down Expand Up @@ -199,9 +202,11 @@ export async function executeDeployChat(
return {
success: true,
output: {
workflowId,
success: true,
action: 'deploy',
isDeployed: true,
isChatDeployed: true,
identifier,
chatUrl: `${baseUrl}/chat/${identifier}`,
apiEndpoint: `${baseUrl}/api/workflows/${workflowId}/run`,
Expand Down Expand Up @@ -355,6 +360,7 @@ export async function executeRedeploy(
success: true,
output: {
workflowId,
isDeployed: true,
deployedAt: result.deployedAt || null,
version: result.version,
apiEndpoint: `${baseUrl}/api/workflows/${workflowId}/run`,
Expand Down