Emit StartedEvent per continuation cycle in the lockstep event stream#733
Emit StartedEvent per continuation cycle in the lockstep event stream#733PratikDhanave wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
Aligns Lockstep streaming semantics with OffThread by emitting workflow.StartedEvent{} at the start of each continuation (input → processing → halt) cycle when there is actual work to process, so event consumers can consistently delimit cycles across execution modes.
Changes:
- Enqueue a
StartedEventfor Lockstep continuation cycles after resuming fromRunStatusPendingRequests, guarded byHasUnprocessedMessages(). - Add a new cross-mode test asserting
StartedEventis emitted once per continuation cycle when an external response resumes a halted workflow.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| workflow/internal/execution/eventstream.go | Updates Lockstep event stream to emit StartedEvent for continuation cycles after pending-request resumes. |
| workflow/inproc/events_test.go | Adds a test covering per-cycle StartedEvent emission across Lockstep and OffThread streaming runs. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| startRunActivity() | ||
| // Emit a StartedEvent for the continuation cycle, mirroring the | ||
| // streaming run loop which raises one per input → processing → | ||
| // halt cycle. Only when there is actual work to process, so | ||
| // no-work wakeups (e.g. spurious signals) stay event-free. The | ||
| // event is drained and yielded before the cycle's supersteps. | ||
| if l.stepRunner.HasUnprocessedMessages() { | ||
| l.eventQueue.Enqueue(workflow.StartedEvent{}) | ||
| } |
There was a problem hiding this comment.
Good catch — fixed in d1e7e21: I now call drainAndFilterEvents right after enqueuing the continuation StartedEvent, so it is yielded before the cycle's supersteps run instead of being held in the queue and drained alongside the first superstep's events. This matches the comment and the initial-run behavior.
| resp := msg.(*workflow.ExternalResponse) | ||
| data, _ := resp.Data.As(port.Response) | ||
| return nil, wctx.YieldOutput(data) |
There was a problem hiding this comment.
Good catch — fixed in d1e7e21. As returns (any, bool) rather than an error, so I now check the ok flag and return an explicit error when the response payload isn't the expected type, rather than silently yielding a nil output.
| t.Fatalf("Build: %v", err) | ||
| } | ||
|
|
||
| ctx := context.Background() |
There was a problem hiding this comment.
Good catch — fixed in d1e7e21: the test now uses a 30s context.WithTimeout (with defer cancel), so respondWhenPending self-terminates via ctx.Err() and can't leak into subsequent tests if the run never reaches PendingRequests.
This comment has been minimized.
This comment has been minimized.
The lockstep event stream enqueued StartedEvent once per TakeEventStream call, but a single blocking WatchStream call drives multiple input->processing->halt cycles: after halting on pending requests it waits for input, then resumes the continuation cycle in place. The resume branch called startRunActivity() (telemetry only) without raising a StartedEvent, so a lockstep consumer saw exactly one StartedEvent regardless of how many external responses arrived. The streaming (OffThread) run loop already raises one StartedEvent per cycle, and workflow.StartedEvent documents that it fires again whenever new messages or external responses arrive after a halt. Emit the event in the lockstep resume branch, guarded by HasUnprocessedMessages so no-work wakeups stay event-free, restoring cross-mode parity.
d1e7e21 to
4a09917
Compare
Cross-Repo Parity Review — PR #733Scope: SummaryThis PR fixes a real behavioral inconsistency inside the Go codebase (lockstep vs streaming However, after checking the upstream .NET and Python implementations, I found a cross-repo semantic divergence worth flagging: Parity Finding:
|
What
In
lockstepRunEventStream.TakeEventStream,StartedEventwas enqueued exactly once, immediately before the run loop. But a single blockingWatchStreamcall (blockOnPendingRequest=true) drives more than one input -> processing -> halt cycle: when the workflow halts withRunStatusPendingRequeststhe resume branch waits for input and then continues the continuation cycle in place. That branch calledstartRunActivity()(which only adds a telemetry span event) and looped back to process the continuation without enqueuing a secondStartedEvent.As a result a lockstep consumer saw exactly one
StartedEventno matter how many external responses arrived, while the streaming (OffThread) run loop emits one per cycle.This change enqueues
workflow.StartedEvent{}after the resume branch, guarded byHasUnprocessedMessages()so no-work wakeups (e.g. spurious signals) stay event-free, matching the streaming loop's semantics. The event is drained and yielded before the continuation cycle's supersteps.Why (parity)
workflow.StartedEventis documented as firing "once per cycle in which there is actual work to process -- typically once at the start of a run and again whenever new messages or external responses arrive after a halt." The streaming run loop already honors this. This aligns the lockstep execution mode with that contract and with the .NET/PythonWorkflowStartedEventsemantics, where the event marks the start of each processing cycle rather than the lifetime of a single stream subscription. Consumers that key offStartedEventto delimit cycles (progress UIs, per-cycle bookkeeping) now behave identically across execution modes.Testing
Added
TestStartedEvent_EmittedPerContinuationCycleinworkflow/inproc/events_test.go. It builds a workflow that posts an external request and halts withRunStatusPendingRequests, runs it under bothinproc.Lockstepandinproc.OffThreadviaRunStreaming, iteratesWatchStream, supplies anExternalResponseafter the first halt to trigger the continuation cycle, and asserts exactly twoStartedEvents (one per cycle) plus the expected output. Before the fix the lockstep case yielded one; the OffThread case asserts cross-mode parity. Verified the test fails before the change and passes after.go build ./...go vet ./workflow/...go test ./workflow/inproc/... ./workflow/internal/execution/...(StartedEvent tests also run with-race)