feat(a2a-client): persist watches in pending.json

extensions/Volcanic.AgenticCore.Extensions.A2A/PendingFile.cs

@@ -0,0 +1,69 @@
+using System.Collections.Concurrent;
+using System.Text.Json;
+
+namespace Volcanic.AgenticCore.Extensions.A2A;
+
+public static class WatchSources
+{
+    public const string Push = "push";
+    public const string Subscribe = "subscribe";
+    public const string Poll = "poll";
+}
+
+/// <param name="Target">The host's delivery target (Telegram: the chat id).</param>
+/// <param name="PushToken">Our secret for this watch's push webhook; null when not using push.</param>
+/// <param name="Undelivered">A finished result that couldn't be delivered yet (sent on the target's next session start).</param>
+public sealed record WatchRecord(
+    string WatchId, string Target, string Agent, string TaskId, string ContextId, string? PushToken,
+    DateTimeOffset StartedAt, string Source, string? Undelivered = null);
+
+/// <summary><c>&lt;workspace&gt;/.volcanic/a2a/pending.json</c>: the watches of a workspace, so they survive restarts.</summary>
+public static class PendingFile
+{
+    private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web) { WriteIndented = true };
+    private static readonly ConcurrentDictionary<string, object> Locks = new(StringComparer.Ordinal);
+
+    public static string PathFor(string workspace) => Path.Combine(workspace, ".volcanic", "a2a", "pending.json");
+
+    public static IReadOnlyList<WatchRecord> Read(string workspace)
+    {
+        lock (LockFor(workspace)) return ReadUnlocked(workspace);
+    }
+
+    public static void Upsert(string workspace, WatchRecord record) => Change(workspace, list =>
+    {
+        list.RemoveAll(r => r.WatchId == record.WatchId);
+        list.Add(record);
+    });
+
+    public static void Remove(string workspace, string watchId) => Change(workspace, list => list.RemoveAll(r => r.WatchId == watchId));
+
+    private static void Change(string workspace, Action<List<WatchRecord>> change)
+    {
+        lock (LockFor(workspace))
+        {
+            var list = ReadUnlocked(workspace).ToList();
+            change(list);
+            var path = PathFor(workspace);
+            Directory.CreateDirectory(Path.GetDirectoryName(path)!);
+            var temp = path + "." + Guid.NewGuid().ToString("N") + ".tmp";
+            File.WriteAllText(temp, JsonSerializer.Serialize(list, Json));
+            File.Move(temp, path, overwrite: true);
+        }
+    }
+
+    private static List<WatchRecord> ReadUnlocked(string workspace)
+    {
+        var path = PathFor(workspace);
+        try
+        {
+            return File.Exists(path) ? JsonSerializer.Deserialize<List<WatchRecord>>(File.ReadAllText(path), Json) ?? [] : [];
+        }
+        catch (Exception ex) when (ex is JsonException or IOException or UnauthorizedAccessException)
+        {
+            return [];
+        }
+    }
+
+    private static object LockFor(string workspace) => Locks.GetOrAdd(Path.GetFullPath(workspace), _ => new object());
+}

tests/Volcanic.AgenticCore.Extensions.Tests/A2A/PendingFileTests.cs

@@ -0,0 +1,35 @@
+using Volcanic.AgenticCore.Extensions.A2A;
+
+namespace Volcanic.AgenticCore.Extensions.Tests.A2A;
+
+public sealed class PendingFileTests : IDisposable
+{
+    private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-pending", Guid.NewGuid().ToString("N"));
+
+    private static WatchRecord Record(string id, string target = "1") =>
+        new(id, target, "travel-bot", "task-" + id, "ctx", null, DateTimeOffset.UnixEpoch, WatchSources.Poll);
+
+    [Fact]
+    public void AddUpdateRemove_RoundTrips()
+    {
+        PendingFile.Upsert(_ws, Record("a"));
+        PendingFile.Upsert(_ws, Record("b"));
+        PendingFile.Upsert(_ws, Record("a") with { Undelivered = "text" });
+        var all = PendingFile.Read(_ws);
+        Assert.Equal(2, all.Count);
+        Assert.Equal("text", all.Single(r => r.WatchId == "a").Undelivered);
+        PendingFile.Remove(_ws, "a");
+        Assert.Equal("b", Assert.Single(PendingFile.Read(_ws)).WatchId);
+    }
+
+    [Fact]
+    public void Read_MissingOrMalformed_Empty()
+    {
+        Assert.Empty(PendingFile.Read(_ws));
+        Directory.CreateDirectory(Path.GetDirectoryName(PendingFile.PathFor(_ws))!);
+        File.WriteAllText(PendingFile.PathFor(_ws), "{ broken");
+        Assert.Empty(PendingFile.Read(_ws));
+    }
+
+    public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } }
+}