.NET: [Bug]: A Workflow stays owned after an executor fails to initialize, so every later run fails #9584

Observed Behavior

If creating an executor fails during a run, because its InitializeAsync (or its factory) throws or is cancelled, the run never releases the Workflow. Disposing that run rethrows the initialization error, and every later run on the same Workflow instance fails with InvalidOperationException: Cannot use a Workflow that is already owned by another runner or parent workflow. until the process restarts.

There are two causes:

• InProcessRunnerContext.EndRunAsync (dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs:530-555) awaits every cached executor creation task (line 537) before it releases ownership (line 551). A creation task that faulted or was cancelled rethrows at line 537, so the method exits before ReleaseOwnershipAsync, and Run.DisposeAsync / StreamingRun.DisposeAsync throws the old error.
• InProcessExecutionEnvironment.RunStreamingAsync (InProcessExecutionEnvironment.cs:75-85) and BeginRunHandlingChatProtocolAsync (:136-152, used by RunAsync) take ownership in BeginRunAsync and then enqueue the first input. Enqueueing creates the start executor when the input type is not already known to be valid (InProcessRunner.cs:104), which RunStreamingAsync always hits. If that throws, the run handle is dropped without being disposed, so nothing ever ends the run.

The failure itself is reported (an ExecutorFailedEvent / WorkflowErrorEvent during the run, or the exception from RunStreamingAsync), but the workflow cannot be used again. A workflow is usually built once and reused, and AddWorkflow registers it as a singleton by default, so one transient failure, such as a timeout while an executor connects to a backend in InitializeAsync or a request that is abandoned while an executor is still initializing, leaves that workflow unusable for the rest of the process.

Expected Behavior

Ending a run releases the workflow even when one of its executors failed to initialize, and disposing the run does not rethrow an error that was already reported. The next run on the same workflow starts normally.

Steps to Reproduce

1. Build a workflow once with an executor that overrides InitializeAsync and fails on its first call (a timeout opening a connection, say).
2. Run it with InProcessExecution.RunStreamingAsync or RunAsync, and dispose the run.
3. Run the same workflow again.

The same happens when the first run is disposed while that executor's InitializeAsync is still awaiting the cancellation token it was given.

Minimal Reproduction

using Microsoft.Agents.AI.Workflows;

Console.WriteLine("== InitializeAsync throws once");
await RunThreeRequestsAsync(new SearchExecutor(firstInit: FirstInit.Throws), abandonFirst: false);

Console.WriteLine("== first request abandoned while InitializeAsync is still running");
await RunThreeRequestsAsync(new SearchExecutor(firstInit: FirstInit.WaitsForCancellation), abandonFirst: true);

static async Task RunThreeRequestsAsync(SearchExecutor search, bool abandonFirst)
{
    // Built once and reused for every request.
    IntakeExecutor intake = new();
    Workflow workflow = new WorkflowBuilder(intake).AddEdge(intake, search).WithOutputFrom(search).Build();

    for (int request = 1; request <= 3; request++)
    {
        try
        {
            await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, $"question {request}");
            if (abandonFirst && request == 1)
            {
                await search.FirstInitStarted.Task;
                await run.DisposeAsync(); // the client disconnected
                Console.WriteLine($"request {request}: abandoned");
                continue;
            }

            await foreach (WorkflowEvent evt in run.WatchStreamAsync())
            {
                if (evt is WorkflowOutputEvent output)
                {
                    Console.WriteLine($"request {request}: {output.Data}");
                }
                else if (evt is WorkflowErrorEvent or ExecutorFailedEvent)
                {
                    Console.WriteLine($"request {request}: {evt.GetType().Name}");
                }
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"request {request}: {ex.GetType().Name}: {ex.Message}");
        }
    }
}

internal enum FirstInit { Throws, WaitsForCancellation }

[SendsMessage(typeof(string))]
internal sealed class IntakeExecutor() : Executor<string>("intake")
{
    public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
        => context.SendMessageAsync(message, cancellationToken: cancellationToken);
}

[YieldsOutput(typeof(string))]
internal sealed class SearchExecutor(FirstInit firstInit) : Executor<string>("search")
{
    private int _initCalls;

    public TaskCompletionSource FirstInitStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);

    protected override async ValueTask InitializeAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Open a connection to the search index.
        if (Interlocked.Increment(ref this._initCalls) > 1)
        {
            return;
        }

        this.FirstInitStarted.SetResult();
        if (firstInit == FirstInit.Throws)
        {
            throw new TimeoutException("Connecting to the search index timed out.");
        }

        await Task.Delay(Timeout.Infinite, cancellationToken);
    }

    public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
        => context.YieldOutputAsync($"results for {message}", cancellationToken);
}

