diff --git a/src/Wiki.Adapters.Anthropic/AnthropicOptions.cs b/src/Wiki.Adapters.Anthropic/AnthropicOptions.cs index a425a22..d46b319 100644 --- a/src/Wiki.Adapters.Anthropic/AnthropicOptions.cs +++ b/src/Wiki.Adapters.Anthropic/AnthropicOptions.cs @@ -27,6 +27,12 @@ public sealed class AnthropicOptions /// Per-request timeout. public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(120); + /// + /// Longest wait for the next line of a streamed response. The API sends ping events while a reply is + /// pending, so a longer silence is a stalled stream. + /// + public TimeSpan StreamIdleTimeout { get; set; } = TimeSpan.FromSeconds(90); + /// How many times a 429/529 response is retried before giving up. public int MaxRetries { get; set; } = 5; diff --git a/src/Wiki.Adapters.Anthropic/MessagesTransport.cs b/src/Wiki.Adapters.Anthropic/MessagesTransport.cs index 52e88fb..d620455 100644 --- a/src/Wiki.Adapters.Anthropic/MessagesTransport.cs +++ b/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 } } + /// + /// The next line, waiting at most . A stalled stream throws an + /// without a status code, the same as a dropped connection. + /// + private async Task 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 CheckAsync(string name, CancellationToken ct) { var stopwatch = Stopwatch.StartNew(); diff --git a/test/Wiki.Application.Tests/EncyclopediaReconcilerTests.cs b/test/Wiki.Application.Tests/EncyclopediaReconcilerTests.cs index 5648793..fcf1902 100644 --- a/test/Wiki.Application.Tests/EncyclopediaReconcilerTests.cs +++ b/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() { diff --git a/test/Wiki.Contracts/Anthropic/AnthropicTextModelTests.cs b/test/Wiki.Contracts/Anthropic/AnthropicTextModelTests.cs index 0dd8bf3..61623d6 100644 --- a/test/Wiki.Contracts/Anthropic/AnthropicTextModelTests.cs +++ b/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( + () => model.CompleteAsync("sys", "user", 16000, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(10))); + + Assert.IsType(ex.InnerException); + Assert.Null(ex.StatusCode); + } + [Fact] public async Task Complete_PassesMaxTokensAndSystem() { diff --git a/test/Wiki.Contracts/Anthropic/FakeAnthropicHandler.cs b/test/Wiki.Contracts/Anthropic/FakeAnthropicHandler.cs index a7bf583..79ebb78 100644 --- a/test/Wiki.Contracts/Anthropic/FakeAnthropicHandler.cs +++ b/test/Wiki.Contracts/Anthropic/FakeAnthropicHandler.cs @@ -29,6 +29,9 @@ internal sealed class FakeAnthropicHandler : HttpMessageHandler /// The stop_reason reported in the final message_delta. public string StopReason { get; set; } = "end_turn"; + /// When set, the stream sends its message_start event and then nothing, until the request is cancelled. + public bool StallAfterFirstEvent { get; set; } + /// When set, every POST is answered with this status and . 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(); } + /// Reads prefix, then blocks every read until it is cancelled. + 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 ReadAsync(Memory 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 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();