diff --git a/Directory.Packages.props b/Directory.Packages.props index ee657fe1a..666de522d 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -32,6 +32,7 @@ + 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/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/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 82d3f0890..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); } @@ -1641,9 +1612,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) @@ -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.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..731333836 --- /dev/null +++ b/src/Netclaw.Configuration/ChatRoutingContext.cs @@ -0,0 +1,24 @@ +// ----------------------------------------------------------------------- +// +// 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; 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; } +} 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/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..6441ff682 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,108 @@ 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(); + await foreach (var u in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) + { + 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 + } + + [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(); @@ -170,6 +270,44 @@ [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.C } 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. + 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")] }; + } + + // 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) + { + 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..4670c0904 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs @@ -0,0 +1,265 @@ +// ----------------------------------------------------------------------- +// +// 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); + } + + [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(); + 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..c06cbbdb2 100644 --- a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs +++ b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs @@ -11,43 +11,53 @@ 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, Dictionary providers, - ModelSelection models) + ModelSelection models, + RetryPolicy? retryPolicy = null) { // 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()); - }); + // 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. + 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..2a658e9bd 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,87 @@ 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++; + } + } + } + + 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; - await Task.Delay(delay, _timeProvider, cancellationToken); + // 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); + // 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); } } 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 @@ + 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; } 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); -}