From 4f2abe9daa718ff15a2623c0bb33c683d8e682b7 Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 09:13:42 +0530 Subject: [PATCH 1/2] Emit StartedEvent per continuation cycle in lockstep event stream 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. --- workflow/inproc/events_test.go | 119 +++++++++++++++++++++ workflow/internal/execution/eventstream.go | 8 ++ 2 files changed, 127 insertions(+) diff --git a/workflow/inproc/events_test.go b/workflow/inproc/events_test.go index 0194a791..e7fbccb4 100644 --- a/workflow/inproc/events_test.go +++ b/workflow/inproc/events_test.go @@ -156,6 +156,125 @@ func TestStartedEvent_EmittedInLockstepWithoutWork(t *testing.T) { } } +// askOnceBinding returns an executor that posts a single ExternalRequest on +// string input and yields the response payload as output once it arrives. It +// drives an input → request-halt → response → output → idle run: exactly two +// input → processing → halt cycles. +func askOnceBinding(id string) workflow.ExecutorBinding { + port := workflow.RequestPort{ + ID: "ask", + Request: reflect.TypeFor[string](), + Response: reflect.TypeFor[string](), + } + binding := workflow.ExecutorBinding{ID: id, ImplementationID: "*workflow.Executor"} + binding.NewExecutorFunc = func(_ string) (*workflow.Executor, error) { + return &workflow.Executor{ + ID: id, + + DisableAutoSendMessageHandlerResultObject: true, + DisableAutoYieldOutputHandlerResultObject: true, + ConfigureProtocol: func(rb *workflow.ProtocolBuilder) (*workflow.ProtocolBuilder, error) { + rb.YieldsOutputType(reflect.TypeFor[string]()) + rb.RouteBuilder. + AddHandlerRaw(reflect.TypeFor[string](), nil, func(wctx *workflow.Context, _ any) (any, error) { + req, err := workflow.NewExternalRequest("req-1", port, "what is your name?") + if err != nil { + return nil, err + } + return nil, wctx.PostRequest(req) + }). + AddHandlerRaw(reflect.TypeFor[*workflow.ExternalResponse](), nil, func(wctx *workflow.Context, msg any) (any, error) { + resp := msg.(*workflow.ExternalResponse) + data, _ := resp.Data.As(port.Response) + return nil, wctx.YieldOutput(data) + }) + return rb, nil + }, + }, nil + } + return binding +} + +// respondWhenPending waits until the streaming run has halted with pending +// requests, then sends the response exactly once. In Lockstep mode the run is +// driven synchronously by the WatchStream consumer, so the continuation must be +// injected from a separate goroutine after the halt is observed; sending before +// the halt would let the same cycle absorb the response. +func respondWhenPending(ctx context.Context, stream *inproc.StreamingRun, resp *workflow.ExternalResponse) { + for { + if ctx.Err() != nil { + return + } + if status, err := stream.GetStatus(ctx); err == nil && status == inproc.RunStatusPendingRequests { + _ = stream.SendResponse(ctx, resp) + return + } + time.Sleep(time.Millisecond) + } +} + +// TestStartedEvent_EmittedPerContinuationCycle verifies that a StartedEvent is +// raised for every input → processing → halt cycle, including the continuation +// cycle that runs after an external response is supplied. The streaming +// (OffThread) run loop already emits one StartedEvent per cycle; Lockstep's +// blocking WatchStream drives multiple cycles from a single TakeEventStream +// call and must match that per-cycle semantics. +func TestStartedEvent_EmittedPerContinuationCycle(t *testing.T) { + for _, tc := range []struct { + name string + env *inproc.ExecutionEnvironment + }{ + {"lockstep", inproc.Lockstep}, + {"offthread", inproc.OffThread}, + } { + t.Run(tc.name, func(t *testing.T) { + asker := askOnceBinding("asker") + wf, err := workflow.NewBuilder(asker).WithOutputFrom(asker).Build() + if err != nil { + t.Fatalf("Build: %v", err) + } + + ctx := context.Background() + stream, err := tc.env.RunStreaming(ctx, wf, "kick") + if err != nil { + t.Fatalf("RunStreaming: %v", err) + } + defer func() { _ = stream.CancelRun() }() + + var startedCount int + var responded bool + var gotOutput any + for evt, err := range stream.WatchStream(ctx) { + if err != nil { + t.Fatalf("watch: %v", err) + } + switch e := evt.(type) { + case workflow.StartedEvent: + startedCount++ + case workflow.RequestInfoEvent: + if !responded { + responded = true + resp, err := e.Request.CreateResponse("Alice") + if err != nil { + t.Fatalf("CreateResponse: %v", err) + } + go respondWhenPending(ctx, stream, resp) + } + case workflow.OutputEvent: + gotOutput = e.Output + } + } + + if got, want := gotOutput, any("Alice"); got != want { + t.Errorf("output = %v, want %v", got, want) + } + if startedCount != 2 { + t.Errorf("StartedEvent count = %d, want 2 (one per continuation cycle)", startedCount) + } + }) + } +} + func TestStreamingRun_WaitToTakeStreamDoesNotBlockOffThread(t *testing.T) { ex := minimalEchoBinding("ex") wf, err := workflow.NewBuilder(ex).WithOutputFrom(ex).Build() diff --git a/workflow/internal/execution/eventstream.go b/workflow/internal/execution/eventstream.go index 47bddab3..d7a3a209 100644 --- a/workflow/internal/execution/eventstream.go +++ b/workflow/internal/execution/eventstream.go @@ -549,6 +549,14 @@ func (l *lockstepRunEventStream) TakeEventStream(ctx context.Context, blockOnPen return } 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{}) + } } else { // No more work to do return From 4a099176d11ae6bec88a770e4f34c88d3c1fd42c Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 14:56:10 +0530 Subject: [PATCH 2/2] Drain continuation StartedEvent before supersteps; surface test errors --- workflow/inproc/events_test.go | 9 +++++++-- workflow/internal/execution/eventstream.go | 6 ++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/workflow/inproc/events_test.go b/workflow/inproc/events_test.go index e7fbccb4..9998aa78 100644 --- a/workflow/inproc/events_test.go +++ b/workflow/inproc/events_test.go @@ -5,6 +5,7 @@ package inproc_test import ( "context" "errors" + "fmt" "iter" "reflect" "slices" @@ -185,7 +186,10 @@ func askOnceBinding(id string) workflow.ExecutorBinding { }). AddHandlerRaw(reflect.TypeFor[*workflow.ExternalResponse](), nil, func(wctx *workflow.Context, msg any) (any, error) { resp := msg.(*workflow.ExternalResponse) - data, _ := resp.Data.As(port.Response) + data, ok := resp.Data.As(port.Response) + if !ok { + return nil, fmt.Errorf("response data is not %v", port.Response) + } return nil, wctx.YieldOutput(data) }) return rb, nil @@ -234,7 +238,8 @@ func TestStartedEvent_EmittedPerContinuationCycle(t *testing.T) { t.Fatalf("Build: %v", err) } - ctx := context.Background() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() stream, err := tc.env.RunStreaming(ctx, wf, "kick") if err != nil { t.Fatalf("RunStreaming: %v", err) diff --git a/workflow/internal/execution/eventstream.go b/workflow/internal/execution/eventstream.go index d7a3a209..579f336f 100644 --- a/workflow/internal/execution/eventstream.go +++ b/workflow/internal/execution/eventstream.go @@ -556,6 +556,12 @@ func (l *lockstepRunEventStream) TakeEventStream(ctx context.Context, blockOnPen // event is drained and yielded before the cycle's supersteps. if l.stepRunner.HasUnprocessedMessages() { l.eventQueue.Enqueue(workflow.StartedEvent{}) + // Drain immediately so the StartedEvent is yielded before the + // cycle's supersteps run, rather than being held in the queue + // and drained alongside the first superstep's events. + if !l.drainAndFilterEvents(linkedCtx, yield) { + return + } } } else { // No more work to do