63 lines
2.8 KiB
C#
63 lines
2.8 KiB
C#
using System.Text.Json;
|
|
using Microsoft.Extensions.Logging;
|
|
using Volcanic.AgenticCore.Abstractions.Extensions;
|
|
using Volcanic.AgenticCore.Abstractions.Extensions.Events;
|
|
|
|
namespace Volcanic.AgenticCore.Extensions.A2A;
|
|
|
|
/// <summary>
|
|
/// Per session (the host adds it, e.g. per Telegram chat; never host-wide): the <c>a2a</c> tool bound to the host's
|
|
/// <paramref name="target"/>, the hop counter from <c>a2a.inbound</c>, and the live-session delivery hook that also
|
|
/// flushes results kept while the target had no live conversation.
|
|
/// </summary>
|
|
public sealed class A2AClientExtension(A2AClientSettings settings, string target, A2AWatchRegistry? registry = null) : IExtension
|
|
{
|
|
public const string ExtensionName = "a2a-client";
|
|
|
|
/// <summary>Emitted by the A2A server's per-session extension before each inbound prompt: <c>{"hops": n, "caller": "…"}</c>.</summary>
|
|
public const string InboundChannel = "a2a.inbound";
|
|
|
|
private readonly A2AWatchRegistry _registry = registry ?? A2AWatchRegistry.Shared;
|
|
private volatile int _hops;
|
|
|
|
public string Name => ExtensionName;
|
|
|
|
public ValueTask ActivateAsync(IExtensionApi api, CancellationToken cancellationToken)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(api);
|
|
api.RegisterTool(new A2ATool(settings, target, _registry, () => _hops));
|
|
api.Events.On(InboundChannel, data =>
|
|
{
|
|
if (data is JsonElement { ValueKind: JsonValueKind.Object } e && e.TryGetProperty("hops", out var h) && h.TryGetInt32(out var n))
|
|
_hops = n;
|
|
});
|
|
api.On<SessionStartEvent>((_, ctx, _) =>
|
|
{
|
|
var workspace = ctx.Cwd;
|
|
_registry.Live.Attach(target, workspace, api);
|
|
// Not inline and not at once: delivering starts a run, which mustn't happen inside session creation, and
|
|
// the host subscribes to the session only once creation returned, so the flush waits a moment. (Telegram,
|
|
// Task 8, delivers through its ChatRegistry instead, which waits until the chat's session is created.)
|
|
_ = Task.Run(async () =>
|
|
{
|
|
try
|
|
{
|
|
await Task.Delay(_registry.SessionStartFlushDelay, _registry.Time).ConfigureAwait(false);
|
|
await _registry.FlushUndeliveredAsync(target, workspace, CancellationToken.None).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex) when (ex is not OutOfMemoryException)
|
|
{
|
|
api.Logger.LogWarning(ex, "A2A: flushing undelivered results for {Target} failed", target);
|
|
}
|
|
});
|
|
return ValueTask.CompletedTask;
|
|
});
|
|
api.On<SessionShutdownEvent>((_, _, _) =>
|
|
{
|
|
_registry.Live.Detach(target, api);
|
|
return ValueTask.CompletedTask;
|
|
});
|
|
return ValueTask.CompletedTask;
|
|
}
|
|
}
|