Discard incomplete streaming replies

docs/changes/unreleased/1497-truncated-stream.md

@@ -0,0 +1,10 @@
+---
+kind: changed
+title: a reply cut off mid-stream is dropped and asked again
+pr: 1497
+surface: [chat, engine]
+invalidates:
+  - A streamed reply ending without a finish reason or [DONE] marker was
+    returned as complete, so its partial text could enter the conversation. It
+    is discarded and retried now; either explicit completion signal is enough.
+---

internal/manual/chat/when-the-connection-drops.md

@@ -31,6 +31,13 @@ says `trying again` while it waits. A conversation opened with `--host` has a se
 connection, and that link to the other machine is the one the surface redials for up to five
 minutes.
 
+## The reply stopped halfway through
+
+If the model connection ends before the model reports a finish reason or sends its `[DONE]`
+marker, codeaf drops the unfinished reply and asks again. Text that appeared while it was
+arriving is not kept in the conversation. A reply with either completion signal is accepted,
+even when the other signal is absent.
+
 Over `--host`, the surface redials the machine by itself. You do not have to do anything.
 
 A conversation opened with `--host` runs on the other machine; the link between the two is

internal/provider/client.go

@@ -2066,6 +2066,21 @@ func (c *Client) completeWithMessagesStreaming(
 			}
 		}
 	}
