Emit StartedEvent per continuation cycle in the lockstep event stream

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()

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