fix(providers): detect a mid-stream LLM stall within seconds

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

@@ -45,6 +45,54 @@ internal readonly record struct StreamReadResult(
     ChatResponse Response,
     StreamDiagnostics Diagnostics);
 
+/// <summary>
+/// Classifies a streaming LLM update as substantive model progress or a
+/// content-free keepalive. This is a small public contract because two
+/// independent watchdog layers, in two different assemblies, must agree on
+/// one rule: <see cref="StreamingResponseReader"/> (feeds
+/// <c>ProcessingWatchdog</c>'s main-session and sub-agent inter-delta
+/// budgets) and <c>Netclaw.Daemon.Configuration.StreamStallGuardChatClient</c>
+/// (the uniform mid-stream stall guard for every LLM call path, including the
+/// sidecar paths <c>ProcessingWatchdog</c> does not cover). A change to this
+/// predicate changes the arming behavior of BOTH watchdogs — verify both
+/// when this rule changes.
+/// </summary>
+public static class ChatStreamUpdateClassifier
+{
+    /// <summary>
+    /// True when an update represents real model progress. A finish reason, non-empty
+    /// text/thinking, a tool call, or any non-usage content all count. Only a
+    /// content-free heartbeat or a usage-only chunk (with no finish reason) is treated
+    /// as a non-substantive keepalive. A reasoning delta with text is substantive, the
+    /// same as a text or tool-call delta — it arms a two-phase watchdog's tight budget.
+    /// This mirrors the pre-extraction predicate so a provider that streams an
+    /// error/refusal or other non-text content before the first token is still
+    /// recognized as progress, not silently treated as a hang.
+    /// </summary>
+    public static bool IsSubstantiveUpdate(ChatResponseUpdate update)
+    {
+        if (update.FinishReason is not null)
+            return true;
+
+        foreach (var content in update.Contents)
+        {
+            switch (content)
+            {
+                case TextContent text when !string.IsNullOrEmpty(text.Text):
+                case TextReasoningContent reasoning when !string.IsNullOrEmpty(reasoning.Text):
+                case FunctionCallContent:
+                    return true;
+                case UsageContent:
+                    break;
+                default:
+                    return true;
+            }
+        }
+
+        return false;
+    }
+}
+
 /// <summary>
 /// Single owner of the streaming LLM consumption loop shared by the main-session
 /// (<see cref="SessionLlmInvoker"/>) and sub-agent (<c>SubAgentActor.InvokeLlmAsync</c>)
@@ -167,40 +215,9 @@ public static async Task<StreamReadResult> ReadAsync(
     /// </summary>
     internal static StreamUpdateClassification Classify(ChatResponseUpdate update, bool anySubstantiveSeen)
     {
-        var hasSubstantive = IsSubstantiveUpdate(update);
+        var hasSubstantive = ChatStreamUpdateClassifier.IsSubstantiveUpdate(update);
         return new StreamUpdateClassification(
             HasSubstantiveContent: hasSubstantive,
             IsFirstSubstantive: hasSubstantive && !anySubstantiveSeen);
     }
-
-    /// <summary>
-    /// True when an update represents real model progress. A finish reason, non-empty
-    /// text/thinking, a tool call, or any non-usage content all count. Only a
-    /// content-free heartbeat or a usage-only chunk (with no finish reason) is treated
-    /// as a non-substantive keepalive. This mirrors the pre-extraction predicate so a
-    /// provider that streams an error/refusal or other non-text content before the
-    /// first token is still recognized as progress, not silently treated as a hang.
-    /// </summary>
-    internal static bool IsSubstantiveUpdate(ChatResponseUpdate update)
-    {
-        if (update.FinishReason is not null)
-            return true;
-
-        foreach (var content in update.Contents)
-        {
-            switch (content)
-            {
-                case TextContent text when !string.IsNullOrEmpty(text.Text):
-                case TextReasoningContent reasoning when !string.IsNullOrEmpty(reasoning.Text):
-                case FunctionCallContent:
-                    return true;
-                case UsageContent:
-                    break;
-                default:
-                    return true;
-            }
-        }
-
-        return false;
-    }
 }

src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs

@@ -590,6 +590,40 @@ await File.WriteAllTextAsync(paths.NetclawConfigPath,
         Assert.Equal(DoctorSeverity.Pass, result.Severity);
     }
 
