feat(workflow): add attach endpoint for in-progress streams#1084
Conversation
🦋 Changeset detectedLatest commit: b425093 The changes in this PR will be included in the next version bump. This PR includes changesets to release 5 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
This comment has been minimized.
This comment has been minimized.
📝 WalkthroughWalkthroughAdds Changes
Sequence DiagramsequenceDiagram
participant Client as Reconnecting Client
participant Server as Server Handler
participant SessionStore as Stream Session Store
participant StreamExecution as Active Stream Execution
participant SSEOutput as SSE Output
Client->>Server: GET /workflows/:id/executions/:executionId/stream<br/>(Last-Event-ID or fromSequence)
activate Server
Server->>SessionStore: Lookup active stream session for executionId
alt Session exists and active
SessionStore-->>Server: Return active session with replay buffer
Server->>StreamExecution: Validate execution state (running/suspended)
alt Execution is streamable
StreamExecution-->>Server: Confirm active stream
Server->>SSEOutput: Create new stream from session replay buffer
Server->>SSEOutput: Replay missed events (from Last-Event-ID/fromSequence onwards)
SSEOutput-->>Client: Replay events (workflow-start, step-complete, etc.)
loop Stream active
StreamExecution->>SSEOutput: New event produced
SSEOutput-->>Client: Stream event via SSE
end
Note over Client,SSEOutput: Client receives real-time events from sequence point
else Execution is terminal
Server-->>Client: 409 Conflict (execution not streamable)
end
else No active session
Server-->>Client: 404 Not Found (workflow/execution not found)
end
deactivate Server
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Deploying voltagent with
|
| Latest commit: |
b425093
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://c2683baf.voltagent.pages.dev |
| Branch Preview URL: | https://feat-workflow-stream-attach.voltagent.pages.dev |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 12 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="packages/server-core/src/handlers/workflow.handlers.ts">
<violation number="1" location="packages/server-core/src/handlers/workflow.handlers.ts:772">
P1: Resumed stream events are not broadcast to subscribers. After `activeSession.streamExecution = resumedStreamExecution`, the original `consumeWorkflowStream` loop (started at stream creation) is still iterating the *old* async iterable — `for await...of` captures the iterator once. If that iterator ended when the workflow suspended, no consumer is iterating the new `resumedStreamExecution` to call `broadcastWorkflowStreamEvent`. Consider re-invoking `consumeWorkflowStream(activeSession, deps, logger)` after updating `streamExecution` so resumed events reach attached clients.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| resumeData, | ||
| options?.stepId ? { stepId: options.stepId } : undefined, | ||
| ); | ||
| activeSession.streamExecution = resumedStreamExecution; |
There was a problem hiding this comment.
P1: Resumed stream events are not broadcast to subscribers. After activeSession.streamExecution = resumedStreamExecution, the original consumeWorkflowStream loop (started at stream creation) is still iterating the old async iterable — for await...of captures the iterator once. If that iterator ended when the workflow suspended, no consumer is iterating the new resumedStreamExecution to call broadcastWorkflowStreamEvent. Consider re-invoking consumeWorkflowStream(activeSession, deps, logger) after updating streamExecution so resumed events reach attached clients.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/server-core/src/handlers/workflow.handlers.ts, line 772:
<comment>Resumed stream events are not broadcast to subscribers. After `activeSession.streamExecution = resumedStreamExecution`, the original `consumeWorkflowStream` loop (started at stream creation) is still iterating the *old* async iterable — `for await...of` captures the iterator once. If that iterator ended when the workflow suspended, no consumer is iterating the new `resumedStreamExecution` to call `broadcastWorkflowStreamEvent`. Consider re-invoking `consumeWorkflowStream(activeSession, deps, logger)` after updating `streamExecution` so resumed events reach attached clients.</comment>
<file context>
@@ -485,6 +759,37 @@ export async function handleResumeWorkflow(
+ resumeData,
+ options?.stepId ? { stepId: options.stepId } : undefined,
+ );
+ activeSession.streamExecution = resumedStreamExecution;
+
+ const status = await resumedStreamExecution.status;
</file context>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/core/src/workflow/core.ts (1)
2229-2238:⚠️ Potential issue | 🟡 MinorMissing
lastEventSequenceinresumeFromcauses event sequence counter to reset on every resume, breaking replay for re-suspended executions.
executionContext.eventSequenceis initialized at Line 974 asoptions?.resumeFrom?.lastEventSequence || 0. TheresumeFromobject here does not forwardlastEventSequencefromexecResult.suspension, so the resumed execution always starts its event sequence counter from0. If the resumed workflow suspends a second time, the storedlastEventSequencein memory will be0— not a continuation of the sequence from the initial run.The attach endpoint's replay buffer relies on this stored
lastEventSequenceas thefromSequencebaseline when a client re-attaches. Clients attaching after a second suspension will receive a stale/incorrect sequence starting position.🛠️ Proposed fix
const resumeOptions: WorkflowRunOptions = { executionId: execResult.executionId, resumeFrom: { executionId: execResult.executionId, checkpoint: execResult.suspension.checkpoint, resumeStepIndex, resumeData: input, + lastEventSequence: execResult.suspension.lastEventSequence, }, suspendController: resumedSuspendController, };🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/core/src/workflow/core.ts` around lines 2229 - 2238, The resumeOptions.resumeFrom object is missing lastEventSequence so resumed runs restart their eventSequence at 0; update the resumeOptions construction (the resumeOptions constant) to include resumeFrom.lastEventSequence set from execResult.suspension.lastEventSequence (or the appropriate field on execResult.suspension) so WorkflowRunOptions.resumeFrom carries lastEventSequence into subsequent resumes; keep the rest of the fields (executionId, checkpoint, resumeStepIndex, resumeData, suspendController) unchanged and ensure the property name is exactly lastEventSequence to match executionContext's resume handling.
🧹 Nitpick comments (4)
packages/server-core/src/handlers/workflow.stream-attach.handlers.spec.ts (2)
175-262: Solid integration test for attach + replay flow.The promise-based gating pattern (
releaseSecondEvent) keeps the test deterministic. One gap: onlyfromSequencereplay is tested — thelastEventIdpath (which the handler also supports viaparseReplaySequence) is not covered. Consider adding a small variant.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/server-core/src/handlers/workflow.stream-attach.handlers.spec.ts` around lines 175 - 262, Add a second test variant that exercises the alternate replay path using lastEventId: duplicate the existing "attaches to active stream and replays from sequence cursor" test but call handleAttachWorkflowStream with a payload using lastEventId (set to firstEvent.id) instead of fromSequence, or explicitly invoke parseReplaySequence if the handler expects that form; ensure you create the same deterministic gating (releaseSecondEvent) and assert the same sequence of events from both primaryReader and attachedReader (step-complete and workflow-result) to validate the lastEventId replay path in handleAttachWorkflowStream/parseReplaySequence.
84-115:readSSEEventassumes one SSE event perread()call.This helper will break if the stream implementation ever batches multiple events into a single chunk or splits an event across chunks. For unit tests against the in-memory stream factory this is fine, but consider adding a brief comment documenting this assumption for future maintainers.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/server-core/src/handlers/workflow.stream-attach.handlers.spec.ts` around lines 84 - 115, The helper readSSEEvent assumes each reader.read() yields exactly one complete SSE event and will break if chunks contain multiple events or split an event across reads; add a concise comment above the readSSEEvent function stating this explicit assumption (and its caveat about batching/splitting), and optionally mention that a more robust implementation would accumulate chunks and parse by SSE event boundaries if needed in the future.packages/server-core/src/handlers/workflow.handlers.ts (2)
36-36: Module-level singletonactiveWorkflowStreamSessionswon't work across multiple server instances.This in-memory
Mapis inherently single-process. In serverless or horizontally-scaled deployments, attach requests hitting a different instance won't find the session. This is likely acceptable forserver-core's intended local-dev / single-process use case, but it's worth noting as a limitation in documentation or a code comment.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/server-core/src/handlers/workflow.handlers.ts` at line 36, The module-level Map activeWorkflowStreamSessions is a single-process in-memory store and will not work across multiple server instances; update the code around the declaration of activeWorkflowStreamSessions to either (a) add a clear comment/documentation note that it is intentionally single-process for local/dev use and list the limitation, or (b) replace it with a cross-instance backing store (e.g., Redis) and update the access points that use activeWorkflowStreamSessions (look for functions handling session create/attach/lookup in workflow.handlers.ts) to use the shared store API (SET/GET/DEL or client wrapper) instead so sessions are discoverable across instances. Ensure the chosen approach updates all reads/writes to activeWorkflowStreamSessions to use the new store or includes the explanatory comment near the Map declaration.
762-791: Resume handler blocks until workflow completion when a streaming session is active.When an active streaming session exists,
handleResumeWorkflowawaitsresumedStreamExecution.status,.result, and.endAt(Lines 774-776). For long-running workflows, this keeps the HTTP response pending until the entire workflow finishes. Meanwhile,consumeWorkflowStreamis also running and will attempt to read the same terminal promises.This may be intentional (the caller gets the final result synchronously), but for workflows that take minutes/hours after resume, the HTTP request will likely time out. Consider returning an immediate acknowledgement (e.g.,
{ status: "resumed", executionId }) and letting clients observe completion via the attached SSE stream instead.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/server-core/src/handlers/workflow.handlers.ts` around lines 762 - 791, The resume branch currently awaits terminal promises (resumedStreamExecution.status, .result, .endAt) which blocks the HTTP response until workflow completion; instead, after calling activeSession.streamExecution.resume(...) and updating activeSession.streamExecution, return an immediate acknowledgement (e.g., { success: true, data: { executionId: resumedStreamExecution.executionId, status: "resumed", startAt: resumedStreamExecution.startAt instanceof Date ? resumedStreamExecution.startAt.toISOString() : resumedStreamExecution.startAt } }) without awaiting status/result/endAt so the request doesn't hang; keep the call to activeSession.streamExecution = resumedStreamExecution and let clients consume completion via the existing SSE stream/consumeWorkflowStream.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@packages/core/src/workflow/core.ts`:
- Around line 2229-2238: The resumeOptions.resumeFrom object is missing
lastEventSequence so resumed runs restart their eventSequence at 0; update the
resumeOptions construction (the resumeOptions constant) to include
resumeFrom.lastEventSequence set from execResult.suspension.lastEventSequence
(or the appropriate field on execResult.suspension) so
WorkflowRunOptions.resumeFrom carries lastEventSequence into subsequent resumes;
keep the rest of the fields (executionId, checkpoint, resumeStepIndex,
resumeData, suspendController) unchanged and ensure the property name is exactly
lastEventSequence to match executionContext's resume handling.
---
Nitpick comments:
In `@packages/server-core/src/handlers/workflow.handlers.ts`:
- Line 36: The module-level Map activeWorkflowStreamSessions is a single-process
in-memory store and will not work across multiple server instances; update the
code around the declaration of activeWorkflowStreamSessions to either (a) add a
clear comment/documentation note that it is intentionally single-process for
local/dev use and list the limitation, or (b) replace it with a cross-instance
backing store (e.g., Redis) and update the access points that use
activeWorkflowStreamSessions (look for functions handling session
create/attach/lookup in workflow.handlers.ts) to use the shared store API
(SET/GET/DEL or client wrapper) instead so sessions are discoverable across
instances. Ensure the chosen approach updates all reads/writes to
activeWorkflowStreamSessions to use the new store or includes the explanatory
comment near the Map declaration.
- Around line 762-791: The resume branch currently awaits terminal promises
(resumedStreamExecution.status, .result, .endAt) which blocks the HTTP response
until workflow completion; instead, after calling
activeSession.streamExecution.resume(...) and updating
activeSession.streamExecution, return an immediate acknowledgement (e.g., {
success: true, data: { executionId: resumedStreamExecution.executionId, status:
"resumed", startAt: resumedStreamExecution.startAt instanceof Date ?
resumedStreamExecution.startAt.toISOString() : resumedStreamExecution.startAt }
}) without awaiting status/result/endAt so the request doesn't hang; keep the
call to activeSession.streamExecution = resumedStreamExecution and let clients
consume completion via the existing SSE stream/consumeWorkflowStream.
In `@packages/server-core/src/handlers/workflow.stream-attach.handlers.spec.ts`:
- Around line 175-262: Add a second test variant that exercises the alternate
replay path using lastEventId: duplicate the existing "attaches to active stream
and replays from sequence cursor" test but call handleAttachWorkflowStream with
a payload using lastEventId (set to firstEvent.id) instead of fromSequence, or
explicitly invoke parseReplaySequence if the handler expects that form; ensure
you create the same deterministic gating (releaseSecondEvent) and assert the
same sequence of events from both primaryReader and attachedReader
(step-complete and workflow-result) to validate the lastEventId replay path in
handleAttachWorkflowStream/parseReplaySequence.
- Around line 84-115: The helper readSSEEvent assumes each reader.read() yields
exactly one complete SSE event and will break if chunks contain multiple events
or split an event across reads; add a concise comment above the readSSEEvent
function stating this explicit assumption (and its caveat about
batching/splitting), and optionally mention that a more robust implementation
would accumulate chunks and parse by SSE event boundaries if needed in the
future.
1 participant
PR Checklist
Please check if your PR fulfills the following requirements:
Bugs / Features
What is the current behavior?
POST /workflows/:id/stream.executionId, so reconnect/resume observers cannot receive live events.What is the new behavior?
GET /workflows/:id/executions/:executionId/streamto attach to an active workflow SSE stream.fromSequencequery param andLast-Event-ID.404when workflow/execution is not found409when execution is not streamable (completed/cancelled/error or no active stream session)POST /workflows/:id/streambehavior unchanged for starting new executions.fixes #1081
Notes for reviewers
packages/server-core/src/handlers/workflow.stream-attach.handlers.spec.ts.examples/with-workflowsmoke scenario.Summary by cubic
Adds an attach endpoint to join in-progress workflow SSE streams and support reconnects with replay. Addresses Linear #1081 and keeps the existing start-stream endpoint unchanged.
Written for commit b425093. Summary will update on new commits.
Summary by CodeRabbit
Release Notes