.NET: [BREAKING] Workflows API Review Naming Changes (Part 1?) (#4090)

* refactor: Normalize Run/RunStreaming with AIAgent * refactor: Clarify Session vs. Run -level concepts * Rename RunId to SessionId to better match Run/Session terminology in AIAgent * [BREAKING]: Will break existing checkpointed sessions in CosmosDb due to field rename * refactor: Rename and simplify interface around getting typed data out of ExternalRequest/Response * Also adds hints around using value types in PortableValue * refactor: Rename AddFanInEdge to AddFanInBarrierEdge This will prevent a breaking change later when we introduce a programmable FanIn edge, analogous to the FanOut edge's EdgeSelector. The goal, in the long run is to support a number of different FanIn scenarios, with naive FanIn (no barrier) by default, similar to FanOut. * refactor: AsAgent(this Workflow, ...) => AsAIAgent(...) * misc - part1: SwitchBuilder internal --------- Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com>
2026-06-16 21:04:09 +08:00 · 2026-02-20 02:05:18 +00:00
parent 5fd260e11d
commit 0086d38f58
93 changed files with 397 additions and 458 deletions
@@ -44,38 +44,38 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
    /// <inheritdoc/>
    public bool IsCheckpointingEnabled => this.CheckpointManager != null;

-    internal ValueTask<AsyncRunHandle> BeginRunAsync(Workflow workflow, string? runId, IEnumerable<Type> knownValidInputTypes, CancellationToken cancellationToken)
+    internal ValueTask<AsyncRunHandle> BeginRunAsync(Workflow workflow, string? sessionId, IEnumerable<Type> knownValidInputTypes, CancellationToken cancellationToken)
    {
-        InProcessRunner runner = InProcessRunner.CreateTopLevelRunner(workflow, this.CheckpointManager, runId, this.EnableConcurrentRuns, knownValidInputTypes);
+        InProcessRunner runner = InProcessRunner.CreateTopLevelRunner(workflow, this.CheckpointManager, sessionId, this.EnableConcurrentRuns, knownValidInputTypes);
        return runner.BeginStreamAsync(this.ExecutionMode, cancellationToken);
    }

    internal ValueTask<AsyncRunHandle> ResumeRunAsync(Workflow workflow, CheckpointInfo fromCheckpoint, IEnumerable<Type> knownValidInputTypes, CancellationToken cancellationToken)
    {
-        InProcessRunner runner = InProcessRunner.CreateTopLevelRunner(workflow, this.CheckpointManager, fromCheckpoint.RunId, this.EnableConcurrentRuns, knownValidInputTypes);
+        InProcessRunner runner = InProcessRunner.CreateTopLevelRunner(workflow, this.CheckpointManager, fromCheckpoint.SessionId, this.EnableConcurrentRuns, knownValidInputTypes);
        return runner.ResumeStreamAsync(this.ExecutionMode, fromCheckpoint, cancellationToken);
    }

    /// <inheritdoc/>
-    public async ValueTask<StreamingRun> OpenStreamAsync(
+    public async ValueTask<StreamingRun> OpenStreamingAsync(
        Workflow workflow,
-        string? runId = null,
+        string? sessionId = null,
        CancellationToken cancellationToken = default)
    {
-        AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, runId, [], cancellationToken)
+        AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, sessionId, [], cancellationToken)
                                             .ConfigureAwait(false);

        return new(runHandle);
    }

    /// <inheritdoc/>
-    public async ValueTask<StreamingRun> StreamAsync<TInput>(
+    public async ValueTask<StreamingRun> RunStreamingAsync<TInput>(
        Workflow workflow,
        TInput input,
-        string? runId = null,
+        string? sessionId = null,
        CancellationToken cancellationToken = default) where TInput : notnull
    {
-        AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, runId, [], cancellationToken)
+        AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, sessionId, [], cancellationToken)
                                             .ConfigureAwait(false);

        return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false);
@@ -91,7 +91,7 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
    }

    /// <inheritdoc/>
