From eb9ead7cd11d0e55294640cf49048de96e3a9c4b Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 23 Apr 2026 21:15:05 +0000 Subject: [PATCH 1/2] fix(session): unify two-phase streaming timeout into single reset-on-delta timer (#731) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StreamIdleTimeout (120s) was causing false positives on GPU-contended self-hosted inference servers where a request could be preempted mid-stream by concurrent sessions. The two-phase design (FirstTokenTimeout → StreamIdleTimeout) assumed mid-stream silence means a dead connection, but on self-hosted setups it can mean GPU scheduling delays. Collapse both timeouts into a single FirstTokenTimeout (600s) that resets on every streaming delta. Dead streams are still caught within 600s, while GPU scheduling stalls no longer trigger false timeouts. --- .../Sessions/ErrorCorrelationTests.cs | 1 - .../LlmSessionTwoPhaseTimeoutTests.cs | 36 +++++++++---------- .../Sessions/LlmSessionWatchdogTests.cs | 7 ++-- .../Sessions/LlmSessionActor.cs | 25 +++++++------ .../Schemas/netclaw-config.v1.schema.json | 9 +---- src/Netclaw.Configuration/SessionConfig.cs | 22 +++--------- src/Netclaw.Providers/ProviderPluginBase.cs | 2 +- 7 files changed, 38 insertions(+), 64 deletions(-) diff --git a/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs b/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs index 494ce98b1..1bb4b851f 100644 --- a/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs @@ -34,7 +34,6 @@ protected override void ConfigureServices(HostBuilderContext context, IServiceCo services.AddSingleton(new SessionConfig { FirstTokenTimeout = TimeSpan.FromSeconds(10), - StreamIdleTimeout = TimeSpan.FromSeconds(10), ToolExecutionTimeout = TimeSpan.FromSeconds(10), SidecarLlmTimeout = TimeSpan.FromSeconds(10), Tuning = new SessionTuning diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTwoPhaseTimeoutTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionTwoPhaseTimeoutTests.cs index 9ee145e4c..595b59896 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTwoPhaseTimeoutTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionTwoPhaseTimeoutTests.cs @@ -16,22 +16,21 @@ namespace Netclaw.Actors.Tests.Sessions; -public sealed class LlmSessionTwoPhaseTimeoutTests(ITestOutputHelper output) : TestKit(output: output) +public sealed class LlmSessionStreamingTimeoutTests(ITestOutputHelper output) : TestKit(output: output) { - private readonly TwoPhaseTestChatClient _chatClient = new(); + private readonly StreamingTimeoutTestChatClient _chatClient = new(); protected override void ConfigureServices(HostBuilderContext context, IServiceCollection services) { services.AddSingleton(new SingleClientProvider(_chatClient)); services.AddSingleton(new ModelCapabilities { - ModelId = "two-phase-timeout-test-model", + ModelId = "streaming-timeout-test-model", ContextWindowTokens = 128_000, }); services.AddSingleton(new SessionConfig { FirstTokenTimeout = TimeSpan.FromSeconds(2), - StreamIdleTimeout = TimeSpan.FromSeconds(1), ToolExecutionTimeout = TimeSpan.FromSeconds(10), SidecarLlmTimeout = TimeSpan.FromSeconds(10), Tuning = new SessionTuning @@ -67,13 +66,13 @@ protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IService } [Fact] - public async Task First_token_timeout_fires_when_no_deltas_arrive() + public async Task Timeout_fires_when_no_deltas_arrive() { _chatClient.Mode = StreamMode.HangForever; - var sessionId = new SessionId("two-phase/first-token-timeout"); + var sessionId = new SessionId("streaming-timeout/no-deltas"); var sessionManager = ActorRegistry.Get(); - var subscriber = CreateTestProbe("first-token-sub"); + var subscriber = CreateTestProbe("no-delta-sub"); await sessionManager.Ask(new JoinSession { @@ -92,19 +91,20 @@ await sessionManager.Ask(new SendUserMessage // FirstTokenTimeout is 2s — should fire within ~3s var error = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); Assert.Equal(ErrorCategory.Timeout, error.Category); - Assert.Contains("did not respond", error.Message); + Assert.Contains("timed out", error.Message); await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); } [Fact] - public async Task Stream_idle_timeout_fires_when_stream_stalls_after_deltas() + public async Task Timeout_resets_on_delta_and_fires_after_silence() { - // Emit 2 deltas then hang — the stream-idle timeout (1s) should fire, not first-token (2s) + // Emit deltas then hang — the unified timeout (2s) resets on each delta, + // then fires 2s after the last one _chatClient.Mode = StreamMode.EmitThenHang; - var sessionId = new SessionId("two-phase/stream-idle-timeout"); + var sessionId = new SessionId("streaming-timeout/delta-then-silence"); var sessionManager = ActorRegistry.Get(); - var subscriber = CreateTestProbe("stream-idle-sub"); + var subscriber = CreateTestProbe("delta-silence-sub"); await sessionManager.Ask(new JoinSession { @@ -120,11 +120,10 @@ await sessionManager.Ask(new SendUserMessage Content = "hello" }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); - // We should see text deltas streamed, then a timeout error - // StreamIdleTimeout is 1s — should fire well before FirstTokenTimeout (2s) - var error = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + // Deltas stream, then silence — timeout fires after FirstTokenTimeout (2s) of no activity + var error = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(8), cancellationToken: TestContext.Current.CancellationToken); Assert.Equal(ErrorCategory.Timeout, error.Category); - Assert.Contains("stopped unexpectedly", error.Message); + Assert.Contains("timed out", error.Message); await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); } @@ -133,7 +132,7 @@ public async Task Successful_stream_completes_without_timeout() { _chatClient.Mode = StreamMode.SucceedImmediately; - var sessionId = new SessionId("two-phase/success"); + var sessionId = new SessionId("streaming-timeout/success"); var sessionManager = ActorRegistry.Get(); var subscriber = CreateTestProbe("success-sub"); @@ -158,7 +157,7 @@ await sessionManager.Ask(new SendUserMessage private enum StreamMode { HangForever, EmitThenHang, SucceedImmediately } - private sealed class TwoPhaseTestChatClient : IChatClient + private sealed class StreamingTimeoutTestChatClient : IChatClient { public StreamMode Mode { get; set; } = StreamMode.HangForever; @@ -193,7 +192,6 @@ private static async IAsyncEnumerable NeverCompletesAsync( private static async IAsyncEnumerable EmitThenHangAsync( [EnumeratorCancellation] CancellationToken cancellationToken = default) { - // Emit a few deltas so the actor receives LlmResponseDeltaReceived yield return new ChatResponseUpdate { Role = AiChatRole.Assistant, diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs index 0da72c584..7b47cded3 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs @@ -30,7 +30,6 @@ protected override void ConfigureServices(HostBuilderContext context, IServiceCo services.AddSingleton(new SessionConfig { FirstTokenTimeout = TimeSpan.FromSeconds(1), - StreamIdleTimeout = TimeSpan.FromSeconds(1), ToolExecutionTimeout = TimeSpan.FromSeconds(1), SidecarLlmTimeout = TimeSpan.FromSeconds(1), Tuning = new SessionTuning @@ -87,7 +86,7 @@ await sessionManager.Ask(new SendUserMessage }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); var firstError = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); - Assert.Contains("did not respond", firstError.Message, StringComparison.OrdinalIgnoreCase); + Assert.Contains("timed out", firstError.Message, StringComparison.OrdinalIgnoreCase); await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); await sessionManager.Ask(new SendUserMessage @@ -97,7 +96,7 @@ await sessionManager.Ask(new SendUserMessage }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); var secondError = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); - Assert.Contains("did not respond", secondError.Message, StringComparison.OrdinalIgnoreCase); + Assert.Contains("timed out", secondError.Message, StringComparison.OrdinalIgnoreCase); await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); Assert.True(_chatClient.CallCount >= 2); @@ -133,7 +132,7 @@ await sessionManager.Ask(new SendUserMessage }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); var firstError = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); - Assert.Contains("did not respond", firstError.Message, StringComparison.OrdinalIgnoreCase); + Assert.Contains("timed out", firstError.Message, StringComparison.OrdinalIgnoreCase); await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); var recoveredText = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 699a09fe6..484409c3f 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -106,9 +106,9 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers private readonly ProcessingWatchdog _watchdog = new(); // Actor-owned CTS for the active LLM call. Cancelled by the watchdog on timeout - // or when a response/failure arrives. The session-level watchdog (FirstTokenTimeout / - // StreamIdleTimeout) is the authoritative timeout — this CTS just propagates - // cancellation to the HTTP layer so timed-out connections are released. + // or when a response/failure arrives. The session-level watchdog (FirstTokenTimeout) + // is the authoritative timeout — this CTS just propagates cancellation to the + // HTTP layer so timed-out connections are released. private CancellationTokenSource? _activeLlmCts; // Correlation ID for the active LLM call. Incremented in FireLlmCall. @@ -116,8 +116,9 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // from cancelled calls are ignored when their CallId doesn't match. private long _activeCallId; - // Two-phase timeout: tracks whether we've received any streaming delta this turn - private bool _firstDeltaReceived; + // 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) + private bool _anyContentStreamed; // Per-turn diagnostic correlation (ephemeral) private string? _activeTurnId; @@ -480,8 +481,8 @@ private void Processing() Command(msg => { if (msg.CallId != _activeCallId) return; // stale delta from cancelled call - _firstDeltaReceived = true; - _watchdog.Refresh(_config.StreamIdleTimeout, Timers); + _anyContentStreamed = true; + _watchdog.Refresh(_config.FirstTokenTimeout, Timers); switch (msg.Content) { @@ -830,7 +831,7 @@ await _approvalService.RecordApprovalAsync( // 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 (!_firstDeltaReceived && IsTransientStreamingError(msg.Cause)) + if (!_anyContentStreamed && IsTransientStreamingError(msg.Cause)) { var policy = _config.Tuning.StreamingRetryPolicy; if (_streamingRetryAttempt < policy.MaxRetries) @@ -874,7 +875,7 @@ await _approvalService.RecordApprovalAsync( var timeout = msg.OperationName switch { "tool-execution" => _config.ToolExecutionTimeout, - "llm-call" => _firstDeltaReceived ? _config.StreamIdleTimeout : _config.FirstTokenTimeout, + "llm-call" => _config.FirstTokenTimeout, _ => _config.TurnLlmTimeout }; @@ -2055,9 +2056,7 @@ private string ExtractLlmErrorMessage(Exception? cause) return $"Context window exceeded after compaction — the session has too many tools or a large system prompt for the {_model.ModelId} context window ({_model.ContextWindowTokens} tokens). Try reducing tools or increasing the model's context window."; if (cause is TimeoutException) - return _firstDeltaReceived - ? "The LLM response stream stopped unexpectedly. Please try again." - : "The LLM provider did not respond in time. The model may be overloaded or the context too large. Please try again."; + return "The LLM response stream timed out due to inactivity. The model may be overloaded or the context too large. Please try again."; return "I encountered an error processing your message. Please try again."; } @@ -2227,7 +2226,7 @@ private void SetSystemPrompt() private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) { - _firstDeltaReceived = false; + _anyContentStreamed = false; CancelAndDisposeLlmCts(); _activeLlmCts = new CancellationTokenSource(); _activeCallId++; diff --git a/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json b/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json index ed31c7a5d..e0ce23486 100644 --- a/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json +++ b/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json @@ -118,14 +118,7 @@ "minimum": 10, "maximum": 1800, "default": 600, - "description": "Maximum wait in seconds for the first streaming token. Covers model prefill on large contexts." - }, - "StreamIdleTimeoutSeconds": { - "type": "integer", - "minimum": 10, - "maximum": 600, - "default": 120, - "description": "Maximum silence in seconds between streaming tokens. Resets on each delta." + "description": "Maximum inactivity in seconds for LLM streaming calls. Used as both the initial wait for the first token and the silence threshold between deltas (resets on each delta)." }, "CompactionThreshold": { "type": "number", diff --git a/src/Netclaw.Configuration/SessionConfig.cs b/src/Netclaw.Configuration/SessionConfig.cs index 2d51d34da..e934ac9f1 100644 --- a/src/Netclaw.Configuration/SessionConfig.cs +++ b/src/Netclaw.Configuration/SessionConfig.cs @@ -58,21 +58,13 @@ public sealed record SessionConfig public TimeSpan SidecarLlmTimeout { get; init; } = TimeSpan.FromSeconds(90); /// - /// Maximum wait time for the first streaming token from the LLM provider. - /// Covers the prefill phase where the model processes input context before - /// generating output. Large contexts (200K+ tokens) can take several minutes. + /// Maximum inactivity timeout for LLM streaming calls. Used both as the + /// initial wait for the first token (prefill phase) and as the silence + /// threshold between consecutive deltas — the timer resets on every delta. /// Falls back to if not explicitly configured. /// public TimeSpan FirstTokenTimeout { get; init; } = TimeSpan.FromSeconds(600); - /// - /// Maximum silence between consecutive streaming tokens. Once the first token - /// arrives, the watchdog switches to this tighter timeout and resets on every - /// delta. If no tokens arrive within this window, the stream is considered dead. - /// Falls back to if not explicitly configured. - /// - public TimeSpan StreamIdleTimeout { get; init; } = TimeSpan.FromSeconds(120); - /// /// Internal tuning constants. Bindable from config for development/testing /// but not part of the documented operator surface. @@ -98,15 +90,10 @@ public static SessionConfig BindFromConfiguration(IConfigurationSection section) TurnLlmTimeout = turnLlmTimeout, ToolExecutionTimeout = TimeSpan.FromSeconds(Math.Max(1, raw.ToolExecutionTimeoutSeconds)), SidecarLlmTimeout = TimeSpan.FromSeconds(Math.Max(1, raw.SidecarLlmTimeoutSeconds)), - // Two-phase timeout: explicit value → TurnLlmTimeout fallback (if customized) → default - // If operator set TurnLlmTimeoutSeconds (non-default 180), use it for both phases - // to preserve backward compat. Otherwise use the generous new defaults. + // Explicit value → TurnLlmTimeout fallback (if customized) → default FirstTokenTimeout = raw.FirstTokenTimeoutSeconds > 0 ? TimeSpan.FromSeconds(raw.FirstTokenTimeoutSeconds) : raw.TurnLlmTimeoutSeconds != 180 ? turnLlmTimeout : TimeSpan.FromSeconds(600), - StreamIdleTimeout = raw.StreamIdleTimeoutSeconds > 0 - ? TimeSpan.FromSeconds(raw.StreamIdleTimeoutSeconds) - : raw.TurnLlmTimeoutSeconds != 180 ? turnLlmTimeout : TimeSpan.FromSeconds(120), Tuning = tuning, }; } @@ -158,6 +145,5 @@ private sealed record RawSessionConfig public int ToolExecutionTimeoutSeconds { get; init; } = 90; public int SidecarLlmTimeoutSeconds { get; init; } = 90; public int FirstTokenTimeoutSeconds { get; init; } - public int StreamIdleTimeoutSeconds { get; init; } } } diff --git a/src/Netclaw.Providers/ProviderPluginBase.cs b/src/Netclaw.Providers/ProviderPluginBase.cs index dceceaad0..2daf09b68 100644 --- a/src/Netclaw.Providers/ProviderPluginBase.cs +++ b/src/Netclaw.Providers/ProviderPluginBase.cs @@ -36,7 +36,7 @@ public Task ProbeAsync(ProviderEntry entry, CancellationTok /// Creates an with a generous timeout suitable for LLM calls. /// The default of 100 seconds is far too short for /// large-context models — prefill alone can exceed 100 seconds on self-hosted hardware. - /// Session-level timeouts (FirstTokenTimeout, StreamIdleTimeout via ProcessingWatchdog) + /// Session-level timeouts (FirstTokenTimeout via ProcessingWatchdog) /// are the authoritative timeout layer; the HttpClient timeout is a last-resort safety /// net that should never fire before the watchdog. /// From ffd5080aaeffe101bc89927ae0b3cf778ef6e607 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 23 Apr 2026 21:25:50 +0000 Subject: [PATCH 2/2] chore: rename test file to match class, update OpenSpec for unified timeout --- openspec/specs/netclaw-session/spec.md | 31 +++++++------------ ....cs => LlmSessionStreamingTimeoutTests.cs} | 0 2 files changed, 11 insertions(+), 20 deletions(-) rename src/Netclaw.Actors.Tests/Sessions/{LlmSessionTwoPhaseTimeoutTests.cs => LlmSessionStreamingTimeoutTests.cs} (100%) diff --git a/openspec/specs/netclaw-session/spec.md b/openspec/specs/netclaw-session/spec.md index 224429939..44baa54e8 100644 --- a/openspec/specs/netclaw-session/spec.md +++ b/openspec/specs/netclaw-session/spec.md @@ -508,29 +508,20 @@ forwarding via `self.Tell(LlmResponseDeltaReceived)`, and error packaging as `LlmCallFailed`. Dynamic context layer injection SHALL be a static method on this class. -#### Scenario: Two-phase LLM call timeout +#### Scenario: LLM streaming inactivity timeout -The system SHALL enforce two separate timeout phases for LLM streaming calls: +The system SHALL enforce a single reset-on-delta inactivity timeout for LLM +streaming calls using `FirstTokenTimeout` (default 600s). The timer starts +when the LLM call is fired and resets on every streaming delta received. -- **Phase 1 — First-Token Timeout**: The system SHALL wait up to - `FirstTokenTimeout` (default 600s) for the first streaming delta. This - covers the prefill phase where the model processes input context. -- **Phase 2 — Stream-Idle Timeout**: Once the first delta arrives, the - system SHALL switch to `StreamIdleTimeout` (default 120s). This resets - on every subsequent delta and detects dead streams. - -- **GIVEN** an LLM streaming call is in progress and no deltas have arrived -- **WHEN** the `FirstTokenTimeout` elapses -- **THEN** the invoker sends `LlmCallFailed` with a `TimeoutException` -- **AND** the error message indicates the provider did not respond - -- **GIVEN** an LLM streaming call has produced at least one delta -- **WHEN** no further deltas arrive within `StreamIdleTimeout` +- **GIVEN** an LLM streaming call is in progress +- **WHEN** no deltas arrive within `FirstTokenTimeout` of the call start or + the last received delta - **THEN** the watchdog fires and the turn fails with `ErrorCategory.Timeout` -- **AND** the error message indicates the stream stopped unexpectedly +- **AND** the error message indicates the stream timed out due to inactivity -Backward compat: if `TurnLlmTimeoutSeconds` is configured but the new -properties are not, both phases use `TurnLlmTimeout`. +Backward compat: if `TurnLlmTimeoutSeconds` is configured but +`FirstTokenTimeoutSeconds` is not, `FirstTokenTimeout` uses `TurnLlmTimeout`. #### Scenario: Streaming deltas forwarded to actor @@ -538,7 +529,7 @@ properties are not, both phases use `TurnLlmTimeout`. - **WHEN** text content chunks arrive - **THEN** each chunk after the first is forwarded as `LlmResponseDeltaReceived` - **AND** the first chunk is held until the second arrives (single-chunk optimization) -- **AND** the watchdog refreshes with `StreamIdleTimeout` on each delta +- **AND** the watchdog refreshes with `FirstTokenTimeout` on each delta ### Requirement: Tool execution encapsulation diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTwoPhaseTimeoutTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionStreamingTimeoutTests.cs similarity index 100% rename from src/Netclaw.Actors.Tests/Sessions/LlmSessionTwoPhaseTimeoutTests.cs rename to src/Netclaw.Actors.Tests/Sessions/LlmSessionStreamingTimeoutTests.cs