diff --git a/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs b/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs
index a999f0aea..a61650171 100644
--- a/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs
+++ b/src/Netclaw.Actors/Sessions/Pipelines/StreamingResponseReader.cs
@@ -45,6 +45,54 @@ internal readonly record struct StreamReadResult(
ChatResponse Response,
StreamDiagnostics Diagnostics);
+///
+/// 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: (feeds
+/// ProcessingWatchdog's main-session and sub-agent inter-delta
+/// budgets) and Netclaw.Daemon.Configuration.StreamStallGuardChatClient
+/// (the uniform mid-stream stall guard for every LLM call path, including the
+/// sidecar paths ProcessingWatchdog does not cover). A change to this
+/// predicate changes the arming behavior of BOTH watchdogs — verify both
+/// when this rule changes.
+///
+public static class ChatStreamUpdateClassifier
+{
+ ///
+ /// 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.
+ ///
+ 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;
+ }
+}
+
///
/// Single owner of the streaming LLM consumption loop shared by the main-session
/// () and sub-agent (SubAgentActor.InvokeLlmAsync)
@@ -167,40 +215,9 @@ public static async Task ReadAsync(
///
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);
}
-
- ///
- /// 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.
- ///
- 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;
- }
}
diff --git a/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs b/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs
index f4c25dbe5..249b0847c 100644
--- a/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs
+++ b/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()
{
diff --git a/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs b/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs
index c473a3e8c..4b0703efa 100644
--- a/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs
+++ b/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
+ {
+ ["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);
+ }
}
diff --git a/src/Netclaw.Configuration/RetryPolicy.cs b/src/Netclaw.Configuration/RetryPolicy.cs
index 982349006..f5836426b 100644
--- a/src/Netclaw.Configuration/RetryPolicy.cs
+++ b/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);
+ ///
+ /// 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
+ /// to disable.
+ ///
+ public TimeSpan StreamInactivityTimeout { get; init; } = TimeSpan.FromSeconds(45);
+
///
/// Determines whether the given exception is transient and should be retried.
/// Retries on: status-less network failures, 408/429/5xx responses (whether they
diff --git a/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json b/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json
index a36c794dc..fe347e0fa 100644
--- a/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json
+++ b/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
diff --git a/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs b/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs
index a2491d4a6..ce3e5936d 100644
--- a/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs
+++ b/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 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(() => 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 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 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
diff --git a/src/Netclaw.Daemon.Tests/Configuration/StreamStallGuardChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/StreamStallGuardChatClientTests.cs
new file mode 100644
index 000000000..a7ed2d1b4
--- /dev/null
+++ b/src/Netclaw.Daemon.Tests/Configuration/StreamStallGuardChatClientTests.cs
@@ -0,0 +1,253 @@
+// -----------------------------------------------------------------------
+//
+// Copyright (C) 2026 - 2026 Petabridge, LLC
+//
+// -----------------------------------------------------------------------
+using System.Runtime.CompilerServices;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Time.Testing;
+using Netclaw.Configuration;
+using Netclaw.Daemon.Configuration;
+using Xunit;
+
+namespace Netclaw.Daemon.Tests.Configuration;
+
+///
+/// All tests drive a — no real Task.Delay or
+/// Thread.Sleep waits. A "stall" is simulated with
+/// Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken), which never
+/// completes on its own and resolves only when the guard cancels it — exactly the
+/// dead/half-open-connection shape under test (no more tokens, no error, no close).
+///
+public sealed class StreamStallGuardChatClientTests
+{
+ private static readonly RetryPolicy Policy = new() { StreamInactivityTimeout = TimeSpan.FromSeconds(45) };
+
+ [Fact]
+ public async Task StallAfterFirstDelta_CancelsWithinInactivityWindow_AndThrowsRetryableTimeout()
+ {
+ var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+ var leaf = new FakeChatClient(streamHandler: (_, _, ct) => YieldTwoThenStallForever(ct));
+ var client = new StreamStallGuardChatClient(leaf, Policy, NullLogger.Instance, time);
+
+ await using var enumerator = client
+ .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+ .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+ Assert.True(await enumerator.MoveNextAsync()); // chunk 1
+ Assert.True(await enumerator.MoveNextAsync()); // chunk 2
+
+ // Provider goes silent after the second delta: the guard's inactivity clock,
+ // armed after the first delta, is now the only thing that can unblock this.
+ var pending = enumerator.MoveNextAsync().AsTask();
+ time.Advance(Policy.StreamInactivityTimeout);
+
+ var ex = await Assert.ThrowsAsync(() => pending);
+
+ // The whole point of a fast, real-time detector is that the existing transient-
+ // failure retry policy already knows what to do with the exception it throws —
+ // no new retry mechanism needed.
+ Assert.True(Policy.ShouldRetry(ex, attempt: 0));
+ }
+
+ [Fact]
+ public async Task SlowButProgressingStream_IsNotFalselyAborted()
+ {
+ // Every gap is 5s under the 45s window — a legitimately paced stream (e.g. a
+ // heavily loaded self-hosted backend) must be allowed to finish.
+ var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+ var gap = TimeSpan.FromSeconds(5);
+ var leaf = new FakeChatClient(streamHandler: (_, _, ct) => SlowButProgressing(time, gap, count: 6, ct));
+ var client = new StreamStallGuardChatClient(leaf, Policy, NullLogger.Instance, time);
+
+ await using var enumerator = client
+ .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+ .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+ for (var i = 0; i < 6; i++)
+ {
+ var pending = enumerator.MoveNextAsync().AsTask();
+ time.Advance(gap);
+ Assert.True(await pending);
+ }
+
+ Assert.False(await enumerator.MoveNextAsync()); // clean completion, never aborted
+ }
+
+ [Fact]
+ public async Task StallBeforeFirstDelta_IsNotGovernedByTheGuard()
+ {
+ // Time to first byte is left to the existing, more generous per-call watchdog —
+ // a self-hosted backend can be legitimately silent for minutes during cold
+ // prefill with no keepalive to reset a tighter timer. Advancing well past the
+ // inactivity window before any delta arrives must not trip this guard.
+ var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+ var prefillDelay = Policy.StreamInactivityTimeout * 4;
+ var leaf = new FakeChatClient(streamHandler: (_, _, ct) => DelayThenYieldOnce(time, prefillDelay, ct));
+ var client = new StreamStallGuardChatClient(leaf, Policy, NullLogger.Instance, time);
+
+ await using var enumerator = client
+ .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+ .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+ var pending = enumerator.MoveNextAsync().AsTask();
+
+ // Cross well past what would be the inactivity window if it were armed.
+ time.Advance(Policy.StreamInactivityTimeout * 2);
+ Assert.False(pending.IsCompleted);
+
+ // Finish the (legitimately long) prefill — the first delta still arrives cleanly.
+ time.Advance(prefillDelay - Policy.StreamInactivityTimeout * 2);
+ Assert.True(await pending);
+ }
+
+ [Fact]
+ public async Task KeepaliveOnlyUpdates_DoNotArmTheGuard_BeforeAnySubstantiveContent()
+ {
+ // Content-free keepalives (e.g. llama-server's prompt_progress heartbeat)
+ // must not arm the tight inactivity timer — only a substantive update
+ // (ChatStreamUpdateClassifier.IsSubstantiveUpdate) may promote it. Two keepalives
+ // arrive, then the provider takes a long-but-legitimate time (a cold prefill,
+ // not a stall) to produce the first substantive delta: crossing what would be
+ // the inactivity window must not abort it, because no substantive content has
+ // arrived yet to arm the guard.
+ var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+ var prefillDelay = Policy.StreamInactivityTimeout * 4;
+ var leaf = new FakeChatClient(streamHandler: (_, _, ct) => KeepaliveTwiceThenDelayedSubstantive(time, prefillDelay, ct));
+ var client = new StreamStallGuardChatClient(leaf, Policy, NullLogger.Instance, time);
+
+ await using var enumerator = client
+ .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+ .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+ Assert.True(await enumerator.MoveNextAsync()); // keepalive 1 — non-substantive
+ Assert.True(await enumerator.MoveNextAsync()); // keepalive 2 — still non-substantive
+
+ var pending = enumerator.MoveNextAsync().AsTask();
+
+ // Cross well past what would be the inactivity window if a keepalive had
+ // (wrongly) armed the guard — the leaf is still legitimately working (a long
+ // cold prefill), not stalled.
+ time.Advance(Policy.StreamInactivityTimeout * 2);
+ Assert.False(pending.IsCompleted);
+
+ // Finish the (legitimately long) prefill — the first substantive update still
+ // arrives cleanly, proving a keepalive never armed the guard.
+ time.Advance(prefillDelay - Policy.StreamInactivityTimeout * 2);
+ Assert.True(await pending);
+ }
+
+ [Fact]
+ public async Task SlowConsumer_HoldingAnUpdate_DoesNotCountAgainstTheInactivityWindow()
+ {
+ // The inactivity timer must measure provider silence only, not how long the
+ // downstream consumer holds an already-yielded update before asking for the
+ // next one. Two substantive chunks arm and then satisfy the guard once; the
+ // consumer then sits on chunk2 far longer than the inactivity window before
+ // requesting chunk3 — that gap must not be held against the provider, which
+ // answers chunk3 promptly once asked.
+ var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+ var leaf = new FakeChatClient(streamHandler: (_, _, ct) => YieldThreeQuickly(ct));
+ var client = new StreamStallGuardChatClient(leaf, Policy, NullLogger.Instance, time);
+
+ await using var enumerator = client
+ .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)
+ .GetAsyncEnumerator(TestContext.Current.CancellationToken);
+
+ Assert.True(await enumerator.MoveNextAsync()); // chunk1 — arms the guard
+ Assert.True(await enumerator.MoveNextAsync()); // chunk2 — the wait for this one was itself timer-guarded
+
+ // Consumer holds chunk2 far longer than the inactivity window before asking
+ // for chunk3. The timer that guarded the (already-satisfied) wait for chunk2
+ // must have been disarmed on arrival — it must not still be counting down.
+ time.Advance(Policy.StreamInactivityTimeout * 10);
+
+ Assert.True(await enumerator.MoveNextAsync()); // chunk3 — must still succeed, not a false abort
+ }
+
+ [Fact]
+ public async Task ZeroTimeout_DisablesTheGuard()
+ {
+ var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
+ var leaf = new FakeChatClient(streamHandler: (_, _, ct) => YieldTwoThenStallForever(ct));
+ var client = new StreamStallGuardChatClient(
+ leaf, new RetryPolicy { StreamInactivityTimeout = TimeSpan.Zero }, NullLogger.Instance, time);
+
+ // Owns cancellation directly (rather than TestContext.Current.CancellationToken)
+ // so the still-pending read below can be unwound deterministically instead of
+ // leaving a MoveNextAsync call in flight when the enumerator is disposed.
+ using var cts = new CancellationTokenSource();
+ var enumerator = client
+ .GetStreamingResponseAsync([new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)
+ .GetAsyncEnumerator(cts.Token);
+
+ Assert.True(await enumerator.MoveNextAsync());
+ Assert.True(await enumerator.MoveNextAsync());
+
+ var pending = enumerator.MoveNextAsync().AsTask();
+ time.Advance(Policy.StreamInactivityTimeout * 100);
+ Assert.False(pending.IsCompleted); // disabled — nothing ever cancels this read
+
+ cts.Cancel();
+ await Assert.ThrowsAnyAsync(() => pending);
+ await enumerator.DisposeAsync();
+ }
+
+ private static async IAsyncEnumerable YieldTwoThenStallForever(
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("chunk1")] };
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("chunk2")] };
+
+ // Dead/half-open connection: no more tokens, no error, no close.
+ await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
+ yield break; // unreachable — Task.Delay above only returns via cancellation
+ }
+
+ private static async IAsyncEnumerable 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}")] };
+ }
+ }
+
+ private static async IAsyncEnumerable DelayThenYieldOnce(
+ TimeProvider time, TimeSpan delay,
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ await Task.Delay(delay, time, cancellationToken);
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("first")] };
+ }
+
+ private static async IAsyncEnumerable KeepaliveTwiceThenDelayedSubstantive(
+ TimeProvider time, TimeSpan delay,
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ // A content-free keepalive: no text/thinking/tool-call content and no finish
+ // reason, matching ChatStreamUpdateClassifier.IsSubstantiveUpdate's "false" case.
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [] };
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [] };
+
+ await Task.Delay(delay, time, cancellationToken);
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("first substantive")] };
+ }
+
+ private static async IAsyncEnumerable YieldThreeQuickly(
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ await Task.Yield();
+ cancellationToken.ThrowIfCancellationRequested();
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("chunk1")] };
+ await Task.Yield();
+ cancellationToken.ThrowIfCancellationRequested();
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("chunk2")] };
+ await Task.Yield();
+ cancellationToken.ThrowIfCancellationRequested();
+ yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("chunk3")] };
+ }
+}
diff --git a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs
index 52a48e9ed..1abec6fdb 100644
--- a/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs
+++ b/src/Netclaw.Daemon/Configuration/DaemonProviderServiceExtensions.cs
@@ -72,8 +72,8 @@ public static IServiceCollection AddDaemonLlmProviders(
// (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.
+ // Composes the cross-cutting middleware (Logging → Retry → StreamStallGuard)
+ // around each provider pipeline via ChatClientBuilder.
services.AddSingleton(sp => new PipelineChatClientFactory(
sp.GetRequiredService(),
sp.GetRequiredService(),
diff --git a/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs b/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs
index 17c12e441..ecc05bd13 100644
--- a/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs
+++ b/src/Netclaw.Daemon/Configuration/PipelineChatClientFactory.cs
@@ -11,10 +11,10 @@ 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 .
+/// Logging → Retry → StreamStallGuard → VendorOptions → raw provider client. The
+/// leaf (raw provider client plus any vendor-options wrap) comes from
+/// ; this layer adds the cross-cutting Retry,
+/// stream-stall detection, 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.
@@ -46,7 +46,12 @@ public IChatClient Create(ModelReference model) =>
/// 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 .
+ /// before . is
+ /// registered innermost (directly around the leaf) so a mid-stream stall it detects is
+ /// classified through the same rule as any other
+ /// transient failure — no separate retry path — even though it always reaches
+ /// post-first-chunk (see
+ /// remarks) and so propagates rather than silently re-issuing the request.
///
internal static IChatClient Compose(
IChatClient leaf,
@@ -56,10 +61,12 @@ internal static IChatClient Compose(
{
var loggingLogger = loggerFactory.CreateLogger();
var retryLogger = loggerFactory.CreateLogger();
+ var stallGuardLogger = loggerFactory.CreateLogger();
return new ChatClientBuilder(leaf)
.Use(inner => new LoggingChatClient(inner, loggingLogger, timeProvider))
.Use(inner => new RetryingChatClient(inner, retryPolicy, retryLogger, timeProvider))
+ .Use(inner => new StreamStallGuardChatClient(inner, retryPolicy, stallGuardLogger, timeProvider))
.Build();
}
}
diff --git a/src/Netclaw.Daemon/Configuration/StreamStallGuardChatClient.cs b/src/Netclaw.Daemon/Configuration/StreamStallGuardChatClient.cs
new file mode 100644
index 000000000..2c4ac1854
--- /dev/null
+++ b/src/Netclaw.Daemon/Configuration/StreamStallGuardChatClient.cs
@@ -0,0 +1,139 @@
+// -----------------------------------------------------------------------
+//
+// Copyright (C) 2026 - 2026 Petabridge, LLC
+//
+// -----------------------------------------------------------------------
+using System.Runtime.CompilerServices;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.Logging;
+using Netclaw.Actors.Sessions.Pipelines;
+using Netclaw.Configuration;
+
+namespace Netclaw.Daemon.Configuration;
+
+///
+/// Decorates an so a streaming response that goes silent
+/// mid-stream aborts in seconds instead of waiting on the coarse per-call watchdog
+/// (minutes). A dead or half-open provider connection can leave the socket open
+/// with no more tokens, no error, and no close — the actor-level watchdog still
+/// catches this eventually, but only after burning most of its budget.
+///
+/// Only the gap after the first substantive update is bounded — the
+/// same distinction and
+/// ProcessingWatchdog already draw, reused here rather than re-derived. Time to
+/// first substantive output is left to the existing, more generous per-call watchdog,
+/// because a self-hosted backend can be legitimately silent for minutes during cold
+/// prefill. A reasoning delta with text arms the tight budget, the same as a text or
+/// tool-call delta. Only a content-free keepalive — no text, no thinking, no tool
+/// call, and no finish reason — leaves the budget unarmed. Once armed, every later
+/// update — including a content-free keepalive — resets the inactivity clock, so a
+/// slow-but-alive stream is never falsely aborted.
+///
+///
+/// The timer measures provider silence only. It is disarmed immediately after each
+/// update arrives and re-armed only just before asking for the next one, so time a
+/// downstream consumer spends holding an already-yielded update is never counted
+/// against the provider.
+///
+///
+/// Sits directly below in the composed pipeline
+/// (). By construction a stall this
+/// guard catches has already yielded at least one chunk, so 's
+/// pre-first-chunk retry cannot re-issue the request (the partial output already
+/// streamed cannot be un-sent) — the propagates to the
+/// caller. No new retry mechanism: the exception is a plain ,
+/// so it is classified by the same rule and reaches
+/// the actor's existing failure handling exactly as any other transient failure would —
+/// just in seconds instead of minutes, leaving the caller's own retry budget intact.
+///
+///
+public sealed class StreamStallGuardChatClient : DelegatingChatClient
+{
+ private readonly TimeSpan _inactivityTimeout;
+ private readonly ILogger _logger;
+ private readonly TimeProvider _timeProvider;
+
+ public StreamStallGuardChatClient(
+ IChatClient innerClient,
+ RetryPolicy policy,
+ ILogger logger,
+ TimeProvider? timeProvider = null)
+ : base(innerClient)
+ {
+ _inactivityTimeout = policy.StreamInactivityTimeout;
+ _logger = logger;
+ _timeProvider = timeProvider ?? TimeProvider.System;
+ }
+
+ public override async IAsyncEnumerable GetStreamingResponseAsync(
+ IEnumerable messages,
+ ChatOptions? options = null,
+ [EnumeratorCancellation] CancellationToken cancellationToken = default)
+ {
+ if (_inactivityTimeout <= TimeSpan.Zero)
+ {
+ await foreach (var update in base.GetStreamingResponseAsync(messages, options, cancellationToken))
+ yield return update;
+ yield break;
+ }
+
+ using var stallCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ // Not armed until the first substantive update arrives (see class remarks) —
+ // the initial due time is infinite.
+ using var timer = _timeProvider.CreateTimer(
+ static state => ((CancellationTokenSource)state!).Cancel(),
+ stallCts,
+ Timeout.InfiniteTimeSpan,
+ Timeout.InfiniteTimeSpan);
+
+ var enumerator = base.GetStreamingResponseAsync(messages, options, stallCts.Token)
+ .GetAsyncEnumerator(stallCts.Token);
+ var armed = false;
+ try
+ {
+ while (true)
+ {
+ if (armed)
+ timer.Change(_inactivityTimeout, Timeout.InfiniteTimeSpan);
+
+ bool hasNext;
+ try
+ {
+ hasNext = await enumerator.MoveNextAsync();
+ }
+ catch (OperationCanceledException) when (
+ !cancellationToken.IsCancellationRequested && stallCts.IsCancellationRequested)
+ {
+ // The caller's own token is still live — this cancellation came from
+ // our timer, not a real abort. Surface it as a plain TimeoutException
+ // so it flows through RetryPolicy.ShouldRetry (already retryable) and
+ // the actor's existing TimeoutException -> ErrorCategory.Timeout
+ // classification, unchanged.
+ _logger.LogWarning(
+ "LLM stream stalled — no update for {TimeoutSeconds:F0}s after the first substantive delta, aborting",
+ _inactivityTimeout.TotalSeconds);
+ throw new TimeoutException(
+ $"LLM stream produced no update for {_inactivityTimeout.TotalSeconds:F0}s after the first substantive delta (stall detected)");
+ }
+
+ // The provider produced something (or ended cleanly) within the window —
+ // disarm before yielding so the timer never also measures how long the
+ // downstream consumer holds this update. It is re-armed above, just
+ // before the next MoveNextAsync, so it measures provider silence only.
+ timer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
+
+ if (!hasNext)
+ yield break;
+
+ if (!armed && ChatStreamUpdateClassifier.IsSubstantiveUpdate(enumerator.Current))
+ armed = true;
+
+ yield return enumerator.Current;
+ }
+ }
+ finally
+ {
+ await enumerator.DisposeAsync();
+ }
+ }
+}