+    [Fact]
+    public async Task ReturnsPass_WhenSessionTuningStreamingRetryPolicySet()
+    {
+        // Session.Tuning.StreamingRetryPolicy (including the mid-stream inactivity
+        // guard's StreamInactivityTimeout) must be settable by an operator — it is
+        // not enough for it to bind correctly; Session.Tuning uses
+        // additionalProperties: false, so the schema must list it explicitly.
+        var basePath = CreateTempBasePath();
+        var paths = new NetclawPaths(basePath);
+        paths.EnsureDirectoriesExist();
+
+        await File.WriteAllTextAsync(paths.NetclawConfigPath,
+            """
+            {
+              "configVersion": 1,
+              "Session": {
+                "Tuning": {
+                  "StreamingRetryPolicy": {
+                    "MaxRetries": 5,
+                    "BaseDelay": "00:00:02",
+                    "MaxDelay": "00:01:00",
+                    "StreamInactivityTimeout": "00:00:00"
+                  }
+                }
+              }
+            }
+            """, TestContext.Current.CancellationToken);
+
+        var check = new ConfigSchemaDoctorCheck(paths);
+        var result = await check.RunAsync(TestContext.Current.CancellationToken);
+
+        Assert.Equal(DoctorSeverity.Pass, result.Severity);
+    }
+
     [Fact]
     public async Task ReturnsError_WhenStaleSessionMaxToolCallsPerTurnPresent()
     {

src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs

@@ -87,4 +87,24 @@ public void BindFromConfiguration_prefers_nested_tuning_values_over_legacy_root_
 
         Assert.Equal(0.8, bound.Tuning.CompactionThreshold);
     }
+
+    [Fact]
+    public void BindFromConfiguration_binds_nested_StreamingRetryPolicy_StreamInactivityTimeout()
+    {
+        // Cross-boundary contract: an operator-set Session:Tuning:StreamingRetryPolicy
+        // value (schema-validated as a HH:mm:ss string) must reach the RetryPolicy
+        // record StreamStallGuardChatClient actually reads at runtime.
+        var config = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string?>
+            {
+                ["Session:Tuning:StreamingRetryPolicy:MaxRetries"] = "5",
+                ["Session:Tuning:StreamingRetryPolicy:StreamInactivityTimeout"] = "00:01:30"
+            })
+            .Build();
+
+        var bound = SessionConfig.BindFromConfiguration(config.GetSection("Session"));
+
+        Assert.Equal(5, bound.Tuning.StreamingRetryPolicy.MaxRetries);
+        Assert.Equal(TimeSpan.FromMinutes(1.5), bound.Tuning.StreamingRetryPolicy.StreamInactivityTimeout);
+    }
 }

src/Netclaw.Configuration/RetryPolicy.cs

@@ -16,6 +16,22 @@ public sealed record RetryPolicy
     public TimeSpan BaseDelay { get; init; } = TimeSpan.FromSeconds(1);
     public TimeSpan MaxDelay { get; init; } = TimeSpan.FromSeconds(30);
 
+    /// <summary>
+    /// Maximum gap between streaming updates once a stream has started producing
+    /// substantive output. Once armed, resets on every later update — including
+    /// content-free keepalives — so a backend that paces real tokens slowly is not
+    /// killed, only one that stops emitting anything at all. Does not apply before
+    /// the first substantive update. A reasoning delta with text is substantive and
+    /// arms it, the same as a text or tool-call delta; only a content-free keepalive
+    /// does not. Time to first substantive output stays governed by the coarser
+    /// per-call watchdog, because a self-hosted backend can be legitimately silent
+    /// for minutes during cold prefill. This is the fast detector for a stream that
+    /// goes silent mid-response — a dead or half-open connection that the coarse
+    /// per-call watchdog would otherwise take minutes to catch. Set to
+    /// <see cref="TimeSpan.Zero"/> to disable.
+    /// </summary>
+    public TimeSpan StreamInactivityTimeout { get; init; } = TimeSpan.FromSeconds(45);
+
     /// <summary>
     /// Determines whether the given exception is transient and should be retried.
     /// Retries on: status-less network failures, 408/429/5xx responses (whether they

src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json

@@ -263,6 +263,34 @@
               "minimum": 0,
               "default": 2000,
               "description": "Character budget for memory content injected by automatic recall per turn. Items are admitted in rank order until the budget is exhausted; whole items are dropped, never truncated. Set to 0 to disable."
