feat(providers): streaming-native chat-client stack with composable routing

Directory.Packages.props

@@ -32,6 +32,7 @@
     <PackageVersion Include="HtmlAgilityPack" Version="1.12.4" />
     <PackageVersion Include="Microsoft.AspNetCore.DataProtection.Extensions" Version="$(MicrosoftAspNetCoreVersion)" />
     <PackageVersion Include="Microsoft.AspNetCore.OpenApi" Version="$(MicrosoftAspNetCoreVersion)" />
+    <PackageVersion Include="Microsoft.Extensions.AI" Version="$(MicrosoftExtensionsAIVersion)" />
     <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="$(MicrosoftExtensionsAIVersion)" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="$(MicrosoftAspNetCoreVersion)" />
     <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="$(MicrosoftExtensionsAIVersion)" />

src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs

@@ -1500,71 +1500,11 @@ await retryWatcher.ExpectMsgAsync<TextOutput>(
         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<SessionManagerActorKey>();
-        var subscriber = CreateTestProbe("retry-502-sub");
-
-        await sessionManager.Ask<SessionJoined>(new JoinSession(subscriber)
-        {
-            SessionId = sessionId,
-            Filter = OutputFilter.Full
-        }, cancellationToken: TestContext.Current.CancellationToken);
-        await subscriber.ExpectMsgAsync<SessionJoined>(cancellationToken: TestContext.Current.CancellationToken);
-
-        await sessionManager.Ask<CommandAck>(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<TextOutput>(TimeSpan.FromSeconds(15), cancellationToken: TestContext.Current.CancellationToken);
-        Assert.Contains("fake", text.Text, StringComparison.OrdinalIgnoreCase);
-        var completed = await subscriber.ExpectMsgAsync<TurnCompleted>(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<SessionManagerActorKey>();
-        var subscriber = CreateTestProbe("retry-exhaust-sub");
-
-        await sessionManager.Ask<SessionJoined>(new JoinSession(subscriber)
-        {
-            SessionId = sessionId,
-            Filter = OutputFilter.Full
-        }, cancellationToken: TestContext.Current.CancellationToken);
-        await subscriber.ExpectMsgAsync<SessionJoined>(cancellationToken: TestContext.Current.CancellationToken);
-
-        await sessionManager.Ask<CommandAck>(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<ErrorOutput>(TimeSpan.FromSeconds(20), cancellationToken: TestContext.Current.CancellationToken);
-        Assert.Equal(ErrorCategory.ProviderFailure, error.Category);
-        var completed = await subscriber.ExpectMsgAsync<TurnCompleted>(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()

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<CurationDecision> 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);

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<ProviderException>(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<T>(Exception? ex) where T : Exception
     {
         while (ex is not null)

src/Netclaw.Actors/Sessions/LlmMessages.cs

@@ -220,9 +220,3 @@ internal sealed record PassivationTimeout : INoSerializationVerificationNeeded;
 /// signal. See <c>LlmSessionActor.CompletePassivation</c>.
 /// </summary>
 internal sealed record PassivationFinalStop : INoSerializationVerificationNeeded;
-
-/// <summary>
-/// Timer-fired message that triggers an LLM call retry after exponential backoff.
-/// Carries the attempt number for observability logging.
-/// </summary>
-internal sealed record RetryLlmCallAfterBackoff(int Attempt) : INoSerializationVerificationNeeded;

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<RetryLlmCallAfterBackoff>(msg =>
-        {
-            TurnLog().Info("turn_streaming_retry attempt={Attempt}", msg.Attempt);
-            FireLlmCall();
-        });
-
         Command<ProcessingWatchdogExpired>(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);
 
-    /// <summary>
-    /// Detect transient streaming errors that are safe to retry when no data
-    /// has been streamed yet (5xx server errors, 429 rate limits, network failures).
-    /// </summary>
-    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();

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))
             {

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))
             {

src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs

@@ -56,6 +56,23 @@ internal readonly record struct StreamReadResult(
 /// </summary>
 internal static class StreamingResponseReader
 {
+    private static readonly Action<ChatResponseUpdate, StreamUpdateClassification, StreamDiagnostics> NoOp =
+        static (_, _, _) => { };
+
+    /// <summary>
+    /// 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 <c>result.Response.Text</c> without an empty-Messages guard.
+    /// </summary>
+    public static Task<StreamReadResult> ReadAsync(
+        IChatClient client,
+        IEnumerable<AiChatMessage> messages,
+        ChatOptions? options,
+        CancellationToken ct)
+        => ReadAsync(client, messages, options, NoOp, ct);
+
     public static async Task<StreamReadResult> ReadAsync(
         IChatClient client,
         IEnumerable<AiChatMessage> messages,

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;

src/Netclaw.Configuration/ChatRoutingContext.cs

@@ -0,0 +1,24 @@
+// -----------------------------------------------------------------------
+// <copyright file="ChatRoutingContext.cs" company="Petabridge, LLC">
+//      Copyright (C) 2026 - 2026 Petabridge, LLC <https://petabridge.com>
+// </copyright>
+// -----------------------------------------------------------------------
+namespace Netclaw.Configuration;
+
+/// <summary>
+/// The inputs a chat-client router uses to select which composed pipeline to invoke
+/// for a call. Minimal today — only <see cref="Role"/> 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.
+/// </summary>
+public sealed record ChatRoutingContext
+{
+    /// <summary>The model role being requested (today's only routing signal).</summary>
+    public required ModelRole Role { get; init; }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public string? SessionId { get; init; }
+}