+	if !decoder.done && finishReason == "" {
+		cut := &StreamCut{Reason: CutTruncated}
+		c.stampCut(ctx, cut, served, began, stall.tokens())
+		cut.Rerouted = c.noteCutProvider(ctx, c.modelFor(request), served)
+		c.noteLaneOutcome(c.modelFor(request), served, cut.Reason.word(), false)
+		c.releaseEndpoint(ctx, c.modelFor(request))
+		c.record(recordFacts{
+			ctx: ctx, request: request, knobs: knobs, stream: true,
+			began: logBegan, status: httpResponse.StatusCode, served: served, err: cut,
+			response: response, reasoningTokens: reasoningTokens,
+			ttft: firstTokenAfter(began, firstToken),
+		})
+		c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len())
+		return nil, false, cut
+	}
 	response.Choices = []ai.Choice{{Index: 0, FinishReason: finishReason, Message: ai.Message{
 		Role:      "assistant",
 		Content:   []ai.ContentPart{{Type: "text", Text: content.String()}},

internal/provider/sse.go

@@ -32,6 +32,9 @@ import (
 type sseDecoder struct {
 	reader  *bufio.Reader
 	message []byte
+	// done records the protocol's explicit [DONE] marker, which shares the
+	// decoder's io.EOF return with an underlying connection close.
+	done bool
 	// alive, when set, is called once per SSE comment line — the ": OPENROUTER
 	// PROCESSING" keepalives a router sends while an upstream assembles its
 	// answer. Comments never become chunks (parseSSEMessage drops them), so
@@ -128,6 +131,7 @@ func (d *sseDecoder) next() ([]byte, error) {
 				d.message = d.message[:0]
 				switch {
 				case done:
+					d.done = true
 					return nil, io.EOF
 				case delivered:
 					return payload, nil

internal/provider/sse_test.go

@@ -257,6 +257,9 @@ func TestKeepaliveCommentsReachTheAliveSeam(t *testing.T) {
 	if _, err := decoder.DecodeChunk(); !errors.Is(err, io.EOF) {
 		t.Fatalf("end = %v, want io.EOF", err)
 	}
+	if !decoder.done {
+		t.Fatal("the explicit [DONE] marker was not retained by the decoder")
+	}
 	if alive != 3 {
 		t.Fatalf("alive calls = %d, want one per comment line", alive)
 	}

internal/provider/streamforming_test.go

@@ -145,7 +145,8 @@ func TestAFragmentAfterReadyFormsNothing(t *testing.T) {
 
 // A stream whose call dies half-sent announces nothing — and the forming events
 // that described it are all a surface ever saw, which is the honest record of
-// what happened.
+// what happened. The call itself is still a truncated response and must be
+// retried rather than returned as a usable answer.
 func TestATruncatedCallFormsButNeverReadies(t *testing.T) {
 	client := streamClientForTest(t, http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
 		writer.Header().Set("Content-Type", "text/event-stream")
@@ -156,8 +157,13 @@ func TestATruncatedCallFormsButNeverReadies(t *testing.T) {
 	ctx := WithStreamObserver(context.Background(), func(event StreamEvent) {
 		observed = append(observed, event)
 	})
-	if _, err := client.CompleteWithMessages(ctx, userMessages("write it")); err != nil {
-		t.Fatal(err)
+	response, err := client.CompleteWithMessages(ctx, userMessages("write it"))
+	cut, ok := CutFrom(err)
+	if !ok || cut.Reason != CutTruncated {
+		t.Fatalf("err = %v, want a truncated stream cut", err)
+	}
+	if response != nil {
+		t.Fatalf("truncated call returned a response: %+v", response)
 	}
 	if len(formingEvents(observed, 0)) == 0 {
 		t.Fatal("a call that arrived and stopped formed nothing")

internal/provider/streamguard.go

@@ -504,7 +504,7 @@ func gapFor(rate float64) time.Duration {
 	return gap
 }
 
-// CutReason says which of the two things went wrong, and it is the only thing
+// CutReason says which thing went wrong, and it is the only thing
 // this package decides about a cut. The sentence is composed upstream.
 type CutReason int
 
@@ -529,6 +529,9 @@ const (
 	// model's chat template, and the reply is unusable no matter how healthy
 	// the stream that carried it was. See [MachineryLeak].
 	CutMachinery
+	// CutTruncated is a stream that ended without an explicit completion marker
+	// or a finish reason, so its partial reply cannot be used.
+	CutTruncated
 )
 
 // word is the short machine-readable name of a cut: what the lane's belief
@@ -550,6 +553,8 @@ func (r CutReason) word() string {
 		return "overrun"
 	case CutMachinery:
 		return "machinery"
+	case CutTruncated:
+		return "truncated"
 	}
 	return "cut"
 }
@@ -561,7 +566,8 @@ func (r CutReason) word() string {
 type StreamCut struct {
 	Reason CutReason
 	// Waited is how long the stream was quiet, on the two silence reasons, and
-	// zero on CutBabble. It is the constant that fired rather than a measurement
+	// zero when no timer made the cut (CutBabble and CutTruncated). It is the
+	// constant that fired rather than a measurement
 	// — the plain bound on an outright silence, [bufferedQuietBound] when
 	// keepalives bought the stream its full patience and it still never wrote —
 	// because the timer is what decided, and the timer's own bound is the
@@ -647,6 +653,8 @@ func (c *StreamCut) Error() string {
 		return fmt.Sprintf("the reply ran past %s without finishing and was cut", roundSeconds(c.Waited))
 	case CutMachinery:
 		return "the reply was the model's own internal markup instead of an answer and was cut"
+	case CutTruncated:
+		return "the connection ended before the reply was finished"
 	default:
 		return "the reply stopped being language and was cut"
 	}

internal/provider/streamintel_test.go

@@ -315,10 +315,9 @@ func TestANamelessCallIsNeverAnnounced(t *testing.T) {
 	}
 }
 
-// A stream that stops without saying [DONE] ends the same way it always has —
-// at EOF, with the response built from what arrived — and the announcements
-// follow the same rule they do everywhere else: the whole calls, never the one
-// the stream stopped in the middle of.
+// A stream that stops without saying [DONE] is truncated. It must not return a
+// response, while announcements still follow the same rule: whole calls, never
+// the one the stream stopped in the middle of.
 func TestAStreamThatStopsShortAnnouncesOnlyWholeCalls(t *testing.T) {
 	client := streamClientForTest(t, http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
 		writer.Header().Set("Content-Type", "text/event-stream")
@@ -330,8 +329,13 @@ func TestAStreamThatStopsShortAnnouncesOnlyWholeCalls(t *testing.T) {
 	ctx := WithStreamObserver(context.Background(), func(event StreamEvent) {
 		observed = append(observed, event)
 	})
-	if _, err := client.CompleteWithMessages(ctx, userMessages("read it")); err != nil {
-		t.Fatal(err)
+	response, err := client.CompleteWithMessages(ctx, userMessages("read it"))
+	cut, ok := CutFrom(err)
+	if !ok || cut.Reason != CutTruncated {
+		t.Fatalf("err = %v, want a truncated stream cut", err)
+	}
+	if response != nil {
+		t.Fatalf("truncated stream returned a response: %+v", response)
 	}
 	ready := readyCalls(t, observed)
 	if len(ready) != 1 || ready[0].ID != "call_1" {

internal/provider/truncatedstream_test.go

@@ -0,0 +1,86 @@
+package provider
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestAStreamNeedsACompletionSignalOrFinishReason(t *testing.T) {
+	const contentFrame = `data: {"id":"one","provider":"test-node","choices":[{"index":0,"delta":{"content":"partial answer"}}]}` + "\n\n"
+	tests := []struct {
+		name             string
+		body             string
+		wantCut          bool
+		wantFinishReason string
+	}{
+		{
+			name:    "connection ends without either signal",
+			body:    contentFrame,
+			wantCut: true,
+		},
+		{
+			name:             "finish reason without done marker",
+			body:             contentFrame + `data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}` + "\n\n",
+			wantFinishReason: "stop",
+		},
+		{
+			name: "done marker without finish reason",
+			body: contentFrame + "data: [DONE]\n\n",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.Header().Set("Content-Type", "text/event-stream")
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte(test.body))
+			}))
+			defer server.Close()
+
+			client, err := NewClient(Config{APIKey: "k", BaseURL: server.URL, Model: "sim/model"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var events []StreamEvent
+			ctx := WithStreamObserver(context.Background(), func(event StreamEvent) {
+				events = append(events, event)
+			})
+			response, err := client.CompleteWithMessages(ctx, userMessages("hello"))
+
+			if test.wantCut {
+				cut, ok := CutFrom(err)
+				if !ok || cut.Reason != CutTruncated {
+					t.Fatalf("err = %v, want a truncated stream cut", err)
+				}
+				if response != nil {
+					t.Fatalf("truncated stream returned a response: %+v", response)
+				}
+				var delta, failed, finished bool
+				for _, event := range events {
+					delta = delta || event.Kind == StreamDelta && event.Delta == "partial answer"
+					failed = failed || event.Kind == StreamFailed
+					finished = finished || event.Kind == StreamFinished
+				}
+				if !delta || !failed || finished {
+					t.Fatalf("truncated stream events = %+v, want visible partial text and failure without completion", events)
+				}
+				return
+			}
+
+			if err != nil {
+				t.Fatalf("complete stream: %v", err)
+			}
+			if response == nil || len(response.Choices) != 1 {
+				t.Fatalf("response = %+v, want one completed choice", response)
+			}
+			if response.Choices[0].FinishReason != test.wantFinishReason {
+				t.Errorf("finish reason = %q, want %q", response.Choices[0].FinishReason, test.wantFinishReason)
+			}
+			if len(response.Choices[0].Message.Content) != 1 || response.Choices[0].Message.Content[0].Text != "partial answer" {
+				t.Errorf("response content = %+v, want the streamed answer", response.Choices[0].Message.Content)
+			}
+		})
+	}
+}

internal/session/loop.go

@@ -2696,6 +2696,8 @@ func cutNotice(cut *provider.StreamCut) string {
 		return "the reply kept going and never finished — asking again"
 	case provider.CutMachinery:
 		return "the model answered in its own internal markup instead of words — that text was dropped, asking again"
+	case provider.CutTruncated:
+		return "the connection ended before the reply was finished — that text was dropped, asking again"
 	default:
 		return "nothing came back from the model — asking again"
 	}
@@ -2716,6 +2718,8 @@ func cutWords(cut *provider.StreamCut) string {
 		return "the reply kept going and never finished"
 	case provider.CutMachinery:
 		return "the model answered in its own internal markup instead of words"
+	case provider.CutTruncated:
+		return "the connection ended before the reply was finished"
 	default:
 		return "nothing came back from the model"
 	}
@@ -2745,6 +2749,8 @@ func hopNotice(cut *provider.StreamCut, verdict taxonomy.Verdict, next string) s
 		return "the reply kept running on without finishing — finishing this one on " + next
 	case provider.CutMachinery:
 		return "the model kept answering in its own internal markup — finishing this one on " + next
+	case provider.CutTruncated:
+		return "the connection kept ending before the reply was finished — finishing this one on " + next
 	default:
 		return "nothing kept coming back from the model — finishing this one on " + next
 	}
@@ -2773,6 +2779,12 @@ func cutFailure(cut *provider.StreamCut, attempts int, hopped []string) error {
 		said = "the reply lost its thread " + timesWord(attempts) +
 			" — it came back as repetition and jumbled text, so none of it was kept. " +
 			"a different model may hold it (/model), or /compact to lighten the conversation"
+	case cut.Reason == provider.CutTruncated && len(hopped) > 0:
+		said = fmt.Sprintf("%s, %s — the partial reply was dropped. %s",
+			cut.Error(), timesWord(attempts), alsoTried(hopped))
+	case cut.Reason == provider.CutTruncated:
+		said = fmt.Sprintf("%s, %s — the partial reply was dropped. A different model may answer (/model)",
+			cut.Error(), timesWord(attempts))
 	case len(hopped) > 0:
 		said = fmt.Sprintf("%s, %s. %s — /model to pick another one yourself",
 			cut.Error(), timesWord(attempts), alsoTried(hopped))