diff --git a/workflow/inproc/events_test.go b/workflow/inproc/events_test.go index 0194a791..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" @@ -156,6 +157,129 @@ 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, 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 + }, + }, 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, 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) + } + 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..579f336f 100644 --- a/workflow/internal/execution/eventstream.go +++ b/workflow/internal/execution/eventstream.go @@ -549,6 +549,20 @@ 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{}) + // 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 return