diff --git a/src/Netclaw.Actors.Tests/Channels/ExecutionOutputAccumulatorTests.cs b/src/Netclaw.Actors.Tests/Channels/ExecutionOutputAccumulatorTests.cs index bff76909d..967df490c 100644 --- a/src/Netclaw.Actors.Tests/Channels/ExecutionOutputAccumulatorTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/ExecutionOutputAccumulatorTests.cs @@ -30,6 +30,102 @@ public void TextDeltaOutput_accumulates_text() Assert.Equal("Hello world", acc.GetAccumulatedText()); } + [Fact] + public void TextStreamDiscarded_clears_multi_delta_accumulated_text() + { + var acc = new ExecutionOutputAccumulator(TestNotifyTool); + + // A real multi-delta stall: two substantive deltas before the dead call + // is discarded — a single-delta fake would not exercise the accumulator's + // append path the way a real stalled provider stream does. + acc.ProcessOutput(new TextDeltaOutput("stalled chunk one ") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = TestSessionId }); + + acc.ProcessOutput(new TextStreamDiscarded { SessionId = TestSessionId }); + + acc.ProcessOutput(new TextDeltaOutput("Resumed answer ") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextDeltaOutput("after timeout") { SessionId = TestSessionId }); + + // The delta-accumulated result must contain ONLY the resumed call's text — + // the dead call's partial content must not survive the discard. + Assert.Equal("Resumed answer after timeout", acc.GetAccumulatedText()); + Assert.DoesNotContain("STALLED_PARTIAL_MARKER", acc.GetAccumulatedText(), StringComparison.Ordinal); + } + + [Fact] + public void TextStreamDiscarded_preserves_an_earlier_completed_calls_text_but_discards_only_the_dead_calls_partial() + { + var acc = new ExecutionOutputAccumulator(TestNotifyTool); + + // Call 1: streams a preamble, then completes (a tool round) — TextOutput + // marks the call boundary and commits the preamble (see D1 in the review). + acc.ProcessOutput(new TextDeltaOutput("Checking the files now. ") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextOutput("Checking the files now. ") { SessionId = TestSessionId }); + + // Call 2: streams two real deltas, then dies mid-stream and is discarded. + acc.ProcessOutput(new TextDeltaOutput("stalled chunk one ") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextStreamDiscarded { SessionId = TestSessionId }); + + // Resumed call streams the real final answer. + acc.ProcessOutput(new TextDeltaOutput("Done: the answer is X.") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextOutput("Done: the answer is X.") { SessionId = TestSessionId }); + + // Before the fix, TextStreamDiscarded cleared the whole turn-scoped + // buffer, wiping call 1's already-completed preamble along with call 2's + // dead partial. The result must keep call 1's text AND the resumed + // answer, with none of the dead call's partial. + var result = acc.GetAccumulatedText(); + Assert.Equal("Checking the files now. Done: the answer is X.", result); + Assert.DoesNotContain("STALLED_PARTIAL_MARKER", result, StringComparison.Ordinal); + } + + [Fact] + public void TextStreamDiscarded_lets_TextOutput_repopulate_after_discard() + { + var acc = new ExecutionOutputAccumulator(TestNotifyTool); + + acc.ProcessOutput(new TextDeltaOutput("stalled") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextStreamDiscarded { SessionId = TestSessionId }); + + // No further deltas — the resumed call's answer arrives as a single + // non-streamed TextOutput. Before the fix, _sawTextDelta stayed true from + // the dead call's delta, so this TextOutput would be silently ignored. + acc.ProcessOutput(new TextOutput("Resumed answer") { SessionId = TestSessionId }); + + Assert.Equal("Resumed answer", acc.GetAccumulatedText()); + } + + [Fact] + public void TextOutput_with_IsCallBoundary_false_does_not_move_the_commit_marker_past_a_live_calls_partial_text() + { + // F2: EmitExpiredPromptNotice/EmitWrongRequesterApprovalNotice/ + // EmitUnavailableApprovalOptionNotice send a mid-stream TextOutput + // (IsCallBoundary = false) while another call still streams. + // Before the fix, ANY TextOutput advanced the commit marker over the + // live call's partial text; a subsequent stall+discard then found + // nothing left to remove, and the resumed call's answer glued onto + // the dead partial. + var acc = new ExecutionOutputAccumulator(TestNotifyTool); + + acc.ProcessOutput(new TextDeltaOutput("stalled chunk one ") { SessionId = TestSessionId }); + acc.ProcessOutput(new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = TestSessionId }); + + acc.ProcessOutput(new TextOutput("That approval prompt has expired.") + { + SessionId = TestSessionId, + IsCallBoundary = false + }); + + acc.ProcessOutput(new TextStreamDiscarded { SessionId = TestSessionId }); + + acc.ProcessOutput(new TextDeltaOutput("Done: the answer is X.") { SessionId = TestSessionId }); + + var result = acc.GetAccumulatedText(); + Assert.Equal("Done: the answer is X.", result); + Assert.DoesNotContain("STALLED_PARTIAL_MARKER", result, StringComparison.Ordinal); + } + [Fact] public void TextOutput_accumulates_when_no_prior_delta() { diff --git a/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs b/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs index df67046cc..b7cdadb0d 100644 --- a/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/ErrorCorrelationTests.cs @@ -42,6 +42,10 @@ protected override void ConfigureSessionServices(IServiceCollection services) { SnapshotInterval = 5, TitleGenerationInterval = 0, + // This suite targets ErrorOutput classification/correlation, not + // turn-level resume — disable resume so a timeout fails the turn on + // the first attempt. See LlmTurnResumeTests for resume coverage. + TimeoutResumeRetryBudget = 0, } }); services.AddSingleton(new StaticSystemPromptProvider("You are a test assistant.")); diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionStreamingTimeoutTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionStreamingTimeoutTests.cs index 787691f95..1b5b2b552 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionStreamingTimeoutTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionStreamingTimeoutTests.cs @@ -52,6 +52,11 @@ protected override void ConfigureSessionServices(IServiceCollection services) { SnapshotInterval = 5, TitleGenerationInterval = 0, + // These tests target the watchdog's own timeout-detection behavior + // (arm/promote/fire), not turn-level resume — disable resume so a + // single watchdog expiry fails the turn exactly as before that + // feature existed. See LlmTurnResumeTests for resume coverage. + TimeoutResumeRetryBudget = 0, } }); services.AddSingleton(new StaticSystemPromptProvider("You are a test assistant.")); diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs index e45132ce8..751a515fa 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionWatchdogTests.cs @@ -51,6 +51,11 @@ protected override void ConfigureSessionServices(IServiceCollection services) { SnapshotInterval = 5, TitleGenerationInterval = 0, + // These tests target the watchdog's own timeout/failure/recovery + // behavior, not turn-level resume — disable resume so a single + // watchdog expiry fails the turn exactly as before that feature + // existed. See LlmTurnResumeTests for resume coverage. + TimeoutResumeRetryBudget = 0, } }); services.AddSingleton(new StaticSystemPromptProvider("You are a test assistant.")); diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmTurnResumeTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmTurnResumeTests.cs new file mode 100644 index 000000000..b21b8d009 --- /dev/null +++ b/src/Netclaw.Actors.Tests/Sessions/LlmTurnResumeTests.cs @@ -0,0 +1,988 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Akka.Actor; +using Akka.Hosting; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Netclaw.Actors.Channels; +using Netclaw.Actors.Hosting; +using Netclaw.Actors.Protocol; +using Netclaw.Actors.Sessions; +using Netclaw.Actors.Tools; +using Netclaw.Configuration; +using Netclaw.Tools; +using Xunit; +using AiChatRole = Microsoft.Extensions.AI.ChatRole; +using static Netclaw.Actors.Sessions.SessionProtocol; + +namespace Netclaw.Actors.Tests.Sessions; + +/// +/// Covers bounded turn-level resume after a mid-stream LLM call timeout (see +/// LlmSessionActor.TryResumeAfterTimeout). Evidence: correlated provider +/// stall storms (a few tokens then silence) previously burned the full watchdog +/// budget and then failed the turn terminally — in headless chat -p mode a +/// failed turn is a failed session with no external retry. These tests prove the +/// discard-and-resume mechanism, its retry budget, the structural (not +/// tool-iteration-gated) safety of resuming any call in the turn, and that a +/// resumed call's stream never corrupts a +/// delta-accumulating consumer's final answer — using +/// so the watchdog fires only on +/// an explicit — no wall-clock race. +/// +public sealed class LlmTurnResumeTests(ITestOutputHelper output) : LlmSessionTestBase(output) +{ + private static readonly TimeSpan FirstTokenTimeout = TimeSpan.FromSeconds(2); + private readonly ResumeTestChatClient _chatClient = new(); + private readonly FakeToolExecutor _fakeToolExecutor = new(); + + protected override bool UseTestScheduler => true; + + protected override void ConfigureSessionServices(IServiceCollection services) + { + services.AddSingleton(new SingleClientProvider(_chatClient)); + services.AddSingleton(new ModelCapabilities + { + ModelId = "turn-resume-test-model", + ContextWindowTokens = 128_000, + }); + services.AddSingleton(new SessionConfig + { + PrefillTimeout = FirstTokenTimeout, + FirstTokenTimeout = FirstTokenTimeout, + ToolExecutionTimeout = TimeSpan.FromSeconds(10), + SidecarLlmTimeout = TimeSpan.FromSeconds(10), + Tuning = new SessionTuning + { + SnapshotInterval = 5, + TitleGenerationInterval = 0, + TimeoutResumeRetryBudget = 2, + } + }); + services.AddSingleton(new StaticSystemPromptProvider("You are a test assistant.")); + services.AddSingleton(_fakeToolExecutor); + + var registry = new ToolRegistry(); + registry.Register( + AIFunctionFactory.Create(() => "search result", "web_search"), + "web_search"); + services.AddSingleton(registry); + } + + [Fact] + public async Task Timeout_with_no_tool_call_discards_partial_content_and_resumes_successfully() + { + const string partialMarker = "STALLED_PARTIAL_MARKER_SHOULD_NOT_APPEAR"; + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas("stalled chunk one ", partialMarker)); + // The resumed call also streams multiple real deltas (not a single-shot + // completion) so the C1 proof below exercises the same delta-accumulation + // path the dead call used, not just the TextOutput fallback. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.MultiDeltaTextThenComplete("Resumed answer ", "after timeout")); + + var sessionId = new SessionId("turn-resume/success"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-success-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "hello" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + // First call: wait for genuine partial streaming (proves a real stall, not + // an instant failure), then let the watchdog fire. + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + + // Collect EVERY output the turn emits, in order, through TurnCompleted — the + // exact event stream a delta-accumulating subscriber (headless JSON + // envelope, webhook/reminder ExecutionOutputAccumulator, chat TUI) sees. + var events = new List(); + var advanced = false; + object msg; + do + { + msg = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + events.Add(msg); + + if (!advanced && msg is TextDeltaOutput d && d.Delta.Contains(partialMarker, StringComparison.Ordinal)) + { + advanced = true; + AdvanceScheduler(FirstTokenTimeout); + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + } + } while (msg is not TurnCompleted); + + var completed = Assert.IsType(msg); + Assert.Equal(TurnOutcome.Completed, completed.Outcome); + + // C1 proof: feed the exact production event sequence into the real + // ExecutionOutputAccumulator (shared by ReminderExecutionActor and + // WebhookExecutionActor). Before the TextStreamDiscarded fix this would + // accumulate "stalled chunk one STALLED_PARTIAL_MARKER_SHOULD_NOT_APPEARResumed + // answer after timeout" — the dead call's partial text glued to the + // resumed call's answer. + var accumulator = new ExecutionOutputAccumulator(new ToolName("notify_channel")); + foreach (var evt in events.OfType()) + accumulator.ProcessOutput(evt); + + Assert.Equal("Resumed answer after timeout", accumulator.GetAccumulatedText()); + Assert.DoesNotContain(partialMarker, accumulator.GetAccumulatedText(), StringComparison.Ordinal); + + // The discard signal must land strictly between the dead call's last delta + // and the resumed call's first delta — proving the actor clears + // subscriber buffers before the resumed stream starts, not after. + var discardIndex = events.FindIndex(e => e is TextStreamDiscarded); + var deadMarkerIndex = events.FindIndex(e => e is TextDeltaOutput dd && dd.Delta.Contains(partialMarker, StringComparison.Ordinal)); + var resumedDeltaIndex = events.FindIndex(e => e is TextDeltaOutput rd && rd.Delta.Contains("Resumed answer", StringComparison.Ordinal)); + Assert.True(discardIndex > 0, "Expected a TextStreamDiscarded output for the resumed turn."); + Assert.True(discardIndex > deadMarkerIndex, "Discard signal must arrive after the dead call's partial content."); + Assert.True(resumedDeltaIndex > discardIndex, "Resumed call's deltas must arrive after the discard signal."); + + // The final TextOutput (independent of delta accumulation) must also be clean. + var finalText = Assert.IsType(events.OfType().Single()); + Assert.Equal("Resumed answer after timeout", finalText.Text); + Assert.DoesNotContain(partialMarker, finalText.Text, StringComparison.Ordinal); + + Assert.Equal(2, _chatClient.CallCount); + + // The resumed call re-issued the SAME messages as the dead call: identical + // role/text sequence, proving no mutation and no extra user message. + AssertIdenticalMessageLists(_chatClient.ReceivedMessages[0], _chatClient.ReceivedMessages[1]); + + // Persistence check: the discarded partial content must never have entered + // _state.History — prove it by sending a follow-up turn and confirming the + // marker never resurfaces in the conversation history sent to the LLM. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.InstantText("third response")); + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "second message" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextOutput t && t.Text == "third response", + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + + var thirdCallMessages = _chatClient.ReceivedMessages[2]; + Assert.DoesNotContain(thirdCallMessages, m => m.Text?.Contains(partialMarker, StringComparison.Ordinal) == true); + } + + [Fact] + public async Task Timeout_resume_budget_exhausted_fails_turn_exactly_as_before() + { + // Budget is 2 (configured in ConfigureSessionServices): the initial call + // plus 2 resumes must all stall before the turn fails. + for (var i = 0; i < 3; i++) + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas($"stall {i} chunk one ", $"stall {i} chunk two")); + + var sessionId = new SessionId("turn-resume/budget-exhausted"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-budget-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "hello" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + for (var attempt = 0; attempt < 3; attempt++) + { + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextDeltaOutput d && d.Delta.Contains($"stall {attempt} chunk two", StringComparison.Ordinal), + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + AdvanceScheduler(FirstTokenTimeout); + } + + var error = await subscriber.FishForMessageAsync( + m => m is ErrorOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + var errorOutput = Assert.IsType(error); + Assert.Equal(ErrorCategory.Timeout, errorOutput.Category); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Failed, completed.Outcome); + + // Exactly 3 calls: the original plus the 2-call resume budget. No fourth + // (unbounded) resume attempt. + Assert.Equal(3, _chatClient.CallCount); + } + + [Fact] + public async Task Timeout_during_restart_drain_fails_turn_without_resuming() + { + // Stall with multiple real deltas so the watchdog fire is a genuine stall, + // not an instant failure. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas("chunk one ", "chunk two")); + + var sessionId = new SessionId("turn-resume/restart-drain"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-restart-drain-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "hello" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextDeltaOutput d && d.Delta.Contains("chunk two", StringComparison.Ordinal), + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + + // Request a coordinated daemon restart drain WHILE the call is in flight. + // TryResumeAfterTimeout must refuse once this lands, even though the retry + // budget (2) is not exhausted — resuming would keep the turn alive and + // block the drain. + var escapedId = Uri.EscapeDataString(sessionId.Value); + var child = await Sys.ActorSelection($"/user/session-manager/{escapedId}") + .ResolveOne(TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + Watch(child); + var drainTask = sessionManager.Ask( + new PrepareForDaemonRestart(sessionId, "config-reload"), + TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + // PrepareForDaemonRestart's own ack is deferred until the drain completes + // (which only happens once this turn resolves), so it cannot be awaited + // here without deadlocking. Instead, round-trip a second ask through the + // same (sessionManager -> child) path and wait for ITS reply: sessionManager + // forwards synchronously, so this reply cannot land before the drain + // request was already dequeued and processed by the child, guaranteeing + // _restartDrainRequested is set before the scheduler is advanced below. A + // same-filter rejoin is a no-op that only acks the caller — it does not + // re-emit SessionJoined to the subscriber, so nothing else to await here. + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + AdvanceScheduler(FirstTokenTimeout); + + var error = await subscriber.FishForMessageAsync( + m => m is ErrorOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(ErrorCategory.Timeout, Assert.IsType(error).Category); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Failed, completed.Outcome); + + // No resume attempt happened — exactly the one dead call. + Assert.Equal(1, _chatClient.CallCount); + + // The drain completes and the session actor passivates, proving the failed + // turn (not a resume) let the coordinated restart proceed. No observer is + // configured in this test fixture, so passivation skips straight to its + // short PassivationFinalStopDelay (100ms) grace window. The timer is + // registered asynchronously on the actor's own dispatcher (after this + // test's earlier AdvanceScheduler call already returned), so poll — each + // retry nudges the virtual clock a little further until the actor has + // caught up and the grace window timer fires. + await AwaitAssertAsync(() => + { + AdvanceScheduler(TimeSpan.FromMilliseconds(50)); + Assert.True(drainTask.IsCompleted, "Expected the restart drain to complete once the failed turn released passivation."); + return Task.CompletedTask; + }, duration: TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + Assert.Equal(sessionId, (await drainTask).SessionId); + await ExpectTerminatedAsync(child, TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + } + + [Fact] + public async Task Timeout_after_tool_call_dispatched_resumes_successfully() + { + // First call dispatches a tool call and completes normally. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.InstantToolCall( + new FunctionCallContent("call-1", "web_search", + new Dictionary { ["query"] = "test query" }))); + // Second call — the post-tool follow-up — stalls with multiple real deltas, + // then times out. This is the dominant real-world failure: C2 found the + // pre-fix ToolIterationCount gate refused resume for every one of the + // motivating stall reports because they all happened after at least one + // completed tool iteration. Safety here is structural, not gate-based: tool + // dispatch only happens in HandleLlmResponseReceived on a fully completed + // response, and this call times out mid-stream, so it can never have + // dispatched a tool call itself. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas("post-tool chunk one ", "post-tool chunk two")); + // Third call — the resume — completes normally. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.MultiDeltaTextThenComplete("Resumed ", "after tool call")); + + var sessionId = new SessionId("turn-resume/tool-gate"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-gate-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "Search for something" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + // Drain the tool call/result from the first (successful) call. + await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + + // Second call (post-tool) stalls; let the watchdog fire. + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextDeltaOutput d && d.Delta.Contains("post-tool chunk two", StringComparison.Ordinal), + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + AdvanceScheduler(FirstTokenTimeout); + + // Third call (the resume) completes cleanly — no ErrorOutput/TurnCompleted + // in between, proving the turn resumed instead of failing. + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + var text = await subscriber.FishForMessageAsync( + m => m is TextOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + var finalText = Assert.IsType(text); + Assert.Equal("Resumed after tool call", finalText.Text); + Assert.DoesNotContain("post-tool chunk", finalText.Text, StringComparison.Ordinal); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Completed, completed.Outcome); + + // Exactly 3 calls: the tool-call round, the stalled follow-up, and the + // resumed follow-up. The tool executor ran exactly once — resume never + // dispatches a tool call, so there is no double execution. + Assert.Equal(3, _chatClient.CallCount); + Assert.Equal(1, _fakeToolExecutor.CallCount); + + // The resumed call (index 2) re-issued the SAME messages as the dead call + // (index 1) — including the tool-call/tool-result content from the earlier + // completed iteration — and exposed the same tools. + AssertIdenticalMessageLists(_chatClient.ReceivedMessages[1], _chatClient.ReceivedMessages[2]); + AssertIdenticalTools(_chatClient.ReceivedOptions[1], _chatClient.ReceivedOptions[2]); + } + + [Fact] + public async Task Resumed_calls_discarded_estimate_uses_the_previous_completed_calls_real_input_count() + { + // D4: EstimateInputTokens re-stringified the whole message list on every + // ContinueFireLlmCall — a quadratic hot-path cost paid by every tool + // iteration to serve only the (rare) resume path. The fix reuses + // _lastInputTokenCount, the provider's REAL input count from the most + // recently completed call, as the honest proxy instead. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.InstantToolCall( + new FunctionCallContent("call-1", "web_search", + new Dictionary { ["query"] = "test query" }), + usage: new UsageDetails { InputTokenCount = 500, OutputTokenCount = 20 })); + // Post-tool follow-up stalls and times out. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas("post-tool chunk one ", "post-tool chunk two")); + // The resume completes normally with its own (different, larger) usage — + // proving the reported estimate is call 1's real count, not call 3's. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.MultiDeltaTextThenComplete( + "Resumed ", "after tool call", + usage: new UsageDetails { InputTokenCount = 520, OutputTokenCount = 10 })); + + var sessionId = new SessionId("turn-resume/discarded-estimate-real-proxy"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-discarded-estimate-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "Search for something" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + // Drain the tool call/result from the first (successful) call, checking + // its usage carries the real 500-token count that the resume will proxy. + await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + var firstUsage = await subscriber.FishForMessageAsync( + m => m is UsageOutput, TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(500, Assert.IsType(firstUsage).InputTokens); + await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + + // Second call (post-tool) stalls; let the watchdog fire. + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextDeltaOutput d && d.Delta.Contains("post-tool chunk two", StringComparison.Ordinal), + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + AdvanceScheduler(FirstTokenTimeout); + + // Third call (the resume) completes cleanly. + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + var usage = await subscriber.FishForMessageAsync( + m => m is UsageOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + var finalUsage = Assert.IsType(usage); + + // The discarded call's estimate is the REAL input count from call 1 (the + // most recently completed call before the resume) — not a fabricated + // character-count guess of call 2's (larger, tool-result-laden) list, and + // not call 3's own (different) real count either. + Assert.Equal(500, finalUsage.DiscardedResumeEstimatedInputTokens); + Assert.Equal(1, finalUsage.DiscardedResumeAttempts); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Completed, completed.Outcome); + } + + [Fact] + public async Task Resumed_calls_discarded_estimate_is_null_when_no_prior_real_usage_exists() + { + // The session's FIRST call ever dies — no call has completed yet, so + // _lastInputTokenCount is still 0 (never set). D4: report that honestly + // as "no estimate" rather than fabricating a character-count guess. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas("chunk one ", "chunk two")); + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.MultiDeltaTextThenComplete( + "Resumed ", "answer", + usage: new UsageDetails { InputTokenCount = 300, OutputTokenCount = 5 })); + + var sessionId = new SessionId("turn-resume/discarded-estimate-unknown"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-discarded-estimate-null-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "hello" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextDeltaOutput d && d.Delta.Contains("chunk two", StringComparison.Ordinal), + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + AdvanceScheduler(FirstTokenTimeout); + + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + var usage = await subscriber.FishForMessageAsync( + m => m is UsageOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + var finalUsage = Assert.IsType(usage); + + // No completed call ever reported real usage this session — the estimate + // must be null (an honest "unknown"), not a fabricated number, while the + // attempt count still reports 1. + Assert.Null(finalUsage.DiscardedResumeEstimatedInputTokens); + Assert.Equal(1, finalUsage.DiscardedResumeAttempts); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Completed, completed.Outcome); + } + + /// + /// Asserts two message lists are identical in role, text, and tool-call / + /// tool-result content — used to prove a resumed call re-sends the exact same + /// prompt as the call it replaced, including turns with tool activity. + /// + private static void AssertIdenticalMessageLists( + IReadOnlyList expected, IReadOnlyList actual) + { + Assert.Equal(expected.Count, actual.Count); + for (var i = 0; i < expected.Count; i++) + { + Assert.Equal(expected[i].Role, actual[i].Role); + Assert.Equal(expected[i].Text, actual[i].Text); + + var expectedCalls = expected[i].Contents.OfType().ToList(); + var actualCalls = actual[i].Contents.OfType().ToList(); + Assert.Equal(expectedCalls.Count, actualCalls.Count); + for (var c = 0; c < expectedCalls.Count; c++) + { + Assert.Equal(expectedCalls[c].CallId, actualCalls[c].CallId); + Assert.Equal(expectedCalls[c].Name, actualCalls[c].Name); + } + + var expectedResults = expected[i].Contents.OfType().ToList(); + var actualResults = actual[i].Contents.OfType().ToList(); + Assert.Equal(expectedResults.Count, actualResults.Count); + for (var r = 0; r < expectedResults.Count; r++) + { + Assert.Equal(expectedResults[r].CallId, actualResults[r].CallId); + Assert.Equal(expectedResults[r].Result?.ToString(), actualResults[r].Result?.ToString()); + } + } + } + + /// + /// Asserts two exposed the same tool names — + /// proving a resumed call offered the LLM the same tool surface as the call + /// it replaced. + /// + private static void AssertIdenticalTools(ChatOptions? expected, ChatOptions? actual) + { + var expectedNames = (expected?.Tools ?? []).Select(t => t.Name).OrderBy(n => n, StringComparer.Ordinal).ToList(); + var actualNames = (actual?.Tools ?? []).Select(t => t.Name).OrderBy(n => n, StringComparer.Ordinal).ToList(); + Assert.Equal(expectedNames, actualNames); + } +} + +/// +/// Separate fixture from because it needs +/// and +/// to differ by an order of magnitude — deliberately +/// sets them equal so its tests do not depend on the watchdog's arm timeout. Covers +/// H3/D3: a resumed call carries the dead call's own _anyContentStreamed +/// value forward — a call that already streamed substantive content stays on the +/// promoted (tighter) budget through every keepalive that follows; a call that +/// died during prefill with zero content stays on the full prefill budget instead +/// of being cut short. +/// +public sealed class LlmTurnResumeWatchdogArmingTests(ITestOutputHelper output) : LlmSessionTestBase(output) +{ + private static readonly TimeSpan FirstTokenTimeout = TimeSpan.FromSeconds(2); + private static readonly TimeSpan PrefillTimeout = TimeSpan.FromMinutes(30); + private readonly ResumeTestChatClient _chatClient = new(); + + protected override bool UseTestScheduler => true; + + protected override void ConfigureSessionServices(IServiceCollection services) + { + services.AddSingleton(new SingleClientProvider(_chatClient)); + services.AddSingleton(new ModelCapabilities + { + ModelId = "turn-resume-arming-test-model", + ContextWindowTokens = 128_000, + }); + services.AddSingleton(new SessionConfig + { + PrefillTimeout = PrefillTimeout, + FirstTokenTimeout = FirstTokenTimeout, + // Larger than PrefillTimeout so the prefill-stage-death test below can + // advance the scheduler all the way to PrefillTimeout without the + // keepalive-immune no-progress deadline (default 1200s = 20 minutes) + // firing first and masking which timer actually fired. + NoProgressTimeout = TimeSpan.FromHours(1), + ToolExecutionTimeout = TimeSpan.FromSeconds(10), + SidecarLlmTimeout = TimeSpan.FromSeconds(10), + Tuning = new SessionTuning + { + SnapshotInterval = 5, + TitleGenerationInterval = 0, + // A budget of 1 keeps this a clean two-call scenario: the dead call, + // then the resume whose own expiry exhausts the budget. + TimeoutResumeRetryBudget = 1, + } + }); + services.AddSingleton(new StaticSystemPromptProvider("You are a test assistant.")); + services.AddSingleton(new FakeToolExecutor()); + services.AddSingleton(new ToolRegistry()); + } + + [Fact] + public async Task Resumed_call_after_midstream_stall_stays_on_promoted_budget_through_a_keepalive() + { + // First call: stall after two real deltas so its own watchdog promotes to + // FirstTokenTimeout before firing — a genuine mid-stream stall, not an + // instant failure. _anyContentStreamed is true when this call dies. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallAfterDeltas("chunk one ", "chunk two")); + // Resumed call: one content-free keepalive, then silence. + // StallImmediately() cannot catch the D3 regression — it never reaches + // OnStreamProgress at all, so it only proves the INITIAL arm value (which + // was already correct even with the bug). The bug only shows once the + // resumed call's watchdog is re-armed by a NON-substantive update: a + // substantive delta would "promote" correctly by accident. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.KeepaliveThenStall()); + + var sessionId = new SessionId("turn-resume/watchdog-arming-keepalive"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-arming-keepalive-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "hello" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + await subscriber.FishForMessageAsync( + m => m is TextDeltaOutput d && d.Delta.Contains("chunk two", StringComparison.Ordinal), + TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + AdvanceScheduler(FirstTokenTimeout); + + // Drain the discard signal the resume emits before re-firing — otherwise + // it sits unread in the probe's mailbox and trips the ExpectNoMsgAsync + // check below on an unrelated, already-delivered message. + await subscriber.FishForMessageAsync( + m => m is TextStreamDiscarded, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + + // The resumed call's keepalive carries no observable output (a + // content-free update never emits a TextDeltaOutput), so there is no + // SessionOutput to fish for as a synchronization signal here. Wait + // briefly on real wall-clock time for the invoker's fire-and-forget Tell + // to reach and be processed by the actor before probing the watchdog's + // re-arm — this is not a virtual-clock condition, just letting an + // in-process async handoff settle. + await subscriber.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300), cancellationToken: TestContext.Current.CancellationToken); + + // If the keepalive incorrectly reverted the arm to the 30-minute prefill + // budget (the D3 bug), advancing only FirstTokenTimeout here would never + // fire the watchdog, and the bounded WaitForStreamInvocationAsync (M4) + // below would time out instead of observing the failure. + AdvanceScheduler(FirstTokenTimeout); + + // Budget is 1: the resume's own watchdog expiry exhausts it, so the turn + // fails — proving the resumed call's watchdog stayed armed at + // FirstTokenTimeout through the keepalive, not the 30-minute PrefillTimeout. + var error = await subscriber.FishForMessageAsync( + m => m is ErrorOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(ErrorCategory.Timeout, Assert.IsType(error).Category); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Failed, completed.Outcome); + Assert.Equal(2, _chatClient.CallCount); + } + + [Fact] + public async Task Resumed_call_after_prefill_stage_death_keeps_the_full_prefill_budget() + { + // First call: dies during prefill without streaming a single update — not + // even a keepalive. _anyContentStreamed is false (no evidence this + // provider is even alive) when the watchdog fires. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallImmediately()); + // Resumed call: also dies during prefill with zero content. + _chatClient.Behaviors.Enqueue(ResumeCallBehavior.StallImmediately()); + + var sessionId = new SessionId("turn-resume/watchdog-arming-prefill-death"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("resume-arming-prefill-death-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "hello" + }, TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); + + // First call dies at the full prefill budget — it is a fresh call, not a + // resume, so it is armed on PrefillTimeout regardless of D3. + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + AdvanceScheduler(PrefillTimeout); + + // Drain the discard signal the resume emits before re-firing — otherwise + // it sits unread in the probe's mailbox and trips the ExpectNoMsgAsync + // check below on an unrelated, already-delivered message. + await subscriber.FishForMessageAsync( + m => m is TextStreamDiscarded, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + await _chatClient.WaitForStreamInvocationAsync(TestContext.Current.CancellationToken); + + // Prove the resumed call is armed on the FULL prefill budget, not the + // promoted budget: advancing only FirstTokenTimeout must NOT fail the + // turn yet. If the fix incorrectly forced the promoted budget onto a + // call with no streamed evidence (the "ALSO" half of D3), this would + // already have failed by here. + AdvanceScheduler(FirstTokenTimeout); + await subscriber.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300), cancellationToken: TestContext.Current.CancellationToken); + + // Advance the rest of the way to the full prefill budget — now it fires, + // exhausting the retry budget (1) and failing the turn. + AdvanceScheduler(PrefillTimeout - FirstTokenTimeout); + + var error = await subscriber.FishForMessageAsync( + m => m is ErrorOutput, TimeSpan.FromSeconds(6), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(ErrorCategory.Timeout, Assert.IsType(error).Category); + + var completed = await subscriber.ExpectMsgAsync(TimeSpan.FromSeconds(3), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Failed, completed.Outcome); + Assert.Equal(2, _chatClient.CallCount); + } +} + +/// +/// One configured behavior for a single call, +/// dequeued in call order. +/// +internal enum ResumeCallBehaviorKind { StallAfterDeltas, StallImmediately, InstantText, MultiDeltaTextThenComplete, InstantToolCall, KeepaliveThenStall } + +internal sealed record ResumeCallBehavior( + ResumeCallBehaviorKind Kind, + string? Delta1 = null, + string? Delta2 = null, + string? Text = null, + FunctionCallContent? ToolCall = null, + UsageDetails? Usage = null) +{ + /// + /// Streams two substantive text deltas (needed so the session's + /// buffered-first-delta trick actually flushes visible content — a single + /// delta stays buffered pending a second) then hangs forever, simulating a + /// half-open provider stream: a few tokens, then silence. + /// + public static ResumeCallBehavior StallAfterDeltas(string delta1, string delta2) + => new(ResumeCallBehaviorKind.StallAfterDeltas, Delta1: delta1, Delta2: delta2); + + /// + /// Hangs forever without ever streaming a single update — not even a + /// keepalive. Isolates the watchdog's initial arm timeout from its + /// stream-progress promotion logic. + /// + public static ResumeCallBehavior StallImmediately() + => new(ResumeCallBehaviorKind.StallImmediately); + + /// + /// Streams one content-free keepalive update (empty Contents, no + /// finish reason — mirrors a provider heartbeat like llama.cpp's + /// prompt_progress) then hangs forever. Unlike + /// , this reaches + /// ProcessingWatchdog.OnStreamProgress once with a non-substantive + /// update — the only update kind that can expose a re-arm that reverts a + /// resumed call's promoted budget back to the full prefill budget (a + /// substantive delta would promote correctly by accident, and + /// never reaches the progress handler at all). + /// + public static ResumeCallBehavior KeepaliveThenStall() + => new(ResumeCallBehaviorKind.KeepaliveThenStall); + + public static ResumeCallBehavior InstantText(string text) + => new(ResumeCallBehaviorKind.InstantText, Text: text); + + /// + /// Streams two substantive text deltas (same buffered-first-delta + /// requirement as ) and then completes + /// normally. The final text is + . + /// An optional trailing update lets a test prove + /// what a subsequent resume's discarded-token estimate is computed from. + /// + public static ResumeCallBehavior MultiDeltaTextThenComplete(string delta1, string delta2, UsageDetails? usage = null) + => new(ResumeCallBehaviorKind.MultiDeltaTextThenComplete, Delta1: delta1, Delta2: delta2, Usage: usage); + + /// + /// Dispatches a tool call and completes normally. An optional trailing + /// update lets a test prove a LATER resume's + /// discarded-token estimate is the real count from THIS completed call. + /// + public static ResumeCallBehavior InstantToolCall(FunctionCallContent toolCall, UsageDetails? usage = null) + => new(ResumeCallBehaviorKind.InstantToolCall, ToolCall: toolCall, Usage: usage); +} + +/// +/// Fake with per-call scripted streaming behavior: +/// stall after a couple of substantive deltas (never completes), return text +/// instantly, or return a tool call instantly. Records every call's message list +/// and so tests can assert a resumed call re-sends the +/// identical prompt and tool surface. +/// +internal sealed class ResumeTestChatClient : IChatClient +{ + // Bounds every wait for the next streaming invocation so a regression in the + // production resume/watchdog wiring fails the test with a clear + // TimeoutException instead of hanging the test run indefinitely. + private static readonly TimeSpan InvocationWaitTimeout = TimeSpan.FromSeconds(15); + + private readonly object _gate = new(); + private int _callCount; + private readonly List> _receivedMessages = []; + private readonly List _receivedOptions = []; + private readonly Channel _invocations = + Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); + + public int CallCount => _callCount; + + public IReadOnlyList> ReceivedMessages + { + get { lock (_gate) { return _receivedMessages.ToArray(); } } + } + + public IReadOnlyList ReceivedOptions + { + get { lock (_gate) { return _receivedOptions.ToArray(); } } + } + + public Queue Behaviors { get; } = new(); + + /// + /// Awaits the next streaming invocation. The watchdog is already armed by + /// then. Bounded by — a regression that + /// stops the actor from re-firing the call (e.g. resume silently not + /// happening) fails with a instead of hanging. + /// + public async Task WaitForStreamInvocationAsync(CancellationToken cancellationToken) + { + using var timeoutCts = new CancellationTokenSource(InvocationWaitTimeout); + using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token); + try + { + await _invocations.Reader.ReadAsync(linkedCts.Token); + } + catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested) + { + throw new TimeoutException( + $"Timed out after {InvocationWaitTimeout} waiting for the next streaming invocation " + + $"(callCount so far: {_callCount})."); + } + } + + public Task GetResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + CancellationToken cancellationToken = default) + => Task.FromException(new NotSupportedException("Streaming path only.")); + + public IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + CancellationToken cancellationToken = default) + { + var messageList = messages.ToList(); + ResumeCallBehavior behavior; + int callNumber; + lock (_gate) + { + _receivedMessages.Add(messageList); + _receivedOptions.Add(options); + callNumber = ++_callCount; + behavior = Behaviors.Count > 0 + ? Behaviors.Dequeue() + : ResumeCallBehavior.InstantText($"[fake] default response #{callNumber}"); + } + + _invocations.Writer.TryWrite(callNumber); + + var updates = behavior.Kind switch + { + ResumeCallBehaviorKind.StallAfterDeltas => StallAfterDeltasAsync(behavior.Delta1!, behavior.Delta2!), + ResumeCallBehaviorKind.StallImmediately => TestStreamingHelpers.NeverCompletesAsync(cancellationToken), + ResumeCallBehaviorKind.KeepaliveThenStall => KeepaliveThenStallAsync(), + ResumeCallBehaviorKind.InstantText => TestStreamingHelpers.ReturnTextAsync(behavior.Text!, cancellationToken), + ResumeCallBehaviorKind.MultiDeltaTextThenComplete => MultiDeltaTextThenCompleteAsync(behavior.Delta1!, behavior.Delta2!), + ResumeCallBehaviorKind.InstantToolCall => InstantToolCallAsync(behavior.ToolCall!, cancellationToken), + _ => throw new InvalidOperationException($"Unhandled behavior kind {behavior.Kind}") + }; + + // A behavior that never completes (a stall) never reaches the trailing + // usage update either — this only ever appends usage after a real + // completion, matching what a real provider does. + return AppendUsageIfPresent(updates, behavior.Usage); + } + + private static async IAsyncEnumerable AppendUsageIfPresent( + IAsyncEnumerable updates, UsageDetails? usage) + { + await foreach (var update in updates) + yield return update; + + if (usage is not null) + yield return new ChatResponseUpdate { Contents = [new UsageContent(usage)] }; + } + + private static async IAsyncEnumerable StallAfterDeltasAsync(string delta1, string delta2) + { + yield return new ChatResponseUpdate + { + Role = AiChatRole.Assistant, + Contents = [new TextContent(delta1)] + }; + await Task.Yield(); + + yield return new ChatResponseUpdate + { + Contents = [new TextContent(delta2)] + }; + await Task.Yield(); + + // Stream is now silent — never completes on its own; the actor's watchdog + // is the only thing that ends this turn. + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await gate.Task; + yield break; + } + + private static async IAsyncEnumerable KeepaliveThenStallAsync() + { + // Content-free keepalive — empty Contents, no finish reason. Mirrors a + // provider heartbeat that proves the socket is alive but carries no + // model output (see StreamingResponseReader.IsSubstantiveUpdate). + yield return new ChatResponseUpdate + { + Role = AiChatRole.Assistant, + Contents = [] + }; + await Task.Yield(); + + // Stream is now silent — never completes on its own; the actor's watchdog + // is the only thing that ends this turn. + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await gate.Task; + yield break; + } + + private static async IAsyncEnumerable MultiDeltaTextThenCompleteAsync(string delta1, string delta2) + { + yield return new ChatResponseUpdate + { + Role = AiChatRole.Assistant, + Contents = [new TextContent(delta1)] + }; + await Task.Yield(); + + yield return new ChatResponseUpdate + { + Contents = [new TextContent(delta2)] + }; + await Task.CompletedTask; + } + + private static async IAsyncEnumerable InstantToolCallAsync( + FunctionCallContent toolCall, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.CompletedTask; + + var response = new ChatResponse(new ChatMessage(AiChatRole.Assistant, [toolCall])); + foreach (var update in response.ToChatResponseUpdates()) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return update; + } + } + + public object? GetService(Type serviceType, object? serviceKey = null) => null; + + public void Dispose() { } +} diff --git a/src/Netclaw.Actors.Tests/Sessions/TurnStateTrackerTests.cs b/src/Netclaw.Actors.Tests/Sessions/TurnStateTrackerTests.cs index f1cb69682..b41a867ff 100644 --- a/src/Netclaw.Actors.Tests/Sessions/TurnStateTrackerTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/TurnStateTrackerTests.cs @@ -156,4 +156,29 @@ public void RawCallVolume_DoesNotControlTheLimit() Assert.Equal(1, tracker.ToolIterationCount); Assert.Equal(100, tracker.ToolCallCount); } + + [Fact] + public void TimeoutResumeCount_StartsAtZeroAndIncrementsPerResume() + { + var tracker = new TurnStateTracker(); + Assert.Equal(0, tracker.TimeoutResumeCount); + + tracker.RecordTimeoutResume(); + Assert.Equal(1, tracker.TimeoutResumeCount); + + tracker.RecordTimeoutResume(); + Assert.Equal(2, tracker.TimeoutResumeCount); + } + + [Fact] + public void TimeoutResumeCount_ResetsOnNewTurn() + { + var tracker = new TurnStateTracker(); + tracker.RecordTimeoutResume(); + tracker.RecordTimeoutResume(); + + tracker.ResetForNewTurn(); + + Assert.Equal(0, tracker.TimeoutResumeCount); + } } diff --git a/src/Netclaw.Actors/Channels/ExecutionOutputAccumulator.cs b/src/Netclaw.Actors/Channels/ExecutionOutputAccumulator.cs index a8074f2fa..9beccabe2 100644 --- a/src/Netclaw.Actors/Channels/ExecutionOutputAccumulator.cs +++ b/src/Netclaw.Actors/Channels/ExecutionOutputAccumulator.cs @@ -36,6 +36,16 @@ public sealed class ExecutionOutputAccumulator private readonly ToolName _notificationToolName; private readonly Action? _onNotifyTracked; private readonly StringBuilder _buffer = new(); + + // Length of _buffer already committed by an earlier call's TextOutput this + // turn. TextStreamDiscarded truncates back to this point instead of clearing + // the whole buffer, so a later call's discard cannot erase an earlier + // COMPLETED call's text (see the TextStreamDiscarded case below). + private int _committedLength; + + // Whether the CURRENT (not-yet-completed) call has streamed a delta. Reset + // at every call boundary — a TextOutput (call completed) or a + // TextStreamDiscarded (call died) — so it never leaks across calls. private bool _sawTextDelta; private bool _notifyAttempted; private bool _notifyFailed; @@ -104,9 +114,32 @@ public OutputAction ProcessOutput(SessionOutput output) _sawTextDelta = true; return OutputAction.Continue; + case TextStreamDiscarded: + // A timed-out call was discarded. The actor re-issues it. Truncate + // back to the last committed call boundary so only the dead call's + // own, not-yet-committed text is removed — text from an earlier + // call that already completed this turn (committed at its own + // TextOutput below) survives. See SessionProtocol.TextStreamDiscarded. + _buffer.Remove(_committedLength, _buffer.Length - _committedLength); + _sawTextDelta = false; + return OutputAction.Continue; + case TextOutput text: + // TextOutput marks one LLM call's text as complete, whether that + // call ended in tool calls (a preamble) or the final answer. Commit + // everything accumulated for it so a LATER call's discard can never + // erase it. Some notices (approval-expired etc.) reuse TextOutput to + // reach the channel while an earlier call still streams. + // IsCallBoundary is false for those. They must not move the + // commit marker or clear the live call's delta flag (see + // SessionProtocol.TextOutput.IsCallBoundary). if (!_sawTextDelta) _buffer.Append(text.Text); + if (text.IsCallBoundary) + { + _committedLength = _buffer.Length; + _sawTextDelta = false; + } return OutputAction.Continue; case ToolResultOutput toolResult: diff --git a/src/Netclaw.Actors/Protocol/SessionOutputDto.cs b/src/Netclaw.Actors/Protocol/SessionOutputDto.cs index f42cf6994..59bb2c0a9 100644 --- a/src/Netclaw.Actors/Protocol/SessionOutputDto.cs +++ b/src/Netclaw.Actors/Protocol/SessionOutputDto.cs @@ -16,6 +16,7 @@ public static class SessionOutputTypes { public const string Text = "text"; public const string TextDelta = "text_delta"; + public const string TextStreamDiscarded = "text_stream_discarded"; public const string Thinking = "thinking"; public const string ThinkingDelta = "thinking_delta"; public const string ToolCall = "tool_call"; @@ -61,6 +62,13 @@ public sealed record SessionOutputDto // Text / Thinking public string? Text { get; init; } + /// + /// Mirrors . Null on + /// the wire (older payloads, or non-Text output types) maps back to true — + /// see . + /// + public bool? IsCallBoundary { get; init; } + // Tool Call / Tool Result public string? CallId { get; init; } public string? ToolName { get; init; } @@ -77,6 +85,8 @@ public sealed record SessionOutputDto public double? UsagePercent { get; init; } public double? PromptMs { get; init; } public double? PredictedPerSecond { get; init; } + public long? DiscardedResumeEstimatedInputTokens { get; init; } + public int? DiscardedResumeAttempts { get; init; } // Turn Completed [System.Text.Json.Serialization.JsonConverter(typeof(NullableTurnNumberJsonConverter))] diff --git a/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs b/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs index 0a60f7565..daa52bcf8 100644 --- a/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs +++ b/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs @@ -23,7 +23,8 @@ public static class SessionOutputDtoMapper Type = SessionOutputTypes.Text, SessionId = msg.SessionId.Value, TimestampMs = msg.TimestampMs, - Text = msg.Text + Text = msg.Text, + IsCallBoundary = msg.IsCallBoundary }, TextDeltaOutput msg => new SessionOutputDto @@ -34,6 +35,13 @@ public static class SessionOutputDtoMapper Text = msg.Delta }, + TextStreamDiscarded msg => new SessionOutputDto + { + Type = SessionOutputTypes.TextStreamDiscarded, + SessionId = msg.SessionId.Value, + TimestampMs = msg.TimestampMs + }, + ThinkingOutput msg => new SessionOutputDto { Type = SessionOutputTypes.Thinking, @@ -84,6 +92,8 @@ public static class SessionOutputDtoMapper UsagePercent = msg.UsagePercent, PromptMs = msg.PromptMs, PredictedPerSecond = msg.PredictedPerSecond, + DiscardedResumeEstimatedInputTokens = msg.DiscardedResumeEstimatedInputTokens, + DiscardedResumeAttempts = msg.DiscardedResumeAttempts, }, TurnCompleted msg => new SessionOutputDto @@ -221,13 +231,19 @@ public static SessionOutput FromDto(SessionOutputDto dto) SessionOutputTypes.Text => new TextOutput(dto.Text ?? string.Empty) { SessionId = sessionId, - TimestampMs = dto.TimestampMs + TimestampMs = dto.TimestampMs, + IsCallBoundary = dto.IsCallBoundary ?? true }, SessionOutputTypes.TextDelta => new TextDeltaOutput(dto.Text ?? string.Empty) { SessionId = sessionId, TimestampMs = dto.TimestampMs }, + SessionOutputTypes.TextStreamDiscarded => new TextStreamDiscarded + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs + }, SessionOutputTypes.Thinking => new ThinkingOutput(dto.Text ?? string.Empty) { SessionId = sessionId, @@ -267,6 +283,8 @@ public static SessionOutput FromDto(SessionOutputDto dto) UsagePercent = dto.UsagePercent, PromptMs = dto.PromptMs, PredictedPerSecond = dto.PredictedPerSecond, + DiscardedResumeEstimatedInputTokens = dto.DiscardedResumeEstimatedInputTokens, + DiscardedResumeAttempts = dto.DiscardedResumeAttempts, }, SessionOutputTypes.TurnCompleted => new TurnCompleted { diff --git a/src/Netclaw.Actors/Sessions/Handlers/TurnStateTracker.cs b/src/Netclaw.Actors/Sessions/Handlers/TurnStateTracker.cs index f2adbc85b..4c7f3178b 100644 --- a/src/Netclaw.Actors/Sessions/Handlers/TurnStateTracker.cs +++ b/src/Netclaw.Actors/Sessions/Handlers/TurnStateTracker.cs @@ -48,6 +48,27 @@ internal sealed class TurnStateTracker public int ToolIterationCount { get; private set; } public bool ForceNoToolsActive { get; set; } + /// + /// Number of timeout-triggered LLM call resumes used so far this turn. Compared + /// against + /// by the actor before resuming again. + /// + public int TimeoutResumeCount { get; private set; } + + /// + /// Estimated input tokens sent to the provider by every LLM call discarded via + /// a timeout resume this turn. The provider billed for this input even though + /// the call never completed and reported no usage — see + /// LlmSessionActor.TryResumeAfterTimeout for the estimation method. + /// Null once any recorded resume had no real prior count to proxy from — the + /// running total then reports "unknown", not a silently incomplete sum. + /// + public long? DiscardedResumeEstimatedInputTokens => + _discardedResumeEstimateHasUnknownContribution ? null : _discardedResumeEstimatedInputTokensSum; + + private long _discardedResumeEstimatedInputTokensSum; + private bool _discardedResumeEstimateHasUnknownContribution; + private bool _budgetNudgeSent; private int _postToolEmptyResponseCount; private int _preToolEmptyResponseCount; @@ -66,6 +87,37 @@ public void ResetForNewTurn() ForceNoToolsActive = false; _toolCallCounts.Clear(); _duplicateNudgeSent = false; + TimeoutResumeCount = 0; + _discardedResumeEstimatedInputTokensSum = 0; + _discardedResumeEstimateHasUnknownContribution = false; + } + + /// + /// Record a timeout-triggered LLM call resume. Called by the actor after it + /// decides to re-issue a timed-out call. + /// + public void RecordTimeoutResume() + { + TimeoutResumeCount++; + } + + /// + /// Add the discarded call's estimated input tokens to this turn's running + /// total. Called once per resume, right before the actor re-issues the call. + /// A null means the actor had no real + /// prior count to proxy from (the discarded call was the session's first) — + /// that poisons to null for + /// the rest of the turn rather than reporting a silently incomplete sum. + /// + public void RecordDiscardedResumeEstimatedInputTokens(long? estimatedTokens) + { + if (estimatedTokens is not { } value) + { + _discardedResumeEstimateHasUnknownContribution = true; + return; + } + + _discardedResumeEstimatedInputTokensSum += value; } /// diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index ac804cbbd..971ed726e 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -97,7 +97,9 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // Owns the exposed tool list (base + discovered) and lease-based eviction. private readonly DiscoveredToolCache _discoveredToolCache = new(); - // Last observed input token count from LLM response (for compaction trigger) + // Last observed REAL input token count from a completed LLM response — drives + // the compaction trigger and, per TryResumeAfterTimeout, also doubles as the + // honest proxy for a discarded (never-completed) call's input size. private long _lastInputTokenCount; // When compaction triggers mid-tool-loop, the turn is still in-progress. @@ -172,6 +174,26 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // Startup context layers: injected on first LLM call, re-injected after compaction private bool _startupContextInjected; + // True when the in-flight call's own ContinueFireLlmCall() invocation was the + // one that flipped _startupContextInjected false→true. A timeout resume rolls + // this back before re-firing so the resumed call's message list is + // byte-identical to the one that just died — otherwise the OnceAtStart context + // layers would silently vanish from the retry (SessionMessageAssembler renders + // them only while StartupContextInjected is false). See TryResumeAfterTimeout. + private bool _startupContextPendingFirstDelivery; + + // Set immediately before a FireLlmCall triggered by TryResumeAfterTimeout. + // FireLlmCall reads and clears it in the same synchronous step, before any + // async gap: when true, FireLlmCall skips its normal _anyContentStreamed + // reset, so the resumed call carries the dead call's own value forward + // instead of restarting the two-phase watchdog budget from scratch. A call + // that already streamed substantive content keeps the promoted (tighter) + // budget through the resumed call's keepalives; a call that died during + // prefill with zero content keeps the full prefill budget, so a genuinely + // slow failover prefill is not killed early. See TryResumeAfterTimeout, + // FireLlmCall, and ProcessingWatchdog.OnStreamProgress. + private bool _resumingAfterTimeout; + // Guards against infinite compaction loops: if a post-compaction buffer drain // overflows again, fail the turn. Reset at the start of each new user turn. private int _compactionOverflowRetryCount; @@ -662,6 +684,16 @@ private void Processing() return; } + // A timeout that surfaced directly as a TimeoutException (not routed + // through the watchdog's own cancellation — e.g. a transport-level + // request timeout firing before our watchdog budget elapses) gets the + // same bounded resume as the watchdog path below. + if (msg.Cause is TimeoutException llmCallFailedTimeout + && TryResumeAfterTimeout(llmCallFailedTimeout, "llm_call_failed")) + { + return; + } + // 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. @@ -711,6 +743,10 @@ private void Processing() _log.Error("Processing watchdog expired for operation {OperationName} (opId={OperationId}, noProgress={NoProgress})", msg.OperationName, msg.OperationId, msg.NoProgress); + + if (TryResumeAfterTimeout(timeoutCause, "watchdog")) + return; + var errorMessage = ExtractLlmErrorMessage(timeoutCause); FailCurrentTurn(errorMessage, timeoutCause, ErrorCategory.Timeout); }); @@ -1273,6 +1309,7 @@ private void HandleCompactionWorkCompleted(CompactionWorkCompleted msg) _state = _state.Apply(evt); _lastInputTokenCount = 0; _startupContextInjected = false; + _startupContextPendingFirstDelivery = false; _recallManager.ResetForCompaction(); _discoveredToolCache.EvictAll(); @@ -2691,7 +2728,15 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) // injects a volatile system notice. The session stays usable. } - _anyContentStreamed = false; + // A resumed call carries the dead call's _anyContentStreamed forward + // instead of resetting it — see _resumingAfterTimeout and + // TryResumeAfterTimeout for why. Consumed here, synchronously, before any + // async gap, so there is no window for a later unrelated FireLlmCall to + // see a stale true value. + if (_resumingAfterTimeout) + _resumingAfterTimeout = false; + else + _anyContentStreamed = false; CancelAndDisposeLlmCts(); _activeLlmCts = new CancellationTokenSource(); _activeCallId++; @@ -2843,6 +2888,7 @@ private void ContinueFireLlmCall(bool forceNoTools) // startup injection complete after the first call to preserve the // existing OnceAtStart semantics. var skillHint = BuildSkillHint(); + _startupContextPendingFirstDelivery = !_startupContextInjected; var messages = SessionMessageAssembler.Assemble(new ContextAssemblyInput( State: _state, ContextLayers: _contextLayers, @@ -2875,7 +2921,20 @@ private void ContinueFireLlmCall(bool forceNoTools) if (!forceNoTools && exposedTools.Count > 0) options.Tools = [.. exposedTools]; - _watchdog.Start(ProcessingWatchdog.LlmCall, _config.PrefillTimeout, Timers, _config.NoProgressTimeout); + // The initial arm follows the same _anyContentStreamed rule as the + // in-stream promotion in HandleLlmResponseDeltaReceived (via + // ProcessingWatchdog.OnStreamProgress) and the watchdog-expiry error + // message below — one flag, one rule, everywhere the actor picks between + // the prefill and promoted budgets. FireLlmCall carries _anyContentStreamed + // forward on a resume instead of resetting it (see _resumingAfterTimeout), + // so a call that already streamed substantive content before stalling + // stays on the tighter promoted budget here and through every keepalive + // that follows; a call that died during prefill with zero content stays on + // the full prefill budget instead of being cut short on unproven ground. + var initialWatchdogTimeout = _anyContentStreamed + ? _config.FirstTokenTimeout + : _config.PrefillTimeout; + _watchdog.Start(ProcessingWatchdog.LlmCall, initialWatchdogTimeout, Timers, _config.NoProgressTimeout); TurnLog().Info("turn_llm_call_start messages={MessageCount} toolsEnabled={ToolsEnabled} forceNoTools={ForceNoTools} callId={CallId}", messages.Count, @@ -2886,6 +2945,88 @@ private void ContinueFireLlmCall(bool forceNoTools) _ = SessionLlmInvoker.InvokeAsync(client, messages, options, self, _activeCallId, _sessionId, _activeLlmCts!.Token); } + /// + /// Bounded, turn-scoped resume after an LLM call times out mid-stream. The dead + /// call's partial output is discarded by construction: streamed deltas only ever + /// reach transient UI output () and + /// the invoker's local buffer (StreamingResponseReader) — nothing from a + /// call that never reaches is written to + /// , so there is nothing to roll back there. Resuming means + /// re-issuing the exact same call: reassembles the + /// message list from the unchanged and the same + /// flag the dead call used. + /// + /// Safety is structural, not gated on tool-iteration count. Tool dispatch only + /// happens in , on a fully completed + /// response, after the watchdog stops. A call that times out mid-stream never + /// reaches that handler, so it can never have dispatched a tool call this turn — + /// resume is safe to allow on any call, including one after an earlier tool + /// iteration completed. Netclaw runs all tools locally through + /// ; there is no provider-hosted tool + /// mechanism whose committed context a re-issued call could replay, so there is + /// no double-execution risk to guard against. Only the retry budget + /// () + /// bounds resume. + /// + /// + private bool TryResumeAfterTimeout(TimeoutException cause, string source) + { + // A coordinated daemon restart is draining this session so it can passivate + // cleanly. Resuming would keep the turn alive and block the drain — fail the + // turn immediately, exactly as every other mid-turn continuation does under + // restart drain (see the _restartDrainRequested checks in Processing/Compacting). + if (_restartDrainRequested) + { + TurnLog().Warning(cause, + "turn_llm_timeout_resume_blocked source={Source} reason=restart_drain_pending", + source); + return false; + } + + if (_turnState.TimeoutResumeCount >= _config.Tuning.TimeoutResumeRetryBudget) + { + TurnLog().Warning(cause, + "turn_llm_timeout_resume_budget_exhausted source={Source} attempts={Attempts} budget={Budget}", + source, _turnState.TimeoutResumeCount, _config.Tuning.TimeoutResumeRetryBudget); + return false; + } + + _turnState.RecordTimeoutResume(); + + // Honest proxy for the discarded call's real input size: the provider + // never reports usage for a call that times out before completion, but + // _lastInputTokenCount (the previous completed call's real, provider- + // reported count) already flows through this seam and approximates it far + // better than a fabricated guess. Null when no completed call has reported + // real usage yet this session (the discarded call was the first) — report + // that honestly instead of fabricating a number. + long? discardedEstimatedInputTokens = _lastInputTokenCount > 0 ? _lastInputTokenCount : null; + _turnState.RecordDiscardedResumeEstimatedInputTokens(discardedEstimatedInputTokens); + TurnLog().Warning(cause, + "turn_llm_timeout_resume source={Source} attempt={Attempt} budget={Budget} discardedEstimatedInputTokens={DiscardedEstimatedInputTokens}", + source, _turnState.TimeoutResumeCount, _config.Tuning.TimeoutResumeRetryBudget, discardedEstimatedInputTokens); + + // Tell every delta-accumulating subscriber (headless JSON envelope, channel + // execution accumulators, the chat TUI) to clear the dead call's partial text + // before the resumed call streams its own deltas — otherwise the two answers + // concatenate into one corrupted reply. Lifecycle output, so it reaches every + // subscriber regardless of their OutputFilter. See TextStreamDiscarded. + EmitOutput(new TextStreamDiscarded { SessionId = _sessionId }); + + // Undo the dead call's speculative "startup context delivered" flip so the + // resumed call's message list matches the one that just died exactly — see + // _startupContextPendingFirstDelivery. + if (_startupContextPendingFirstDelivery) + { + _startupContextInjected = false; + _startupContextPendingFirstDelivery = false; + } + + _resumingAfterTimeout = true; + FireLlmCall(forceNoTools: _turnState.ForceNoToolsActive); + return true; + } + private async Task CreateWorkingContextContinuationAsync( long generation, bool forceNoTools, @@ -3804,6 +3945,12 @@ private void EmitUsageOutput(UsageDetails usage) UsagePercent = usagePercent, PromptMs = promptMs, PredictedPerSecond = predictedPerSec, + DiscardedResumeEstimatedInputTokens = _turnState.TimeoutResumeCount > 0 + ? _turnState.DiscardedResumeEstimatedInputTokens + : null, + DiscardedResumeAttempts = _turnState.TimeoutResumeCount > 0 + ? _turnState.TimeoutResumeCount + : null, }, OutputFilter.Usage); } @@ -3831,13 +3978,17 @@ private bool HasApprovalHistory /// Emits the channel-visible "approval prompt expired" notice. Used when a /// tool interaction response cannot be honored — fail loud instead of /// silently dropping the click (constitution: no silent fallbacks). + /// IsCallBoundary is false. This notice can fire while an unrelated LLM + /// call still streams. It must not move a subscriber's call-boundary + /// marker (see SessionProtocol.TextOutput.IsCallBoundary). /// private void EmitExpiredPromptNotice() => EmitOutput(new TextOutput( "That approval prompt has expired — the session moved on or restarted. " + "Please re-issue the request and I'll ask again if approval is needed.") { - SessionId = _sessionId + SessionId = _sessionId, + IsCallBoundary = false }, OutputFilter.Text); /// @@ -3937,18 +4088,24 @@ or ApprovalDecision.ApprovedEverywhere } } + // IsCallBoundary is false. This notice can fire while an unrelated LLM + // call still streams. It must not move a subscriber's call-boundary + // marker (see SessionProtocol.TextOutput.IsCallBoundary). private void EmitWrongRequesterApprovalNotice() => EmitOutput(new TextOutput( "Approval response ignored: only the requesting user can approve this tool action.") { - SessionId = _sessionId + SessionId = _sessionId, + IsCallBoundary = false }, OutputFilter.Text); + // IsCallBoundary is false — see EmitWrongRequesterApprovalNotice above. private void EmitUnavailableApprovalOptionNotice() => EmitOutput(new TextOutput( "Approval response ignored: that option is not available for this tool action.") { - SessionId = _sessionId + SessionId = _sessionId, + IsCallBoundary = false }, OutputFilter.Text); private bool TryResolveTextApprovalResponse( diff --git a/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs b/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs index 8fa0e4612..869c7aa65 100644 --- a/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs +++ b/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs @@ -37,10 +37,38 @@ public abstract record SessionOutput : IWithSessionId, INoSerializationVerificat } /// - /// User-facing text reply from the assistant. - /// Requires . + /// User-facing text reply from the assistant. Marks the end of one LLM + /// call's text — the actor sends it once a call fully completes, whether + /// that call ended in tool calls (a preamble) or the final answer (see + /// LlmSessionActor.EmitAndDispatchToolBatch and + /// LlmSessionActor.EmitResponseOutputs). A subscriber that + /// accumulates text across a turn can use + /// this message as the call-completion boundary. Requires + /// . + /// + /// EXCEPTION: is false for a small set of + /// channel-visible notices (approval-expired, wrong-requester, + /// unavailable-approval-option — see + /// LlmSessionActor.EmitExpiredPromptNotice and its siblings). These + /// notices can fire while an unrelated LLM call still streams. A + /// subscriber that tracks a call-boundary marker (a committed-length + /// offset, or a tracked segment) MUST NOT move that marker for a + /// whose is false. + /// A later then finds nothing left to + /// remove, and the live call's dead partial text glues onto the next + /// call's answer. + /// /// - public sealed record TextOutput(string Text) : SessionOutput; + public sealed record TextOutput(string Text) : SessionOutput + { + /// + /// True when this marks the end of an LLM + /// call — the normal case. False only for the mid-stream notices + /// described above. The default is true, so every current emitter + /// keeps the call-boundary contract unchanged. + /// + public bool IsCallBoundary { get; init; } = true; + } /// /// Incremental text delta from the assistant while a turn is streaming. @@ -48,6 +76,30 @@ public sealed record TextOutput(string Text) : SessionOutput; /// public sealed record TextDeltaOutput(string Delta) : SessionOutput; + /// + /// A timed-out LLM call was discarded. The actor re-issues the same call + /// (see LlmSessionActor.TryResumeAfterTimeout). The dead call may + /// have already sent one or more events for + /// this turn. + /// + /// A subscriber that accumulates text across + /// a turn MUST follow two rules on receipt of this message: + /// + /// + /// Clear the dead call's own, not-yet-committed text. + /// Keep text from an earlier call that already reached its own + /// completion boundary this turn (for example, a + /// tool-round preamble). + /// + /// The resumed call streams the full answer again, from the start. A + /// subscriber that keeps the dead call's partial text glues two unrelated + /// answers together. Lifecycle — always delivered regardless of + /// , so it reaches every buffering subscriber + /// even one that filters on instead of + /// . + /// + public sealed record TextStreamDiscarded : SessionOutput; + /// /// Thinking/reasoning tokens from the model (e.g., Claude extended thinking). /// Requires . @@ -134,6 +186,26 @@ public sealed record UsageOutput : SessionOutput /// Sourced from llama.cpp timings.predicted_per_second. /// public double? PredictedPerSecond { get; init; } + + /// + /// Estimated input tokens billed for LLM calls that timed out and were + /// discarded this turn (see LlmSessionActor.TryResumeAfterTimeout). + /// The provider never returns real usage for a call that times out before + /// completion, so this is the real, provider-reported input count from the + /// most recently completed call this session — an honest proxy, not a + /// fabricated figure. The provider likely still billed for the discarded + /// call's input even though it is excluded from + /// and above. Null when no call was discarded + /// this turn, or when no completed call has reported real usage yet this + /// session. + /// + public long? DiscardedResumeEstimatedInputTokens { get; init; } + + /// + /// Number of timed-out calls discarded and re-issued this turn. Null when + /// no resume happened this turn. + /// + public int? DiscardedResumeAttempts { get; init; } } /// diff --git a/src/Netclaw.Cli.Tests/Cli/DaemonClientMappingTests.cs b/src/Netclaw.Cli.Tests/Cli/DaemonClientMappingTests.cs index 2bf5926ee..bc25d3634 100644 --- a/src/Netclaw.Cli.Tests/Cli/DaemonClientMappingTests.cs +++ b/src/Netclaw.Cli.Tests/Cli/DaemonClientMappingTests.cs @@ -213,6 +213,49 @@ public void SubAgentOutput_roundtrips_through_dto_completed() Assert.Equal(2, result.FindingsCount); } + [Fact] + public void TextOutput_roundtrips_IsCallBoundary_false_through_dto() + { + // F2: the notice emitters (EmitExpiredPromptNotice and siblings) set + // IsCallBoundary = false on the wire so a resuming daemon client + // doesn't treat the notice as the live call's completion. + var original = new TextOutput("That approval prompt has expired.") + { + SessionId = new SessionId("signalr/test"), + TimestampMs = 900, + IsCallBoundary = false + }; + + var dto = SessionOutputDtoMapper.ToDto(original); + Assert.Equal(SessionOutputTypes.Text, dto.Type); + Assert.False(dto.IsCallBoundary); + + var roundTripped = DaemonClient.FromDto(dto); + var result = Assert.IsType(roundTripped); + Assert.False(result.IsCallBoundary); + Assert.Equal("That approval prompt has expired.", result.Text); + } + + [Fact] + public void TextOutput_FromDto_defaults_IsCallBoundary_true_for_older_payloads_without_the_field() + { + // A daemon running an older version never sends IsCallBoundary on the + // wire — the field deserializes as null. It must default to true so + // every pre-existing TextOutput keeps the call-boundary contract. + var dto = new SessionOutputDto + { + Type = SessionOutputTypes.Text, + SessionId = "signalr/test", + TimestampMs = 901, + Text = "Full answer.", + IsCallBoundary = null + }; + + var output = DaemonClient.FromDto(dto); + var result = Assert.IsType(output); + Assert.True(result.IsCallBoundary); + } + [Theory] [InlineData(ErrorCategory.ToolFailure)] [InlineData(ErrorCategory.ProviderFailure)] @@ -263,6 +306,48 @@ public void BufferFlush_roundtrips_through_dto() Assert.Equal(777, result.TimestampMs); } + [Fact] + public void TextStreamDiscarded_roundtrips_through_dto() + { + var original = new TextStreamDiscarded + { + SessionId = new SessionId("signalr/test"), + TimestampMs = 779 + }; + + var dto = SessionOutputDtoMapper.ToDto(original); + Assert.Equal(SessionOutputTypes.TextStreamDiscarded, dto.Type); + Assert.Equal("signalr/test", dto.SessionId); + Assert.Equal(779, dto.TimestampMs); + + var roundTripped = DaemonClient.FromDto(dto); + var result = Assert.IsType(roundTripped); + Assert.Equal("signalr/test", result.SessionId.Value); + Assert.Equal(779, result.TimestampMs); + } + + [Fact] + public void UsageOutput_roundtrips_discarded_resume_fields_through_dto() + { + var original = new UsageOutput + { + SessionId = new SessionId("signalr/test"), + TimestampMs = 780, + InputTokens = 100, + DiscardedResumeEstimatedInputTokens = 42, + DiscardedResumeAttempts = 2 + }; + + var dto = SessionOutputDtoMapper.ToDto(original); + Assert.Equal(42, dto.DiscardedResumeEstimatedInputTokens); + Assert.Equal(2, dto.DiscardedResumeAttempts); + + var roundTripped = DaemonClient.FromDto(dto); + var result = Assert.IsType(roundTripped); + Assert.Equal(42, result.DiscardedResumeEstimatedInputTokens); + Assert.Equal(2, result.DiscardedResumeAttempts); + } + [Fact] public void ProcessingStateOutput_roundtrips_through_dto() { diff --git a/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs b/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs index f4c25dbe5..b683db16c 100644 --- a/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs +++ b/src/Netclaw.Cli.Tests/Doctor/ConfigSchemaDoctorCheckTests.cs @@ -618,6 +618,31 @@ await File.WriteAllTextAsync(paths.NetclawConfigPath, Assert.Contains("MaxToolCallsPerTurn", result.Message, StringComparison.OrdinalIgnoreCase); } + [Fact] + public async Task ReturnsPass_WhenSessionTuningTimeoutResumeRetryBudgetSet() + { + var basePath = CreateTempBasePath(); + var paths = new NetclawPaths(basePath); + paths.EnsureDirectoriesExist(); + + await File.WriteAllTextAsync(paths.NetclawConfigPath, + """ + { + "configVersion": 1, + "Session": { + "Tuning": { + "TimeoutResumeRetryBudget": 3 + } + } + } + """, TestContext.Current.CancellationToken); + + var check = new ConfigSchemaDoctorCheck(paths); + var result = await check.RunAsync(TestContext.Current.CancellationToken); + + Assert.Equal(DoctorSeverity.Pass, result.Severity); + } + private static string CreateTempBasePath() { var path = Path.Combine(Path.GetTempPath(), "netclaw-tests", Guid.NewGuid().ToString("N")); diff --git a/src/Netclaw.Cli.Tests/HeadlessChannelTests.cs b/src/Netclaw.Cli.Tests/HeadlessChannelTests.cs new file mode 100644 index 000000000..51f8d4e0b --- /dev/null +++ b/src/Netclaw.Cli.Tests/HeadlessChannelTests.cs @@ -0,0 +1,271 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Actors.Protocol; +using Netclaw.Cli.Daemon; +using Netclaw.Configuration; +using Xunit; +using static Netclaw.Actors.Sessions.SessionProtocol; + +namespace Netclaw.Cli.Tests; + +/// +/// Covers C1: the discard-and-resume mechanism (LlmSessionActor.TryResumeAfterTimeout) +/// emits before a resumed call streams its own +/// deltas. accumulates +/// text into its JSON envelope response buffer — without honoring the discard +/// signal, a dead call's partial text glues onto the resumed call's answer. These +/// tests drive directly (an internal +/// test seam) with a real multi-delta stall, then assert on the DELTA-accumulated +/// buffer, not . +/// +public sealed class HeadlessChannelTests +{ + private static HeadlessChannel CreateChannel(bool jsonOutput) => new( + new DaemonClient("http://127.0.0.1:1"), // never dialed in this test + new NetclawPaths(), + new FakeApplicationLifetime(), + TimeProvider.System, + new HeadlessOptions("test prompt") { JsonOutput = jsonOutput }, + NullLogger.Instance); + + [Fact] + public void TextStreamDiscarded_clears_json_envelope_buffer_between_dead_and_resumed_deltas() + { + var channel = CreateChannel(jsonOutput: true); + var sessionId = new SessionId("headless/test"); + + // Real multi-delta stall — two substantive deltas before discard, matching + // a genuine half-open provider stream (a single delta would not exercise + // the buffered-first-delta path the way a real stall does). + channel.HandleOutput(new TextDeltaOutput("stalled chunk one ") { SessionId = sessionId }, null); + channel.HandleOutput(new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = sessionId }, null); + + channel.HandleOutput(new TextStreamDiscarded { SessionId = sessionId }, null); + + channel.HandleOutput(new TextDeltaOutput("Resumed answer ") { SessionId = sessionId }, null); + channel.HandleOutput(new TextDeltaOutput("after timeout") { SessionId = sessionId }, null); + + // The JSON envelope's Response field is built from this buffer — it must + // contain ONLY the resumed call's text. + Assert.Equal("Resumed answer after timeout", channel.ResponseBufferForTesting); + Assert.DoesNotContain("STALLED_PARTIAL_MARKER", channel.ResponseBufferForTesting, StringComparison.Ordinal); + } + + [Fact] + public void TextStreamDiscarded_is_a_no_op_when_no_deltas_streamed_yet() + { + var channel = CreateChannel(jsonOutput: true); + var sessionId = new SessionId("headless/test-empty"); + + channel.HandleOutput(new TextStreamDiscarded { SessionId = sessionId }, null); + channel.HandleOutput(new TextDeltaOutput("first answer") { SessionId = sessionId }, null); + + Assert.Equal("first answer", channel.ResponseBufferForTesting); + } + + [Fact] + public void TextStreamDiscarded_preserves_an_earlier_completed_calls_text_but_discards_only_the_dead_calls_partial() + { + // D1: the JSON envelope buffer was TURN-scoped, so TextStreamDiscarded's + // Clear() wiped an earlier COMPLETED call's already-delivered text along + // with the dead call's partial. This mirrors ChatPage's call-scoped + // segment semantics: TextOutput marks a call boundary and commits + // everything before it, so a later discard can only ever remove text + // from the call that is actually dying. + var channel = CreateChannel(jsonOutput: true); + var sessionId = new SessionId("headless/preamble-preserved"); + + // Call 1: streams a preamble, then completes (a tool round). + channel.HandleOutput(new TextDeltaOutput("Checking the files now. ") { SessionId = sessionId }, null); + channel.HandleOutput(new TextOutput("Checking the files now. ") { SessionId = sessionId }, null); + + // Call 2: streams two real deltas, then dies mid-stream and is discarded. + channel.HandleOutput(new TextDeltaOutput("stalled chunk one ") { SessionId = sessionId }, null); + channel.HandleOutput(new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = sessionId }, null); + channel.HandleOutput(new TextStreamDiscarded { SessionId = sessionId }, null); + + // Resumed call streams the real final answer. + channel.HandleOutput(new TextDeltaOutput("Done: the answer is X.") { SessionId = sessionId }, null); + + Assert.Equal("Checking the files now. Done: the answer is X.", channel.ResponseBufferForTesting); + Assert.DoesNotContain("STALLED_PARTIAL_MARKER", channel.ResponseBufferForTesting, StringComparison.Ordinal); + } + + [Fact] + public void TextStreamDiscarded_lets_TextOutput_repopulate_after_discard() + { + // D2: the discard arm cleared _responseBuffer but left + // _receivedTextDeltaInCurrentTurn true, so a resumed call that never + // streams a delta (SessionLlmInvoker withholds the first delta until a + // second arrives — a single-chunk response never gets a second) hit the + // "already streamed this turn" branch of the TextOutput case and skipped + // the append entirely. A successful turn then reported an EMPTY response + // — worse than the pre-fix loud failure. + var channel = CreateChannel(jsonOutput: true); + var sessionId = new SessionId("headless/single-chunk-resume"); + + channel.HandleOutput(new TextDeltaOutput("stalled") { SessionId = sessionId }, null); + channel.HandleOutput(new TextStreamDiscarded { SessionId = sessionId }, null); + + // No further deltas — the resumed call's answer arrives as a single, + // non-streamed TextOutput. + channel.HandleOutput(new TextOutput("Resumed answer") { SessionId = sessionId }, null); + + Assert.Equal("Resumed answer", channel.ResponseBufferForTesting); + } + + [Fact] + public void TextOutput_with_IsCallBoundary_false_does_not_move_the_commit_marker_past_a_live_calls_partial_text() + { + // F2: EmitExpiredPromptNotice/EmitWrongRequesterApprovalNotice/ + // EmitUnavailableApprovalOptionNotice send a mid-stream TextOutput + // (IsCallBoundary = false) while another call still streams. + // Before the fix, ANY TextOutput advanced the commit marker over the + // live call's partial text; a subsequent stall+discard then found + // nothing left to remove, and the resumed call's answer glued onto + // the dead partial. + var channel = CreateChannel(jsonOutput: true); + var sessionId = new SessionId("headless/notice-mid-stream"); + + // Call: streams two real deltas. + channel.HandleOutput(new TextDeltaOutput("stalled chunk one ") { SessionId = sessionId }, null); + channel.HandleOutput(new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = sessionId }, null); + + // A notice fires mid-stream — a non-call-boundary TextOutput. Its own + // text is handled exactly as today (logged only, since a delta is + // already in flight); only its effect on the commit marker changes. + channel.HandleOutput(new TextOutput("That approval prompt has expired.") + { + SessionId = sessionId, + IsCallBoundary = false + }, null); + + // The live call then dies and is discarded. + channel.HandleOutput(new TextStreamDiscarded { SessionId = sessionId }, null); + + // Resumed call streams the real final answer. + channel.HandleOutput(new TextDeltaOutput("Done: the answer is X.") { SessionId = sessionId }, null); + + Assert.Equal("Done: the answer is X.", channel.ResponseBufferForTesting); + Assert.DoesNotContain("STALLED_PARTIAL_MARKER", channel.ResponseBufferForTesting, StringComparison.Ordinal); + } + + [Fact] + public void UsageOutput_starts_on_its_own_line_after_a_streamed_turn() + { + // F1 regression: the "any text printed this turn" flag was narrowed to + // a CALL-scoped flag that the final TextOutput always resets before + // UsageOutput arrives (the actor always emits text then usage for a + // completed call), so the newline guard never fired and [usage] glued + // onto the end of the streamed answer with no line break between them. + var channel = CreateChannel(jsonOutput: false); + var sessionId = new SessionId("headless/usage-newline"); + + using var stdout = new StringWriter(); + var originalOut = Console.Out; + Console.SetOut(stdout); + try + { + channel.HandleOutput(new TextDeltaOutput("Hello ") { SessionId = sessionId }, null); + channel.HandleOutput(new TextDeltaOutput("world.") { SessionId = sessionId }, null); + channel.HandleOutput(new TextOutput("Hello world.") { SessionId = sessionId }, null); + channel.HandleOutput(new UsageOutput + { + SessionId = sessionId, + InputTokens = 10, + OutputTokens = 5, + TotalTokens = 15 + }, null); + } + finally + { + Console.SetOut(originalOut); + } + + var output = stdout.ToString().Replace("\r\n", "\n", StringComparison.Ordinal); + Assert.DoesNotContain("world.[usage]", output, StringComparison.Ordinal); + Assert.Contains("\n[usage]", output, StringComparison.Ordinal); + } + + [Fact] + public void UsageOutput_omits_discarded_est_in_when_null_but_keeps_discarded_attempts() + { + // F3: a resume can happen without any completed call in the session + // ever reporting real usage (see D4 in LlmSessionActor) — the + // estimate is then null. The console line must omit discarded_est_in= + // entirely rather than print it as an empty token, while + // discarded_attempts= (a real, always-known count) still prints. + var channel = CreateChannel(jsonOutput: false); + var sessionId = new SessionId("headless/discarded-null-estimate"); + + using var stdout = new StringWriter(); + var originalOut = Console.Out; + Console.SetOut(stdout); + try + { + channel.HandleOutput(new UsageOutput + { + SessionId = sessionId, + InputTokens = 10, + OutputTokens = 5, + TotalTokens = 15, + DiscardedResumeEstimatedInputTokens = null, + DiscardedResumeAttempts = 1 + }, null); + } + finally + { + Console.SetOut(originalOut); + } + + var output = stdout.ToString(); + Assert.DoesNotContain("discarded_est_in=", output, StringComparison.Ordinal); + Assert.Contains("discarded_attempts=1", output, StringComparison.Ordinal); + } + + [Fact] + public void UsageOutput_prints_discarded_est_in_when_a_real_estimate_is_available() + { + // Companion to the null case above: once an earlier call in the + // session has reported real usage, the estimate must still print. + var channel = CreateChannel(jsonOutput: false); + var sessionId = new SessionId("headless/discarded-real-estimate"); + + using var stdout = new StringWriter(); + var originalOut = Console.Out; + Console.SetOut(stdout); + try + { + channel.HandleOutput(new UsageOutput + { + SessionId = sessionId, + InputTokens = 10, + OutputTokens = 5, + TotalTokens = 15, + DiscardedResumeEstimatedInputTokens = 42, + DiscardedResumeAttempts = 2 + }, null); + } + finally + { + Console.SetOut(originalOut); + } + + var output = stdout.ToString(); + Assert.Contains("discarded_est_in=42 discarded_attempts=2", output, StringComparison.Ordinal); + } + + private sealed class FakeApplicationLifetime : IHostApplicationLifetime + { + public CancellationToken ApplicationStarted => CancellationToken.None; + public CancellationToken ApplicationStopping => CancellationToken.None; + public CancellationToken ApplicationStopped => CancellationToken.None; + + public void StopApplication() { } + } +} diff --git a/src/Netclaw.Cli.Tests/Tui/ChatPageTests.cs b/src/Netclaw.Cli.Tests/Tui/ChatPageTests.cs index eb6fcc6db..cb158f2ea 100644 --- a/src/Netclaw.Cli.Tests/Tui/ChatPageTests.cs +++ b/src/Netclaw.Cli.Tests/Tui/ChatPageTests.cs @@ -251,6 +251,94 @@ public async Task Escape_WithNoPendingInteraction_IsNoOp_CtrlQQuits() Assert.Equal(new[] { "shutdown" }, vm.LifecycleEvents); } + [Fact] + public async Task TextStreamDiscarded_ClearsDeadCallTextBeforeResumedDeltasRender() + { + // Covers C1: a timed-out LLM call is discarded and re-issued + // (LlmSessionActor.TryResumeAfterTimeout). The dead call already streamed + // TextDeltaOutput events for this turn — TextStreamDiscarded must clear + // ChatPage's assistant text buffer before the resumed call's deltas + // render, or the two answers glue into one corrupted reply. Real + // multi-delta stall (two deltas) before the discard, matching a genuine + // half-open provider stream. + var testSessionId = new SessionId("chat-page/resume-test"); + var outputSequence = new SessionOutput[] + { + new TextDeltaOutput("stalled chunk one ") { SessionId = testSessionId }, + new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = testSessionId }, + new TextStreamDiscarded { SessionId = testSessionId }, + new TextDeltaOutput("Resumed answer ") { SessionId = testSessionId }, + new TextDeltaOutput("after timeout") { SessionId = testSessionId }, + }; + + var (terminal, app, _) = CreateHeadlessApp(seed: null, out var input, outputSequence: outputSequence); + input.EnqueueKey(ConsoleKey.Q, false, false, true); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await app.RunAsync(cts.Token); + + var screen = terminal.ToString(); + + // The resumed call's answer must render as its OWN clean segment. + Assert.True(screen.Contains("Resumed answer after timeout", StringComparison.Ordinal), + $"Expected the resumed call's text on screen. Screen:\n{terminal}"); + + // ChatPage finalizes the dead call's segment as "interrupted" (kept + // visible for the user's record) rather than deleting it outright, so + // the marker itself may still appear — what must NEVER happen is the + // dead call's text gluing directly onto the resumed answer as one + // continuous string (the exact corruption C1 found). + Assert.False(screen.Contains("STALLED_PARTIAL_MARKERResumed", StringComparison.Ordinal), + $"Expected the dead call's partial text to NOT be glued directly onto the resumed answer. Screen:\n{terminal}"); + + // The dead call's segment must be marked as interrupted, not presented + // as if it were itself a completed, trustworthy answer. + Assert.True(screen.Contains("interrupted", StringComparison.Ordinal), + $"Expected the dead call's segment to be marked as interrupted. Screen:\n{terminal}"); + } + + [Fact] + public async Task NoticeTextOutput_DuringLiveStream_DoesNotFinalizeOrCorruptTheLiveSegment() + { + // F2: EmitExpiredPromptNotice/EmitWrongRequesterApprovalNotice/ + // EmitUnavailableApprovalOptionNotice send a mid-stream TextOutput + // (IsCallBoundary = false) while another call still streams. + // Before the fix, ANY TextOutput finalized the live assistant segment + // early — rendering the dead call's partial text as if it were the + // complete answer and untracking the segment, so the later discard + // found nothing to mark interrupted and the resumed call's deltas + // started a brand-new segment instead of replacing the corrupted one. + var testSessionId = new SessionId("chat-page/notice-mid-stream"); + var outputSequence = new SessionOutput[] + { + new TextDeltaOutput("stalled chunk one ") { SessionId = testSessionId }, + new TextDeltaOutput("STALLED_PARTIAL_MARKER") { SessionId = testSessionId }, + new TextOutput("That approval prompt has expired.") + { + SessionId = testSessionId, + IsCallBoundary = false + }, + new TextStreamDiscarded { SessionId = testSessionId }, + new TextDeltaOutput("Resumed answer ") { SessionId = testSessionId }, + new TextDeltaOutput("after timeout") { SessionId = testSessionId }, + }; + + var (terminal, app, _) = CreateHeadlessApp(seed: null, out var input, outputSequence: outputSequence); + input.EnqueueKey(ConsoleKey.Q, false, false, true); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await app.RunAsync(cts.Token); + + var screen = terminal.ToString(); + + Assert.True(screen.Contains("Resumed answer after timeout", StringComparison.Ordinal), + $"Expected the resumed call's text on screen. Screen:\n{terminal}"); + Assert.False(screen.Contains("STALLED_PARTIAL_MARKERResumed", StringComparison.Ordinal), + $"Expected the dead call's partial text to NOT be glued directly onto the resumed answer. Screen:\n{terminal}"); + Assert.True(screen.Contains("interrupted", StringComparison.Ordinal), + $"Expected the dead call's segment to still be marked as interrupted after the mid-stream notice. Screen:\n{terminal}"); + } + [Fact] public async Task DenyPendingInteraction_NoDenyOption_DoesNotSubmit() { @@ -579,7 +667,8 @@ private static (VirtualTerminal Terminal, TerminaApplication App, TestChatViewMo CreateHeadlessApp(ToolInteractionRequest? seed, out VirtualInputSource input, int width = 120, int height = 40, IReadOnlyList? options = null, - bool startGenerating = false) + bool startGenerating = false, + IReadOnlyList? outputSequence = null) { var terminal = new VirtualTerminal(width, height); var virtualInput = new VirtualInputSource(); @@ -602,7 +691,7 @@ private static (VirtualTerminal Terminal, TerminaApplication App, TestChatViewMo : options is null ? seed : seed with { Options = options }; - capturedVm = new TestChatViewModel(effectiveSeed, startGenerating); + capturedVm = new TestChatViewModel(effectiveSeed, startGenerating, outputSequence); return capturedVm; }); }); @@ -624,6 +713,7 @@ private sealed class TestChatViewModel : ChatViewModel { private readonly ToolInteractionRequest? _seed; private readonly bool _startGenerating; + private readonly IReadOnlyList? _outputSequence; /// /// Set when the page routes Escape to app shutdown (the pre-#1757 @@ -647,7 +737,8 @@ private sealed class TestChatViewModel : ChatViewModel /// public string? LastSubmittedInteractionKey { get; private set; } - public TestChatViewModel(ToolInteractionRequest? seed, bool startGenerating = false) + public TestChatViewModel(ToolInteractionRequest? seed, bool startGenerating = false, + IReadOnlyList? outputSequence = null) : base( // 127.0.0.1:1 is never dialed: InitializeSessionAsync is // overridden to no-op, so the underlying HubConnection stays @@ -661,6 +752,7 @@ public TestChatViewModel(ToolInteractionRequest? seed, bool startGenerating = fa { _seed = seed; _startGenerating = startGenerating; + _outputSequence = outputSequence; } protected override Task InitializeSessionAsync() => Task.CompletedTask; @@ -675,6 +767,14 @@ public override void OnActivated() // clears it, but the UI thread can observe the pre-clear value). if (_startGenerating) IsGenerating.Value = true; + // Replays a scripted SessionOutput sequence as if the daemon had + // emitted it live — lets tests drive ChatPage's streaming/text + // rendering (e.g. the timeout-resume discard signal) headlessly. + if (_outputSequence is not null) + { + foreach (var output in _outputSequence) + EmitOutputForTesting(output); + } } public override void RequestAppShutdown() diff --git a/src/Netclaw.Cli/HeadlessChannel.cs b/src/Netclaw.Cli/HeadlessChannel.cs index 5e80a2885..7a756080d 100644 --- a/src/Netclaw.Cli/HeadlessChannel.cs +++ b/src/Netclaw.Cli/HeadlessChannel.cs @@ -36,11 +36,33 @@ public sealed class HeadlessChannel : IChannel private readonly ILogger _logger; private bool _isConnected; + + // Whether the CURRENT (not-yet-completed) LLM call has streamed a delta. + // Reset at every call boundary — a TextOutput (call completed) or a + // TextStreamDiscarded (call died) — so it never leaks across calls in the + // same turn. Drives the buffer/commit logic in the TextOutput, + // TextDeltaOutput, and TextStreamDiscarded cases below. + private bool _receivedTextDeltaInCurrentCall; + + // True when any call in this turn streamed a delta. The first delta sets + // it. Only TurnCompleted clears it — a call boundary (TextOutput) does + // NOT reset it, unlike the call-scoped flag above. The pre-[usage] + // newline guard uses this flag only. The actor always sends a call's + // final TextOutput before UsageOutput, and that TextOutput resets the + // call-scoped flag. So only the turn-scoped flag can still tell + // UsageOutput whether the console cursor sits mid-line. private bool _receivedTextDeltaInCurrentTurn; private bool _receivedThinkingDeltaInCurrentTurn; // JSON output accumulation private readonly StringBuilder _responseBuffer = new(); + + // Length of _responseBuffer already committed by an earlier call's + // TextOutput this turn. TextStreamDiscarded truncates back to this point + // instead of clearing the whole buffer, so a later call's discard cannot + // erase an earlier COMPLETED call's text (see the TextStreamDiscarded case + // in HandleOutput). + private int _responseBufferCommittedLength; private readonly List _toolCalls = []; private JsonUsage? _usage; private string? _resolvedSessionId; @@ -52,6 +74,13 @@ public sealed class HeadlessChannel : IChannel public Actors.Channels.ChannelType ChannelType => Actors.Channels.ChannelType.Headless; public string DisplayName => "Headless Prompt"; + /// + /// Test seam: the accumulated JSON envelope response buffer. Lets + /// Netclaw.Cli.Tests assert on the delta-accumulated result of a + /// sequence of calls without parsing stdout. + /// + internal string ResponseBufferForTesting => _responseBuffer.ToString(); + public ValueTask GetHealthAsync(CancellationToken cancellationToken = default) { var health = _isConnected @@ -171,7 +200,12 @@ private async Task RunHeadlessAsync(CancellationToken stopping) } } - private void HandleOutput(SessionOutput output, StreamWriter? log) + /// + /// Dispatches one from the live daemon subscription. + /// Internal (not private) so Netclaw.Cli.Tests can drive it directly + /// without standing up a daemon connection. + /// + internal void HandleOutput(SessionOutput output, StreamWriter? log) { switch (output) { @@ -180,22 +214,39 @@ private void HandleOutput(SessionOutput output, StreamWriter? log) break; case TextOutput msg: - if (_receivedTextDeltaInCurrentTurn) + // TextOutput marks one LLM call's text as complete, whether that + // call ended in tool calls (a preamble) or the final answer. Commit + // everything accumulated for it — via deltas, or (when the call + // never streamed one, e.g. a single-chunk response) via msg.Text + // directly — so a LATER call's discard can never erase it. Some + // notices (approval-expired etc.) reuse TextOutput to reach the + // console while an earlier call still streams. IsCallBoundary + // is false for those. They must not move the commit marker or + // clear the live call's delta flag (see + // SessionProtocol.TextOutput.IsCallBoundary). + if (_receivedTextDeltaInCurrentCall) { Log(log, $"ASSISTANT_FINAL: {msg.Text}"); - break; } - - if (_jsonOutput) - _responseBuffer.Append(msg.Text); else - Console.WriteLine(msg.Text); - Log(log, $"ASSISTANT: {msg.Text}"); + { + if (_jsonOutput) + _responseBuffer.Append(msg.Text); + else + Console.WriteLine(msg.Text); + Log(log, $"ASSISTANT: {msg.Text}"); + } + if (msg.IsCallBoundary) + { + _responseBufferCommittedLength = _responseBuffer.Length; + _receivedTextDeltaInCurrentCall = false; + } break; case TextDeltaOutput msg: - if (!_receivedTextDeltaInCurrentTurn && _promptSentTicks > 0) + if (!_receivedTextDeltaInCurrentCall && _promptSentTicks > 0) Interlocked.CompareExchange(ref _firstDeltaTicks, Stopwatch.GetTimestamp(), 0); + _receivedTextDeltaInCurrentCall = true; _receivedTextDeltaInCurrentTurn = true; if (_jsonOutput) _responseBuffer.Append(msg.Delta); @@ -204,6 +255,28 @@ private void HandleOutput(SessionOutput output, StreamWriter? log) Log(log, $"ASSISTANT_DELTA: {msg.Delta}"); break; + case TextStreamDiscarded: + // A timed-out call was discarded. The actor re-issues it. Truncate + // the JSON envelope buffer back to the last committed call boundary + // — only the dead call's own, not-yet-committed text is removed. + // Text from an earlier call that already completed this turn (see + // the TextOutput case above) survives (see + // SessionProtocol.TextStreamDiscarded). A plain-text console stream + // cannot un-print what is already on screen, so mark the boundary + // instead — otherwise the two answers would read as one. + if (_jsonOutput) + { + _responseBuffer.Remove(_responseBufferCommittedLength, _responseBuffer.Length - _responseBufferCommittedLength); + } + else if (_receivedTextDeltaInCurrentCall) + { + Console.WriteLine(); + Console.WriteLine("[response interrupted by a provider stall — retrying]"); + } + _receivedTextDeltaInCurrentCall = false; + Log(log, "ASSISTANT_STREAM_DISCARDED"); + break; + case ThinkingOutput msg: if (_receivedThinkingDeltaInCurrentTurn) { @@ -244,6 +317,22 @@ private void HandleOutput(SessionOutput output, StreamWriter? log) break; case UsageOutput msg: + // discarded_est_in=/discarded_attempts= are the previous completed + // call's real, provider-reported input count, used as an honest + // proxy for a call that timed out and was discarded this turn — the + // provider likely billed for this input but never reported usage + // for it, so it is NOT included in the in=/total= figures. The whole + // suffix is omitted entirely (not printed as empty) when no resume + // happened this turn. discarded_est_in= alone drops when a resume + // happens but no completed call in this session reports a real + // estimate yet (see D4 in LlmSessionActor). discarded_attempts= + // still prints — that count is always real. + var discardedSuffix = msg.DiscardedResumeAttempts is > 0 + ? msg.DiscardedResumeEstimatedInputTokens is { } estimatedInputTokens + ? $" discarded_est_in={estimatedInputTokens} discarded_attempts={msg.DiscardedResumeAttempts}" + : $" discarded_attempts={msg.DiscardedResumeAttempts}" + : string.Empty; + if (_jsonOutput) { _usage = new JsonUsage @@ -255,6 +344,8 @@ private void HandleOutput(SessionOutput output, StreamWriter? log) ReasoningTokens = msg.ReasoningTokens, PromptMs = msg.PromptMs, PredictedPerSecond = msg.PredictedPerSecond, + DiscardedResumeEstimatedInputTokens = msg.DiscardedResumeEstimatedInputTokens, + DiscardedResumeAttempts = msg.DiscardedResumeAttempts, }; } else @@ -262,11 +353,15 @@ private void HandleOutput(SessionOutput output, StreamWriter? log) // If the turn streamed text deltas, they did NOT end with a newline // (each delta is Console.Write). Force the usage line onto its own // line so downstream parsers (evals, humans) can anchor on ^[usage]. + // Turn-scoped, not call-scoped: TextOutput (the call's completion + // marker) always arrives before UsageOutput and resets the + // call-scoped flag. Only the turn-scoped flag still shows + // whether the console cursor sits mid-line. if (_receivedTextDeltaInCurrentTurn) Console.WriteLine(); - Console.WriteLine($"[usage] in={msg.InputTokens} out={msg.OutputTokens} total={msg.TotalTokens} cached={msg.CachedInputTokens} prompt_ms={msg.PromptMs} tok_s={msg.PredictedPerSecond}"); + Console.WriteLine($"[usage] in={msg.InputTokens} out={msg.OutputTokens} total={msg.TotalTokens} cached={msg.CachedInputTokens} prompt_ms={msg.PromptMs} tok_s={msg.PredictedPerSecond}{discardedSuffix}"); } - Log(log, $"USAGE: in={msg.InputTokens} out={msg.OutputTokens} total={msg.TotalTokens} cached={msg.CachedInputTokens} reasoning={msg.ReasoningTokens} context_window={msg.ContextWindowTokens} prompt_ms={msg.PromptMs} predicted_tok_s={msg.PredictedPerSecond}"); + Log(log, $"USAGE: in={msg.InputTokens} out={msg.OutputTokens} total={msg.TotalTokens} cached={msg.CachedInputTokens} reasoning={msg.ReasoningTokens} context_window={msg.ContextWindowTokens} prompt_ms={msg.PromptMs} predicted_tok_s={msg.PredictedPerSecond}{discardedSuffix}"); break; case ErrorOutput msg: @@ -287,8 +382,10 @@ private void HandleOutput(SessionOutput output, StreamWriter? log) } Log(log, $"TURN_COMPLETED: turn={msg.TurnNumber}"); Log(log, "SESSION_ENDED"); + _receivedTextDeltaInCurrentCall = false; _receivedTextDeltaInCurrentTurn = false; _receivedThinkingDeltaInCurrentTurn = false; + _responseBufferCommittedLength = 0; break; case FileOutput msg: @@ -394,5 +491,7 @@ private sealed class JsonUsage public long? ReasoningTokens { get; init; } public double? PromptMs { get; init; } public double? PredictedPerSecond { get; init; } + public long? DiscardedResumeEstimatedInputTokens { get; init; } + public int? DiscardedResumeAttempts { get; init; } } } diff --git a/src/Netclaw.Cli/Tui/ChatPage.cs b/src/Netclaw.Cli/Tui/ChatPage.cs index 5baeffc6f..ca8f0df86 100644 --- a/src/Netclaw.Cli/Tui/ChatPage.cs +++ b/src/Netclaw.Cli/Tui/ChatPage.cs @@ -397,8 +397,17 @@ private void HandleOutput(SessionOutput output) // final full snapshot for compatibility. Finalize without duplicating. if (_assistantSegmentId.Value != 0) { - FinalizeAssistantSegmentIfNeeded(); - _chatHistory.ScrollToBottom(); + // Some notices (approval-expired etc.) reuse TextOutput to reach + // the chat while an earlier call still streams. IsCallBoundary + // is false for those. They must not finalize (and untrack) the + // live segment — the call keeps streaming, and its own + // boundary TextOutput closes the segment later (see + // SessionProtocol.TextOutput.IsCallBoundary). + if (msg.IsCallBoundary) + { + FinalizeAssistantSegmentIfNeeded(); + _chatHistory.ScrollToBottom(); + } break; } @@ -417,6 +426,25 @@ private void HandleOutput(SessionOutput output) _chatHistory.ScrollToBottom(); break; + case TextStreamDiscarded: + // A timed-out call was discarded. The actor re-issues it. Finalize + // the dead call's partial segment as interrupted (in place, so the + // history keeps a record of it) and untrack it — the next + // TextDeltaOutput from the resumed call starts a fresh segment, so + // the two answers never render as one (see + // SessionProtocol.TextStreamDiscarded). + if (_assistantSegmentId.Value != 0) + { + _chatHistory.Replace(_assistantSegmentId, + new StaticTextSegment($"Netclaw: {_assistantBuffer} [interrupted — retrying]", Color.BrightBlack), + keepTracked: false); + _assistantSegmentId = default; + _assistantBuffer.Clear(); + _chatHistory.AppendLine(""); + _chatHistory.ScrollToBottom(); + } + break; + case ThinkingOutput: // Hidden — reasoning output is too verbose for the chat view. // TODO: collapsible thinking sections when Termina supports it. diff --git a/src/Netclaw.Cli/Tui/ChatViewModel.cs b/src/Netclaw.Cli/Tui/ChatViewModel.cs index a32a641a1..9e1135ffc 100644 --- a/src/Netclaw.Cli/Tui/ChatViewModel.cs +++ b/src/Netclaw.Cli/Tui/ChatViewModel.cs @@ -396,6 +396,14 @@ internal void SeedPendingInteractionForTesting(ToolInteractionRequest interactio RequestRedraw(); } + /// + /// Test seam: push one as if the daemon had + /// emitted it, without staging any approval-interaction bookkeeping. Used by + /// ChatPageTests to exercise streaming/text rendering (e.g. the + /// timeout-resume discard signal) without a daemon connection. + /// + internal void EmitOutputForTesting(SessionOutput output) => _outputSubject.OnNext(output); + /// /// Opens the per-session USAGE log file if not already open. Matches /// HeadlessChannel's filename and append semantics so a single session diff --git a/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs b/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs index c473a3e8c..77d097313 100644 --- a/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs +++ b/src/Netclaw.Configuration.Tests/SessionConfigDefaultsTests.cs @@ -51,6 +51,13 @@ public void Max_tool_iterations_per_turn_defaults_to_60() Assert.Equal(60, config.MaxToolIterationsPerTurn); } + [Fact] + public void Timeout_resume_retry_budget_defaults_to_2() + { + var tuning = new SessionTuning(); + Assert.Equal(2, tuning.TimeoutResumeRetryBudget); + } + [Fact] public void BindFromConfiguration_supports_legacy_root_level_tuning_keys() { @@ -87,4 +94,26 @@ public void BindFromConfiguration_prefers_nested_tuning_values_over_legacy_root_ Assert.Equal(0.8, bound.Tuning.CompactionThreshold); } + + /// + /// Cross-Boundary Contract Rule round-trip: the config-file shape a user would + /// write (validated against netclaw-config.v1.schema.json — see + /// ConfigSchemaDoctorCheckTests.ReturnsPass_WhenSessionTuningTimeoutResumeRetryBudgetSet) + /// must bind to the exact runtime value LlmSessionActor.TryResumeAfterTimeout + /// reads via SessionConfig.Tuning.TimeoutResumeRetryBudget. + /// + [Fact] + public void BindFromConfiguration_binds_TimeoutResumeRetryBudget() + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["Session:Tuning:TimeoutResumeRetryBudget"] = "5" + }) + .Build(); + + var bound = SessionConfig.BindFromConfiguration(config.GetSection("Session")); + + Assert.Equal(5, bound.Tuning.TimeoutResumeRetryBudget); + } } diff --git a/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json b/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json index a36c794dc..4fc2e84e8 100644 --- a/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json +++ b/src/Netclaw.Configuration/Schemas/netclaw-config.v1.schema.json @@ -263,6 +263,12 @@ "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." + }, + "TimeoutResumeRetryBudget": { + "type": "integer", + "minimum": 0, + "default": 2, + "description": "Per-turn budget for LLM call resumes after a mid-stream timeout. On a timeout, the session actor discards the dead call's partial output and re-issues the same call, up to this many times per turn, then fails the turn. Applies to any call in the turn, including one after an earlier tool iteration completed. Set to 0 to disable resume." } }, "additionalProperties": false diff --git a/src/Netclaw.Configuration/SessionTuning.cs b/src/Netclaw.Configuration/SessionTuning.cs index 09ac7c9e3..28b121b08 100644 --- a/src/Netclaw.Configuration/SessionTuning.cs +++ b/src/Netclaw.Configuration/SessionTuning.cs @@ -135,4 +135,18 @@ public sealed record SessionTuning /// are not retried because the partial response cannot be reconstructed. /// public RetryPolicy StreamingRetryPolicy { get; init; } = new(); + + /// + /// Per-turn budget for LLM call resumes after a mid-stream timeout. A + /// timeout is the watchdog's inter-delta or no-progress deadline, or a + /// transport . On a timeout, the session + /// actor discards the dead call's partial output and re-issues the same + /// call. This can happen on any call in the turn, including a call after + /// an earlier tool iteration completed — resume never dispatches a tool + /// call, so it cannot double-execute one. The actor allows up to this + /// many resumes per turn, then it fails the turn. Set the budget to 0 to + /// disable resume: the actor then fails the turn on the first timeout, + /// the same as before this feature existed. + /// + public int TimeoutResumeRetryBudget { get; init; } = 2; }