diff --git a/pkg/runtime/live_sessions_test.go b/pkg/runtime/live_sessions_test.go index 2baf59e08..be891d58e 100644 --- a/pkg/runtime/live_sessions_test.go +++ b/pkg/runtime/live_sessions_test.go @@ -328,9 +328,13 @@ func TestCompactLiveSession_ExecutesAtIterationBoundary(t *testing.T) { started := make(chan struct{}) release := make(chan struct{}) prov := &stepProvider{id: "test/mock-model", steps: []providerStep{ - // Turn 1: content without a finish reason keeps the loop running, - // so the queued request executes at the next iteration boundary. - {stream: newStreamBuilder().AddContent("working on it").Build(), started: started, release: release}, + // Turn 1: a tool call keeps the loop running (Stopped=false), so the + // queued request executes at the next iteration boundary. + {stream: newStreamBuilder(). + AddToolCallName("call_1", "unknown_tool"). + AddToolCallArguments("call_1", "{}"). + AddToolCallStopWithUsage(1, 1). + Build(), started: started, release: release}, // The compaction summary call. {stream: newStreamBuilder().AddContent("a compact summary").AddStopWithUsage(10, 5).Build()}, // Turn 2: natural stop ends the stream. @@ -446,14 +450,28 @@ func TestCompactLiveSession_DuplicateSessionIDsCompactOnlyTargetEntry(t *testing startedB := make(chan struct{}) releaseB := make(chan struct{}) prov := &stepProvider{id: "test/mock-model", steps: []providerStep{ - // Older stream turn 1: kept in flight while the newer stream - // registers under the same session ID. - {stream: newStreamBuilder().AddContent("older working").Build(), started: startedA, release: releaseA}, - // Newer stream turn 1, gated so it stays live throughout. - {stream: newStreamBuilder().AddContent("newer working").Build(), started: startedB, release: releaseB}, + // Older stream turn 1: a tool call keeps the stream live (the loop + // continues to execute the call) while the newer stream registers + // under the same session ID. A tool-call turn is used rather than a + // bare content turn so the older stream deterministically reaches a + // second model call regardless of the bare-EOF stop rule. + {stream: newStreamBuilder(). + AddToolCallName("call_older", "unknown_tool"). + AddToolCallArguments("call_older", "{}"). + AddToolCallStopWithUsage(1, 1). + Build(), started: startedA, release: releaseA}, + // Newer stream turn 1, gated so it stays live throughout. A tool-call + // turn is used (matching the older stream) so the newer stream reaches + // an iteration boundary where the pending compaction can run. + {stream: newStreamBuilder(). + AddToolCallName("call_newer", "unknown_tool"). + AddToolCallArguments("call_newer", "{}"). + AddToolCallStopWithUsage(1, 1). + Build(), started: startedB, release: releaseB}, // Older stream turn 2: natural stop. With the request left alone this - // is the older stream's next model call; stealing the request would - // consume this step as the compaction summary instead. + // is the older stream's next model call (after the tool result feeds + // back in); stealing the request would consume this step as the + // compaction summary instead. {stream: newStreamBuilder().AddStopWithUsage(1, 1).Build()}, // The compaction summary call, drained by the newer stream's own // iteration boundary. @@ -551,7 +569,14 @@ func TestCompactLiveSession_HookVetoSynthesizesSkipped(t *testing.T) { started := make(chan struct{}) release := make(chan struct{}) prov := &stepProvider{id: "test/mock-model", steps: []providerStep{ - {stream: newStreamBuilder().AddContent("working on it").Build(), started: started, release: release}, + // Turn 1: tool call keeps the loop running (Stopped=false) so the + // hook veto fires at the iteration boundary, not at teardown. + {stream: newStreamBuilder(). + AddToolCallName("call_1", "unknown_tool"). + AddToolCallArguments("call_1", "{}"). + AddToolCallStopWithUsage(1, 1). + Build(), started: started, release: release}, + // Turn 2: natural stop (model call after the tool result, compaction vetoed). {stream: newStreamBuilder().AddStopWithUsage(1, 1).Build()}, }} rt := newLiveSessionsRuntime(t, prov, mockModelStoreWithLimit{limit: 100_000}, diff --git a/pkg/runtime/streaming.go b/pkg/runtime/streaming.go index 4c07ba4c0..abd43bb02 100644 --- a/pkg/runtime/streaming.go +++ b/pkg/runtime/streaming.go @@ -359,14 +359,11 @@ mainLoop: applyXMLFallback() - // If the stream completed without producing any usable content or tool - // calls, likely because of a token limit, stop to avoid breaking the request - // loop. Whitespace-only content counts as no output: it carries no answer, - // and runTurn's empty-turn detection uses the same trimmed-empty test, so - // treating it as stopped here guarantees that an empty-turn warning is always - // followed by a turn exit rather than an identical-message re-entry (#3145). + // Invariant: a bare-EOF turn (no per-choice finish_reason) is terminal + // whenever there are no tool calls — the outer loop has nothing to continue + // on. Turns with tool calls keep Stopped=false so the loop executes them. // NOTE(krissetto): this can likely be removed once compaction works properly with all providers (aka dmr) - stoppedDueToNoOutput := strings.TrimSpace(fullContent.String()) == "" && len(toolCalls) == 0 + stoppedNoToolCalls := len(toolCalls) == 0 // Prefer the provider's explicit finish reason when available (e.g. // tool_calls). Only fall back to inference when no explicit reason was @@ -399,7 +396,7 @@ mainLoop: ReasoningContent: fullReasoningContent.String(), ThinkingSignature: thinkingSignature, ThoughtSignature: thoughtSignature, - Stopped: stoppedDueToNoOutput, + Stopped: stoppedNoToolCalls, FinishReason: finishReason, Usage: messageUsage, }, nil diff --git a/pkg/runtime/streaming_test.go b/pkg/runtime/streaming_test.go index f287faa3d..129aa66e7 100644 --- a/pkg/runtime/streaming_test.go +++ b/pkg/runtime/streaming_test.go @@ -190,6 +190,33 @@ func TestHandleStream_WhitespaceOnlyContentStops(t *testing.T) { "a whitespace-only, bare-EOF turn must stop so the empty-turn warning is followed by a turn exit, not an identical re-entry (#3145)") } +// TestHandleStream_ContentOnlyBareEOFStops guards the loop that OpenAI-compatible +// gateways (litellm/CBORG) trigger: they close the SSE stream with a bare EOF and +// never send a per-choice finish_reason. A turn that produced real content but no +// tool calls has nothing left for the run loop to execute, so it must report +// Stopped=true. Keying the stop decision on empty content instead of "no tool +// calls" left such a final message with Stopped=false, so runTurn re-entered the +// model with identical messages and re-emitted the same completion text forever. +func TestHandleStream_ContentOnlyBareEOFStops(t *testing.T) { + stream := newStreamBuilder(). + AddContent("Frontmatter ingestion complete."). // real content, no tool calls + Build() // no terminal chunk: bare EOF, no finish reason + + a := agent.New("root", "test", agent.WithModel(&mockProvider{id: "test/mock-model", stream: stream})) + sess := session.New(session.WithUserMessage("go")) + + evCh := make(chan Event, 64) + res, err := handleStream( + t.Context(), nil, stream, a, nil, sess, nil, + defaultTelemetry{}, NewChannelSink(evCh), defaultStreamIdleTimeout, + ) + require.NoError(t, err) + + assert.Empty(t, res.Calls) + assert.True(t, res.Stopped, + "a content-bearing, bare-EOF turn with no tool calls must stop so the run loop exits instead of re-entering with identical messages") +} + // stalledStream is a chat.MessageStream that blocks in Recv() until // either unblocked or the stream is closed. It is used to simulate a // half-open TCP connection where the remote side stops sending data.