.NET: Pass through external input request and handle response conversion for workflow as agent scenario

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs

@@ -53,6 +53,9 @@ internal AsyncRunHandle(ISuperStepRunner stepRunner, ICheckpointingHandle checkp
     public ValueTask<RunStatus> GetStatusAsync(CancellationToken cancellationToken = default)
         => this._eventStream.GetStatusAsync(cancellationToken);
 
+    internal bool TryGetResponsePortExecutorId(string portId, out string? executorId)
+        => this._stepRunner.TryGetResponsePortExecutorId(portId, out executorId);
+
     public async IAsyncEnumerable<WorkflowEvent> TakeEventStreamAsync(bool blockOnPendingRequest, [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         //Debug.Assert(breakOnHalt);

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/EdgeMap.cs

@@ -3,6 +3,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -95,6 +96,18 @@ public bool TryRegisterPort(IRunnerContext runContext, string executorId, Reques
         return portRunner.ChaseEdgeAsync(new MessageEnvelope(response, ExecutorIdentity.None), this._stepTracer, cancellationToken);
     }
 
+    internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId)
+    {
+        if (this._portEdgeRunners.TryGetValue(portId, out ResponseEdgeRunner? portRunner))
+        {
+            executorId = portRunner.ExecutorId;
+            return true;
+        }
+
+        executorId = null;
+        return false;
+    }
+
     internal async ValueTask<Dictionary<EdgeId, PortableValue>> ExportStateAsync()
     {
         Dictionary<EdgeId, PortableValue> exportedStates = [];

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs

@@ -19,6 +19,7 @@ internal interface ISuperStepRunner
     bool HasUnprocessedMessages { get; }
 
     ValueTask EnqueueResponseAsync(ExternalResponse response, CancellationToken cancellationToken = default);
+    bool TryGetResponsePortExecutorId(string portId, out string? executorId);
 
     ValueTask<bool> IsValidInputTypeAsync<T>(CancellationToken cancellationToken = default);
     ValueTask<bool> EnqueueMessageAsync<T>(T message, CancellationToken cancellationToken = default);

dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs

@@ -160,6 +160,8 @@ public async ValueTask<AsyncRunHandle> ResumeStreamAsync(ExecutionMode mode, Che
 
     bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests;
     bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions;
+    bool ISuperStepRunner.TryGetResponsePortExecutorId(string portId, out string? executorId)
+        => this.RunContext.TryGetResponsePortExecutorId(portId, out executorId);
 
     public bool IsCheckpointingEnabled => this.RunContext.IsCheckpointingEnabled;
 

dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs

@@ -296,6 +296,9 @@ public bool CompleteRequest(string requestId)
         return this._externalRequests.TryRemove(requestId, out _);
     }
 
+    internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId)
+        => this._edgeMap.TryGetResponsePortExecutorId(portId, out executorId);
+
     private IEventSink OutgoingEvents { get; }
 
     internal StateManager StateManager { get; } = new();

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs

@@ -68,10 +68,17 @@ private ValueTask HandleUserInputResponseAsync(
             throw new InvalidOperationException($"No pending ToolApprovalRequest found with id '{response.RequestId}'.");
         }
 
-        List<ChatMessage> implicitTurnMessages = [new ChatMessage(ChatRole.User, [response])];
+        // Merge the external response with any already-buffered regular messages so mixed-content
+        // resumes can be processed in one invocation.
+        return this.ProcessTurnMessagesAsync(async (pendingMessages, ctx, ct) =>
+        {
+            pendingMessages.Add(new ChatMessage(ChatRole.User, [response]));
+
+            await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false);
 
-        // ContinueTurnAsync owns failing to emit a TurnToken if this response does not clear up all remaining outstanding requests.
-        return this.ContinueTurnAsync(implicitTurnMessages, context, this._currentTurnEmitEvents ?? false, cancellationToken);
+            // Clear the buffered turn messages because they were consumed by ContinueTurnAsync.
+            return null;
+        }, context, cancellationToken);
     }
 
     private ValueTask HandleFunctionResultAsync(
@@ -84,8 +91,17 @@ private ValueTask HandleFunctionResultAsync(
             throw new InvalidOperationException($"No pending FunctionCall found with id '{result.CallId}'.");
         }
 
-        List<ChatMessage> implicitTurnMessages = [new ChatMessage(ChatRole.Tool, [result])];
-        return this.ContinueTurnAsync(implicitTurnMessages, context, this._currentTurnEmitEvents ?? false, cancellationToken);
+        // Merge the external response with any already-buffered regular messages so mixed-content
+        // resumes can be processed in one invocation.
+        return this.ProcessTurnMessagesAsync(async (pendingMessages, ctx, ct) =>
+        {
+            pendingMessages.Add(new ChatMessage(ChatRole.Tool, [result]));
+
+            await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false);
+
+            // Clear the buffered turn messages because they were consumed by ContinueTurnAsync.
+            return null;
+        }, context, cancellationToken);
     }
 
     public bool ShouldEmitStreamingEvents(bool? emitEvents)
