fix(a2a): continuations emit the resumed task, guarded clients bypass proxies
A continued task (input-required -> new message) produced only status updates, so the SDK answered 'Agent handler did not produce any response events'. The handler now enqueues the resumed task (caller message and question in history, working) as the first event. The guarded SocketsHttpHandler sets UseProxy = false: a proxy would make the connection and skip the ConnectCallback address check.
This commit is contained in:
@@ -40,7 +40,7 @@ internal sealed class A2AAgentHandler(
|
||||
var token = runCts.Token;
|
||||
try
|
||||
{
|
||||
result = await RunAsync(context, updater, token).ConfigureAwait(false);
|
||||
result = await RunAsync(context, eventQueue, updater, token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (token.IsCancellationRequested)
|
||||
{
|
||||
@@ -57,10 +57,18 @@ internal sealed class A2AAgentHandler(
|
||||
Log(context, label, started);
|
||||
}
|
||||
|
||||
private async Task<Result> RunAsync(RequestContext context, TaskUpdater updater, CancellationToken token)
|
||||
private async Task<Result> RunAsync(RequestContext context, AgentEventQueue eventQueue, TaskUpdater updater, CancellationToken token)
|
||||
{
|
||||
if (context.Task is null) await updater.SubmitAsync(token).ConfigureAwait(false);
|
||||
await updater.StartWorkAsync(null, token).ConfigureAwait(false);
|
||||
if (context.Task is null)
|
||||
{
|
||||
await updater.SubmitAsync(token).ConfigureAwait(false);
|
||||
await updater.StartWorkAsync(null, token).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
// The SDK answers a request with the first Task event it sees, and a continuation produces none of its own.
|
||||
await eventQueue.EnqueueTaskAsync(Resumed(context), token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var hops = Hops.Read(context.Metadata, context.Message.Metadata);
|
||||
if (hops > options.Value.MaxHops)
|
||||
@@ -119,6 +127,21 @@ internal sealed class A2AAgentHandler(
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The continued task as the store holds it after the SDK appended the caller's message, moved to working: the
|
||||
/// question (status message) goes to history, as a status update would put it.
|
||||
/// </summary>
|
||||
private AgentTask Resumed(RequestContext context)
|
||||
{
|
||||
var json = System.Text.Json.JsonSerializer.Serialize(context.Task!, A2AJsonUtilities.DefaultOptions);
|
||||
var task = System.Text.Json.JsonSerializer.Deserialize<AgentTask>(json, A2AJsonUtilities.DefaultOptions)!;
|
||||
var history = task.History ??= [];
|
||||
if (!history.Any(m => m.MessageId == context.Message.MessageId)) history.Add(context.Message);
|
||||
if (task.Status.Message is { } question) history.Add(question);
|
||||
task.Status = new global::A2A.TaskStatus { State = TaskState.Working, Timestamp = time.GetUtcNow() };
|
||||
return task;
|
||||
}
|
||||
|
||||
/// <summary>Write the final status (and answer artifact) with a bounded token; returns the label to log.</summary>
|
||||
private async Task<string> WriteFinalAsync(RequestContext context, TaskUpdater updater, Result result)
|
||||
{
|
||||
|
||||
@@ -69,7 +69,7 @@ internal sealed class PushUrlGuard(IOptions<A2AHostOptions> options, Func<string
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A handler for caller-chosen URLs: no redirects, and every connection goes only to addresses checked with
|
||||
/// A handler for caller-chosen URLs: no redirects, no proxy, and every connection goes only to addresses checked with
|
||||
/// <see cref="IsPrivate"/> at connect time (so DNS rebinding can't reach internal hosts), unless
|
||||
/// <paramref name="allowPrivate"/> returns true.
|
||||
/// </summary>
|
||||
@@ -80,6 +80,7 @@ internal sealed class PushUrlGuard(IOptions<A2AHostOptions> options, Func<string
|
||||
return new SocketsHttpHandler
|
||||
{
|
||||
AllowAutoRedirect = false,
|
||||
UseProxy = false, // a proxy would make the connection, bypassing the ConnectCallback check
|
||||
PooledConnectionLifetime = TimeSpan.FromMinutes(5),
|
||||
ConnectCallback = async (context, cancellationToken) =>
|
||||
{
|
||||
|
||||
@@ -63,6 +63,36 @@ public class PushUrlGuardAddressTests
|
||||
public void GuardedHandler_DoesNotFollowRedirects() =>
|
||||
Assert.False(PushUrlGuard.CreateGuardedHandler(() => false).AllowAutoRedirect);
|
||||
|
||||
[Fact]
|
||||
public void GuardedHandler_BypassesProxies() =>
|
||||
Assert.False(PushUrlGuard.CreateGuardedHandler(() => false).UseProxy);
|
||||
|
||||
[Fact]
|
||||
public async Task GuardedHandler_AllowPrivate_ConnectsToLoopback()
|
||||
{
|
||||
using var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
var server = Task.Run(async () =>
|
||||
{
|
||||
using var client = await listener.AcceptTcpClientAsync();
|
||||
var stream = client.GetStream();
|
||||
var buffer = new byte[4096];
|
||||
var request = "";
|
||||
while (!request.Contains("\r\n\r\n"))
|
||||
{
|
||||
var read = await stream.ReadAsync(buffer);
|
||||
if (read == 0) break;
|
||||
request += System.Text.Encoding.ASCII.GetString(buffer, 0, read);
|
||||
}
|
||||
await stream.WriteAsync("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"u8.ToArray());
|
||||
});
|
||||
using var http = new HttpClient(PushUrlGuard.CreateGuardedHandler(() => true,
|
||||
(_, _) => Task.FromResult(new[] { IPAddress.Loopback })));
|
||||
Assert.Equal("ok", await http.GetStringAsync($"http://hooks.example:{port}/x"));
|
||||
await server;
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("tok\r\nX-Evil: 1", null, "token")]
|
||||
[InlineData(null, "Bearer x", "scheme")]
|
||||
|
||||
Reference in New Issue
Block a user