Output with Microsoft.Agents.AI.Workflows 1.25.0, and the same with a project reference to main at 0200599:

== InitializeAsync throws once
request 1: ExecutorFailedEvent
request 1: WorkflowErrorEvent
request 1: TimeoutException: Connecting to the search index timed out.
request 2: InvalidOperationException: Cannot use a Workflow that is already owned by another runner or parent workflow.
request 3: InvalidOperationException: Cannot use a Workflow that is already owned by another runner or parent workflow.
== first request abandoned while InitializeAsync is still running
request 1: TaskCanceledException: A task was canceled.
request 2: InvalidOperationException: Cannot use a Workflow that is already owned by another runner or parent workflow.
request 3: InvalidOperationException: Cannot use a Workflow that is already owned by another runner or parent workflow.

Expected: requests 2 and 3 print results for question 2 and results for question 3, and disposing request 1 does not throw.

Error Messages and Stack Traces

The first request's dispose (1.25.0):

System.TimeoutException: Connecting to the search index timed out.
   at SearchExecutor.InitializeAsync(IWorkflowContext context, CancellationToken cancellationToken) in Program.cs:line 74
   at Microsoft.Agents.AI.Workflows.InProc.InProcessRunnerContext.<>c__DisplayClass16_0.<<EnsureExecutorAsync>g__CreateExecutorAsync|0>d.MoveNext()
--- End of stack trace from previous location ---
   at Microsoft.Agents.AI.Workflows.InProc.InProcessRunnerContext.EndRunAsync()
   at Microsoft.Agents.AI.Workflows.Execution.AsyncRunHandle.DisposeAsync()
   at Program.<<Main>$>g__RunThreeRequestsAsync|0_0(SearchExecutor search, Boolean abandonFirst) in Program.cs:line 28

Every later request:

System.InvalidOperationException: Cannot use a Workflow that is already owned by another runner or parent workflow.
   at Microsoft.Agents.AI.Workflows.Workflow.TakeOwnership(Object ownerToken, Boolean subworkflow, Object existingOwnershipSignoff)
   at Microsoft.Agents.AI.Workflows.InProc.InProcessRunnerContext..ctor(Workflow workflow, String sessionId, Boolean checkpointingEnabled, IEventSink outgoingEvents, IStepTracer stepTracer, Object existingOwnershipSignoff, Boolean subworkflow, Boolean enableConcurrentRuns, ILogger logger)
   at Microsoft.Agents.AI.Workflows.InProc.InProcessRunner..ctor(Workflow workflow, ICheckpointManager checkpointManager, String sessionId, Object existingOwnerSignoff, Boolean subworkflow, Boolean enableConcurrentRuns, IEnumerable`1 knownValidInputTypes)
   at Microsoft.Agents.AI.Workflows.InProc.InProcessRunner.CreateTopLevelRunner(Workflow workflow, ICheckpointManager checkpointManager, String sessionId, Boolean enableConcurrentRuns, IEnumerable`1 knownValidInputTypes)
   at Microsoft.Agents.AI.Workflows.InProc.InProcessExecutionEnvironment.BeginRunAsync(Workflow workflow, String sessionId, IEnumerable`1 knownValidInputTypes, CancellationToken cancellationToken)
   at Microsoft.Agents.AI.Workflows.InProc.InProcessExecutionEnvironment.RunStreamingAsync[TInput](Workflow workflow, TInput input, String sessionId, CancellationToken cancellationToken)
   at Program.<<Main>$>g__RunThreeRequestsAsync|0_0(SearchExecutor search, Boolean abandonFirst) in Program.cs:line 19

Package Versions

Microsoft.Agents.AI.Workflows: 1.25.0 (also reproduced on 1.0.0 and on main at 0200599)

.NET Version

.NET 10.0 (SDK 10.0.402)

Operating System

macOS

Regression

No

Additional Context

The same thing happens with RunAsync. When the start executor is the one that fails to initialize while RunStreamingAsync (or RunAsync, for an input whose type the start executor does not declare) enqueues the first input, the call itself throws the initialization error and the second request fails the same way.

Skipping executors whose creation task faulted or was cancelled in EndRunAsync, releasing ownership in a finally, and disposing the run handle when enqueueing the first input throws fixes all of these. I am opening a PR with this fix and tests.

Acknowledgements

• I searched existing issues and did not find a duplicate.
• I personally verified this behavior and the reproduction details are authentic.
• I will wait for explicit maintainer agreement before starting implementation of a non-trivial change.