admin/VolcanicAgenticCore
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(a2a): fresh response artifact per run, atomic continuation, sturdier e2e tests

Browse Source 
- Each run's first text delta replaces the response artifact (append:false) and a resumed task
  drops the previous run's response, so a failed/canceled continuation never keeps old text.
- Resumed is built from a fresh store read, history chronological (question, then answer),
  timestamps on the TaskUpdater clock.
- Pool.Track refuses a second unfinished run of the same task; the pre-check rejects
  continuations of a running task with UnsupportedOperation.
- Tests: continuation history, streaming continuation, response replacement, concurrent
  continuation; busy test waits for the model call; TimeoutException in waits; push test
  matches TASK_STATE_COMPLETED; harness cleans up on failed start.
This commit is contained in:
EugeneTes
2026-10-07 13:38:36 +00:00
parent ebdf4a1aa4
commit baddd320c7
10 changed files with 177 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs
View File

@@ -3,6 +3,7 @@ using A2A;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Volcanic.AgenticCore.A2A.Server.Callers;
using Volcanic.AgenticCore.A2A.Server.Tasks;
using Volcanic.AgenticCore.Abstractions.Events;
using Volcanic.AgenticCore.Abstractions.Processes;
@@ -13,9 +14,10 @@ namespace Volcanic.AgenticCore.A2A.Server.Agent;
/// input-required (the agent called <c>request_input</c>), failed (model error, hop limit, bad file, busy context, any
/// unexpected exception) or canceled (the request token or a CancelTask). The task is registered as live in the pool
/// before its first event and stays live until its final status is written, so restart recovery never touches it.
/// A second run of a task whose run is unfinished (a racing continuation) throws before writing anything.
/// </summary>
internal sealed class A2AAgentHandler(
A2ACaller caller, A2ASessionPool pool, IOptions<A2AHostOptions> options, ISecretRedactor redactor, A2AFileDownloader downloader,
A2ACaller caller, WorkspaceTaskStore store, A2ASessionPool pool, IOptions<A2AHostOptions> options, ISecretRedactor redactor, A2AFileDownloader downloader,
TimeProvider time, ILogger<A2AAgentHandler> logger) : IAgentHandler
{
/// <summary>Bound on the final status write: after a client disconnect the bounded event queue may never be read.</summary>
@@ -128,20 +130,27 @@ 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.
/// The continued task, read fresh from the store (the SDK has just appended the caller's message), moved to working:
/// history gets the question (the status message) and then the caller's answer, and the previous run's
/// <c>response</c> artifact is dropped so this run's answer starts empty.
/// </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 task = store.Peek(context.TaskId) ?? Clone(context.Task!);
var history = task.History ??= [];
if (!history.Any(m => m.MessageId == context.Message.MessageId)) history.Add(context.Message);
history.RemoveAll(m => m.MessageId == context.Message.MessageId);
if (task.Status.Message is { } question) history.Add(question);
task.Status = new global::A2A.TaskStatus { State = TaskState.Working, Timestamp = time.GetUtcNow() };
history.Add(context.Message);
task.Artifacts?.RemoveAll(a => a.ArtifactId == A2ARunMapper.ResponseArtifactId);
// Same clock as the SDK's TaskUpdater, so the task's timestamps stay in order.
task.Status = new global::A2A.TaskStatus { State = TaskState.Working, Timestamp = DateTimeOffset.UtcNow };
return task;
}
private static AgentTask Clone(AgentTask task) =>
System.Text.Json.JsonSerializer.Deserialize<AgentTask>(
System.Text.Json.JsonSerializer.Serialize(task, A2AJsonUtilities.DefaultOptions), A2AJsonUtilities.DefaultOptions)!;
/// <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)
{

6
src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs
View File

@@ -28,6 +28,7 @@ internal sealed class A2ARunMapper : IAsyncDisposable
private readonly CancellationTokenSource _cts;
private readonly CancellationToken _ct;
private readonly Task _pump;
private bool _textStarted;
public A2ARunMapper(TaskUpdater updater, ISecretRedactor redactor, ILogger logger, CancellationToken cancellationToken)
{
@@ -48,8 +49,11 @@ internal sealed class A2ARunMapper : IAsyncDisposable
switch (e)
{
case AgentSessionEvent.MessageUpdate { AssistantMessageEvent: AssistantMessageEvent.TextDelta delta }:
// The run's first chunk replaces whatever response artifact the task already has.
var append = _textStarted;
_textStarted = true;
Post(() => _updater.AddArtifactAsync([Part.FromText(delta.Delta)], ResponseArtifactId, "response", null,
lastChunk: false, append: true, _ct));
lastChunk: false, append: append, _ct));
break;
case AgentSessionEvent.MessageEnd { Message: AssistantMessage message }:
if (message.StopReason == StopReason.Aborted)

15
src/Volcanic.AgenticCore.A2A.Server/Agent/A2ASessionPool.cs
View File

@@ -87,19 +87,28 @@ internal sealed class A2ASessionPool : IA2ALiveRuns, IAsyncDisposable
/// <summary>
/// Mark the task live (so restart recovery leaves it alone) until <see cref="FinishedGrace"/> after the returned handle
/// is disposed. Call before the task's first event is queued and dispose after its final status is written.
/// Throws UnsupportedOperation while another run of the same task is unfinished (two concurrent continuations).
/// </summary>
public IDisposable Track(string taskId, out LiveRun run)
{
foreach (var (id, old) in _live)
if (IsExpired(old)) _live.TryRemove(new KeyValuePair<string, LiveRun>(id, old));
var created = new LiveRun();
// A continuation can start while the previous run of the same task is still in its grace period: the newest run
// replaces it.
_live[taskId] = created;
lock (_live)
{
if (_live.TryGetValue(taskId, out var current) && current.FinishedAt is null)
throw new A2AException($"Task '{taskId}' is already running.", A2AErrorCode.UnsupportedOperation);
// A continuation can start while the previous run of the same task is still in its grace period: the newest
// run replaces it.
_live[taskId] = created;
}
run = created;
return new Tracking(this, created);
}
/// <summary>Whether a run of the task is in progress (tracked and not finished).</summary>
public bool IsRunning(string taskId) => _live.TryGetValue(taskId, out var run) && run.FinishedAt is null;
/// <summary>Take the context's session for one task. Throws UnsupportedOperation while another task runs in it.</summary>
public async Task<Lease> AcquireAsync(A2ACaller caller, string contextId, string taskId, CancellationToken cancellationToken)
{

2
src/Volcanic.AgenticCore.A2A.Server/Hosting/A2AServerCache.cs
View File

@@ -23,7 +23,7 @@ internal sealed class A2AServerCache(IServiceProvider services) : IAsyncDisposab
private VolcanicA2AServer Create(A2ACaller caller)
{
var store = ActivatorUtilities.CreateInstance<WorkspaceTaskStore>(services, caller, (IA2ALiveRuns)_pool);
var handler = ActivatorUtilities.CreateInstance<A2AAgentHandler>(services, caller);
var handler = ActivatorUtilities.CreateInstance<A2AAgentHandler>(services, caller, store);
return ActivatorUtilities.CreateInstance<VolcanicA2AServer>(services, caller, handler, store);
}

2
src/Volcanic.AgenticCore.A2A.Server/Hosting/VolcanicA2AServer.cs
View File

@@ -44,7 +44,7 @@ internal sealed class VolcanicA2AServer(
{
if (message.ContextId is { Length: > 0 } given && given != task.ContextId)
throw new A2AException("The task belongs to another context.", A2AErrorCode.InvalidParams);
if (task.Status.State != TaskState.InputRequired && !task.Status.State.IsTerminal())
if ((task.Status.State != TaskState.InputRequired && !task.Status.State.IsTerminal()) || pool.IsRunning(task.Id))
throw new A2AException($"Task '{task.Id}' is not waiting for input.", A2AErrorCode.UnsupportedOperation);
}
var contextId = message.ContextId is { Length: > 0 } c ? c : task?.ContextId;

7
tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerPushTests.cs
View File

@@ -21,10 +21,9 @@ public class ServerPushTests
Config = new PushNotificationConfig { Url = "https://hooks.example/cb", Token = "tok-1" },
}, CancellationToken.None);
await client.SendMessageAsync(Requests.Text("now", contextId: task.ContextId, taskId: task.Id), CancellationToken.None);
var deadline = DateTime.UtcNow.AddSeconds(5);
while (!host.Push.Requests.Any(r => r.Body.Contains("TASK_STATE_COMPLETED") || r.Body.Contains("completed", StringComparison.OrdinalIgnoreCase)) && DateTime.UtcNow < deadline)
await Task.Delay(20);
var push = host.Push.Requests.Last();
// The completed push: StreamResponse { task } with the state's proto JSON name.
await Requests.WaitUntilAsync(() => host.Push.Requests.Any(r => r.Body.Contains("\"TASK_STATE_COMPLETED\"")), "the completed push");
var push = host.Push.Requests.First(r => r.Body.Contains("\"TASK_STATE_COMPLETED\""));
Assert.Equal("https://hooks.example/cb", push.Url.ToString());
Assert.Equal("tok-1", push.Token);
Assert.Contains(task.Id, push.Body);

71
tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerTaskTests.cs
View File

@@ -77,6 +77,73 @@ public class ServerTaskTests
Assert.Equal(first.Id, second.Id);
Assert.Equal(TaskState.Completed, second.Status.State);
Assert.Equal("Booked for Friday", Requests.ResponseText(second));
var stored = await client.GetTaskAsync(new GetTaskRequest { Id = first.Id }, CancellationToken.None);
var history = stored.History!.Select(m => string.Concat(m.Parts.Select(p => p.Text))).ToList();
Assert.Single(history, t => t == "Friday");
var question = history.IndexOf("Which date?");
Assert.True(question >= 0 && question < history.IndexOf("Friday"), string.Join(" | ", history));
}
[Fact]
public async Task RequestInput_StreamingContinuation_StartsWithWorkingTask()
{
await using var host = await A2ATestServer.StartAsync();
host.Faux.Enqueue(FauxTurn.Tool("request_input", new { question = "Which date?" }), FauxTurn.Say("Booked for Friday"));
var client = host.Client();
var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!;
var events = await CollectAsync(client.SendStreamingMessageAsync(
Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None));
var resumed = events[0].Task;
Assert.NotNull(resumed);
Assert.Equal(first.Id, resumed.Id);
Assert.Equal(TaskState.Working, resumed.Status.State);
Assert.Equal(TaskState.Completed, events.Last(e => e.StatusUpdate is not null).StatusUpdate!.Status.State);
}
[Fact]
public async Task RequestInput_FailedContinuation_DropsEarlierResponse()
{
await using var host = await A2ATestServer.StartAsync();
host.Faux.Enqueue(FauxTurn.Say("Let me check.").WithTool("request_input", new { question = "Which date?" }),
FauxTurn.Error("model exploded"));
var client = host.Client();
var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!;
Assert.Equal(TaskState.InputRequired, first.Status.State);
Assert.Equal("Let me check.", Requests.ResponseText(first));
var second = (await client.SendMessageAsync(Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None)).Task!;
Assert.Equal(TaskState.Failed, second.Status.State);
Assert.DoesNotContain("Let me check.", Requests.ResponseText(second));
}
[Fact]
public async Task RequestInput_ContinuationAnswer_ReplacesEarlierResponse()
{
await using var host = await A2ATestServer.StartAsync();
host.Faux.Enqueue(FauxTurn.Say("Let me check.").WithTool("request_input", new { question = "Which date?" }),
FauxTurn.Say("Booked for Friday"));
var client = host.Client();
var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!;
var events = await CollectAsync(client.SendStreamingMessageAsync(
Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None));
var chunks = events.Where(e => e.ArtifactUpdate is { Artifact.ArtifactId: "response" }).Select(e => e.ArtifactUpdate!).ToList();
Assert.False(chunks[0].Append); // the run's first chunk starts a fresh answer
var stored = await client.GetTaskAsync(new GetTaskRequest { Id = first.Id }, CancellationToken.None);
Assert.Equal("Booked for Friday", Requests.ResponseText(stored));
}
[Fact]
public async Task RequestInput_SecondContinuationWhileRunning_Rejected()
{
await using var host = await A2ATestServer.StartAsync();
host.Faux.Enqueue(FauxTurn.Tool("request_input", new { question = "Which date?" }), FauxTurn.Say("slow").WithDelay(TimeSpan.FromSeconds(5)));
var client = host.Client();
var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!;
await client.SendMessageAsync(Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id, returnImmediately: true), CancellationToken.None);
await Requests.WaitUntilAsync(() => host.Faux.Requests.Count == 2, "the continuation's model call");
var ex = await Assert.ThrowsAsync<A2AException>(() =>
client.SendMessageAsync(Requests.Text("Saturday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None));
Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode);
await client.CancelTaskAsync(new CancelTaskRequest { Id = first.Id }, CancellationToken.None);
}
[Fact]
@@ -88,6 +155,8 @@ public class ServerTaskTests
var first = (await client.SendMessageAsync(Requests.Text("one"), CancellationToken.None)).Task!;
var slow = (await client.SendMessageAsync(Requests.Text("two", contextId: first.ContextId, returnImmediately: true), CancellationToken.None)).Task!;
await Requests.WaitForStateAsync(client, slow.Id, s => s == TaskState.Working);
// Working is written before the run holds the context: wait until its model call has started.
await Requests.WaitUntilAsync(() => host.Faux.Requests.Count == 2, "the second model call");
var ex = await Assert.ThrowsAsync<A2AException>(() =>
client.SendMessageAsync(Requests.Text("three", contextId: first.ContextId), CancellationToken.None));
Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode);
@@ -111,7 +180,7 @@ public class ServerTaskTests
public async Task Subscribe_LiveTask_GetsEvents()
{
await using var host = await A2ATestServer.StartAsync();
host.Faux.Enqueue(FauxTurn.Say("eventually").WithDelay(TimeSpan.FromMilliseconds(500)));
host.Faux.Enqueue(FauxTurn.Say("eventually").WithDelay(TimeSpan.FromSeconds(2)));
var client = host.Client();
var task = (await client.SendMessageAsync(Requests.Text("go", returnImmediately: true), CancellationToken.None)).Task!;
var events = await CollectAsync(client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = task.Id }, CancellationToken.None));