+            },
+            "StreamingRetryPolicy": {
+              "type": "object",
+              "description": "Retry policy for transient streaming LLM failures (5xx, 429), plus the mid-stream inactivity guard. Only pre-first-chunk failures are retried; mid-stream failures are not, because the partial response cannot be reconstructed.",
+              "properties": {
+                "MaxRetries": {
+                  "type": "integer",
+                  "minimum": 0,
+                  "default": 3,
+                  "description": "Maximum retry attempts for a transient streaming failure."
+                },
+                "BaseDelay": {
+                  "type": "string",
+                  "default": "00:00:01",
+                  "description": "Base retry backoff delay in HH:mm:ss format, before jitter and exponential growth."
+                },
+                "MaxDelay": {
+                  "type": "string",
+                  "default": "00:00:30",
+                  "description": "Retry backoff delay cap in HH:mm:ss format."
+                },
+                "StreamInactivityTimeout": {
+                  "type": "string",
+                  "default": "00:00:45",
+                  "description": "Maximum gap between streaming updates once a stream has started producing substantive output, in HH:mm:ss format. Resets on every later update, including a content-free keepalive. Does not apply before the first substantive update — time to first substantive output stays governed by the coarser per-call watchdog. Set to 00:00:00 to disable this guard."
+                }
+              },
+              "additionalProperties": false
             }
           },
           "additionalProperties": false

src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs

@@ -8,6 +8,7 @@
 using Microsoft.Extensions.AI;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Time.Testing;
 using Netclaw.Actors.Sessions;
 using Netclaw.Configuration;
 using Netclaw.Daemon.Configuration;
@@ -119,6 +120,81 @@ private static async IAsyncEnumerable<ChatResponseUpdate> ThrowFirstThenYield(
         yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("ok")] };
     }
 
+    [Fact]
+    public async Task Compose_wires_StreamStallGuard_soMidStreamStall_AbortsInSeconds_NotMinutes()
+    {
+        // End-to-end proof: a provider that streams a chunk and then goes silent (dead or
+        // half-open connection — no more tokens, no error, no close) surfaces a
+        // TimeoutException through the full Logging -> Retry -> StreamStallGuard -> leaf
+        // pipeline once the fast inactivity window elapses, not the slow per-call watchdog.
+        var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+        var stallPolicy = _policy with { StreamInactivityTimeout = TimeSpan.FromSeconds(30) };
+        var attempts = 0;
+        var leaf = new FakeChatClient(streamHandler: (_, _, ct) =>
+        {
+            attempts++;
+            return YieldOneThenStallForever(ct);
+        });
+        var pipeline = PipelineChatClientFactory.Compose(leaf, stallPolicy, NullLoggerFactory.Instance, time);
+
+        await using var enumerator = pipeline
+            .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+            .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+        Assert.True(await enumerator.MoveNextAsync()); // chunk1 flows through Logging -> Retry -> StreamStallGuard -> leaf
+
+        var pending = enumerator.MoveNextAsync().AsTask();
+        time.Advance(stallPolicy.StreamInactivityTimeout);
+
+        var ex = await Assert.ThrowsAsync<TimeoutException>(() => pending);
+        Assert.Contains("stall detected", ex.Message, StringComparison.OrdinalIgnoreCase);
+
+        // Post-first-chunk: RetryingChatClient must not have silently re-issued the request.
+        Assert.Equal(1, attempts);
+    }
+
+    [Fact]
+    public async Task Compose_wires_StreamStallGuard_soSlowButProgressingStream_CompletesNormally()
+    {
+        var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+        var stallPolicy = _policy with { StreamInactivityTimeout = TimeSpan.FromSeconds(30) };
+        var gap = TimeSpan.FromSeconds(5);
+        var leaf = new FakeChatClient(streamHandler: (_, _, ct) => SlowButProgressing(time, gap, count: 4, ct));
+        var pipeline = PipelineChatClientFactory.Compose(leaf, stallPolicy, NullLoggerFactory.Instance, time);
+
+        await using var enumerator = pipeline
+            .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+            .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+        for (var i = 0; i < 4; i++)
+        {
+            var pending = enumerator.MoveNextAsync().AsTask();
+            time.Advance(gap);
+            Assert.True(await pending);
+        }
+
+        Assert.False(await enumerator.MoveNextAsync());
+    }
+
+    private static async IAsyncEnumerable<ChatResponseUpdate> YieldOneThenStallForever(
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("chunk1")] };
+        await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
+        yield break; // unreachable — Task.Delay above only returns via cancellation
+    }
+
+    private static async IAsyncEnumerable<ChatResponseUpdate> SlowButProgressing(
+        TimeProvider time, TimeSpan gap, int count,
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        for (var i = 0; i < count; i++)
+        {
+            await Task.Delay(gap, time, cancellationToken);
+            yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent($"chunk{i}")] };
+        }
+    }
+
     // Records, for each log call, the session id on the active scope stack — so a test can
     // assert that a specific line (not just the call as a whole) was emitted in scope.
     private sealed class MessageScopeLogger : ILogger