-    public async ValueTask<StreamingRun> ResumeStreamAsync(
+    public async ValueTask<StreamingRun> ResumeStreamingAsync(
        Workflow workflow,
        CheckpointInfo fromCheckpoint,
        CancellationToken cancellationToken = default)
@@ -106,11 +106,11 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen

    private async ValueTask<AsyncRunHandle> BeginRunHandlingChatProtocolAsync<TInput>(Workflow workflow,
        TInput input,
-        string? runId = null,
+        string? sessionId = null,
        CancellationToken cancellationToken = default)
    {
        ProtocolDescriptor descriptor = await workflow.DescribeProtocolAsync(cancellationToken).ConfigureAwait(false);
-        AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, runId, descriptor.Accepts, cancellationToken)
+        AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, sessionId, descriptor.Accepts, cancellationToken)
                                             .ConfigureAwait(false);

        await runHandle.EnqueueMessageAsync(input, cancellationToken).ConfigureAwait(false);
@@ -127,13 +127,13 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
    public async ValueTask<Run> RunAsync<TInput>(
        Workflow workflow,
        TInput input,
-        string? runId = null,
+        string? sessionId = null,
        CancellationToken cancellationToken = default) where TInput : notnull
    {
        AsyncRunHandle runHandle = await this.BeginRunHandlingChatProtocolAsync(
                                                workflow,
                                                input,
-                                                runId,
+                                                sessionId,
                                                cancellationToken)
                                             .ConfigureAwait(false);

@@ -22,27 +22,27 @@ namespace Microsoft.Agents.AI.Workflows.InProc;
 /// scenarios where workflow execution does not require executor distribution. </para></remarks>
 internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
 {
-    public static InProcessRunner CreateTopLevelRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? runId = null, bool enableConcurrentRuns = false, IEnumerable<Type>? knownValidInputTypes = null)
+    public static InProcessRunner CreateTopLevelRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, bool enableConcurrentRuns = false, IEnumerable<Type>? knownValidInputTypes = null)
    {
        return new InProcessRunner(workflow,
                                   checkpointManager,
-                                   runId,
+                                   sessionId,
                                   enableConcurrentRuns: enableConcurrentRuns,
                                   knownValidInputTypes: knownValidInputTypes);
    }

-    public static InProcessRunner CreateSubworkflowRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? runId = null, object? existingOwnerSignoff = null, bool enableConcurrentRuns = false, IEnumerable<Type>? knownValidInputTypes = null)
+    public static InProcessRunner CreateSubworkflowRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, object? existingOwnerSignoff = null, bool enableConcurrentRuns = false, IEnumerable<Type>? knownValidInputTypes = null)
    {
        return new InProcessRunner(workflow,
                                   checkpointManager,
-                                   runId,
+                                   sessionId,
                                   existingOwnerSignoff: existingOwnerSignoff,
                                   enableConcurrentRuns: enableConcurrentRuns,
                                   knownValidInputTypes: knownValidInputTypes,
                                   subworkflow: true);
    }

-    private InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? runId = null, object? existingOwnerSignoff = null, bool subworkflow = false, bool enableConcurrentRuns = false, IEnumerable<Type>? knownValidInputTypes = null)
+    private InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, object? existingOwnerSignoff = null, bool subworkflow = false, bool enableConcurrentRuns = false, IEnumerable<Type>? knownValidInputTypes = null)
    {
        if (enableConcurrentRuns && !workflow.AllowConcurrent)
        {
@@ -50,11 +50,11 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
                $"not supporting concurrent: {string.Join(", ", workflow.NonConcurrentExecutorIds)}");
        }

-        this.RunId = runId ?? Guid.NewGuid().ToString("N");
+        this.SessionId = sessionId ?? Guid.NewGuid().ToString("N");
        this.StartExecutorId = workflow.StartExecutorId;

        this.Workflow = Throw.IfNull(workflow);
-        this.RunContext = new InProcessRunnerContext(workflow, this.RunId, checkpointingEnabled: checkpointManager != null, this.OutgoingEvents, this.StepTracer, existingOwnerSignoff, subworkflow, enableConcurrentRuns);
+        this.RunContext = new InProcessRunnerContext(workflow, this.SessionId, checkpointingEnabled: checkpointManager != null, this.OutgoingEvents, this.StepTracer, existingOwnerSignoff, subworkflow, enableConcurrentRuns);
        this.CheckpointManager = checkpointManager;

        this._knownValidInputTypes = knownValidInputTypes != null
