fix(anthropic): fail a stalled response stream after an idle timeout

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/Wiki.Adapters.Anthropic/AnthropicOptions.cs

@@ -27,6 +27,12 @@ public sealed class AnthropicOptions
     /// <summary>Per-request timeout.</summary>
     public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(120);
 
+    /// <summary>
+    /// Longest wait for the next line of a streamed response. The API sends <c>ping</c> events while a reply is
+    /// pending, so a longer silence is a stalled stream.
+    /// </summary>
+    public TimeSpan StreamIdleTimeout { get; set; } = TimeSpan.FromSeconds(90);
+
     /// <summary>How many times a 429/529 response is retried before giving up.</summary>
     public int MaxRetries { get; set; } = 5;
 

src/Wiki.Adapters.Anthropic/MessagesTransport.cs

@@ -50,6 +50,7 @@ internal sealed class MessagesTransport
         var stream = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
         await using (stream.ConfigureAwait(false))
         using (var reader = new StreamReader(stream, Encoding.UTF8))
+        using (var idle = CancellationTokenSource.CreateLinkedTokenSource(ct))
         {
             long inputTokens = 0;
             long outputTokens = 0;
@@ -57,7 +58,7 @@ internal sealed class MessagesTransport
             string? stopReason = null;
 
             string? line;
-            while ((line = await reader.ReadLineAsync(ct).ConfigureAwait(false)) is not null)
+            while ((line = await ReadLineAsync(reader, idle, ct).ConfigureAwait(false)) is not null)
             {
                 if (line.Length == 0)
                 {
@@ -128,6 +129,30 @@ internal sealed class MessagesTransport
         }
     }
 
+    /// <summary>
+    /// The next line, waiting at most <see cref="AnthropicOptions.StreamIdleTimeout"/>. A stalled stream throws an
+    /// <see cref="HttpRequestException"/> without a status code, the same as a dropped connection.
+    /// </summary>
+    private async Task<string?> ReadLineAsync(StreamReader reader, CancellationTokenSource idle, CancellationToken ct)
+    {
+        idle.CancelAfter(_options.StreamIdleTimeout);
+        try
+        {
+            return await reader.ReadLineAsync(idle.Token).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException e) when (!ct.IsCancellationRequested)
+        {
+            throw new HttpRequestException(HttpRequestError.ResponseEnded,
+                $"anthropic: no data for {_options.StreamIdleTimeout.TotalSeconds:0.#} s; the stream stalled",
+                new TimeoutException("The streamed response stalled.", e));
+        }
+        finally
+        {
+            // the consumer's time between reads does not count as idle
+            idle.CancelAfter(Timeout.InfiniteTimeSpan);
+        }
+    }
+
     public async Task<AdapterHealth> CheckAsync(string name, CancellationToken ct)
     {
         var stopwatch = Stopwatch.StartNew();

test/Wiki.Application.Tests/EncyclopediaReconcilerTests.cs

@@ -280,6 +280,20 @@ public class EncyclopediaReconcilerTests
         Assert.Equal(["heating", "heating"], _writer.Calls);
     }
 
+    [Fact]
+    public async Task Reconcile_StalledStream_CommitsOthersNotRemembered()
+    {
+        await SeedAsync();
+        _planModel.Reply(PlanReply(HeatingAndKitchen));
+        _writer.Throws["heating"] = new HttpRequestException(HttpRequestError.ResponseEnded, "stream stalled", new TimeoutException());
+
+        var summary = await Reconciler().RunAsync(CancellationToken.None);
+
+        Assert.Equal((1, 1), (summary.Written, summary.Rejected));
+        Assert.NotNull(summary.Commit);
+        Assert.Equal(["kitchen"], State.Subjects.Keys);
+    }
+
     [Fact]
     public async Task Reconcile_PlanFailedSameNotes_DoesNotReplan()
     {

test/Wiki.Contracts/Anthropic/AnthropicTextModelTests.cs

@@ -80,6 +80,25 @@ public sealed class AnthropicTextModelTests
         Assert.Equal(HttpStatusCode.BadRequest, ex.StatusCode);
     }
 
+    [Fact]
+    public async Task Complete_StalledStream_ThrowsHttpRequestAfterIdleTimeout()
+    {
+        var model = new AnthropicTextModel(
+            new HttpClient(new FakeAnthropicHandler { StallAfterFirstEvent = true }),
+            new AnthropicOptions
+            {
+                BaseUrl = "https://anthropic.test", Model = "claude-sonnet-5", ApiKey = "test-key",
+                StreamIdleTimeout = TimeSpan.FromMilliseconds(100),
+            },
+            delay: (_, _) => Task.CompletedTask);
+
+        var ex = await Assert.ThrowsAsync<HttpRequestException>(
+            () => model.CompleteAsync("sys", "user", 16000, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(10)));
+
+        Assert.IsType<TimeoutException>(ex.InnerException);
+        Assert.Null(ex.StatusCode);
+    }
+
     [Fact]
     public async Task Complete_PassesMaxTokensAndSystem()
     {

test/Wiki.Contracts/Anthropic/FakeAnthropicHandler.cs

@@ -29,6 +29,9 @@ internal sealed class FakeAnthropicHandler : HttpMessageHandler
     /// <summary>The stop_reason reported in the final message_delta.</summary>
     public string StopReason { get; set; } = "end_turn";
 
+    /// <summary>When set, the stream sends its message_start event and then nothing, until the request is cancelled.</summary>
+    public bool StallAfterFirstEvent { get; set; }
+
     /// <summary>When set, every POST is answered with this status and <see cref="FailureBody"/>.</summary>
     public HttpStatusCode? FailWithStatus { get; set; }
 
@@ -96,6 +99,15 @@ internal sealed class FakeAnthropicHandler : HttpMessageHandler
 
         var sse = RespondWithErrorEvent ? BuildErrorSse(ErrorEventType) : BuildSuccessSse(body.Length, ResponseText, StopReason);
 
+        if (StallAfterFirstEvent)
+        {
+            var first = sse[..(sse.IndexOf("\n\n", StringComparison.Ordinal) + 2)];
+            return new HttpResponseMessage(HttpStatusCode.OK)
+            {
+                Content = new StreamContent(new StallingStream(Encoding.UTF8.GetBytes(first))),
+            };
+        }
+
         return new HttpResponseMessage(HttpStatusCode.OK)
         {
             Content = new StringContent(sse, Encoding.UTF8, "text/event-stream"),
@@ -142,6 +154,40 @@ internal sealed class FakeAnthropicHandler : HttpMessageHandler
         return sb.ToString();
     }
 
+    /// <summary>Reads <c>prefix</c>, then blocks every read until it is cancelled.</summary>
+    private sealed class StallingStream(byte[] prefix) : Stream
+    {
+        private int _position;
+
+        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 async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            if (_position < prefix.Length)
+            {
+                var n = Math.Min(buffer.Length, prefix.Length - _position);
+                prefix.AsMemory(_position, n).CopyTo(buffer);
+                _position += n;
+                return n;
+            }
+            await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false);
+            return 0;
+        }
+
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
+            ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
+
+        public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+        public override void Flush() { }
+        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();
+    }
+
     private static string BuildErrorSse(string errorType)
     {
         var sb = new StringBuilder();