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

fix(a2a-client): deliver outside the workspace lock

Browse Source
This commit is contained in:
EugeneTes
2026-10-08 13:31:27 +00:00
parent 24e2ad9877
commit e1e0f35bb9
3 changed files with 105 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResultDelivery.cs
View File

@@ -7,6 +7,12 @@ namespace Volcanic.AgenticCore.Extensions.A2A;
public interface IA2AResultDelivery
{
/// <summary>Run <paramref name="text"/> as a follow-up prompt in the target's conversation. False: not deliverable now.</summary>
/// <remarks>
/// May run a whole agent turn before returning, and that turn may call back into the <see cref="A2AWatchRegistry"/>
/// (cancel a watch, start one): the registry never holds a lock while delivering. A delivery in flight can't be
/// interrupted, so shutting the registry down waits for it. A false result keeps the text in <c>pending.json</c> for
/// the next flush; delivery is at least once.
/// </remarks>
Task<bool> DeliverAsync(string target, string text, CancellationToken cancellationToken);
/// <summary>The target's workspace. A path computation: must not create or wake a session.</summary>

54
extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs
View File

@@ -16,9 +16,11 @@ namespace Volcanic.AgenticCore.Extensions.A2A;
/// is persisted in the workspace's <c>pending.json</c> until delivered.
/// </summary>
/// <remarks>
/// Delivery is at least once: a result is delivered, then removed from <c>pending.json</c>, so a crash in between
/// delivers it again after the restart. Within a process, a per-workspace lock serialises finishing, flushing, restoring
/// and stopping, so one result is delivered once and a cancelled watch's record is never written back.
/// Delivery is at least once. A finished watch first stores its text as <c>Undelivered</c> in <c>pending.json</c>, then
/// delivers outside any lock (a delivery may run a whole agent turn that calls back into this registry), then removes the
/// record; a crash in between delivers it again on the next flush. A per-workspace lock guards only the bookkeeping
/// (claiming a watch, writing <c>pending.json</c>), and records being delivered are claimed in <c>_delivering</c>, so
/// finishing, flushing and restoring deliver one result once, and a cancel during a delivery is never undone.
/// A push only means "check now": its body is never trusted.
/// </remarks>
public sealed class A2AWatchRegistry : IAsyncDisposable
@@ -40,6 +42,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
private readonly HttpMessageHandler _handler;
private readonly ConcurrentDictionary<string, Watch> _watches = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, SemaphoreSlim> _workspaceLocks = new(StringComparer.Ordinal);
/// <summary>Watch ids whose result is being delivered right now (claimed under the workspace lock).</summary>
private readonly ConcurrentDictionary<string, byte> _delivering = new(StringComparer.Ordinal);
private volatile IA2AResultDelivery _delivery;
public A2AWatchRegistry(TimeProvider time, HttpMessageHandler handler, ILogger? logger = null)
@@ -161,11 +166,25 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
/// <summary>Deliver results that were kept because the target had no live conversation.</summary>
public async Task FlushUndeliveredAsync(string target, string workspace, CancellationToken cancellationToken)
{
List<WatchRecord> claimed;
using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false))
{
foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is not null))
if (await TryDeliverAsync(target, record.Undelivered!, cancellationToken).ConfigureAwait(false))
claimed = PendingFile.Read(workspace)
.Where(r => r.Target == target && r.Undelivered is not null && _delivering.TryAdd(r.WatchId, 0))
.ToList();
}
try
{
foreach (var record in claimed)
{
if (!await TryDeliverAsync(target, record.Undelivered!, cancellationToken).ConfigureAwait(false)) continue;
using (await LockAsync(workspace).ConfigureAwait(false))
PendingFile.Remove(workspace, record.WatchId);
}
}
finally
{
foreach (var record in claimed) _delivering.TryRemove(record.WatchId, out _);
}
}
@@ -354,17 +373,29 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
return watch.Record;
}
/// <summary>Delivers (or keeps undelivered) under the workspace lock, then drops the push config best-effort.</summary>
/// <summary>
/// Claims the watch and stores its text as undelivered (under the lock), delivers outside the lock, removes the record
/// once delivered (under the lock), then drops the push config best-effort.
/// </summary>
private async Task FinishAsync(Watch watch, string text, RemoteAgent? agent)
{
var record = watch.Record;
using (await LockAsync(watch.Workspace).ConfigureAwait(false))
{
if (!_watches.TryRemove(new(record.WatchId, watch))) return; // stopped (cancel or shutdown) meanwhile
_delivering[record.WatchId] = 0;
PendingFile.Upsert(watch.Workspace, record with { Undelivered = text });
}
try
{
// A failed delivery writes nothing: the text is stored, and a cancel meanwhile has removed it for good.
if (await TryDeliverAsync(record.Target, text, CancellationToken.None).ConfigureAwait(false))
PendingFile.Remove(watch.Workspace, record.WatchId);
else
PendingFile.Upsert(watch.Workspace, record with { Undelivered = text });
using (await LockAsync(watch.Workspace).ConfigureAwait(false))
PendingFile.Remove(watch.Workspace, record.WatchId);
}
finally
{
_delivering.TryRemove(record.WatchId, out _);
}
if (agent is null || record.PushToken is null) return;
try
@@ -401,7 +432,10 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b;
/// <summary>Stops every watch (the records stay in pending.json for the next restore).</summary>
/// <summary>
/// Stops every watch (the records stay in pending.json for the next restore). Waits for in-flight deliveries: a
/// delivery can't be interrupted.
/// </summary>
public async Task StopAsync()
{
var watches = _watches.ToArray();

56
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
View File

@@ -46,6 +46,22 @@ public sealed class A2AWatchTests : IDisposable
public IEnumerable<string> KnownTargets() => ["1"];
}
/// <summary>Delivery that runs <see cref="OnDeliver"/> first, like a whole agent turn calling back into the registry.</summary>
private sealed class ReentrantDelivery(string workspace) : IA2AResultDelivery
{
public ConcurrentQueue<string> Delivered { get; } = new();
public Func<Task>? OnDeliver { get; set; }
public async Task<bool> DeliverAsync(string target, string text, CancellationToken cancellationToken)
{
if (Interlocked.Exchange(ref _onDeliverTaken, 1) == 0 && OnDeliver is { } turn) await turn();
Delivered.Enqueue(text);
return true;
}
private int _onDeliverTaken;
public string? WorkspaceFor(string target) => workspace;
public IEnumerable<string> KnownTargets() => ["1"];
}
private sealed class ListLogger : ILogger
{
public ConcurrentQueue<string> Lines { get; } = new();
@@ -231,7 +247,12 @@ public sealed class A2AWatchTests : IDisposable
await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None);
await WaitUntilAsync(() => PendingFile.Read(_ws).Any(r => r.Undelivered is not null));
delivery.Accept = true;
await registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None);
// The finishing watch holds its claim until its failed delivery returns: flush until the result goes out.
await WaitUntilAsync(() =>
{
registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None).GetAwaiter().GetResult();
return !delivery.Delivered.IsEmpty;
});
Assert.Contains("Later", Assert.Single(delivery.Delivered).Text);
Assert.Empty(PendingFile.Read(_ws));
}
@@ -406,6 +427,39 @@ public sealed class A2AWatchTests : IDisposable
Assert.Contains("Followed", Assert.Single(delivery.Delivered).Text);
}
[Fact]
public async Task Delivery_ReenteringTheRegistry_DoesNotDeadlock()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("never").WithDelay(TimeSpan.FromSeconds(10)));
WriteAgents();
var delivery = new ReentrantDelivery(_ws);
await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(50)] };
registry.UseDelivery(delivery);
var settings = new A2AClientSettings();
using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None);
var slow = await StartRemoteTaskAsync(agent, "slow");
await registry.StartAsync("1", _ws, agent, slow, settings, CancellationToken.None);
var turnDone = false;
// The delivered result's "turn" cancels the slow task's watch and starts a new watch in the same workspace.
delivery.OnDeliver = async () =>
{
Assert.True(await registry.StopWatchingAsync("1", _ws, slow.Id));
await registry.StartAsync("1", _ws, agent, slow, settings, CancellationToken.None);
turnDone = true;
};
// The remote runs both tasks at once: script the quick answer only once the slow task has taken its turn.
await WaitUntilAsync(() => remote.Faux.Remaining == 0);
remote.Faux.Enqueue(FauxTurn.Say("Quick"));
var quick = await StartRemoteTaskAsync(agent, "quick");
await registry.StartAsync("1", _ws, agent, quick, settings, CancellationToken.None);
await WaitUntilAsync(() => Volatile.Read(ref turnDone) && !delivery.Delivered.IsEmpty, timeoutMs: 5_000);
Assert.Contains("Quick", Assert.Single(delivery.Delivered));
var watched = Assert.Single(registry.List("1"));
Assert.Equal(slow.Id, watched.TaskId);
Assert.Contains(PendingFile.Read(_ws), r => r.WatchId == watched.WatchId);
}
[Fact]
public void PollDelays_BackOffToFiveMinutes()
{