@@ -198,7 +214,7 @@ await this.EnsureSessionAsync(context, cancellationToken).ConfigureAwait(false),
             ExtractUnservicedRequests(response.Messages.SelectMany(message => message.Contents));
         }
 
-        if (this._options.EmitAgentResponseEvents == true)
+        if (this._options.EmitAgentResponseEvents)
         {
             await context.YieldOutputAsync(response, cancellationToken).ConfigureAwait(false);
         }

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIContentExternalHandler.cs

@@ -16,10 +16,12 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
     where TResponseContent : AIContent
 {
     private readonly PortBinding? _portBinding;
+    private readonly string _portId;
     private ConcurrentDictionary<string, TRequestContent> _pendingRequests = new();
 
     public AIContentExternalHandler(ref ProtocolBuilder protocolBuilder, string portId, bool intercepted, Func<TResponseContent, IWorkflowContext, CancellationToken, ValueTask> handler)
     {
+        this._portId = portId;
         PortBinding? portBinding = null;
         protocolBuilder = protocolBuilder.ConfigureRoutes(routeBuilder => ConfigureRoutes(routeBuilder, out portBinding));
         this._portBinding = portBinding;
@@ -58,12 +60,14 @@ public ValueTask ProcessRequestContentAsync(string id, TRequestContent requestCo
     {
         if (!this._pendingRequests.TryAdd(id, requestContent))
         {
-            throw new InvalidOperationException($"A pending request with ID '{id}' already exists.");
+            // Request is already pending; treat as an idempotent re-emission.
+            // Do not repost to the sink because request IDs must remain unique while pending.
+            return default;
         }
 
         return this.IsIntercepted
              ? context.SendMessageAsync(requestContent, cancellationToken: cancellationToken)
-             : this._portBinding.PostRequestAsync(requestContent, id, cancellationToken);
+             : this._portBinding.PostRequestAsync(requestContent, this.CreateExternalRequestId(id), cancellationToken);
     }
 
     public bool MarkRequestAsHandled(string id)
@@ -74,6 +78,8 @@ public bool MarkRequestAsHandled(string id)
     [MemberNotNullWhen(false, nameof(_portBinding))]
     private bool IsIntercepted => this._portBinding == null;
 
+    private string CreateExternalRequestId(string requestId) => $"{this._portId.Length}:{this._portId}:{requestId}";
+
     private static string MakeKey(string id) => $"{id}_PendingRequests";
 
     public async ValueTask OnCheckpointingAsync(string id, IWorkflowContext context, CancellationToken cancellationToken = default)

dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs

@@ -60,6 +60,9 @@ public ValueTask<bool> TrySendMessageAsync<TMessage>(TMessage message)
     internal ValueTask<bool> TrySendMessageUntypedAsync(object message, Type? declaredType = null)
         => this._runHandle.EnqueueMessageUntypedAsync(message, declaredType);
 
+    internal bool TryGetResponsePortExecutorId(string portId, out string? executorId)
+        => this._runHandle.TryGetResponsePortExecutorId(portId, out executorId);
+
     /// <summary>
     /// Asynchronously streams workflow events as they occur during workflow execution.
     /// </summary>