From 0028967c14cf5317b027628f9936705647316d21 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 3 Jun 2026 08:51:04 +0000 Subject: [PATCH 1/4] feat(providers): streaming-native chat-client stack with composable routing Redesign the IChatClient stack so the streaming-vs-non-streaming transport distinction no longer leaks into callers, and resilience/observability are no longer entangled with transport. Supersedes #1289 and #1309. Transport: Netclaw now issues only streaming requests. RetryingChatClient retries the streaming path (pre-first-chunk; a mid-stream failure propagates so emitted output is never duplicated), reusing RetryPolicy. The auxiliary callers (title generation, memory distillation, compaction observation, memory curation, memory extraction) aggregate the stream via the existing StreamingResponseReader (empty-Messages guard + diagnostics) and read response.Text, instead of the non-streaming GetResponseAsync path that the OpenAI Codex backend rejects (400 'Stream must be set to true') and that drops reasoning content on some providers. Composition: per-(provider,model) pipelines are composed with Microsoft.Extensions.AI ChatClientBuilder (Logging -> Retry -> VendorOptions -> raw) in PipelineChatClientFactory; the .Use ordering is asserted by test. Routing: a router selects which composed pipeline to invoke per call. RoutingChatClient walks an ordered candidate list, so failover is 'more than one candidate' and single-provider alerting is the terminal failure of a one-entry walk -- absorbing and deleting FailoverChatClient, AlertingChatClientDecorator, NetclawChatClientProvider and ResilientChatClientProviderDecorator. The router takes a ChatRoutingContext so future per-session/per-provider routing slots in as a new policy; today's RoleBasedFailoverRouter reproduces the prior wiring. Logging: LoggingChatClient is now stateless (the per-session-intended delta/cumulative token counters on the process singleton are gone) and tags log lines with the ambient session id via SessionLoggingScope, keyed 'SessionId' to match the WithContext('SessionId', ...) key used across actors so one Seq attribute correlates actor and chat-client logs. Verified: full solution build (0 warnings); Netclaw.Daemon.Tests 668/668; Netclaw.Actors.Tests 2186/2186; slopwatch clean; file headers present. --- Directory.Packages.props | 1 + .../Memory/MemoryCurationActor.cs | 5 +- .../Sessions/LlmSessionActor.cs | 6 +- .../Pipelines/SessionCompactionPipeline.cs | 6 +- .../Pipelines/SessionTitleGenerator.cs | 4 +- .../Pipelines/StreamingResponseReader.cs | 17 ++ .../Sessions/SessionMemoryObserverActor.cs | 5 +- .../ChatRoutingContext.cs | 30 +++ .../Configuration/FailoverChatClientTests.cs | 194 --------------- .../Configuration/LoggingChatClientTests.cs | 196 ++++----------- .../PipelineChatClientFactoryTests.cs | 73 ++++++ ...silientChatClientProviderDecoratorTests.cs | 128 ---------- .../Configuration/RetryingChatClientTests.cs | 105 +++++++- .../RoutingChatClientProviderTests.cs | 82 +++++++ .../Configuration/RoutingChatClientTests.cs | 216 ++++++++++++++++ .../Configuration/ScopeCapturingLogger.cs | 45 ++++ .../AlertingChatClientDecorator.cs | 95 -------- .../Configuration/ChatClientRouter.cs | 71 ++++++ .../DaemonProviderServiceExtensions.cs | 43 ++-- .../Configuration/FailoverChatClient.cs | 202 --------------- .../Configuration/LoggingChatClient.cs | 89 +------ .../NetclawChatClientProvider.cs | 37 --- .../PipelineChatClientFactory.cs | 65 +++++ .../ResilientChatClientProviderDecorator.cs | 85 ------- .../Configuration/RetryingChatClient.cs | 87 ++++++- .../Configuration/RoutingChatClient.cs | 230 ++++++++++++++++++ .../Configuration/SessionLoggingScope.cs | 44 ++++ src/Netclaw.Daemon/Netclaw.Daemon.csproj | 1 + 28 files changed, 1156 insertions(+), 1006 deletions(-) create mode 100644 src/Netclaw.Configuration/ChatRoutingContext.cs delete mode 100644 src/Netclaw.Daemon.Tests/Configuration/FailoverChatClientTests.cs create mode 100644 src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs delete mode 100644 src/Netclaw.Daemon.Tests/Configuration/ResilientChatClientProviderDecoratorTests.cs create mode 100644 src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientProviderTests.cs create mode 100644 src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs create mode 100644 src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs delete mode 100644 src/Netclaw.Daemon/Configuration/AlertingChatClientDecorator.cs create mode 100644 src/Netclaw.Daemon/Configuration/ChatClientRouter.cs delete mode 100644 src/Netclaw.Daemon/Configuration/FailoverChatClient.cs delete mode 100644 src/Netclaw.Daemon/Configuration/NetclawChatClientProvider.cs create mode 100644 src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs delete mode 100644 src/Netclaw.Daemon/Configuration/ResilientChatClientProviderDecorator.cs create mode 100644 src/Netclaw.Daemon/Configuration/RoutingChatClient.cs create mode 100644 src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index e2c815ea6..40230730a 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -32,6 +32,7 @@ + diff --git a/src/Netclaw.Actors/Memory/MemoryCurationActor.cs b/src/Netclaw.Actors/Memory/MemoryCurationActor.cs index 4615c8845..ae7b9fda9 100644 --- a/src/Netclaw.Actors/Memory/MemoryCurationActor.cs +++ b/src/Netclaw.Actors/Memory/MemoryCurationActor.cs @@ -6,6 +6,7 @@ using Akka.Actor; using Akka.Event; using Microsoft.Extensions.AI; +using Netclaw.Actors.Sessions.Pipelines; using Netclaw.Configuration; using SessionId = Netclaw.Actors.Protocol.SessionId; @@ -315,8 +316,8 @@ private async Task EvaluateSingleAsync( MaxOutputTokens = 512 }; - var response = await llmClient.GetResponseAsync(messages, options, cts.Token); - var responseText = response.Text?.Trim(); + var result = await StreamingResponseReader.ReadAsync(llmClient, messages, options, cts.Token); + var responseText = result.Response.Text?.Trim(); var decision = string.IsNullOrWhiteSpace(responseText) ? null : CurationPromptBuilder.ParseResponse(responseText); diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 82d3f0890..37a716a70 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -1641,9 +1641,9 @@ internal static async Task InvokeMemoryExtractionCoreAsync( new(Microsoft.Extensions.AI.ChatRole.User, CompactionPromptBuilder.BuildMemoryExtractionUserPrompt(history)) }; - var extractionResponse = await client.GetResponseAsync(extractionMessages, - cancellationToken: cts.Token); - var extractedText = extractionResponse.Messages[^1].Text ?? string.Empty; + var extractionResult = await StreamingResponseReader.ReadAsync( + client, extractionMessages, options: null, cts.Token); + var extractedText = extractionResult.Response.Text ?? string.Empty; self.Tell(new MemoryExtractionCompleted { ExtractedMemories = extractedText }); } catch (Exception ex) diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs index 4fc8aa90a..660bd8cc7 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs @@ -182,9 +182,9 @@ public static async Task ExecuteAsync( ObservationPromptBuilder.BuildObservationUserPrompt(remainingDiscarded)) }; - var response = await client.GetResponseAsync( - observerMessages, cancellationToken: cts.Token); - var text = response.Messages[^1].Text; + var result = await StreamingResponseReader.ReadAsync( + client, observerMessages, options: null, cts.Token); + var text = result.Response.Text; if (string.IsNullOrWhiteSpace(text)) { diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs index dd4648fe4..022e6e24f 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs @@ -49,8 +49,8 @@ public static async Task GenerateAsync( new(Microsoft.Extensions.AI.ChatRole.User, CompactionPromptBuilder.BuildTitleGenerationPrompt(history)) }; - var response = await client.GetResponseAsync(messages, cancellationToken: cts.Token); - var title = response.Messages[^1].Text ?? string.Empty; + var result = await StreamingResponseReader.ReadAsync(client, messages, options: null, cts.Token); + var title = result.Response.Text ?? string.Empty; if (string.IsNullOrWhiteSpace(title)) { diff --git a/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs b/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs index 4d4840977..a999f0aea 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs @@ -56,6 +56,23 @@ internal readonly record struct StreamReadResult( /// internal static class StreamingResponseReader { + private static readonly Action NoOp = + static (_, _, _) => { }; + + /// + /// Convenience overload for callers that only want the aggregated response with + /// no per-update dispatch — title generation, memory distillation, compaction + /// observation, memory curation. Streams under the hood (the only transport + /// Netclaw issues) and aggregates with the same empty-response fallback, so the + /// caller can read result.Response.Text without an empty-Messages guard. + /// + public static Task ReadAsync( + IChatClient client, + IEnumerable messages, + ChatOptions? options, + CancellationToken ct) + => ReadAsync(client, messages, options, NoOp, ct); + public static async Task ReadAsync( IChatClient client, IEnumerable messages, diff --git a/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs b/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs index fcbfe4f39..fc12b30c5 100644 --- a/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs +++ b/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs @@ -12,6 +12,7 @@ using Netclaw.Actors.Memory; using Netclaw.Actors.Protocol; using Netclaw.Actors.Serialization; +using Netclaw.Actors.Sessions.Pipelines; using Netclaw.Actors.SubAgents; using Netclaw.Configuration; @@ -354,8 +355,8 @@ internal static async Task RunDistillationAsync( sessionId, turnCount, transcript, existingProposals)) }; - var response = await client.GetResponseAsync(messages, cancellationToken: cts.Token); - var text = response.Messages[^1].Text ?? string.Empty; + var response = (await StreamingResponseReader.ReadAsync(client, messages, options: null, cts.Token)).Response; + var text = response.Text ?? string.Empty; inputTokens = response.Usage?.InputTokenCount; outputTokens = response.Usage?.OutputTokenCount; diff --git a/src/Netclaw.Configuration/ChatRoutingContext.cs b/src/Netclaw.Configuration/ChatRoutingContext.cs new file mode 100644 index 000000000..f5119604c --- /dev/null +++ b/src/Netclaw.Configuration/ChatRoutingContext.cs @@ -0,0 +1,30 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +namespace Netclaw.Configuration; + +/// +/// The inputs a chat-client router uses to select which composed pipeline to invoke +/// for a call. Minimal today — only is consulted — but shaped so +/// per-session / per-provider routing slots in later as a new router policy without +/// changing this type's shape or any caller. +/// +public sealed record ChatRoutingContext +{ + /// The model role being requested (today's only routing signal). + public required ModelRole Role { get; init; } + + /// + /// Owning session id. Unused today; a future per-session router policy reads this + /// to route a session's chats to a specific model/provider. + /// + public string? SessionId { get; init; } + + /// + /// Index of the current failover attempt within a single call. Unused today; a + /// future policy can use it to re-rank candidates across attempts. + /// + public int AttemptIndex { get; init; } +} diff --git a/src/Netclaw.Daemon.Tests/Configuration/FailoverChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/FailoverChatClientTests.cs deleted file mode 100644 index eb2dedcfd..000000000 --- a/src/Netclaw.Daemon.Tests/Configuration/FailoverChatClientTests.cs +++ /dev/null @@ -1,194 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.AI; -using Microsoft.Extensions.Logging.Abstractions; -using Netclaw.Configuration; -using Netclaw.Daemon.Configuration; -using Xunit; - -namespace Netclaw.Daemon.Tests.Configuration; - -public sealed class FailoverChatClientTests -{ - [Fact] - public async Task UsesPrimary_WhenHealthy() - { - var primary = new FakeChatClient((_,_,_) => - Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "primary")]))); - var fallback = new FakeChatClient((_,_,_) => - Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")]))); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - var response = await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - - Assert.Equal("primary", response.Messages[0].Text); - } - - [Fact] - public async Task FallsBack_WhenPrimaryFails() - { - var primary = new FakeChatClient((_,_,_) => - throw new HttpRequestException("primary down")); - var fallback = new FakeChatClient((_,_,_) => - Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")]))); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - var response = await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - - Assert.Equal("fallback", response.Messages[0].Text); - } - - [Fact] - public async Task PropagatesException_WhenBothFail() - { - var primary = new FakeChatClient((_,_,_) => - throw new HttpRequestException("primary down")); - var fallback = new FakeChatClient((_,_,_) => - throw new HttpRequestException("fallback down")); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - - var ex = await Assert.ThrowsAsync(() => - client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)); - Assert.Contains("fallback down", ex.Message); - } - - [Fact] - public async Task DoesNotFallback_OnCancellation() - { - var cts = new CancellationTokenSource(); - cts.Cancel(); - - var primary = new FakeChatClient((_,_,ct) => - throw new OperationCanceledException(ct)); - var fallback = new FakeChatClient((_,_,_) => - Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")]))); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - - await Assert.ThrowsAnyAsync(() => - client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], - cancellationToken: cts.Token)); - } - - [Fact] - public async Task Streaming_UsesPrimary_WhenHealthy() - { - var primary = new FakeChatClient(streaming: true); - var fallback = new FakeChatClient(streaming: true); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - - var updates = new List(); - await foreach (var u in client.GetStreamingResponseAsync( - [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) - { - updates.Add(u); - } - - Assert.Single(updates); - Assert.Equal("streamed", updates[0].Text); - } - - [Fact] - public async Task Streaming_FallsBack_WhenPrimaryFailsBeforeFirstChunk() - { - var fallbackStreamCalls = 0; - - var primary = new FakeChatClient(streamHandler: (_, _, ct) => - ThrowBeforeFirstChunkAsync(ct)); - var fallback = new FakeChatClient(streamHandler: (_, _, ct) => - { - fallbackStreamCalls++; - return SingleTextUpdateAsync("fallback", ct); - }); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - - var updates = new List(); - await foreach (var u in client.GetStreamingResponseAsync( - [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) - { - updates.Add(u); - } - - Assert.Single(updates); - Assert.Equal("fallback", updates[0].Text); - Assert.Equal(1, fallbackStreamCalls); - } - - [Fact] - public async Task Streaming_DoesNotFallback_AfterPrimaryAlreadyYielded() - { - var fallbackStreamCalls = 0; - - var primary = new FakeChatClient(streamHandler: (_, _, ct) => - YieldThenThrowAsync(ct)); - var fallback = new FakeChatClient(streamHandler: (_, _, ct) => - { - fallbackStreamCalls++; - return SingleTextUpdateAsync("fallback", ct); - }); - - var client = new FailoverChatClient(primary, fallback, NullLogger.Instance, NullNotificationSink.Instance, TimeProvider.System); - var updates = new List(); - - var ex = await Assert.ThrowsAsync(async () => - { - await foreach (var u in client.GetStreamingResponseAsync( - [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) - { - updates.Add(u); - } - }); - - Assert.Contains("primary stream failed", ex.Message); - Assert.Single(updates); - Assert.Equal("primary", updates[0].Text); - Assert.Equal(0, fallbackStreamCalls); - } - - private static async IAsyncEnumerable ThrowBeforeFirstChunkAsync( - [System.Runtime.CompilerServices.EnumeratorCancellation] - CancellationToken cancellationToken) - { - await Task.Yield(); - cancellationToken.ThrowIfCancellationRequested(); - throw new HttpRequestException("primary stream failed before first chunk"); -#pragma warning disable CS0162 // Unreachable code - yield break; -#pragma warning restore CS0162 // Unreachable code - } - - private static async IAsyncEnumerable YieldThenThrowAsync( - [System.Runtime.CompilerServices.EnumeratorCancellation] - CancellationToken cancellationToken) - { - await Task.Yield(); - cancellationToken.ThrowIfCancellationRequested(); - yield return new ChatResponseUpdate - { - Role = ChatRole.Assistant, - Contents = [new TextContent("primary")] - }; - - throw new HttpRequestException("primary stream failed after first chunk"); - } - - private static async IAsyncEnumerable SingleTextUpdateAsync( - string text, - [System.Runtime.CompilerServices.EnumeratorCancellation] - CancellationToken cancellationToken) - { - await Task.Yield(); - cancellationToken.ThrowIfCancellationRequested(); - yield return new ChatResponseUpdate - { - Role = ChatRole.Assistant, - Contents = [new TextContent(text)] - }; - } -} diff --git a/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs index 541cd079c..9a7ec0c0e 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs @@ -1,91 +1,67 @@ -// ----------------------------------------------------------------------- +// ----------------------------------------------------------------------- // // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Configuration; using Netclaw.Daemon.Configuration; using Xunit; +// Netclaw's LoggingChatClient collides with Microsoft.Extensions.AI.LoggingChatClient +// (the core package's own logging decorator) now that the core package is referenced. +using LoggingChatClient = Netclaw.Daemon.Configuration.LoggingChatClient; namespace Netclaw.Daemon.Tests.Configuration; public sealed class LoggingChatClientTests { [Fact] - public async Task LogsCompletionTime() + public async Task Streaming_LogsCompletion() { var logs = new List(); - var logger = new CapturingLogger(logs); - var fake = new FakeChatClient(); + var client = new LoggingChatClient(new FakeChatClient(streaming: true), new CapturingLogger(logs)); - var client = new LoggingChatClient(fake, logger); - var response = await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); + await Drain(client); - Assert.NotNull(response); - Assert.Contains(logs, l => l.Contains("LLM call completed")); + Assert.Contains(logs, l => l.Contains("LLM streaming call completed")); } [Fact] - public async Task LogsTokenUsage() + public async Task Streaming_LogsTokenUsageWithoutDeltaOrCumulative() { var logs = new List(); - var logger = new CapturingLogger(logs); - var fake = new FakeChatClient((_,_,_) => - { - var response = new ChatResponse([new ChatMessage(ChatRole.Assistant, "ok")]) - { - Usage = new UsageDetails { InputTokenCount = 10, OutputTokenCount = 20 } - }; - return Task.FromResult(response); - }); + var client = new LoggingChatClient( + new FakeChatClient(streamHandler: (_, _, _) => StreamWithUsage(100, 20)), + new CapturingLogger(logs)); - var client = new LoggingChatClient(fake, logger); - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); + await Drain(client); - Assert.Contains(logs, l => l.Contains("input: 10") && l.Contains("output: 20")); + Assert.Contains(logs, l => l.Contains("input: 100") && l.Contains("output: 20")); + // The cross-call delta/cumulative counters were removed (stateless client). + Assert.DoesNotContain(logs, l => l.Contains("delta:") || l.Contains("cumulative:")); } [Fact] - public async Task LogsErrorOnFailure() + public async Task Streaming_LogsErrorOnInitFailure() { var logs = new List(); - var logger = new CapturingLogger(logs); - var fake = new FakeChatClient((_,_,_) => - throw new HttpRequestException("boom")); + var client = new LoggingChatClient( + new FakeChatClient(streamHandler: (_, _, _) => throw new HttpRequestException("boom")), + new CapturingLogger(logs)); - var client = new LoggingChatClient(fake, logger); + await Assert.ThrowsAsync(() => Drain(client)); - await Assert.ThrowsAsync(() => - client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)); - - Assert.Contains(logs, l => l.Contains("LLM call failed")); + Assert.Contains(logs, l => l.Contains("LLM streaming call failed")); } [Fact] - public async Task Streaming_LogsCompletion() + public async Task Streaming_LogsPromptSummaryInDebugMode() { var logs = new List(); - var logger = new CapturingLogger(logs); - var fake = new FakeChatClient(streaming: true); + var client = new LoggingChatClient(new FakeChatClient(streaming: true), new CapturingLogger(logs)); - var client = new LoggingChatClient(fake, logger); await foreach (var _ in client.GetStreamingResponseAsync( - [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) { } - - Assert.Contains(logs, l => l.Contains("LLM streaming call completed")); - } - - [Fact] - public async Task LogsPromptSummaryInDebugMode() - { - var logs = new List(); - var logger = new CapturingLogger(logs); - var fake = new FakeChatClient(); - - var client = new LoggingChatClient(fake, logger); - await client.GetResponseAsync( [ new ChatMessage(ChatRole.System, "sys"), new ChatMessage(ChatRole.User, "hello") @@ -97,7 +73,9 @@ await client.GetResponseAsync( AIFunctionFactory.Create((string query) => query, "search_tools"), AIFunctionFactory.Create((string url) => url, "browser_playwright/browser_navigate") ] - }, cancellationToken: TestContext.Current.CancellationToken); + }, TestContext.Current.CancellationToken)) + { + } Assert.Contains(logs, l => l.Contains("LLM prompt summary")); Assert.Contains(logs, l => l.Contains("promptSha256=")); @@ -105,125 +83,55 @@ await client.GetResponseAsync( } [Fact] - public async Task LogsPromptDumpWhenTraceEnabled() + public async Task Streaming_LogsPromptDumpWhenTraceEnabled() { var logs = new List(); - var logger = new CapturingLogger(logs); - var fake = new FakeChatClient(); + var client = new LoggingChatClient(new FakeChatClient(streaming: true), new CapturingLogger(logs)); - var client = new LoggingChatClient(fake, logger); - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hello")], cancellationToken: TestContext.Current.CancellationToken); + await Drain(client); Assert.Contains(logs, l => l.Contains("LLM prompt dump:")); Assert.Contains(logs, l => l.Contains("role=user")); } [Fact] - public async Task LogsTokenDeltaAcrossMultipleCalls() + public async Task Streaming_attaches_SessionId_scope_from_diagnostics_context() { - var logs = new List(); - var logger = new CapturingLogger(logs); - var callCount = 0; + var logger = new ScopeCapturingLogger(); + var client = new LoggingChatClient(new FakeChatClient(streaming: true), logger); - var fake = new FakeChatClient((_, _, _) => + using (SessionDiagnosticsContext.Push("ch/thread")) { - callCount++; - // Return increasing input tokens: 100, 150, 200 - var inputTokens = callCount * 50; - return Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "ok")]) - { - Usage = new UsageDetails { InputTokenCount = inputTokens, OutputTokenCount = 10 } - }); - }); - - var client = new LoggingChatClient(fake, logger); - - // First call: 50 tokens - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - // Second call: 100 tokens - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - // Third call: 150 tokens - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - - // First call has no previous, so delta is N/A - Assert.Contains(logs, l => l.Contains("delta: N/A")); - // Second and third calls have delta of +50 each - Assert.Equal(2, logs.Count(l => l.Contains("delta: +50"))); + await Drain(client); + } + + Assert.True(logger.HasSessionScope("ch/thread")); } [Fact] - public async Task LogsCumulativeTokensAcrossMultipleCalls() + public async Task Streaming_without_session_context_attaches_no_scope() { - var logs = new List(); - var logger = new CapturingLogger(logs); - var callCount = 0; + Assert.Null(SessionDiagnosticsContext.SessionId); + var logger = new ScopeCapturingLogger(); + var client = new LoggingChatClient(new FakeChatClient(streaming: true), logger); - var fake = new FakeChatClient((_, _, _) => - { - callCount++; - // Return input tokens: 50, 100, 150 - var inputTokens = callCount * 50; - return Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "ok")]) - { - Usage = new UsageDetails { InputTokenCount = inputTokens, OutputTokenCount = 10 } - }); - }); - - var client = new LoggingChatClient(fake, logger); - - // First call: 50 tokens - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - // Second call: 100 tokens - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - // Third call: 150 tokens - await client.GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); - - // Verify cumulative: 50, 150, 300 - Assert.Contains(logs, l => l.Contains("cumulative: 50")); - Assert.Contains(logs, l => l.Contains("cumulative: 150")); - Assert.Contains(logs, l => l.Contains("cumulative: 300")); + await Drain(client); + + Assert.False(logger.HasAnySessionScope()); } - [Fact] - public async Task Streaming_LogsTokenDeltaAndCumulative() + private static async Task Drain(IChatClient client) { - var logs = new List(); - var logger = new CapturingLogger(logs); - var callCount = 0; - - var fake = new FakeChatClient(streaming: true, streamHandler: (messages, options, ct) => + await foreach (var _ in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) { - callCount++; - // Return 100 input tokens on each call - return StreamWithUsage(messages, 100, 10); - }); - - var client = new LoggingChatClient(fake, logger); - - // First streaming call - await foreach (var _ in client.GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) { } - // Second streaming call - await foreach (var _ in client.GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) { } - - // Verify delta and cumulative appear in streaming logs - Assert.Contains(logs, l => l.Contains("delta:") && l.Contains("cumulative:")); - // First call should have delta N/A (no previous), cumulative 100 - Assert.Contains(logs, l => l.Contains("delta: N/A") && l.Contains("cumulative: 100")); - // Second call should have delta 0, cumulative 200 - Assert.Contains(logs, l => l.Contains("delta: 0") && l.Contains("cumulative: 200")); + } } - private static async IAsyncEnumerable StreamWithUsage( - IEnumerable messages, int inputTokens, int outputTokens) + private static async IAsyncEnumerable StreamWithUsage(int inputTokens, int outputTokens) { await Task.CompletedTask; - // Yield a text update - yield return new ChatResponseUpdate - { - Role = ChatRole.Assistant, - Contents = [new TextContent("response")] - }; - // Yield usage details + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("response")] }; yield return new ChatResponseUpdate { Contents = [new UsageContent(new UsageDetails { InputTokenCount = inputTokens, OutputTokenCount = outputTokens })] diff --git a/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs b/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs new file mode 100644 index 000000000..b92516be7 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs @@ -0,0 +1,73 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Configuration; +using Netclaw.Daemon.Configuration; +using Xunit; +using LoggingChatClient = Netclaw.Daemon.Configuration.LoggingChatClient; + +namespace Netclaw.Daemon.Tests.Configuration; + +public sealed class PipelineChatClientFactoryTests +{ + private readonly RetryPolicy _policy = new() + { + MaxRetries = 3, + BaseDelay = TimeSpan.FromMilliseconds(1), + MaxDelay = TimeSpan.FromMilliseconds(10) + }; + + [Fact] + public void Compose_puts_Logging_outermost() + { + var pipeline = PipelineChatClientFactory.Compose( + new FakeChatClient(streaming: true), _policy, NullLoggerFactory.Instance, TimeProvider.System); + + // ChatClientBuilder applies the first-registered factory outermost. Logging must + // wrap Retry so a single completion log spans the whole retried operation — guard + // against the .Use() ordering silently flipping on a package bump. + Assert.IsType(pipeline); + } + + [Fact] + public async Task Compose_streams_through_and_logs_completion() + { + var logs = new List(); + var pipeline = PipelineChatClientFactory.Compose( + new FakeChatClient(streaming: true), _policy, new ListLoggerFactory(logs), TimeProvider.System); + + var updates = new List(); + await foreach (var u in pipeline.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) + { + updates.Add(u); + } + + Assert.Single(updates); // leaf reached, output flows through + Assert.Contains(logs, l => l.Contains("LLM streaming call completed")); // Logging middleware wired + } + + private sealed class ListLoggerFactory : ILoggerFactory + { + private readonly List _logs; + public ListLoggerFactory(List logs) => _logs = logs; + public ILogger CreateLogger(string categoryName) => new ListLogger(_logs); + public void AddProvider(ILoggerProvider provider) { } + public void Dispose() { } + + private sealed class ListLogger(List logs) : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log( + LogLevel logLevel, EventId eventId, TState state, + Exception? exception, Func formatter) + => logs.Add(formatter(state, exception)); + } + } +} diff --git a/src/Netclaw.Daemon.Tests/Configuration/ResilientChatClientProviderDecoratorTests.cs b/src/Netclaw.Daemon.Tests/Configuration/ResilientChatClientProviderDecoratorTests.cs deleted file mode 100644 index 6a18cf6a3..000000000 --- a/src/Netclaw.Daemon.Tests/Configuration/ResilientChatClientProviderDecoratorTests.cs +++ /dev/null @@ -1,128 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.Logging.Abstractions; -using Netclaw.Configuration; -using Netclaw.Daemon.Configuration; -using Xunit; - -namespace Netclaw.Daemon.Tests.Configuration; - -public sealed class ResilientChatClientProviderDecoratorTests -{ - private static readonly RetryPolicy Policy = new() - { - MaxRetries = 3, - BaseDelay = TimeSpan.FromMilliseconds(1), - MaxDelay = TimeSpan.FromMilliseconds(10) - }; - - [Fact] - public void MainRole_UsesFailover_WhenDistinctFallbackConfigured() - { - var rawMain = new FakeChatClient(); - var rawFallback = new FakeChatClient(); - var inner = new StubChatClientProvider(rawMain, rawFallback, rawMain); - - var models = new ModelSelection - { - Main = new ModelReference { Provider = "main", ModelId = "main-model" }, - Fallback = new ModelReference { Provider = "fallback", ModelId = "fallback-model" } - }; - - var decorated = new ResilientChatClientProviderDecorator( - inner, - Policy, - models, - NullLoggerFactory.Instance, - NullNotificationSink.Instance); - - var main = decorated.GetClient(ModelRole.Main); - var fallbackRole = decorated.GetClient(ModelRole.Fallback); - var compaction = decorated.GetClient(ModelRole.Compaction); - - Assert.IsType(main); - Assert.Same(main, fallbackRole); - Assert.Same(main, compaction); - } - - [Fact] - public void MainRole_SkipsFailover_WhenFallbackResolvesToSameRawClient() - { - var rawMain = new FakeChatClient(); - var inner = new StubChatClientProvider(rawMain, rawMain, rawMain); - - var models = new ModelSelection - { - Main = new ModelReference { Provider = "main", ModelId = "main-model" }, - Fallback = new ModelReference { Provider = "fallback", ModelId = "fallback-model" } - }; - - var decorated = new ResilientChatClientProviderDecorator( - inner, - Policy, - models, - NullLoggerFactory.Instance, - NullNotificationSink.Instance); - - var main = decorated.GetClient(ModelRole.Main); - - Assert.IsType(main); - Assert.IsNotType(main); - Assert.Same(main, decorated.GetClient(ModelRole.Fallback)); - } - - [Fact] - public void CompactionRole_UsesSeparateDecoratedClient_WhenRawCompactionIsDistinct() - { - var rawMain = new FakeChatClient(); - var rawCompaction = new FakeChatClient(); - var inner = new StubChatClientProvider(rawMain, rawMain, rawCompaction); - - var models = new ModelSelection - { - Main = new ModelReference { Provider = "main", ModelId = "main-model" }, - Compaction = new ModelReference { Provider = "compaction", ModelId = "compaction-model" } - }; - - var decorated = new ResilientChatClientProviderDecorator( - inner, - Policy, - models, - NullLoggerFactory.Instance, - NullNotificationSink.Instance); - - var main = decorated.GetClient(ModelRole.Main); - var compaction = decorated.GetClient(ModelRole.Compaction); - - Assert.IsType(main); - Assert.IsType(compaction); - Assert.NotSame(main, compaction); - } - - private sealed class StubChatClientProvider : IChatClientProvider - { - private readonly Microsoft.Extensions.AI.IChatClient _main; - private readonly Microsoft.Extensions.AI.IChatClient _fallback; - private readonly Microsoft.Extensions.AI.IChatClient _compaction; - - public StubChatClientProvider( - Microsoft.Extensions.AI.IChatClient main, - Microsoft.Extensions.AI.IChatClient fallback, - Microsoft.Extensions.AI.IChatClient compaction) - { - _main = main; - _fallback = fallback; - _compaction = compaction; - } - - public Microsoft.Extensions.AI.IChatClient GetClient(ModelRole role) => role switch - { - ModelRole.Fallback => _fallback, - ModelRole.Compaction => _compaction, - _ => _main - }; - } -} diff --git a/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs index e938b87d0..559c5fad3 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs @@ -4,6 +4,7 @@ // // ----------------------------------------------------------------------- using System.Net; +using System.Runtime.CompilerServices; using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging.Abstractions; using Netclaw.Configuration; @@ -157,9 +158,14 @@ await Assert.ThrowsAsync(() => } [Fact] - public async Task StreamingIsPassedThrough_NoRetry() + public async Task StreamingRetriesPreFirstChunk_ThenSucceeds() { - var fake = new FakeChatClient(streaming: true); + var attempts = 0; + var fake = new FakeChatClient(streamHandler: (_, _, ct) => + { + attempts++; + return ThrowBeforeChunkThenYield(attempts, failUntil: 3, ct); + }); var client = new RetryingChatClient(fake, _policy, NullLogger.Instance); var updates = new List(); @@ -169,7 +175,102 @@ [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.C updates.Add(u); } + Assert.Single(updates); // only the successful attempt yields + Assert.Equal(3, attempts); // 2 pre-chunk failures + 1 success, each re-initiates the stream + } + + [Fact] + public async Task StreamingDoesNotRetryAfterFirstChunk() + { + var attempts = 0; + var fake = new FakeChatClient(streamHandler: (_, _, ct) => + { + attempts++; + return YieldThenThrow(ct); + }); + var client = new RetryingChatClient(fake, _policy, NullLogger.Instance); + + var updates = new List(); + await Assert.ThrowsAsync(async () => + { + await foreach (var u in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) + { + updates.Add(u); + } + }); + + // The 500 is retryable by policy, but a chunk was already emitted, so the + // failure propagates instead of restarting (no duplicate output). Assert.Single(updates); + Assert.Equal(1, attempts); + } + + [Fact] + public async Task StreamingStopsAfterMaxRetries() + { + var attempts = 0; + var fake = new FakeChatClient(streamHandler: (_, _, ct) => + { + attempts++; + return ThrowBeforeChunkThenYield(attempts, failUntil: 100, ct); + }); + var client = new RetryingChatClient(fake, _policy, NullLogger.Instance); + + await Assert.ThrowsAsync(async () => + { + await foreach (var _ in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) { } + }); + + Assert.Equal(4, attempts); // 1 initial + 3 retries + } + + [Fact] + public async Task StreamingDoesNotRetry_WhenCancelled() + { + var cts = new CancellationTokenSource(); + cts.Cancel(); + + var attempts = 0; + var fake = new FakeChatClient(streamHandler: (_, _, ct) => + { + attempts++; + return ThrowBeforeChunkThenYield(attempts, failUntil: 100, ct); + }); + var client = new RetryingChatClient(fake, _policy, NullLogger.Instance); + + await Assert.ThrowsAnyAsync(async () => + { + await foreach (var _ in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)) { } + }); + + Assert.Equal(1, attempts); // cancellation is not retried + } + + // Throws a retryable 429 before yielding any chunk while attemptNumber < failUntil, + // otherwise yields one chunk. The runtime-dependent condition keeps the yield + // reachable (no CS0162) so no warning suppression is needed. + private static async IAsyncEnumerable ThrowBeforeChunkThenYield( + int attemptNumber, int failUntil, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + if (attemptNumber < failUntil) + throw new HttpRequestException("rate limited", null, HttpStatusCode.TooManyRequests); + + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("ok")] }; + } + + private static async IAsyncEnumerable YieldThenThrow( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("partial")] }; + throw new HttpRequestException("mid-stream failure", null, HttpStatusCode.InternalServerError); } } diff --git a/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientProviderTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientProviderTests.cs new file mode 100644 index 000000000..631a30f11 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientProviderTests.cs @@ -0,0 +1,82 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Configuration; +using Netclaw.Daemon.Configuration; +using Xunit; + +namespace Netclaw.Daemon.Tests.Configuration; + +public sealed class RoutingChatClientProviderTests +{ + [Fact] + public void GetClient_caches_per_role_and_shares_Main_and_Fallback() + { + var provider = new RoutingChatClientProvider( + new StubRouter(), NullNotificationSink.Instance, NullLoggerFactory.Instance); + + var main = provider.GetClient(ModelRole.Main); + + Assert.Same(main, provider.GetClient(ModelRole.Main)); // cached per role + Assert.Same(main, provider.GetClient(ModelRole.Fallback)); // Fallback shares the Main routing client + Assert.NotSame(main, provider.GetClient(ModelRole.Compaction));// Compaction is its own routing client + Assert.Same(provider.GetClient(ModelRole.Compaction), provider.GetClient(ModelRole.Compaction)); + } + + private sealed class StubRouter : IChatClientRouter + { + private readonly IReadOnlyList _candidates = [new FakeChatClient()]; + public IReadOnlyList Route(ChatRoutingContext context) => _candidates; + } +} + +public sealed class RoleBasedFailoverRouterTests +{ + [Fact] + public void Main_has_two_candidates_when_fallback_configured() + { + var models = new ModelSelection + { + Main = new ModelReference { Provider = "p", ModelId = "main" }, + Fallback = new ModelReference { Provider = "p", ModelId = "fb" } + }; + var router = new RoleBasedFailoverRouter(_ => new FakeChatClient(), models); + + var main = router.Route(new ChatRoutingContext { Role = ModelRole.Main }); + + Assert.Equal(2, main.Count); + Assert.Same(main, router.Route(new ChatRoutingContext { Role = ModelRole.Fallback })); + // No distinct compaction model → compaction reuses the main candidate list. + Assert.Same(main, router.Route(new ChatRoutingContext { Role = ModelRole.Compaction })); + } + + [Fact] + public void Main_has_one_candidate_when_no_fallback() + { + var models = new ModelSelection { Main = new ModelReference { Provider = "p", ModelId = "main" } }; + var router = new RoleBasedFailoverRouter(_ => new FakeChatClient(), models); + + Assert.Single(router.Route(new ChatRoutingContext { Role = ModelRole.Main })); + } + + [Fact] + public void Compaction_is_a_distinct_single_pipeline_when_configured() + { + var models = new ModelSelection + { + Main = new ModelReference { Provider = "p", ModelId = "main" }, + Compaction = new ModelReference { Provider = "p", ModelId = "comp" } + }; + var router = new RoleBasedFailoverRouter(_ => new FakeChatClient(), models); + + var main = router.Route(new ChatRoutingContext { Role = ModelRole.Main }); + var compaction = router.Route(new ChatRoutingContext { Role = ModelRole.Compaction }); + + Assert.Single(compaction); + Assert.NotSame(main, compaction); + } +} diff --git a/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs new file mode 100644 index 000000000..ab4016055 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs @@ -0,0 +1,216 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using System.Runtime.CompilerServices; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Configuration; +using Netclaw.Daemon.Configuration; +using Xunit; + +namespace Netclaw.Daemon.Tests.Configuration; + +public sealed class RoutingChatClientTests +{ + private static RoutingChatClient Client(IOperationalNotificationSink sink, params IChatClient[] candidates) => + new(new StubRouter(candidates), new ChatRoutingContext { Role = ModelRole.Main }, + sink, NullLogger.Instance, TimeProvider.System); + + [Fact] + public async Task UsesPrimary_WhenHealthy() + { + var sink = new CapturingSink(); + var primary = new FakeChatClient((_, _, _) => + Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "primary")]))); + var fallback = new FakeChatClient((_, _, _) => + Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")]))); + + var response = await Client(sink, primary, fallback) + .GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); + + Assert.Equal("primary", response.Messages[0].Text); + Assert.Empty(sink.Alerts); + } + + [Fact] + public async Task FailsOverToNextCandidate_WhenPrimaryFails() + { + var sink = new CapturingSink(); + var primary = new FakeChatClient((_, _, _) => throw new HttpRequestException("primary down")); + var fallback = new FakeChatClient((_, _, _) => + Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")]))); + + var response = await Client(sink, primary, fallback) + .GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken); + + Assert.Equal("fallback", response.Messages[0].Text); + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderFailover); + Assert.DoesNotContain(sink.Alerts, a => a.Category == AlertType.ProviderUnreachable); + } + + [Fact] + public async Task EmitsUnreachable_WhenAllCandidatesFail() + { + var sink = new CapturingSink(); + var primary = new FakeChatClient((_, _, _) => throw new HttpRequestException("primary down")); + var fallback = new FakeChatClient((_, _, _) => throw new HttpRequestException("fallback down")); + + var ex = await Assert.ThrowsAsync(() => + Client(sink, primary, fallback) + .GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)); + + Assert.Contains("fallback down", ex.Message); + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderFailover); + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderUnreachable); + } + + [Fact] + public async Task SingleCandidateFailure_EmitsUnreachable_NotFailover() + { + var sink = new CapturingSink(); + var only = new FakeChatClient((_, _, _) => throw new HttpRequestException("down")); + + await Assert.ThrowsAsync(() => + Client(sink, only) + .GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)); + + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderUnreachable); + Assert.DoesNotContain(sink.Alerts, a => a.Category == AlertType.ProviderFailover); + } + + [Fact] + public async Task DoesNotFailover_OnCancellation() + { + var cts = new CancellationTokenSource(); + cts.Cancel(); + var sink = new CapturingSink(); + var fallbackCalls = 0; + var primary = new FakeChatClient((_, _, ct) => throw new OperationCanceledException(ct)); + var fallback = new FakeChatClient((_, _, _) => + { + fallbackCalls++; + return Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")])); + }); + + await Assert.ThrowsAnyAsync(() => + Client(sink, primary, fallback) + .GetResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)); + + Assert.Equal(0, fallbackCalls); + Assert.Empty(sink.Alerts); + } + + [Fact] + public async Task Streaming_UsesPrimary_WhenHealthy() + { + var sink = new CapturingSink(); + var primary = new FakeChatClient(streamHandler: (_, _, ct) => SingleTextUpdateAsync("primary", ct)); + var fallback = new FakeChatClient(streamHandler: (_, _, ct) => SingleTextUpdateAsync("fallback", ct)); + + var texts = await CollectText(Client(sink, primary, fallback)); + + Assert.Equal(["primary"], texts); + Assert.Empty(sink.Alerts); + } + + [Fact] + public async Task Streaming_FailsOver_WhenPrimaryFailsBeforeFirstChunk() + { + var sink = new CapturingSink(); + var fallbackCalls = 0; + var primary = new FakeChatClient(streamHandler: (_, _, ct) => ThrowBeforeFirstChunkAsync(true, ct)); + var fallback = new FakeChatClient(streamHandler: (_, _, ct) => + { + fallbackCalls++; + return SingleTextUpdateAsync("fallback", ct); + }); + + var texts = await CollectText(Client(sink, primary, fallback)); + + Assert.Equal(["fallback"], texts); + Assert.Equal(1, fallbackCalls); + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderFailover); + } + + [Fact] + public async Task Streaming_DoesNotFailover_AfterPrimaryAlreadyYielded() + { + var sink = new CapturingSink(); + var fallbackCalls = 0; + var primary = new FakeChatClient(streamHandler: (_, _, ct) => YieldThenThrowAsync(ct)); + var fallback = new FakeChatClient(streamHandler: (_, _, ct) => + { + fallbackCalls++; + return SingleTextUpdateAsync("fallback", ct); + }); + var texts = new List(); + + var ex = await Assert.ThrowsAsync(async () => + { + await foreach (var u in Client(sink, primary, fallback).GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) + { + foreach (var c in u.Contents) + if (c is TextContent t) texts.Add(t.Text); + } + }); + + Assert.Contains("after first chunk", ex.Message); + Assert.Equal(["primary"], texts); + Assert.Equal(0, fallbackCalls); + } + + private static async Task> CollectText(IChatClient client) + { + var texts = new List(); + await foreach (var u in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) + { + foreach (var c in u.Contents) + if (c is TextContent t) texts.Add(t.Text); + } + + return texts; + } + + private static async IAsyncEnumerable ThrowBeforeFirstChunkAsync( + bool shouldThrow, [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + if (shouldThrow) + throw new HttpRequestException("primary stream failed before first chunk"); + + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("unused")] }; + } + + private static async IAsyncEnumerable YieldThenThrowAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("primary")] }; + throw new HttpRequestException("primary stream failed after first chunk"); + } + + private static async IAsyncEnumerable SingleTextUpdateAsync( + string text, [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent(text)] }; + } + + private sealed class StubRouter(IReadOnlyList candidates) : IChatClientRouter + { + public IReadOnlyList Route(ChatRoutingContext context) => candidates; + } + + private sealed class CapturingSink : IOperationalNotificationSink + { + public List Alerts { get; } = []; + public void Emit(OperationalAlert alert) => Alerts.Add(alert); + } +} diff --git a/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs b/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs new file mode 100644 index 000000000..2b93e2bba --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs @@ -0,0 +1,45 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.Logging; + +namespace Netclaw.Daemon.Tests.Configuration; + +/// +/// Test logger that records the state objects passed to +/// so tests can assert which scopes were opened around a logging call. Returns +/// null from BeginScope (a valid no-op for using var); only the +/// captured state matters. +/// +internal sealed class ScopeCapturingLogger : ILogger +{ + private const string SessionIdKey = "SessionId"; + + public List Scopes { get; } = []; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, + Exception? exception, Func formatter) + { + } + + public bool IsEnabled(LogLevel logLevel) => true; + + public IDisposable? BeginScope(TState state) where TState : notnull + { + Scopes.Add(state); + return null; + } + + /// True if a scope tagging the given session id was opened. + public bool HasSessionScope(string expectedId) => + Scopes.Any(s => s is IEnumerable> kvps + && kvps.Any(kv => kv.Key == SessionIdKey && kv.Value is string v && v == expectedId)); + + /// True if any session-id scope (regardless of value) was opened. + public bool HasAnySessionScope() => + Scopes.Any(s => s is IEnumerable> kvps + && kvps.Any(kv => kv.Key == SessionIdKey)); +} diff --git a/src/Netclaw.Daemon/Configuration/AlertingChatClientDecorator.cs b/src/Netclaw.Daemon/Configuration/AlertingChatClientDecorator.cs deleted file mode 100644 index 3691f5ae8..000000000 --- a/src/Netclaw.Daemon/Configuration/AlertingChatClientDecorator.cs +++ /dev/null @@ -1,95 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.AI; -using Netclaw.Configuration; - -namespace Netclaw.Daemon.Configuration; - -/// -/// Thin decorator that emits a provider.unreachable alert when the -/// underlying throws. Used for single-provider setups -/// where no is in the chain. -/// -/// The exception is always re-thrown after emitting the alert — this decorator -/// does not change error-handling behavior, only adds notification. -/// -public sealed class AlertingChatClientDecorator : IChatClient -{ - private readonly IChatClient _inner; - private readonly IOperationalNotificationSink _notificationSink; - private readonly TimeProvider _timeProvider; - - public AlertingChatClientDecorator( - IChatClient inner, - IOperationalNotificationSink notificationSink, - TimeProvider timeProvider) - { - _inner = inner; - _notificationSink = notificationSink; - _timeProvider = timeProvider; - } - - public async Task GetResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - { - try - { - return await _inner.GetResponseAsync(messages, options, cancellationToken); - } - catch (Exception ex) when (!cancellationToken.IsCancellationRequested) - { - EmitUnreachableAlert(ex); - throw; - } - } - - public IAsyncEnumerable GetStreamingResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - { - return StreamWithAlertingAsync(messages, options, cancellationToken); - } - - private async IAsyncEnumerable StreamWithAlertingAsync( - IEnumerable messages, - ChatOptions? options, - [System.Runtime.CompilerServices.EnumeratorCancellation] - CancellationToken cancellationToken) - { - IAsyncEnumerable stream; - try - { - stream = _inner.GetStreamingResponseAsync(messages, options, cancellationToken); - } - catch (Exception ex) when (!cancellationToken.IsCancellationRequested) - { - EmitUnreachableAlert(ex); - throw; - } - - await foreach (var update in stream) - yield return update; - } - - private void EmitUnreachableAlert(Exception ex) - { - _notificationSink.Emit(OperationalAlert.Create( - _timeProvider, - "provider.unreachable", - AlertType.ProviderUnreachable, - "LLM provider unreachable — no fallback configured", - AlertSeverity.Critical, - context: new Dictionary { ["error"] = ex.Message })); - } - - public object? GetService(Type serviceType, object? serviceKey = null) - => _inner.GetService(serviceType, serviceKey); - - public void Dispose() => _inner.Dispose(); -} diff --git a/src/Netclaw.Daemon/Configuration/ChatClientRouter.cs b/src/Netclaw.Daemon/Configuration/ChatClientRouter.cs new file mode 100644 index 000000000..23ed4c8ba --- /dev/null +++ b/src/Netclaw.Daemon/Configuration/ChatClientRouter.cs @@ -0,0 +1,71 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.AI; +using Netclaw.Configuration; + +namespace Netclaw.Daemon.Configuration; + +/// +/// Selects which composed chat-client pipeline(s) to invoke for a given routing +/// context. Returns candidates in priority order; +/// walks them, so failover is "the candidate list has more than one entry" rather +/// than a bespoke decorator. The router is the seam where future per-session / +/// per-provider routing slots in as a different policy. +/// +public interface IChatClientRouter +{ + /// + /// Returns the composed pipelines to try, in order, for this context. Never empty. + /// + IReadOnlyList Route(ChatRoutingContext context); +} + +/// +/// Today's policy: route by with primary→fallback failover. +/// Builds the main/fallback/compaction pipelines once via +/// and maps each role to an ordered candidate +/// list. Reproduces the previous resilient-decorator wiring: +/// +/// Main / Fallback → [main], plus fallback when a distinct +/// fallback model is configured (failover across both). +/// Compaction → its own single pipeline when a distinct compaction model is +/// configured; otherwise it reuses the main candidate list (inheriting failover). +/// +/// +public sealed class RoleBasedFailoverRouter : IChatClientRouter +{ + private readonly IReadOnlyList _mainCandidates; + private readonly IReadOnlyList _compactionCandidates; + + public RoleBasedFailoverRouter(PipelineChatClientFactory factory, ModelSelection models) + : this(factory.Create, models) + { + } + + // Test seam: build candidates from any create function, independent of the provider + // plumbing PipelineChatClientFactory needs. + internal RoleBasedFailoverRouter(Func create, ModelSelection models) + { + var main = create(models.Main); + _mainCandidates = models.Fallback is not null + ? [main, create(models.Fallback)] + : [main]; + + // A distinct compaction model gets its own (single-candidate) pipeline; without + // one, compaction reuses the main candidates so it inherits failover. + _compactionCandidates = models.Compaction is not null + ? [create(models.Compaction)] + : _mainCandidates; + } + + public IReadOnlyList Route(ChatRoutingContext context) => context.Role switch + { + ModelRole.Compaction => _compactionCandidates, + // Main and Fallback both resolve to the main candidate list (which already + // contains the fallback when configured), matching the prior provider contract. + _ => _mainCandidates + }; +} diff --git a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs index cf8141c6e..a4e36ed58 100644 --- a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs +++ b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs @@ -11,14 +11,15 @@ namespace Netclaw.Daemon.Configuration; /// -/// Daemon-level provider wiring: plugin factory, retry, resilient decorator. -/// Chains on top of . +/// Daemon-level provider wiring: plugin factory, retry, composed pipelines, and a +/// router-backed chat client provider. Chains on top of +/// . /// public static class DaemonProviderServiceExtensions { /// - /// Registers provider plugins (via Netclaw.Providers) plus daemon-specific - /// factory, retry, and resilient chat client provider. + /// Registers provider plugins (via Netclaw.Providers) plus the daemon-specific + /// plugin factory, retry policy, pipeline composition, and routing. /// public static IServiceCollection AddDaemonLlmProviders( this IServiceCollection services, @@ -28,26 +29,32 @@ public static IServiceCollection AddDaemonLlmProviders( // Register plugins and OAuth from Netclaw.Providers services.AddLlmProviders(); - // Register the plugin factory and chat client provider + // Raw provider client factory (raw client + vendor options per model) services.AddSingleton(sp => new ProviderPluginFactory(providers, sp.GetServices())); // Retry policy (TODO: make configurable via netclaw.json Resilience section) services.AddSingleton(new RetryPolicy()); - // Raw provider → Resilient decorator (Logging → Retry → Failover → Alerting) - services.AddSingleton(sp => - { - var raw = new NetclawChatClientProvider( - sp.GetRequiredService(), models); - return new ResilientChatClientProviderDecorator( - raw, - sp.GetRequiredService(), - models, - sp.GetRequiredService(), - sp.GetRequiredService(), - sp.GetService()); - }); + // Composes the cross-cutting middleware (Logging → Retry) around each provider + // pipeline via ChatClientBuilder. + services.AddSingleton(sp => new PipelineChatClientFactory( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetService())); + + // Routing policy. Today: role-based selection with primary→fallback failover. + // Per-session / per-provider routing slots in here later as a different policy. + services.AddSingleton(sp => new RoleBasedFailoverRouter( + sp.GetRequiredService(), models)); + + // Router-backed provider the actor layer consumes via GetClient(role). + services.AddSingleton(sp => new RoutingChatClientProvider( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetService())); return services; } diff --git a/src/Netclaw.Daemon/Configuration/FailoverChatClient.cs b/src/Netclaw.Daemon/Configuration/FailoverChatClient.cs deleted file mode 100644 index c68db7036..000000000 --- a/src/Netclaw.Daemon/Configuration/FailoverChatClient.cs +++ /dev/null @@ -1,202 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.AI; -using Microsoft.Extensions.Logging; -using Netclaw.Configuration; - -namespace Netclaw.Daemon.Configuration; - -/// -/// An that tries a primary client first, -/// then fails over to a fallback if the primary throws after all retries -/// are exhausted. Both primary and fallback should already be wrapped -/// in their own retry/logging decorators. -/// -/// Emits provider.failover alerts when the primary fails and we -/// switch to the fallback, and provider.unreachable alerts when -/// the fallback also fails. -/// -public sealed class FailoverChatClient : IChatClient -{ - private readonly IChatClient _primary; - private readonly IChatClient _fallback; - private readonly ILogger _logger; - private readonly IOperationalNotificationSink _notificationSink; - private readonly TimeProvider _timeProvider; - - public FailoverChatClient( - IChatClient primary, - IChatClient fallback, - ILogger logger, - IOperationalNotificationSink notificationSink, - TimeProvider timeProvider) - { - _primary = primary; - _fallback = fallback; - _logger = logger; - _notificationSink = notificationSink; - _timeProvider = timeProvider; - } - - public async Task GetResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - { - try - { - return await _primary.GetResponseAsync(messages, options, cancellationToken); - } - catch (Exception ex) when (!cancellationToken.IsCancellationRequested) - { - _logger.LogWarning(ex, - "Primary LLM failed, failing over to fallback provider"); - EmitFailoverAlert(ex); - - try - { - return await _fallback.GetResponseAsync(messages, options, cancellationToken); - } - catch (Exception fallbackEx) when (!cancellationToken.IsCancellationRequested) - { - _logger.LogError(fallbackEx, - "Fallback LLM also failed — all providers unreachable"); - EmitUnreachableAlert(fallbackEx); - throw; - } - } - } - - public IAsyncEnumerable GetStreamingResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - { - return StreamWithFailoverAsync(messages, options, cancellationToken); - } - - private async IAsyncEnumerable StreamWithFailoverAsync( - IEnumerable messages, - ChatOptions? options, - [System.Runtime.CompilerServices.EnumeratorCancellation] - CancellationToken cancellationToken) - { - // Try primary stream creation first. - IAsyncEnumerable stream; - Exception? primaryInitiationFailure = null; - try - { - stream = _primary.GetStreamingResponseAsync(messages, options, cancellationToken); - } - catch (Exception ex) when (!cancellationToken.IsCancellationRequested) - { - primaryInitiationFailure = ex; - EmitFailoverAlert(ex); - - IAsyncEnumerable fallbackStream; - try - { - fallbackStream = _fallback.GetStreamingResponseAsync(messages, options, cancellationToken); - } - catch (Exception fallbackEx) when (!cancellationToken.IsCancellationRequested) - { - EmitUnreachableAlert(fallbackEx); - throw; - } - - stream = fallbackStream; - } - - if (primaryInitiationFailure is not null) - { - _logger.LogWarning(primaryInitiationFailure, - "Primary LLM streaming failed on initiation, failing over to fallback"); - - await foreach (var fallbackUpdate in stream) - yield return fallbackUpdate; - - yield break; - } - - // Primary stream started. If the first MoveNextAsync fails, fail over. - // Once any primary chunk is emitted, failures propagate to avoid mixed-output artifacts. - await using var primaryEnumerator = stream.GetAsyncEnumerator(cancellationToken); - var yieldedPrimaryChunk = false; - Exception? primaryPreFirstChunkFailure = null; - - while (true) - { - ChatResponseUpdate update; - try - { - if (!await primaryEnumerator.MoveNextAsync()) - break; - - update = primaryEnumerator.Current; - } - catch (Exception ex) when (!cancellationToken.IsCancellationRequested && !yieldedPrimaryChunk) - { - primaryPreFirstChunkFailure = ex; - break; - } - - yieldedPrimaryChunk = true; - yield return update; - } - - if (primaryPreFirstChunkFailure is null) - yield break; - - _logger.LogWarning(primaryPreFirstChunkFailure, - "Primary LLM streaming failed before first chunk, failing over to fallback"); - EmitFailoverAlert(primaryPreFirstChunkFailure); - - IAsyncEnumerable preChunkFallbackStream; - try - { - preChunkFallbackStream = _fallback.GetStreamingResponseAsync(messages, options, cancellationToken); - } - catch (Exception fallbackEx) when (!cancellationToken.IsCancellationRequested) - { - EmitUnreachableAlert(fallbackEx); - throw; - } - - await foreach (var fallbackUpdate in preChunkFallbackStream) - yield return fallbackUpdate; - } - - private void EmitFailoverAlert(Exception ex) - { - _notificationSink.Emit(OperationalAlert.Create( - _timeProvider, - "provider.failover", - AlertType.ProviderFailover, - "Primary LLM provider failed, failing over to fallback", - AlertSeverity.Warning, - context: new Dictionary { ["error"] = ex.Message })); - } - - private void EmitUnreachableAlert(Exception ex) - { - _notificationSink.Emit(OperationalAlert.Create( - _timeProvider, - "provider.unreachable", - AlertType.ProviderUnreachable, - "All LLM providers failed — primary and fallback both unreachable", - AlertSeverity.Critical, - context: new Dictionary { ["error"] = ex.Message })); - } - - public object? GetService(Type serviceType, object? serviceKey = null) - => _primary.GetService(serviceType, serviceKey); - - public void Dispose() - { - _primary.Dispose(); - _fallback.Dispose(); - } -} diff --git a/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs b/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs index c41158564..ffa618b33 100644 --- a/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs +++ b/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs @@ -12,20 +12,17 @@ namespace Netclaw.Daemon.Configuration; /// -/// Decorates an with logging for elapsed time, -/// token usage, and errors. This class is not thread-safe and is -/// intended to be scoped per-actor or per-session (Akka actors are -/// single-threaded, so no synchronization is needed). +/// Decorates an with logging for elapsed time, token +/// usage, and errors, and tags log lines with the ambient session id. Stateless +/// and safe to share across sessions. Netclaw issues only streaming requests, so +/// only the streaming path is instrumented; the inherited non-streaming +/// pass-through is unused. /// public sealed class LoggingChatClient : DelegatingChatClient { private readonly ILogger _logger; private readonly TimeProvider _timeProvider; - // Track token usage across calls (single-threaded, no lock needed) - private long? _lastInputTokens; - private long _cumulativeInputTokens; - public LoggingChatClient( IChatClient innerClient, ILogger logger, @@ -36,36 +33,13 @@ public LoggingChatClient( _timeProvider = timeProvider ?? TimeProvider.System; } - public override async Task GetResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - { - var messageList = messages as IReadOnlyList ?? messages.ToList(); - LogPromptDiagnostics(messageList, options); - - var start = _timeProvider.GetTimestamp(); - try - { - var response = await base.GetResponseAsync(messageList, options, cancellationToken); - var elapsed = _timeProvider.GetElapsedTime(start); - LogCompletion(elapsed, response.Usage); - return response; - } - catch (Exception ex) - { - var elapsed = _timeProvider.GetElapsedTime(start); - _logger.LogError(ex, "LLM call failed after {ElapsedMs:F0}ms", elapsed.TotalMilliseconds); - throw; - } - } - public override async IAsyncEnumerable GetStreamingResponseAsync( IEnumerable messages, ChatOptions? options = null, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) { + using var sessionScope = SessionLoggingScope.Begin(_logger); var messageList = messages as IReadOnlyList ?? messages.ToList(); LogPromptDiagnostics(messageList, options); @@ -130,12 +104,9 @@ public override async IAsyncEnumerable GetStreamingResponseA if (inputTokens > 0 || outputTokens > 0) { - var delta = RecordInputTokens(inputTokens); - var deltaFormatted = delta is null ? "N/A" : FormatDelta(delta.Value); - _logger.LogInformation( - "LLM streaming call completed in {ElapsedMs:F0}ms (input: {InputTokens}, delta: {Delta}, cumulative: {Cumulative}, output: {OutputTokens})", - totalElapsed.TotalMilliseconds, inputTokens, deltaFormatted, _cumulativeInputTokens, outputTokens); + "LLM streaming call completed in {ElapsedMs:F0}ms (input: {InputTokens}, output: {OutputTokens})", + totalElapsed.TotalMilliseconds, inputTokens, outputTokens); } else { @@ -301,48 +272,4 @@ private sealed record PromptSummary( int BrowserToolOptionCount, string PromptHash); - /// - /// Records input token usage and returns the delta from the previous call, - /// or null if this is the first call. - /// - private long? RecordInputTokens(long inputTokens) - { - var delta = _lastInputTokens.HasValue - ? inputTokens - _lastInputTokens.Value - : (long?)null; - _cumulativeInputTokens += inputTokens; - _lastInputTokens = inputTokens; - return delta; - } - - private static string FormatDelta(long delta) => delta switch - { - > 0 => $"+{delta}", - _ => delta.ToString() - }; - - private void LogCompletion(TimeSpan elapsed, UsageDetails? usage) - { - if (usage is not null) - { - var inputTokens = usage.InputTokenCount ?? 0; - var outputTokens = usage.OutputTokenCount ?? 0; - var delta = RecordInputTokens(inputTokens); - var deltaFormatted = delta is null ? "N/A" : FormatDelta(delta.Value); - - _logger.LogInformation( - "LLM call completed in {ElapsedMs:F0}ms (input: {InputTokens}, delta: {Delta}, cumulative: {Cumulative}, output: {OutputTokens})", - elapsed.TotalMilliseconds, - inputTokens, - deltaFormatted, - _cumulativeInputTokens, - outputTokens); - } - else - { - _logger.LogInformation( - "LLM call completed in {ElapsedMs:F0}ms", - elapsed.TotalMilliseconds); - } - } } diff --git a/src/Netclaw.Daemon/Configuration/NetclawChatClientProvider.cs b/src/Netclaw.Daemon/Configuration/NetclawChatClientProvider.cs deleted file mode 100644 index 7434164ae..000000000 --- a/src/Netclaw.Daemon/Configuration/NetclawChatClientProvider.cs +++ /dev/null @@ -1,37 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.AI; -using Netclaw.Configuration; - -namespace Netclaw.Daemon.Configuration; - -/// -/// Resolves instances by -/// using a and . -/// Clients are created once at construction and reused for all requests. -/// -public sealed class NetclawChatClientProvider : IChatClientProvider -{ - private readonly IChatClient _main; - private readonly IChatClient? _fallback; - private readonly IChatClient? _compaction; - - public NetclawChatClientProvider(ProviderPluginFactory factory, ModelSelection models) - { - _main = factory.Create(models.Main); - _fallback = models.Fallback is not null - ? factory.Create(models.Fallback) : null; - _compaction = models.Compaction is not null - ? factory.Create(models.Compaction) : null; - } - - public IChatClient GetClient(ModelRole role) => role switch - { - ModelRole.Fallback => _fallback ?? _main, - ModelRole.Compaction => _compaction ?? _main, - _ => _main - }; -} diff --git a/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs b/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs new file mode 100644 index 000000000..17c12e441 --- /dev/null +++ b/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs @@ -0,0 +1,65 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging; +using Netclaw.Configuration; + +namespace Netclaw.Daemon.Configuration; + +/// +/// Builds the fully-composed middleware pipeline for a single (provider, model): +/// Logging → Retry → VendorOptions → raw provider client. The leaf +/// (raw provider client plus any vendor-options wrap) comes from +/// ; this layer adds the cross-cutting Retry and +/// Logging decorators via . +/// +/// One pipeline is built per configured model; routing (selecting which pipeline to +/// invoke per call) is a separate concern owned by the router. +/// +public sealed class PipelineChatClientFactory +{ + private readonly ProviderPluginFactory _factory; + private readonly RetryPolicy _retryPolicy; + private readonly ILoggerFactory _loggerFactory; + private readonly TimeProvider _timeProvider; + + public PipelineChatClientFactory( + ProviderPluginFactory factory, + RetryPolicy retryPolicy, + ILoggerFactory loggerFactory, + TimeProvider? timeProvider = null) + { + _factory = factory; + _retryPolicy = retryPolicy; + _loggerFactory = loggerFactory; + _timeProvider = timeProvider ?? TimeProvider.System; + } + + public IChatClient Create(ModelReference model) => + Compose(_factory.Create(model), _retryPolicy, _loggerFactory, _timeProvider); + + /// + /// Composes the cross-cutting middleware around a leaf client. Internal so the + /// middleware order can be asserted in isolation: + /// applies the first-registered factory outermost, so + /// — which must capture the total elapsed time including retries — is registered + /// before . + /// + internal static IChatClient Compose( + IChatClient leaf, + RetryPolicy retryPolicy, + ILoggerFactory loggerFactory, + TimeProvider timeProvider) + { + var loggingLogger = loggerFactory.CreateLogger(); + var retryLogger = loggerFactory.CreateLogger(); + + return new ChatClientBuilder(leaf) + .Use(inner => new LoggingChatClient(inner, loggingLogger, timeProvider)) + .Use(inner => new RetryingChatClient(inner, retryPolicy, retryLogger, timeProvider)) + .Build(); + } +} diff --git a/src/Netclaw.Daemon/Configuration/ResilientChatClientProviderDecorator.cs b/src/Netclaw.Daemon/Configuration/ResilientChatClientProviderDecorator.cs deleted file mode 100644 index 8a868eb89..000000000 --- a/src/Netclaw.Daemon/Configuration/ResilientChatClientProviderDecorator.cs +++ /dev/null @@ -1,85 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.AI; -using Microsoft.Extensions.Logging; -using Netclaw.Configuration; - -namespace Netclaw.Daemon.Configuration; - -/// -/// Wraps an and decorates each client with -/// logging and retry. For the role, additionally -/// wraps in if a distinct fallback is configured, -/// or for single-provider setups. -/// -public sealed class ResilientChatClientProviderDecorator : IChatClientProvider -{ - private readonly IChatClient _main; - private readonly IChatClient _compaction; - - public ResilientChatClientProviderDecorator( - IChatClientProvider inner, - RetryPolicy retryPolicy, - ModelSelection models, - ILoggerFactory loggerFactory, - IOperationalNotificationSink notificationSink, - TimeProvider? timeProvider = null) - { - var tp = timeProvider ?? TimeProvider.System; - var retryLogger = loggerFactory.CreateLogger(); - var loggingLogger = loggerFactory.CreateLogger(); - var failoverLogger = loggerFactory.CreateLogger(); - - // Decorate main: Logging → Retry → raw - var rawMain = inner.GetClient(ModelRole.Main); - var decoratedMain = Decorate(rawMain, retryPolicy, retryLogger, loggingLogger, tp); - - // If fallback is a distinct provider, wrap in FailoverChatClient - if (models.Fallback is not null) - { - var rawFallback = inner.GetClient(ModelRole.Fallback); - // Only wrap in failover if fallback is actually a different client - if (!ReferenceEquals(rawFallback, rawMain)) - { - var decoratedFallback = Decorate(rawFallback, retryPolicy, retryLogger, loggingLogger, tp); - _main = new FailoverChatClient( - decoratedMain, decoratedFallback, failoverLogger, notificationSink, tp); - } - else - { - _main = new AlertingChatClientDecorator(decoratedMain, notificationSink, tp); - } - } - else - { - _main = new AlertingChatClientDecorator(decoratedMain, notificationSink, tp); - } - - // Decorate compaction - var rawCompaction = inner.GetClient(ModelRole.Compaction); - _compaction = ReferenceEquals(rawCompaction, rawMain) - ? _main // reuse the decorated main if compaction falls back to it - : Decorate(rawCompaction, retryPolicy, retryLogger, loggingLogger, tp); - } - - public IChatClient GetClient(ModelRole role) => role switch - { - ModelRole.Compaction => _compaction, - _ => _main - }; - - private static IChatClient Decorate( - IChatClient raw, - RetryPolicy policy, - ILogger retryLogger, - ILogger loggingLogger, - TimeProvider tp) - { - // Inner → Retry → Logging (logging is outermost so it captures retry time) - var retrying = new RetryingChatClient(raw, policy, retryLogger, tp); - return new LoggingChatClient(retrying, loggingLogger, tp); - } -} diff --git a/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs b/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs index b38c96860..d710bfec4 100644 --- a/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs +++ b/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs @@ -1,8 +1,9 @@ -// ----------------------------------------------------------------------- +// ----------------------------------------------------------------------- // // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- +using System.Runtime.CompilerServices; using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging; using Netclaw.Configuration; @@ -11,8 +12,10 @@ namespace Netclaw.Daemon.Configuration; /// /// Decorates an with retry logic for transient failures. -/// Wraps with a retry loop. -/// Streaming calls are NOT retried (mid-stream restart is deferred to a follow-up). +/// Both the non-streaming and streaming calls are retried. Streaming is retried +/// only before the first chunk is yielded — once any update has been emitted +/// downstream, a later failure propagates unchanged so already-streamed output is +/// never duplicated by a restart. /// public sealed class RetryingChatClient : DelegatingChatClient { @@ -47,14 +50,82 @@ public override async Task GetResponseAsync( catch (Exception ex) when (!cancellationToken.IsCancellationRequested && _policy.ShouldRetry(ex, attempt)) { - var delay = _policy.GetDelay(attempt); - _logger.LogWarning(ex, - "LLM call failed (attempt {Attempt}/{Max}), retrying in {Delay:F1}s", - attempt + 1, _policy.MaxRetries, delay.TotalSeconds); + await BackoffAsync(ex, attempt, cancellationToken); + attempt++; + } + } + } - await Task.Delay(delay, _timeProvider, cancellationToken); + public override async IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + // Materialize once so re-initiation on retry re-sends the same prompt. + var messageList = messages as IReadOnlyList ?? messages.ToList(); + var attempt = 0; + + while (true) + { + var yieldedChunk = false; + Exception? preFirstChunkFailure = null; + + // Initiation can throw before any enumerator is produced. + IAsyncEnumerable stream; + try + { + stream = base.GetStreamingResponseAsync(messageList, options, cancellationToken); + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested + && _policy.ShouldRetry(ex, attempt)) + { + await BackoffAsync(ex, attempt, cancellationToken); attempt++; + continue; } + + await using var enumerator = stream.GetAsyncEnumerator(cancellationToken); + while (true) + { + ChatResponseUpdate update; + try + { + if (!await enumerator.MoveNextAsync()) + break; + + update = enumerator.Current; + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested + && !yieldedChunk + && _policy.ShouldRetry(ex, attempt)) + { + // Pre-first-chunk failure: safe to restart (nothing emitted yet). + preFirstChunkFailure = ex; + break; + } + + // Past this point a chunk has been emitted; a later failure is NOT + // caught above (yieldedChunk is true) and propagates to the consumer. + yieldedChunk = true; + yield return update; + } + + if (preFirstChunkFailure is null) + yield break; // clean completion (or a post-first-chunk throw already unwound) + + await BackoffAsync(preFirstChunkFailure, attempt, cancellationToken); + attempt++; + // outer loop re-initiates the stream } } + + private async Task BackoffAsync(Exception ex, int attempt, CancellationToken cancellationToken) + { + var delay = _policy.GetDelay(attempt); + _logger.LogWarning(ex, + "LLM call failed (attempt {Attempt}/{Max}), retrying in {Delay:F1}s", + attempt + 1, _policy.MaxRetries, delay.TotalSeconds); + + await Task.Delay(delay, _timeProvider, cancellationToken); + } } diff --git a/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs b/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs new file mode 100644 index 000000000..bffd4dc1c --- /dev/null +++ b/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs @@ -0,0 +1,230 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using System.Collections.Concurrent; +using System.Runtime.CompilerServices; +using System.Runtime.ExceptionServices; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging; +using Netclaw.Configuration; + +namespace Netclaw.Daemon.Configuration; + +/// +/// Invokes the composed pipelines chosen by an , +/// walking the ordered candidate list with failover semantics. This single walk +/// generalizes the previous FailoverChatClient (a two-candidate list) and +/// AlertingChatClientDecorator (a one-candidate list whose terminal failure +/// raises provider.unreachable). +/// +/// Streaming fails over only before the first chunk is yielded from a candidate +/// — once any update has been emitted, a later failure propagates so already-streamed +/// output is never duplicated by switching providers (the same invariant the streaming +/// retry decorator below it enforces per provider). +/// +public sealed class RoutingChatClient : IChatClient +{ + private readonly IChatClientRouter _router; + private readonly ChatRoutingContext _context; + private readonly IOperationalNotificationSink _sink; + private readonly ILogger _logger; + private readonly TimeProvider _timeProvider; + + public RoutingChatClient( + IChatClientRouter router, + ChatRoutingContext context, + IOperationalNotificationSink sink, + ILogger logger, + TimeProvider timeProvider) + { + _router = router; + _context = context; + _sink = sink; + _logger = logger; + _timeProvider = timeProvider; + } + + public async Task GetResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + CancellationToken cancellationToken = default) + { + var messageList = messages as IReadOnlyList ?? messages.ToList(); + var candidates = _router.Route(_context); + + for (var i = 0; i < candidates.Count; i++) + { + var isLast = i == candidates.Count - 1; + try + { + return await candidates[i].GetResponseAsync(messageList, options, cancellationToken); + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested && !isLast) + { + EmitFailover(ex); + LogWithSession(LogLevel.Warning, ex, "LLM provider failed, failing over to next candidate"); + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + EmitUnreachable(ex, candidates.Count); + LogWithSession(LogLevel.Error, ex, UnreachableMessage(candidates.Count)); + throw; + } + } + + throw new InvalidOperationException("Router returned no chat-client candidates."); + } + + public async IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + var messageList = messages as IReadOnlyList ?? messages.ToList(); + var candidates = _router.Route(_context); + if (candidates.Count == 0) + throw new InvalidOperationException("Router returned no chat-client candidates."); + + for (var i = 0; i < candidates.Count; i++) + { + var isLast = i == candidates.Count - 1; + var yieldedChunk = false; + Exception? failure = null; + + // Initiation can throw before any enumerator is produced. + IAsyncEnumerable? stream = null; + try + { + stream = candidates[i].GetStreamingResponseAsync(messageList, options, cancellationToken); + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + failure = ex; + } + + if (stream is not null) + { + await using var enumerator = stream.GetAsyncEnumerator(cancellationToken); + while (true) + { + ChatResponseUpdate update; + try + { + if (!await enumerator.MoveNextAsync()) + break; + + update = enumerator.Current; + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested && !yieldedChunk) + { + // Pre-first-chunk failure: safe to fail over (nothing emitted yet). + failure = ex; + break; + } + + // Past here a chunk has been emitted; a later failure is NOT caught + // above (yieldedChunk is true) and propagates to the consumer. + yieldedChunk = true; + yield return update; + } + } + + if (failure is null) + yield break; // candidate completed (or a post-first-chunk throw already unwound) + + if (isLast) + { + EmitUnreachable(failure, candidates.Count); + LogWithSession(LogLevel.Error, failure, UnreachableMessage(candidates.Count)); + ExceptionDispatchInfo.Capture(failure).Throw(); + } + + EmitFailover(failure); + LogWithSession(LogLevel.Warning, failure, "LLM provider failed, failing over to next candidate"); + // outer loop advances to the next candidate + } + } + + public object? GetService(Type serviceType, object? serviceKey = null) + { + var candidates = _router.Route(_context); + return candidates.Count > 0 ? candidates[0].GetService(serviceType, serviceKey) : null; + } + + // Candidate pipelines are process-lifetime singletons owned by the router and may + // be shared across roles; disposal is not this wrapper's responsibility. + public void Dispose() { } + + private static string UnreachableMessage(int candidateCount) => + candidateCount == 1 + ? "LLM provider unreachable — no fallback configured" + : "All LLM providers failed — every candidate unreachable"; + + private void EmitFailover(Exception ex) => + _sink.Emit(OperationalAlert.Create( + _timeProvider, + "provider.failover", + AlertType.ProviderFailover, + "Primary LLM provider failed, failing over to fallback", + AlertSeverity.Warning, + context: new Dictionary { ["error"] = ex.Message })); + + private void EmitUnreachable(Exception ex, int candidateCount) => + _sink.Emit(OperationalAlert.Create( + _timeProvider, + "provider.unreachable", + AlertType.ProviderUnreachable, + UnreachableMessage(candidateCount), + AlertSeverity.Critical, + context: new Dictionary { ["error"] = ex.Message })); + + // Failover/outage events are logged here, outside any per-pipeline LoggingChatClient + // scope, so attach the session id so they correlate by session in Seq. + private void LogWithSession(LogLevel level, Exception ex, string message) + { + using (SessionLoggingScope.Begin(_logger)) + _logger.Log(level, ex, message); + } +} + +/// +/// backed by an . +/// Returns a per (cached so a +/// role keeps a stable instance); candidate selection and failover happen per call +/// inside the routing client, so per-call routing is preserved. +/// +public sealed class RoutingChatClientProvider : IChatClientProvider +{ + private readonly IChatClientRouter _router; + private readonly IOperationalNotificationSink _sink; + private readonly ILoggerFactory _loggerFactory; + private readonly TimeProvider _timeProvider; + private readonly ConcurrentDictionary _cache = new(); + + public RoutingChatClientProvider( + IChatClientRouter router, + IOperationalNotificationSink sink, + ILoggerFactory loggerFactory, + TimeProvider? timeProvider = null) + { + _router = router; + _sink = sink; + _loggerFactory = loggerFactory; + _timeProvider = timeProvider ?? TimeProvider.System; + } + + public IChatClient GetClient(ModelRole role) + { + // Main and Fallback share one routing client (the main candidate list already + // contains the fallback), preserving the prior provider's identity contract. + var key = role == ModelRole.Fallback ? ModelRole.Main : role; + return _cache.GetOrAdd(key, r => new RoutingChatClient( + _router, + new ChatRoutingContext { Role = r }, + _sink, + _loggerFactory.CreateLogger(), + _timeProvider)); + } +} diff --git a/src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs b/src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs new file mode 100644 index 000000000..14e1da57c --- /dev/null +++ b/src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs @@ -0,0 +1,44 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.Logging; +using Netclaw.Configuration; + +namespace Netclaw.Daemon.Configuration; + +/// +/// Shared helper for tagging chat-client log lines with the ambient session id. +/// +/// The id lives in (an AsyncLocal pushed by +/// every session-owned path). The OTLP log exporter has IncludeScopes enabled +/// but cannot read that AsyncLocal directly, so without a logging scope LLM logs reach +/// Seq with no session correlation. Opening a scope keyed SessionId surfaces it +/// as a filterable attribute — matching the WithContext("SessionId", …) key the +/// session/channel actors already use, so a single attribute correlates actor and +/// chat-client logs. +/// +/// MEL scopes are per-, so each chat-client decorator that emits +/// its own log lines (e.g. and the routing/failover +/// client) opens its own scope from this helper. +/// +internal static class SessionLoggingScope +{ + private const string SessionIdKey = "SessionId"; + + /// + /// Opens a scope tagging subsequent log lines on with the + /// ambient session id, or returns null when no session is in scope (a no-op + /// using). A single-entry array avoids a per-call dictionary allocation while + /// still presenting as IEnumerable<KeyValuePair>, which is what the OTLP + /// exporter projects into log attributes. + /// + public static IDisposable? Begin(ILogger logger) + { + var sessionId = SessionDiagnosticsContext.SessionId; + return sessionId is null + ? null + : logger.BeginScope(new[] { new KeyValuePair(SessionIdKey, sessionId) }); + } +} diff --git a/src/Netclaw.Daemon/Netclaw.Daemon.csproj b/src/Netclaw.Daemon/Netclaw.Daemon.csproj index 817f2f836..ea1aa0065 100644 --- a/src/Netclaw.Daemon/Netclaw.Daemon.csproj +++ b/src/Netclaw.Daemon/Netclaw.Daemon.csproj @@ -37,6 +37,7 @@ + From 3c65ac1852cc5e2fc3574793156b81eb80eaa8d0 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 3 Jun 2026 12:36:38 +0000 Subject: [PATCH 2/4] =?UTF-8?q?fix(providers):=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20SessionId=20on=20retry=20logs,=20drop=20AttemptInde?= =?UTF-8?q?x,=20streaming-failover=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review of the stack redesign found three issues: 1. RetryingChatClient.BackoffAsync logged retry warnings without the SessionId scope, so the hottest failure path reached Seq uncorrelated while LoggingChatClient/RoutingChatClient logs carried it. Open the scope around the warning. 2. ChatRoutingContext.AttemptIndex was structurally unfulfillable: the per-role RoutingChatClient context is frozen at construction and the router is called once per call (not per candidate), so it could never be non-zero. Removed it; SessionId remains the genuine per-session-routing seam. 3. The streaming terminal-unreachable branch of RoutingChatClient (last candidate fails pre-first-chunk -> provider.unreachable + rethrow) and single-candidate / streaming-cancellation behavior were untested. Added three streaming tests. --- .../ChatRoutingContext.cs | 10 +--- .../Configuration/RoutingChatClientTests.cs | 49 +++++++++++++++++++ .../Configuration/RetryingChatClient.cs | 11 +++-- 3 files changed, 59 insertions(+), 11 deletions(-) diff --git a/src/Netclaw.Configuration/ChatRoutingContext.cs b/src/Netclaw.Configuration/ChatRoutingContext.cs index f5119604c..731333836 100644 --- a/src/Netclaw.Configuration/ChatRoutingContext.cs +++ b/src/Netclaw.Configuration/ChatRoutingContext.cs @@ -17,14 +17,8 @@ public sealed record ChatRoutingContext public required ModelRole Role { get; init; } /// - /// Owning session id. Unused today; a future per-session router policy reads this - /// to route a session's chats to a specific model/provider. + /// Owning session id. Unused today; the explicit seam for a future per-session + /// router policy that routes a session's chats to a specific model/provider. /// public string? SessionId { get; init; } - - /// - /// Index of the current failover attempt within a single call. Unused today; a - /// future policy can use it to re-rank candidates across attempts. - /// - public int AttemptIndex { get; init; } } diff --git a/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs index ab4016055..4670c0904 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs @@ -162,6 +162,55 @@ [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.C Assert.Equal(0, fallbackCalls); } + [Fact] + public async Task Streaming_EmitsUnreachable_WhenAllCandidatesFailBeforeFirstChunk() + { + var sink = new CapturingSink(); + var primary = new FakeChatClient(streamHandler: (_, _, ct) => ThrowBeforeFirstChunkAsync(true, ct)); + var fallback = new FakeChatClient(streamHandler: (_, _, ct) => ThrowBeforeFirstChunkAsync(true, ct)); + + await Assert.ThrowsAsync(() => CollectText(Client(sink, primary, fallback))); + + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderFailover); + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderUnreachable); + } + + [Fact] + public async Task Streaming_SingleCandidateFailure_EmitsUnreachable_NotFailover() + { + var sink = new CapturingSink(); + var only = new FakeChatClient(streamHandler: (_, _, ct) => ThrowBeforeFirstChunkAsync(true, ct)); + + await Assert.ThrowsAsync(() => CollectText(Client(sink, only))); + + Assert.Contains(sink.Alerts, a => a.Category == AlertType.ProviderUnreachable); + Assert.DoesNotContain(sink.Alerts, a => a.Category == AlertType.ProviderFailover); + } + + [Fact] + public async Task Streaming_DoesNotFailover_OnCancellation() + { + var cts = new CancellationTokenSource(); + cts.Cancel(); + var sink = new CapturingSink(); + var fallbackCalls = 0; + var primary = new FakeChatClient(streamHandler: (_, _, ct) => ThrowBeforeFirstChunkAsync(true, ct)); + var fallback = new FakeChatClient(streamHandler: (_, _, ct) => + { + fallbackCalls++; + return SingleTextUpdateAsync("fallback", ct); + }); + + await Assert.ThrowsAnyAsync(async () => + { + await foreach (var _ in Client(sink, primary, fallback).GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)) { } + }); + + Assert.Equal(0, fallbackCalls); + Assert.Empty(sink.Alerts); + } + private static async Task> CollectText(IChatClient client) { var texts = new List(); diff --git a/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs b/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs index d710bfec4..2a658e9bd 100644 --- a/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs +++ b/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs @@ -122,9 +122,14 @@ public override async IAsyncEnumerable GetStreamingResponseA private async Task BackoffAsync(Exception ex, int attempt, CancellationToken cancellationToken) { var delay = _policy.GetDelay(attempt); - _logger.LogWarning(ex, - "LLM call failed (attempt {Attempt}/{Max}), retrying in {Delay:F1}s", - attempt + 1, _policy.MaxRetries, delay.TotalSeconds); + // Retry is the hottest failure path; tag the warning with the session id (like + // the other chat-client decorators) so retry storms correlate by session in Seq. + using (SessionLoggingScope.Begin(_logger)) + { + _logger.LogWarning(ex, + "LLM call failed (attempt {Attempt}/{Max}), retrying in {Delay:F1}s", + attempt + 1, _policy.MaxRetries, delay.TotalSeconds); + } await Task.Delay(delay, _timeProvider, cancellationToken); } From 7f66ece406408d8693d51b87bb49b765d3fba68d Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 3 Jun 2026 13:19:07 +0000 Subject: [PATCH 3/4] =?UTF-8?q?refactor(providers):=20single=20retry=20lay?= =?UTF-8?q?er=20=E2=80=94=20transport=20owns=20LLM=20transient=20retry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconcile the two pre-first-chunk retry layers (review follow-up): the actor's per-turn streaming retry and the transport's RetryingChatClient overlapped on the main session loop. Make the transport the single owner. RetryPolicy.ShouldRetry now also recognizes the curated ProviderException (StatusCode 408/429/5xx, walking inner exceptions) that provider transports throw — previously only a raw HttpRequestException was retried, so the transport retry barely fired for real provider 429/5xx. That was the broader coverage the actor's LlmFailureClassifier.IsTransientStreaming provided. The transport RetryPolicy is now configured from Session:Tuning:StreamingRetryPolicy (resolves the prior TODO; preserves the knob), making RetryingChatClient the single, configurable, uniform retry layer for every caller (main loop + sidecars). Remove the actor per-turn retry: the LlmCallFailed pre-content retry block, the RetryLlmCallAfterBackoff message+handler, _streamingRetryAttempt, the streaming-retry timer, and the now-unused IsTransientStreaming. Context-overflow rollback and terminal FailCurrentTurn handling are unchanged. Re-home coverage: the actor-level Streaming_retry_recovers/_exhaustion integration tests are removed (the actor no longer retries); a ProviderException(502) recover case is added to RetryingChatClientTests at the transport layer. Verified: full build (0 warnings); Daemon 672/672; Actors 2184/2184; slopwatch + headers clean. --- .../Sessions/LlmSessionIntegrationTests.cs | 70 ++----------------- .../Sessions/LlmFailureClassifier.cs | 14 ---- src/Netclaw.Actors/Sessions/LlmMessages.cs | 6 -- .../Sessions/LlmSessionActor.cs | 51 ++------------ src/Netclaw.Configuration/RetryPolicy.cs | 25 ++++++- .../Configuration/RetryingChatClientTests.cs | 37 ++++++++++ .../DaemonProviderServiceExtensions.cs | 9 ++- src/Netclaw.Daemon/Program.cs | 8 ++- 8 files changed, 84 insertions(+), 136 deletions(-) diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs index 4270b88b6..8c6702d8e 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs @@ -1500,71 +1500,11 @@ await retryWatcher.ExpectMsgAsync( Assert.DoesNotContain(sessionId.Value, _lifecycleObserver.DeactivatedSessionIds); } - [Fact] - public async Task Streaming_retry_recovers_from_transient_502() - { - // First 2 calls throw 502, third succeeds - _fakeChatClient.PlannedExceptions.Enqueue( - new ProviderException("server error (502)", "HTTP 502", statusCode: 502)); - _fakeChatClient.PlannedExceptions.Enqueue( - new ProviderException("server error (502)", "HTTP 502", statusCode: 502)); - - var sessionId = new SessionId("test-channel/streaming-retry-502"); - var sessionManager = ActorRegistry.Get(); - var subscriber = CreateTestProbe("retry-502-sub"); - - await sessionManager.Ask(new JoinSession(subscriber) - { - SessionId = sessionId, - Filter = OutputFilter.Full - }, cancellationToken: TestContext.Current.CancellationToken); - await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); - - await sessionManager.Ask(new SendUserMessage - { - SessionId = sessionId, - Content = "Message that should succeed after retries" - }, cancellationToken: TestContext.Current.CancellationToken); - - // Should succeed after retries — response arrives - var text = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(15), cancellationToken: TestContext.Current.CancellationToken); - Assert.Contains("fake", text.Text, StringComparison.OrdinalIgnoreCase); - var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); - Assert.Equal(TurnOutcome.Completed, completed.Outcome); - } - - [Fact] - public async Task Streaming_retry_exhaustion_fails_turn() - { - // All calls throw 502 — retries exhaust and turn fails. - // Default RetryPolicy.MaxRetries=3, so we need initial + 3 retries = 4 exceptions. - for (var i = 0; i < 5; i++) - _fakeChatClient.PlannedExceptions.Enqueue( - new ProviderException("server error (502)", "HTTP 502", statusCode: 502)); - - var sessionId = new SessionId("test-channel/streaming-retry-exhaust"); - var sessionManager = ActorRegistry.Get(); - var subscriber = CreateTestProbe("retry-exhaust-sub"); - - await sessionManager.Ask(new JoinSession(subscriber) - { - SessionId = sessionId, - Filter = OutputFilter.Full - }, cancellationToken: TestContext.Current.CancellationToken); - await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); - - await sessionManager.Ask(new SendUserMessage - { - SessionId = sessionId, - Content = "Message that will fail after exhausting retries" - }, cancellationToken: TestContext.Current.CancellationToken); - - // Should fail with error after exhausting retries - var error = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(20), cancellationToken: TestContext.Current.CancellationToken); - Assert.Equal(ErrorCategory.ProviderFailure, error.Category); - var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); - Assert.Equal(TurnOutcome.Failed, completed.Outcome); - } + // NOTE: transient-failure retry (including ProviderException 5xx recover/exhaust) moved + // from this actor to the transport RetryingChatClient and is covered by + // RetryingChatClientTests; the actor no longer retries, so the former + // Streaming_retry_recovers_from_transient_502 / _exhaustion_fails_turn integration tests + // were removed. [Fact] public async Task Reminder_redelivery_is_deduped_in_Ready_phase() diff --git a/src/Netclaw.Actors/Sessions/LlmFailureClassifier.cs b/src/Netclaw.Actors/Sessions/LlmFailureClassifier.cs index 06ecf70cd..627c9c6dc 100644 --- a/src/Netclaw.Actors/Sessions/LlmFailureClassifier.cs +++ b/src/Netclaw.Actors/Sessions/LlmFailureClassifier.cs @@ -107,20 +107,6 @@ public static bool IsContextOverflow(Exception? ex) return false; } - public static bool IsTransientStreaming(Exception? ex) - { - if (ex is null) - return false; - - var providerEx = FindException(ex); - if (providerEx?.StatusCode is >= 500) - return true; - if (providerEx?.StatusCode is 429) - return true; - - return ex is HttpRequestException { StatusCode: null }; - } - private static T? FindException(Exception? ex) where T : Exception { while (ex is not null) diff --git a/src/Netclaw.Actors/Sessions/LlmMessages.cs b/src/Netclaw.Actors/Sessions/LlmMessages.cs index a03a63e23..ccd1aff8f 100644 --- a/src/Netclaw.Actors/Sessions/LlmMessages.cs +++ b/src/Netclaw.Actors/Sessions/LlmMessages.cs @@ -220,9 +220,3 @@ internal sealed record PassivationTimeout : INoSerializationVerificationNeeded; /// signal. See LlmSessionActor.CompletePassivation. /// internal sealed record PassivationFinalStop : INoSerializationVerificationNeeded; - -/// -/// Timer-fired message that triggers an LLM call retry after exponential backoff. -/// Carries the attempt number for observability logging. -/// -internal sealed record RetryLlmCallAfterBackoff(int Attempt) : INoSerializationVerificationNeeded; diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 37a716a70..9cfa94554 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -143,8 +143,9 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // from cancelled calls are ignored when their CallId doesn't match. private long _activeCallId; - // Tracks whether any content was streamed this call — used to gate transient retry logic - // (mid-stream failures can't be retried because partial output was already emitted) + // Tracks whether any content was streamed this call — selects the watchdog timeout + // (the generous prefill budget before the first token vs the tighter inter-delta + // budget after). private bool _anyContentStreamed; // Per-turn diagnostic correlation (ephemeral) @@ -161,10 +162,6 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // overflows again, fail the turn. Reset at the start of each new user turn. private int _compactionOverflowRetryCount; - // Per-turn retry counter for transient streaming failures (5xx, 429) - private int _streamingRetryAttempt; - private static readonly object StreamingRetryTimerKey = new(); - // Skill registry for slash-command dispatch private readonly Skills.SkillRegistry? _skillRegistry; private readonly SubAgentDefinitionRegistry? _subAgentRegistry; @@ -603,33 +600,14 @@ private void Processing() // Use the configured context window as the token count estimate since // the provider rejected the request without returning usage stats. - Timers.Cancel(StreamingRetryTimerKey); Self.Tell(new CompactionTriggered(_model.ContextWindowTokens)); TransitionTo(SessionPhase.Compacting); return; } - // Pre-stream transient failure: retry with backoff if no data was streamed yet. - // IsTransientStreamingError handles ProviderException (which wraps HTTP status codes) - // while RetryPolicy only provides the attempt limit and backoff delay. - if (!_anyContentStreamed && IsTransientStreamingError(msg.Cause)) - { - var policy = _config.Tuning.StreamingRetryPolicy; - if (_streamingRetryAttempt < policy.MaxRetries) - { - var delay = policy.GetDelay(_streamingRetryAttempt); - _streamingRetryAttempt++; - TurnLog().Warning(msg.Cause, - "turn_llm_transient_failure — retrying in {DelayMs:F0}ms (attempt {Attempt}/{Max})", - delay.TotalMilliseconds, _streamingRetryAttempt, policy.MaxRetries); - Timers.StartSingleTimer( - StreamingRetryTimerKey, - new RetryLlmCallAfterBackoff(_streamingRetryAttempt), - delay); - return; // Stay in Processing, watchdog is already stopped - } - } - + // Transient-failure retry is owned entirely by the transport + // (RetryingChatClient, pre-first-chunk) and is already exhausted by the time + // the failure reaches here, so a failed turn is terminal. TurnLog().Error(msg.Cause, "turn_llm_call_failed"); // Evict discovered tools to prevent a poisoned tool set from cascading @@ -642,12 +620,6 @@ private void Processing() FailCurrentTurn(errorMessage, msg.Cause, category); }); - Command(msg => - { - TurnLog().Info("turn_streaming_retry attempt={Attempt}", msg.Attempt); - FireLlmCall(); - }); - Command(msg => { if (!_watchdog.IsCurrent(msg)) @@ -1369,7 +1341,6 @@ private void DrainBufferOrReady() if (resumeToolLoop || hadBufferedMessages) { - _streamingRetryAttempt = 0; FireLlmCall(); TransitionTo(SessionPhase.Processing); } @@ -2007,7 +1978,6 @@ private void DrainBufferedMessagesOrBecomeReady() _buffer.Clear(); _recallManager.ResetForNewTurn(); // New user input — resolve recall fresh - _streamingRetryAttempt = 0; FireLlmCall(); // Already in Processing — no transition needed, just fired a new LLM call return; @@ -2213,7 +2183,6 @@ private void ContinueIncomingUserMessage(SendUserMessage cmd) _state = _state.AddUserMessage(userContent, mediaRefs.Count > 0 ? mediaRefs : null); TryReplyAck(); _recallManager.ResetForNewTurn(); - _streamingRetryAttempt = 0; _compactionOverflowRetryCount = 0; FireInitialTurnLlmCall(executableUserContent); TransitionTo(SessionPhase.Processing); @@ -2481,13 +2450,6 @@ private string ExtractLlmErrorMessage(Exception? cause) internal static bool IsContextOverflowError(Exception? ex) => LlmFailureClassifier.IsContextOverflow(ex); - /// - /// Detect transient streaming errors that are safe to retry when no data - /// has been streamed yet (5xx server errors, 429 rate limits, network failures). - /// - internal static bool IsTransientStreamingError(Exception? ex) - => LlmFailureClassifier.IsTransientStreaming(ex); - private string GetSessionDirectory() => SessionDirectoryHelper.GetSessionDirectory(_sessionId, _sessionsBasePath); @@ -4153,7 +4115,6 @@ private void FailCurrentTurn(string errorMessage, Exception cause, ErrorCategory _pendingToolInteractions.Clear(); _resolvedToolApprovals.Clear(); ClearApprovalTurnState(); - Timers.Cancel(StreamingRetryTimerKey); _state = _state.AddErrorReply(errorMessage); var correlationId = Guid.NewGuid(); diff --git a/src/Netclaw.Configuration/RetryPolicy.cs b/src/Netclaw.Configuration/RetryPolicy.cs index d272bb30b..a8f7a3beb 100644 --- a/src/Netclaw.Configuration/RetryPolicy.cs +++ b/src/Netclaw.Configuration/RetryPolicy.cs @@ -18,14 +18,23 @@ public sealed record RetryPolicy /// /// Determines whether the given exception is transient and should be retried. - /// Retries on: status-less network failures, 408/429/5xx responses, - /// and timeout-style cancellations. + /// Retries on: status-less network failures, 408/429/5xx responses (whether they + /// surface as a raw or are curated into a + /// by a provider transport layer), and + /// timeout-style cancellations. /// public bool ShouldRetry(Exception ex, int attempt) { if (attempt >= MaxRetries) return false; + // Curated provider errors (e.g. the self-hosted OpenAI-compatible client) carry + // the HTTP status on a ProviderException rather than a raw HttpRequestException, + // and it may be nested under an inner exception. Without this, the retry layer + // would miss the provider 429/5xx it most needs to retry. + if (FindInner(ex) is { StatusCode: 408 or 429 or (>= 500 and <= 599) }) + return true; + return ex switch { HttpRequestException { StatusCode: null } => true, @@ -42,6 +51,18 @@ HttpStatusCode.ServiceUnavailable or }; } + private static T? FindInner(Exception? ex) where T : Exception + { + while (ex is not null) + { + if (ex is T match) + return match; + ex = ex.InnerException; + } + + return null; + } + /// /// Returns the delay before the next retry attempt using exponential backoff with jitter. /// diff --git a/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs index 559c5fad3..6441ff682 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs @@ -249,6 +249,30 @@ [new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)) { } Assert.Equal(1, attempts); // cancellation is not retried } + [Fact] + public async Task StreamingRetries_ProviderException5xx_ThenSucceeds() + { + // Curated provider errors carry the status on a ProviderException, not a raw + // HttpRequestException — the transport must still recognize them as transient. + var attempts = 0; + var fake = new FakeChatClient(streamHandler: (_, _, ct) => + { + attempts++; + return ThrowProviderExceptionThenYield(attempts, failUntil: 3, ct); + }); + var client = new RetryingChatClient(fake, _policy, NullLogger.Instance); + + var updates = new List(); + await foreach (var u in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) + { + updates.Add(u); + } + + Assert.Single(updates); + Assert.Equal(3, attempts); // 2 ProviderException(502) failures + 1 success + } + // Throws a retryable 429 before yielding any chunk while attemptNumber < failUntil, // otherwise yields one chunk. The runtime-dependent condition keeps the yield // reachable (no CS0162) so no warning suppression is needed. @@ -264,6 +288,19 @@ private static async IAsyncEnumerable ThrowBeforeChunkThenYi yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("ok")] }; } + // Same shape as ThrowBeforeChunkThenYield but throws a curated ProviderException(502). + private static async IAsyncEnumerable ThrowProviderExceptionThenYield( + int attemptNumber, int failUntil, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + if (attemptNumber < failUntil) + throw new ProviderException("server error (502)", "HTTP 502", statusCode: 502); + + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("ok")] }; + } + private static async IAsyncEnumerable YieldThenThrow( [EnumeratorCancellation] CancellationToken cancellationToken) { diff --git a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs index a4e36ed58..c06cbbdb2 100644 --- a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs +++ b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs @@ -24,7 +24,8 @@ public static class DaemonProviderServiceExtensions public static IServiceCollection AddDaemonLlmProviders( this IServiceCollection services, Dictionary providers, - ModelSelection models) + ModelSelection models, + RetryPolicy? retryPolicy = null) { // Register plugins and OAuth from Netclaw.Providers services.AddLlmProviders(); @@ -33,8 +34,10 @@ public static IServiceCollection AddDaemonLlmProviders( services.AddSingleton(sp => new ProviderPluginFactory(providers, sp.GetServices())); - // Retry policy (TODO: make configurable via netclaw.json Resilience section) - services.AddSingleton(new RetryPolicy()); + // Transport retry budget/backoff. The RetryingChatClient layer is the single + // owner of LLM transient-failure retry; this is its configured policy + // (Session:Tuning:StreamingRetryPolicy), defaulting to the standard policy. + services.AddSingleton(retryPolicy ?? new RetryPolicy()); // Composes the cross-cutting middleware (Logging → Retry) around each provider // pipeline via ChatClientBuilder. diff --git a/src/Netclaw.Daemon/Program.cs b/src/Netclaw.Daemon/Program.cs index b9dc4d33f..afd8c60ad 100644 --- a/src/Netclaw.Daemon/Program.cs +++ b/src/Netclaw.Daemon/Program.cs @@ -362,7 +362,13 @@ static NetclawPaths ConfigureConfigServices(IServiceCollection services, IConfig var models = configuration.GetSection("Models") .Get() ?? new ModelSelection(); - services.AddDaemonLlmProviders(providers, models); + // The transport RetryingChatClient is the single owner of LLM transient-failure + // retry, so it uses the configured streaming-retry budget. + var streamingRetryPolicy = SessionConfig + .BindFromConfiguration(configuration.GetSection("Session")) + .Tuning.StreamingRetryPolicy; + + services.AddDaemonLlmProviders(providers, models, streamingRetryPolicy); return paths; } From e836e23d0569f8888bb2a10af1994a1cdd0dd6e2 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 3 Jun 2026 15:34:06 +0000 Subject: [PATCH 4/4] =?UTF-8?q?refactor(openai):=20drop=20StreamingOnlyCha?= =?UTF-8?q?tClient=20shim=20=E2=80=94=20Netclaw=20is=20streaming-only?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The streaming-native chat-client stack converts every auxiliary caller (title generation, memory extraction, compaction) to the streaming transport, so no code path invokes the non-streaming GetResponseAsync on the Codex client anymore. The StreamingOnlyChatClient wrapper (added in 0e5ba60c to serve those calls via streaming under the hood) is now dead; the Codex 'Stream must be set to true' 400 is structurally unreachable. A comment on the OAuth path records why no special handling is needed. --- .../OpenAi/OpenAiProviderPlugin.cs | 36 +++++-------------- 1 file changed, 9 insertions(+), 27 deletions(-) diff --git a/src/Netclaw.Providers/OpenAi/OpenAiProviderPlugin.cs b/src/Netclaw.Providers/OpenAi/OpenAiProviderPlugin.cs index 2cd13aa75..1bd99b25c 100644 --- a/src/Netclaw.Providers/OpenAi/OpenAiProviderPlugin.cs +++ b/src/Netclaw.Providers/OpenAi/OpenAiProviderPlugin.cs @@ -37,15 +37,15 @@ public override IChatClient CreateChatClient(ProviderEntry entry, ModelReference }; options.AddPolicy(new OpenAiCodexRequestPolicy(accountId), PipelinePosition.PerCall); - // The Codex backend rejects non-streaming Responses calls with - // 400 {"detail":"Stream must be set to true"}. Netclaw's session loop - // streams, but auxiliary calls (title generation, memory extraction, - // compaction) use the non-streaming GetResponseAsync path. Wrap the - // client so those calls are served by streaming under the hood. - return new StreamingOnlyChatClient( - new OpenAI.Responses.ResponsesClient( - new ApiKeyCredential(token.Value), options) - .AsIChatClient(model.ModelId)); + // No non-streaming wrapper is needed here: Netclaw issues streaming-only + // LLM calls everywhere (the session loop and every auxiliary caller — + // title generation, memory extraction, compaction — go through the + // streaming transport), so the Codex backend's + // 400 {"detail":"Stream must be set to true"} on non-streaming Responses + // calls is structurally unreachable. + return new OpenAI.Responses.ResponsesClient( + new ApiKeyCredential(token.Value), options) + .AsIChatClient(model.ModelId); } // API key path → standard endpoint @@ -54,21 +54,3 @@ public override IChatClient CreateChatClient(ProviderEntry entry, ModelReference .AsIChatClient(model.ModelId); } } - -/// -/// Serves non-streaming calls by consuming the -/// underlying streaming endpoint and aggregating the updates. Required for the OpenAI Codex -/// backend, which rejects non-streaming Responses requests with -/// 400 {"detail":"Stream must be set to true"}. Streaming calls pass straight through. -/// -internal sealed class StreamingOnlyChatClient : DelegatingChatClient -{ - public StreamingOnlyChatClient(IChatClient innerClient) : base(innerClient) { } - - public override Task GetResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - => base.GetStreamingResponseAsync(messages, options, cancellationToken) - .ToChatResponseAsync(cancellationToken); -}