diff --git a/workflow/internal/execution/eventstream.go b/workflow/internal/execution/eventstream.go index 47bddab3..d0b96dcb 100644 --- a/workflow/internal/execution/eventstream.go +++ b/workflow/internal/execution/eventstream.go @@ -176,8 +176,8 @@ func (s *streamingRunEventStream) runLoop() { default: } - cycleCtx, runActivity := telemetry.StartWorkflowRun(ctx, workflowMetadata(wf, s.stepRunner.SessionID())) - runActivity.AddEvent(observability.EventWorkflowStarted) + cycleCtx := ctx + var runActivity *observability.Activity // Run all available supersteps continuously // Events are streamed out in real-time as they happen via the event handler @@ -189,6 +189,12 @@ func (s *streamingRunEventStream) runLoop() { // (e.g. Run.RunToNextHalt returning after reading an Idle halt signal). s.setStatus(RunStatusRunning) + // Open the WorkflowRun span only when there's actual work to + // process, to avoid spurious zero-superstep spans on no-work loop + // iterations. This mirrors the lockstep implementation. + cycleCtx, runActivity = telemetry.StartWorkflowRun(ctx, workflowMetadata(wf, s.stepRunner.SessionID())) + runActivity.AddEvent(observability.EventWorkflowStarted) + // Emit StartedEvent only when there's actual work to process, // to avoid spurious events on no-work loop iterations. s.sendEvent(cycleCtx, workflow.StartedEvent{}) @@ -213,8 +219,10 @@ func (s *streamingRunEventStream) runLoop() { } } } - runActivity.AddEvent(observability.EventWorkflowCompleted) - runActivity.End() + if runActivity != nil { + runActivity.AddEvent(observability.EventWorkflowCompleted) + runActivity.End() + } // Update status based on what's waiting if s.stepRunner.HasUnservicedRequests() { diff --git a/workflow/internal/execution/eventstream_test.go b/workflow/internal/execution/eventstream_test.go index dd955a7f..55fb35cc 100644 --- a/workflow/internal/execution/eventstream_test.go +++ b/workflow/internal/execution/eventstream_test.go @@ -10,6 +10,8 @@ import ( "time" "github.com/microsoft/agent-framework-go/workflow" + "github.com/microsoft/agent-framework-go/workflow/internal/observability" + "github.com/microsoft/agent-framework-go/workflow/internal/workflowtest" ) func TestStreamingRunEventStream_RemovesEventHandlerOnStop(t *testing.T) { @@ -55,6 +57,46 @@ func TestStreamingRunEventStream_ErrorEventCancelsRunLoop(t *testing.T) { waitForEventHandlerCount(t, runner.outgoingEvents, 0) } +func TestStreamingRunEventStream_NoWorkWakeupDoesNotOpenWorkflowRunSpan(t *testing.T) { + tracer := workflowtest.NewRecordingTracer() + wf, err := workflow.NewBuilder((&workflow.Executor{ID: "start", ImplementationID: "workflow_test.start"}).Bind()). + WithTelemetry(tracer, workflow.TelemetryOptions{}). + Build() + if err != nil { + t.Fatalf("Build: %v", err) + } + + runner := newTestSuperStepRunner() + runner.wf = wf + stream := newStreamingRunEventStream(runner, false) + + stream.Start() + defer stream.Stop() + waitForEventHandlerCount(t, runner.outgoingEvents, 1) + + // Trigger a no-work wakeup: the run loop is signaled but + // HasUnprocessedMessages reports false, so the iteration has nothing to + // process. Such an iteration must not open a WorkflowRun span, matching + // the lockstep implementation. + stream.SignalInput() + + // The loop enqueues an internalHaltSignal once the (no-work) cycle + // completes; use it to synchronize before asserting on spans. + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + evt, ok := stream.nextEvent(ctx) + if !ok { + t.Fatal("run loop did not complete the no-work iteration") + } + if _, ok := evt.(*internalHaltSignal); !ok { + t.Fatalf("unexpected event on no-work wakeup: %#v", evt) + } + + if count := workflowtest.CountSpansWithPrefix(tracer.Spans(), observability.ActivityWorkflowInvoke); count != 0 { + t.Fatalf("workflow_invoke span count = %d, want 0 on no-work wakeup", count) + } +} + func waitForEventHandlerCount(t *testing.T, sink *ConcurrentEventSink, want int) { t.Helper() deadline := time.After(time.Second) @@ -75,6 +117,7 @@ func waitForEventHandlerCount(t *testing.T, sink *ConcurrentEventSink, want int) type testSuperStepRunner struct { outgoingEvents *ConcurrentEventSink + wf *workflow.Workflow } func newTestSuperStepRunner() *testSuperStepRunner { @@ -84,6 +127,9 @@ func newTestSuperStepRunner() *testSuperStepRunner { func (r *testSuperStepRunner) SessionID() string { return "session" } func (r *testSuperStepRunner) Workflow() *workflow.Workflow { + if r.wf != nil { + return r.wf + } wf, err := workflow.NewBuilder((&workflow.Executor{ID: "start", ImplementationID: "workflow_test.start"}).Bind()).Build() if err != nil { panic(err)