From 5613c3269bcead3a64a03b285992d016810e9147 Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 09:19:03 +0530 Subject: [PATCH 1/2] Do not open a spurious empty WorkflowRun span on no-work wakeups The streaming run loop opened a WorkflowRun span (StartWorkflowRun + EventWorkflowStarted) unconditionally on every wakeup, and emitted EventWorkflowCompleted + End() unconditionally, even when the iteration had no unprocessed messages to run. No-work wakeups are reachable: the loop is signaled when only HasUnservicedRequests() is true, and checkpoint restore always signals the run loop regardless of queued messages. Each such wakeup produced a zero-superstep workflow_invoke span. Move span creation inside the HasUnprocessedMessages branch, co-located with the StartedEvent emission that was already guarded there, and only finalize the span when one was actually started (guard on nil runActivity). This mirrors the lockstep implementation, which already uses a nil runActivity guard, keeping the two paths consistent. --- workflow/internal/execution/eventstream.go | 16 +++++-- .../internal/execution/eventstream_test.go | 46 +++++++++++++++++++ 2 files changed, 58 insertions(+), 4 deletions(-) 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..06e119a7 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, mirroring a checkpoint restore + // whose only state is an unserviced request. 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) From 8e0a576b912077ab4db21793f9761bc867d0e053 Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 14:56:22 +0530 Subject: [PATCH 2/2] Align no-work wakeup test comment with actual runner state --- workflow/internal/execution/eventstream_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/workflow/internal/execution/eventstream_test.go b/workflow/internal/execution/eventstream_test.go index 06e119a7..55fb35cc 100644 --- a/workflow/internal/execution/eventstream_test.go +++ b/workflow/internal/execution/eventstream_test.go @@ -75,9 +75,9 @@ func TestStreamingRunEventStream_NoWorkWakeupDoesNotOpenWorkflowRunSpan(t *testi waitForEventHandlerCount(t, runner.outgoingEvents, 1) // Trigger a no-work wakeup: the run loop is signaled but - // HasUnprocessedMessages reports false, mirroring a checkpoint restore - // whose only state is an unserviced request. Such an iteration must not - // open a WorkflowRun span, matching the lockstep implementation. + // 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