14
tests/Volcanic.AgenticCore.A2A.Server.Tests/SessionPoolTests.cs
View File

@@ -5,6 +5,8 @@ using Microsoft.Extensions.Options;
using Microsoft.Extensions.Time.Testing;
using Volcanic.AgenticCore.A2A.Server.Agent;
using Volcanic.AgenticCore.A2A.Server.Callers;
using Volcanic.AgenticCore.A2A.Server.Push;
using Volcanic.AgenticCore.A2A.Server.Tasks;
using Volcanic.AgenticCore.Features.Agent;
using Volcanic.AgenticCore.Features.Extensions;
using Volcanic.AgenticCore.Features.Processes;
@@ -48,7 +50,8 @@ public sealed class SessionPoolTests : IAsyncDisposable
}
private A2AAgentHandler Handler() =>
new(_caller, _pool, Options.Create(_options), SecretRedactor.Empty, new A2AFileDownloader(new HttpClient()), TimeProvider.System,
new(_caller, new WorkspaceTaskStore(_caller, _pool, A2APushSender.Disabled, _time, NullLogger<WorkspaceTaskStore>.Instance),
_pool, Options.Create(_options), SecretRedactor.Empty, new A2AFileDownloader(new HttpClient()), TimeProvider.System,
NullLogger<A2AAgentHandler>.Instance);
private static RequestContext Context(string text, string taskId = "t1", string contextId = "c1") => new()
@@ -82,6 +85,15 @@ public sealed class SessionPoolTests : IAsyncDisposable
Assert.Null(run.FinishedAt);
}
[Fact]
public void Pool_Track_UnfinishedRunOfSameTask_Refused()
{
using var first = _pool.Track("t1", out _);
var ex = Assert.Throws<A2AException>(() => _pool.Track("t1", out _));
Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode);
Assert.True(_pool.IsRunning("t1"));
}
[Fact]
public async Task Pool_AbortAsync_FinishedRun_IsNoOp()
{

73
tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/A2ATestServer.cs
View File

@@ -33,36 +33,44 @@ public sealed class A2ATestServer : IAsyncDisposable
public static async Task<A2ATestServer> StartAsync(Action<A2AHostOptions>? configure = null, bool withNotifier = true)
{
var host = new A2ATestServer();
Directory.CreateDirectory(host.Workspace);
Directory.CreateDirectory(host.OtherWorkspace);
host.WriteCallers(host.Workspace, $$"""[ { "name": "alpha", "key": "{{Key}}" }, { "name": "gamma", "key": "key-gamma" } ]""");
host.WriteCallers(host.OtherWorkspace, $$"""[ { "name": "beta", "key": "{{OtherKey}}" } ]""");
try
{
Directory.CreateDirectory(host.Workspace);
Directory.CreateDirectory(host.OtherWorkspace);
host.WriteCallers(host.Workspace, $$"""[ { "name": "alpha", "key": "{{Key}}" }, { "name": "gamma", "key": "key-gamma" } ]""");
host.WriteCallers(host.OtherWorkspace, $$"""[ { "name": "beta", "key": "{{OtherKey}}" } ]""");
var builder = WebApplication.CreateBuilder();
builder.WebHost.UseTestServer();
builder.Services.AddAgenticCore(o =>
var builder = WebApplication.CreateBuilder();
builder.WebHost.UseTestServer();
builder.Services.AddAgenticCore(o =>
{
o.AgentDir = Path.Combine(host._root, "agent");
o.DiscoverExtensionAssemblies = false;
o.DefaultModel = FauxProvider.ModelKey;
o.Extensions.Add(ExtensionRegistration.FromInstance(host.Faux.Extension));
o.ProcessEnvironment.HardenHostProcess = false;
o.Retry = new RetrySettings { Enabled = false }; // FauxTurn.Error fails the task at once
});
builder.Services.AddSingleton<A2APushSender>(sp => new HttpA2APushSender(new HttpClient(host.Push), TimeProvider.System, NullLogger<HttpA2APushSender>.Instance));
if (withNotifier) builder.Services.AddSingleton<IA2AOwnerNotifier>(host.Notifier);
builder.Services.AddSingleton<IA2ACallerDirectory>(new WorkspaceCallerDirectory(
() => [new A2AWorkspace(host.Workspace, "test-ws1", "ws1"), new A2AWorkspace(host.OtherWorkspace, "test-ws2", "ws2")],
NullLogger<WorkspaceCallerDirectory>.Instance));
builder.Services.AddA2AServer(o =>
{
o.AllowPrivatePushUrls = true;
configure?.Invoke(o);
});
host.App = builder.Build();
host.App.MapA2AServer();
await host.App.StartAsync();
return host;
}
catch
{
o.AgentDir = Path.Combine(host._root, "agent");
o.DiscoverExtensionAssemblies = false;
o.DefaultModel = FauxProvider.ModelKey;
o.Extensions.Add(ExtensionRegistration.FromInstance(host.Faux.Extension));
o.ProcessEnvironment.HardenHostProcess = false;
o.Retry = new RetrySettings { Enabled = false }; // FauxTurn.Error fails the task at once
});
builder.Services.AddSingleton<A2APushSender>(sp => new HttpA2APushSender(new HttpClient(host.Push), TimeProvider.System, NullLogger<HttpA2APushSender>.Instance));
if (withNotifier) builder.Services.AddSingleton<IA2AOwnerNotifier>(host.Notifier);
builder.Services.AddSingleton<IA2ACallerDirectory>(new WorkspaceCallerDirectory(
() => [new A2AWorkspace(host.Workspace, "test-ws1", "ws1"), new A2AWorkspace(host.OtherWorkspace, "test-ws2", "ws2")],
NullLogger<WorkspaceCallerDirectory>.Instance));
builder.Services.AddA2AServer(o =>
{
o.AllowPrivatePushUrls = true;
configure?.Invoke(o);
});
host.App = builder.Build();
host.App.MapA2AServer();
await host.App.StartAsync();
return host;
await host.DisposeAsync();
throw;
}
}
public void WriteCallers(string workspace, string json)
@@ -82,9 +90,12 @@ public sealed class A2ATestServer : IAsyncDisposable
public async ValueTask DisposeAsync()
{
await App.StopAsync();
await App.DisposeAsync();
try { Directory.Delete(_root, true); } catch (IOException) { }
if (App is not null)
{
await App.StopAsync();
await App.DisposeAsync();
}
try { Directory.Delete(_root, true); } catch (Exception e) when (e is IOException or UnauthorizedAccessException) { }
}
public sealed class RecordingNotifier : IA2AOwnerNotifier

15
tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/Requests.cs
View File

@@ -26,7 +26,20 @@ public static class Requests
while (true)
{
var task = await client.GetTaskAsync(new GetTaskRequest { Id = taskId }, CancellationToken.None);
if (done(task.Status.State) || DateTime.UtcNow > deadline) return task;
if (done(task.Status.State)) return task;
if (DateTime.UtcNow > deadline)
throw new TimeoutException($"Task {taskId} did not reach the expected state in {timeoutMs} ms; last state {task.Status.State}.");
await Task.Delay(20);
}
}
/// <summary>Poll until <paramref name="condition"/> holds; throws TimeoutException naming <paramref name="what"/>.</summary>
public static async Task WaitUntilAsync(Func<bool> condition, string what, int timeoutMs = 10_000)
{
var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
while (!condition())
{
if (DateTime.UtcNow > deadline) throw new TimeoutException($"Timed out after {timeoutMs} ms waiting for {what}.");
await Task.Delay(20);
}
}