@@ -65,8 +65,8 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
        this.EdgeMap = new EdgeMap(this.RunContext, this.Workflow.Edges, this.Workflow.Ports.Values, this.Workflow.StartExecutorId, this.StepTracer);
    }

-    /// <inheritdoc cref="ISuperStepRunner.RunId"/>
-    public string RunId { get; }
+    /// <inheritdoc cref="ISuperStepRunner.SessionId"/>
+    public string SessionId { get; }

    /// <inheritdoc cref="ISuperStepRunner.StartExecutorId"/>
    public string StartExecutorId { get; }
@@ -303,7 +303,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
        Dictionary<ScopeKey, PortableValue> stateData = await this.RunContext.StateManager.ExportStateAsync().ConfigureAwait(false);

        Checkpoint checkpoint = new(this.StepTracer.StepNumber, this._workflowInfoCache, runnerData, stateData, edgeData, this._lastCheckpointInfo);
-        this._lastCheckpointInfo = await this.CheckpointManager.CommitCheckpointAsync(this.RunId, checkpoint).ConfigureAwait(false);
+        this._lastCheckpointInfo = await this.CheckpointManager.CommitCheckpointAsync(this.SessionId, checkpoint).ConfigureAwait(false);
        this.StepTracer.TraceCheckpointCreated(this._lastCheckpointInfo);
        this._checkpoints.Add(this._lastCheckpointInfo);
    }
@@ -317,7 +317,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
            throw new InvalidOperationException("This run was not configured with a CheckpointManager, so it cannot restore checkpoints.");
        }

-        Checkpoint checkpoint = await this.CheckpointManager.LookupCheckpointAsync(this.RunId, checkpointInfo)
+        Checkpoint checkpoint = await this.CheckpointManager.LookupCheckpointAsync(this.SessionId, checkpointInfo)
                                                            .ConfigureAwait(false);

        // Validate the checkpoint is compatible with this workflow
@@ -346,7 +346,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
        async ValueTask UpdateCheckpointIndexAsync()
        {
            this._checkpoints.Clear();
-            this._checkpoints.AddRange(await this.CheckpointManager!.RetrieveIndexAsync(this.RunId).ConfigureAwait(false));
+            this._checkpoints.AddRange(await this.CheckpointManager!.RetrieveIndexAsync(this.SessionId).ConfigureAwait(false));
        }
    }

@@ -22,7 +22,7 @@ namespace Microsoft.Agents.AI.Workflows.InProc;
 internal sealed class InProcessRunnerContext : IRunnerContext
 {
    private int _runEnded;
-    private readonly string _runId;
+    private readonly string _sessionId;
    private readonly Workflow _workflow;
    private readonly object? _previousOwnership;
    private bool _ownsWorkflow;
@@ -40,7 +40,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext

    public InProcessRunnerContext(
        Workflow workflow,
-        string runId,
+        string sessionId,
        bool checkpointingEnabled,
        IEventSink outgoingEvents,
        IStepTracer? stepTracer,
@@ -61,7 +61,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext
        }

        this._workflow = workflow;
-        this._runId = runId;
+        this._sessionId = sessionId;

        this._edgeMap = new(this, this._workflow, stepTracer);
        this._outputFilter = new(workflow);
@@ -94,7 +94,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext
                throw new InvalidOperationException($"Executor with ID '{executorId}' is not registered.");
            }

-            Executor executor = await registration.CreateInstanceAsync(this._runId).ConfigureAwait(false);
+            Executor executor = await registration.CreateInstanceAsync(this._sessionId).ConfigureAwait(false);
            executor.AttachRequestContext(this.BindExternalRequestContext(executorId));

            await executor.InitializeAsync(this.BindWorkflowContext(executorId), cancellationToken: cancellationToken)
@@ -438,7 +438,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext
    {
        if (Volatile.Read(ref this._runEnded) == 1)
        {
-            throw new InvalidOperationException($"Workflow run '{this._runId}' has been ended. Please start a new Run or StreamingRun.");
+            throw new InvalidOperationException($"Workflow run for session '{this._sessionId}' has been ended. Please start a new Run or StreamingRun.");
        }
    }