fix(a2a-client): read 0.3-shaped file parts and report parts the SDK can't read

The 1.0 SDK ignores unknown part fields, so a file part shaped {kind:file, file:{bytes|uri,
name, mimeType}} arrived as an empty part and vanished. A2AWireCompat rewrites such parts
to {raw|url, filename, mediaType} in JSON and SSE responses before the SDK reads them, and
marks any part it still can't read so the result lists it with its field names.
This commit is contained in:
EugeneTes
2026-10-12 16:26:11 +00:00
parent a92623b89b
commit 6893ee8197
5 changed files with 269 additions and 1 deletions

View File

@@ -300,6 +300,11 @@ Files:
- crm.tar.gz (320000000 bytes, application/gzip): not saved, over 209715200 bytes
```
The SDK speaks protocol 1.0 and would silently drop file parts in other shapes, so every A2A response (JSON and
SSE) first passes `A2AWireCompat`: the 0.3 shape `{"kind":"file","file":{"bytes"|"uri","name","mimeType"}}` (or those
fields directly on the part) becomes `{"raw"|"url","filename","mediaType"}`, and a part it still can't read is listed
as `- part: not saved, not understood (fields: ...)` and logged with its field names (never values).
Saving a task again overwrites its folder's files, so `action=status` with the `taskId` of a finished task fetches
its files again (as long as the remote agent still returns them). Watch deliveries (`[A2A result from …]`) carry the
same list.

View File

@@ -53,6 +53,13 @@ internal static class A2AFiles
var folder = $"inbox/a2a/{Sanitize(folderId)}";
foreach (var (part, fallback) in parts)
{
if (part.ContentCase == PartContentCase.None &&
part.Metadata?.GetValueOrDefault(A2AWireCompat.UnrecognizedKey) is { ValueKind: System.Text.Json.JsonValueKind.String } keys)
{
if (seen.Add($"unknown:{keys.GetString()}"))
files.Add(new ReceivedFile("part", null, null, null, $"not understood (fields: {keys.GetString()})"));
continue;
}
if (part.ContentCase is not (PartContentCase.Raw or PartContentCase.Url)) continue;
var name = Sanitize(part.Filename ?? (part.Url is { } u ? UrlName(u) : null) ?? fallback);
var mediaType = string.IsNullOrWhiteSpace(part.MediaType) ? null : part.MediaType;

View File

@@ -40,6 +40,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
private const int SubscribeFailuresBeforePoll = 3;
private readonly HttpMessageHandler _handler;
/// <summary><see cref="_handler"/> behind <see cref="A2AWireCompat"/>, for every A2A call (not for downloads).</summary>
private readonly HttpMessageHandler _wire;
private readonly ConcurrentDictionary<string, Watch> _watches = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, SemaphoreSlim> _workspaceLocks = new(StringComparer.Ordinal);
@@ -63,6 +66,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
Time = time;
_handler = handler;
Logger = logger ?? NullLogger.Instance;
_wire = new A2AWireCompat(handler, () => Logger);
_delivery = Live;
}
@@ -87,7 +91,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
internal HttpMessageHandler Handler => _handler;
internal RemoteAgentConnector Connector(A2AClientSettings settings) =>
new(_handler, TimeSpan.FromSeconds(settings.CardTimeoutSeconds));
new(_wire, TimeSpan.FromSeconds(settings.CardTimeoutSeconds));
internal static TimeSpan DelayAt(TimeSpan[] delays, int attempt) => delays[Math.Clamp(attempt, 0, delays.Length - 1)];

View File

@@ -0,0 +1,159 @@
using System.IO.Pipelines;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json.Nodes;
using Microsoft.Extensions.Logging;
namespace Volcanic.AgenticCore.Extensions.A2A;
/// <summary>
/// Sits between the A2A SDK and the network and rewrites file parts the SDK (protocol 1.0) would drop silently: the
/// 0.3 shape <c>{"kind":"file","file":{"bytes"|"uri","name","mimeType"}}</c> and the same fields directly on the part
/// become <c>{"raw"|"url","filename","mediaType"}</c>. A part with none of <c>text</c>/<c>raw</c>/<c>url</c>/<c>data</c>
/// afterwards gets its field names in metadata <see cref="UnrecognizedKey"/>, so the result can say so instead of
/// losing it. Works on JSON responses and on SSE streams (line by line, each <c>data:</c> line on its own).
/// </summary>
internal sealed class A2AWireCompat(HttpMessageHandler inner, Func<ILogger> logger) : DelegatingHandler(inner)
{
public const string UnrecognizedKey = "volcanic.unrecognizedPart";
private static readonly string[] V1Keys = ["text", "raw", "url", "data"];
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false);
var type = response.Content.Headers.ContentType?.MediaType;
if (type is "application/json")
{
var body = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
var rewritten = Rewrite(body, logger());
if (rewritten is not null) Replace(response, new StringContent(rewritten, Encoding.UTF8, "application/json"), disposeOld: true);
}
else if (type is "text/event-stream")
{
var source = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
var content = new StreamContent(RewriteEvents(source, logger()));
content.Headers.ContentType = new MediaTypeHeaderValue("text/event-stream");
Replace(response, content, disposeOld: false); // its stream now belongs to the new content
}
return response;
}
private static void Replace(HttpResponseMessage response, HttpContent content, bool disposeOld)
{
var old = response.Content;
foreach (var header in old.Headers.Where(h => h.Key is not ("Content-Type" or "Content-Length")))
content.Headers.TryAddWithoutValidation(header.Key, header.Value);
response.Content = content;
if (disposeOld) old.Dispose();
}
/// <summary>The JSON with its parts rewritten, or null when nothing changed (or it isn't JSON).</summary>
internal static string? Rewrite(string json, ILogger logger)
{
if (json.IndexOf("\"parts\"", StringComparison.Ordinal) < 0) return null;
JsonNode? root;
try { root = JsonNode.Parse(json); }
catch (System.Text.Json.JsonException) { return null; }
var (converted, unknown) = (0, new List<string>());
Walk(root, ref converted, unknown);
if (converted == 0 && unknown.Count == 0) return null;
if (converted > 0) logger.LogInformation("A2A: rewrote {Count} file part(s) from the 0.3 shape", converted);
foreach (var keys in unknown) logger.LogWarning("A2A: a part with unknown fields ({Keys}) from a remote agent", keys);
return root!.ToJsonString();
}
private static void Walk(JsonNode? node, ref int converted, List<string> unknown)
{
switch (node)
{
case JsonObject obj:
if (obj["parts"] is JsonArray parts)
foreach (var part in parts.OfType<JsonObject>())
Normalize(part, ref converted, unknown);
foreach (var (_, child) in obj.ToList()) Walk(child, ref converted, unknown);
break;
case JsonArray array:
foreach (var child in array.ToList()) Walk(child, ref converted, unknown);
break;
}
}
private static void Normalize(JsonObject part, ref int converted, List<string> unknown)
{
if (V1Keys.Any(k => Has(part, k))) return;
var file = part["file"] as JsonObject ?? (Has(part, "bytes") || Has(part, "uri") ? part : null);
if (file is not null && (Has(file, "bytes") || Has(file, "uri")))
{
if (Has(file, "bytes")) part["raw"] = file["bytes"]!.DeepClone();
else part["url"] = file["uri"]!.DeepClone();
if (!Has(part, "filename") && Has(file, "name")) part["filename"] = file["name"]!.DeepClone();
if (!Has(part, "mediaType") && Has(file, "mimeType")) part["mediaType"] = file["mimeType"]!.DeepClone();
foreach (var key in new[] { "file", "bytes", "uri", "name", "mimeType", "kind" }) part.Remove(key);
converted++;
return;
}
var keys = string.Join(", ", part.Select(p => p.Value is JsonObject inner ? $"{p.Key}{{{string.Join(", ", inner.Select(i => i.Key))}}}" : p.Key));
unknown.Add(keys);
var metadata = part["metadata"] as JsonObject ?? new JsonObject();
metadata[UnrecognizedKey] = keys;
part["metadata"] = metadata;
}
private static bool Has(JsonObject obj, string key) => obj[key] is { } value && value.GetValueKind() != System.Text.Json.JsonValueKind.Null;
/// <summary>The event stream, each <c>data:</c> line rewritten on its own; the rest passes through.</summary>
private static Stream RewriteEvents(Stream source, ILogger logger)
{
var pipe = new Pipe();
_ = Task.Run(async () =>
{
Exception? error = null;
try
{
using var reader = new StreamReader(source, Encoding.UTF8);
while (await reader.ReadLineAsync().ConfigureAwait(false) is { } line)
{
if (line.StartsWith("data:", StringComparison.Ordinal) && Rewrite(line[5..].TrimStart(), logger) is { } rewritten)
line = "data: " + rewritten;
var result = await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(line + "\n")).ConfigureAwait(false);
if (result.IsCompleted) break; // the reader is gone
}
}
catch (Exception ex) { error = ex; }
finally
{
await source.DisposeAsync().ConfigureAwait(false);
await pipe.Writer.CompleteAsync(error).ConfigureAwait(false);
}
});
return new OwningStream(pipe.Reader.AsStream(), source);
}
/// <summary>Reads the rewritten events; disposing it also closes the source, which stops the pump.</summary>
private sealed class OwningStream(Stream inner, Stream source) : Stream
{
public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => throw new NotSupportedException();
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
public override void Flush() { }
public override int Read(byte[] buffer, int offset, int count) => inner.Read(buffer, offset, count);
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) => inner.ReadAsync(buffer, cancellationToken);
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => inner.ReadAsync(buffer, offset, count, cancellationToken);
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
protected override void Dispose(bool disposing)
{
if (disposing)
{
source.Dispose();
inner.Dispose();
}
base.Dispose(disposing);
}
}
}

View File

@@ -0,0 +1,93 @@
using System.Net;
using System.Text;
using A2A;
using Microsoft.Extensions.Logging.Abstractions;
using Volcanic.AgenticCore.Extensions.A2A;
namespace Volcanic.AgenticCore.Extensions.Tests.A2A;
public sealed class A2AWireCompatTests : IDisposable
{
private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-wire", Guid.NewGuid().ToString("N"));
private static readonly string Skill = Convert.ToBase64String(Encoding.UTF8.GetBytes("# skill"));
private static string TaskJson(string parts) => $$"""
{"id":"t-1","contextId":"c-1","status":{"state":"TASK_STATE_COMPLETED"},
"artifacts":[{"artifactId":"response","parts":[{"kind":"text","text":"Here you go"}]},
{"artifactId":"files","parts":[{{parts}}]}]}
""";
private static IA2AClient Client(Func<HttpRequestMessage, HttpResponseMessage> respond) =>
new A2AClient(new Uri("https://agent.example/a2a"),
new HttpClient(new A2AWireCompat(new Fake(respond), () => NullLogger.Instance)));
private static HttpResponseMessage Json(string result) => new(HttpStatusCode.OK)
{
Content = new StringContent($$"""{"jsonrpc":"2.0","id":1,"result":{{result}}}""", Encoding.UTF8, "application/json"),
};
[Fact]
public async Task GetTask_V03FileParts_BecomeRawAndUrl()
{
var client = Client(_ => Json(TaskJson(
"""{"kind":"file","file":{"bytes":"SKILL","name":"SKILL.md","mimeType":"text/markdown"}},""".Replace("SKILL\"", Skill + "\"") +
"""{"kind":"file","file":{"uri":"https://cdn.example/crm.tar.gz","name":"crm.tar.gz","mimeType":"application/gzip"}}""")));
var task = await client.GetTaskAsync(new GetTaskRequest { Id = "t-1" });
var parts = task.Artifacts![1].Parts;
Assert.Equal("# skill", Encoding.UTF8.GetString(parts[0].Raw!));
Assert.Equal(("SKILL.md", "text/markdown"), (parts[0].Filename!, parts[0].MediaType!));
Assert.Equal("https://cdn.example/crm.tar.gz", parts[1].Url);
Assert.Equal("crm.tar.gz", parts[1].Filename);
Assert.Equal("Here you go", A2AResults.FinalText(task));
}
[Fact]
public async Task UnknownPart_ReportedInFiles_V1PartsUntouched()
{
var client = Client(_ => Json(TaskJson(
"{\"raw\":\"" + Skill + "\",\"filename\":\"SKILL.md\"}," + """{"kind":"blob","blob":{"content":"x","title":"y"}}""")));
var task = await client.GetTaskAsync(new GetTaskRequest { Id = "t-1" });
var files = await A2AFiles.SaveAsync(_ws, task, null, new Fake(_ => new HttpResponseMessage(HttpStatusCode.NotFound)), new A2AClientSettings(), CancellationToken.None);
Assert.Equal("inbox/a2a/t-1/SKILL.md", files[0].Path);
Assert.Equal("not understood (fields: kind, blob{content, title})", files[1].Problem);
}
[Fact]
public async Task Subscribe_SseEvents_Rewritten()
{
var client = Client(_ => new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StringContent(
"event: message\n" +
"data: " + ("""{"jsonrpc":"2.0","id":1,"result":{"task":""" +
TaskJson("{\"kind\":\"file\",\"file\":{\"bytes\":\"" + Skill + "\",\"name\":\"SKILL.md\"}}") + "}}").Replace("\n", "") + "\n\n",
Encoding.UTF8, "text/event-stream"),
});
var events = new List<StreamResponse>();
await foreach (var ev in client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = "t-1" })) events.Add(ev);
var part = Assert.Single(events).Task!.Artifacts![1].Parts.Single();
Assert.Equal("# skill", Encoding.UTF8.GetString(part.Raw!));
Assert.Equal("SKILL.md", part.Filename);
}
[Fact]
public void Rewrite_NoParts_LeavesBodyAlone()
{
Assert.Null(A2AWireCompat.Rewrite("""{"name":"card"}""", NullLogger.Instance));
Assert.Null(A2AWireCompat.Rewrite("""{"parts":[{"text":"hi"}]}""", NullLogger.Instance));
}
public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } }
private sealed class Fake(Func<HttpRequestMessage, HttpResponseMessage> respond) : HttpMessageHandler
{
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) =>
Task.FromResult(respond(request));
}
}