feat(a2a): map one A2A task to one agent session run
This commit is contained in:
EugeneTes
114
src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs
Normal file
114
src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs
Normal file
@@ -0,0 +1,114 @@
|
||||
using System.Diagnostics;
|
||||
using A2A;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Volcanic.AgenticCore.A2A.Server.Callers;
|
||||
using Volcanic.AgenticCore.Abstractions.Processes;
|
||||
|
||||
namespace Volcanic.AgenticCore.A2A.Server.Agent;
|
||||
|
||||
/// <summary>
|
||||
/// One task = one run of the caller context's session. Ends completed (final text as artifact <c>response</c>),
|
||||
/// input-required (the agent called <c>request_input</c>), failed (model error, hop limit, bad file) or canceled.
|
||||
/// </summary>
|
||||
internal sealed class A2AAgentHandler(
|
||||
A2ACaller caller, A2ASessionPool pool, IOptions<A2AHostOptions> options, ISecretRedactor redactor, HttpClient http,
|
||||
TimeProvider time, ILogger<A2AAgentHandler> logger) : IAgentHandler
|
||||
{
|
||||
public async Task ExecuteAsync(RequestContext context, AgentEventQueue eventQueue, CancellationToken cancellationToken)
|
||||
{
|
||||
var updater = new TaskUpdater(eventQueue, context.TaskId, context.ContextId);
|
||||
var started = Stopwatch.GetTimestamp();
|
||||
if (context.Task is null) await updater.SubmitAsync(cancellationToken).ConfigureAwait(false);
|
||||
await updater.StartWorkAsync(null, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var hops = Hops.Read(context.Metadata, context.Message.Metadata);
|
||||
if (hops > options.Value.MaxHops)
|
||||
{
|
||||
await updater.FailAsync(A2AText.Agent($"hop limit reached ({hops} > {options.Value.MaxHops})"), CancellationToken.None).ConfigureAwait(false);
|
||||
Log(context, "failed (hop limit)", started);
|
||||
return;
|
||||
}
|
||||
|
||||
A2ASessionPool.Lease lease;
|
||||
try
|
||||
{
|
||||
lease = await pool.AcquireAsync(caller, context.ContextId, context.TaskId, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (A2AException ex)
|
||||
{
|
||||
await updater.FailAsync(A2AText.Agent(ex.Message), CancellationToken.None).ConfigureAwait(false);
|
||||
Log(context, "failed (busy)", started);
|
||||
return;
|
||||
}
|
||||
|
||||
using (lease)
|
||||
{
|
||||
string prompt;
|
||||
try
|
||||
{
|
||||
prompt = await A2APrompt.BuildAsync(caller, context.Message, options.Value.MaxFileBytes, http, time, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (A2APromptException ex)
|
||||
{
|
||||
await updater.FailAsync(A2AText.Agent(ex.Message), CancellationToken.None).ConfigureAwait(false);
|
||||
Log(context, "failed (file)", started);
|
||||
return;
|
||||
}
|
||||
|
||||
var mapper = new A2ARunMapper(updater, redactor, cancellationToken);
|
||||
lease.Extension.BeginRun(hops);
|
||||
lease.Channel.Bind(updater);
|
||||
try
|
||||
{
|
||||
using var subscription = lease.Session.Subscribe(mapper.OnEvent);
|
||||
using var abort = cancellationToken.Register(() => _ = lease.Session.AbortAsync());
|
||||
await lease.Session.PromptAsync(prompt, cancellationToken: CancellationToken.None).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
lease.Channel.Bind(null);
|
||||
await mapper.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
string state;
|
||||
if (cancellationToken.IsCancellationRequested || mapper.Aborted)
|
||||
{
|
||||
await updater.CancelAsync(CancellationToken.None).ConfigureAwait(false);
|
||||
state = "canceled";
|
||||
}
|
||||
else if (mapper.Error is { } error)
|
||||
{
|
||||
await updater.FailAsync(A2AText.Agent(error), CancellationToken.None).ConfigureAwait(false);
|
||||
state = "failed";
|
||||
}
|
||||
else
|
||||
{
|
||||
if (mapper.FinalText is { } text)
|
||||
await updater.AddArtifactAsync([Part.FromText(text)], A2ARunMapper.ResponseArtifactId, "response", null,
|
||||
lastChunk: true, append: false, CancellationToken.None).ConfigureAwait(false);
|
||||
if (lease.Extension.PendingQuestion is { } question)
|
||||
{
|
||||
await updater.RequireInputAsync(A2AText.Agent(question), CancellationToken.None).ConfigureAwait(false);
|
||||
state = "input-required";
|
||||
}
|
||||
else
|
||||
{
|
||||
await updater.CompleteAsync(null, CancellationToken.None).ConfigureAwait(false);
|
||||
state = "completed";
|
||||
}
|
||||
}
|
||||
Log(context, state, started);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task CancelAsync(RequestContext context, AgentEventQueue eventQueue, CancellationToken cancellationToken)
|
||||
{
|
||||
await pool.AbortAsync(context.TaskId).ConfigureAwait(false);
|
||||
await new TaskUpdater(eventQueue, context.TaskId, context.ContextId).CancelAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private void Log(RequestContext context, string state, long started) =>
|
||||
logger.LogInformation("a2a task {TaskId} owner={Owner} caller={Caller} state={State} in {Duration}",
|
||||
context.TaskId, caller.OwnerId, caller.Name, state, Stopwatch.GetElapsedTime(started));
|
||||
}
|
||||
92
src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs
Normal file
92
src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs
Normal file
@@ -0,0 +1,92 @@
|
||||
using System.Threading.Channels;
|
||||
using A2A;
|
||||
using Volcanic.AgenticCore.Abstractions.Events;
|
||||
using Volcanic.AgenticCore.Abstractions.Messages;
|
||||
using Volcanic.AgenticCore.Abstractions.Processes;
|
||||
using Volcanic.AgenticCore.Abstractions.Providers;
|
||||
|
||||
namespace Volcanic.AgenticCore.A2A.Server.Agent;
|
||||
|
||||
/// <summary>
|
||||
/// Maps session events (raised synchronously on the loop) to task updates, written in order by a pump so the loop never
|
||||
/// waits. Text deltas append to artifact <c>response</c>; tool starts become working-status messages. Tracks how the run
|
||||
/// ended: the last assistant message decides (a retried failure followed by success counts as success).
|
||||
/// </summary>
|
||||
internal sealed class A2ARunMapper : IAsyncDisposable
|
||||
{
|
||||
public const string ResponseArtifactId = "response";
|
||||
|
||||
private readonly Channel<Func<ValueTask>> _work = Channel.CreateUnbounded<Func<ValueTask>>(new UnboundedChannelOptions { SingleReader = true });
|
||||
private readonly TaskUpdater _updater;
|
||||
private readonly ISecretRedactor _redactor;
|
||||
private readonly CancellationToken _ct;
|
||||
private readonly Task _pump;
|
||||
|
||||
public A2ARunMapper(TaskUpdater updater, ISecretRedactor redactor, CancellationToken cancellationToken)
|
||||
{
|
||||
_updater = updater;
|
||||
_redactor = redactor;
|
||||
_ct = cancellationToken;
|
||||
_pump = Task.Run(PumpAsync, CancellationToken.None);
|
||||
}
|
||||
|
||||
public string? FinalText { get; private set; }
|
||||
public string? Error { get; private set; }
|
||||
public bool Aborted { get; private set; }
|
||||
|
||||
public void OnEvent(AgentSessionEvent e)
|
||||
{
|
||||
switch (e)
|
||||
{
|
||||
case AgentSessionEvent.MessageUpdate { AssistantMessageEvent: AssistantMessageEvent.TextDelta delta }:
|
||||
Post(() => _updater.AddArtifactAsync([Part.FromText(delta.Delta)], ResponseArtifactId, "response", null,
|
||||
lastChunk: false, append: true, _ct));
|
||||
break;
|
||||
case AgentSessionEvent.MessageEnd { Message: AssistantMessage message }:
|
||||
if (message.StopReason == StopReason.Aborted)
|
||||
{
|
||||
Aborted = true;
|
||||
}
|
||||
else if (message.IsFailure)
|
||||
{
|
||||
Error = _redactor.Redact(message.ErrorMessage ?? "The model call failed.");
|
||||
}
|
||||
else
|
||||
{
|
||||
Error = null;
|
||||
if (!string.IsNullOrWhiteSpace(message.Text)) FinalText = message.Text;
|
||||
}
|
||||
break;
|
||||
case AgentSessionEvent.ToolExecutionStart start:
|
||||
Post(() => _updater.StartWorkAsync(A2AText.Agent($"using {start.ToolName}"), _ct));
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private void Post(Func<ValueTask> work) => _work.Writer.TryWrite(work);
|
||||
|
||||
private async Task PumpAsync()
|
||||
{
|
||||
await foreach (var work in _work.Reader.ReadAllAsync().ConfigureAwait(false))
|
||||
{
|
||||
try { await work().ConfigureAwait(false); }
|
||||
catch (OperationCanceledException) { }
|
||||
catch (InvalidOperationException) { } // queue already completed (task cancelled)
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Wait until every posted update is written.</summary>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
_work.Writer.TryComplete();
|
||||
await _pump.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
internal static class A2AText
|
||||
{
|
||||
public static global::A2A.Message Agent(string text) => new()
|
||||
{
|
||||
Role = Role.Agent, MessageId = Guid.NewGuid().ToString("N"), Parts = [Part.FromText(text)],
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user