fix(runtime): stop content-only turns that end with a bare EOF (no finish_reason)

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},

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

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.