diff --git a/docs/changes/unreleased/1497-truncated-stream.md b/docs/changes/unreleased/1497-truncated-stream.md new file mode 100644 index 000000000..703b9c013 --- /dev/null +++ b/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. +--- diff --git a/internal/manual/chat/when-the-connection-drops.md b/internal/manual/chat/when-the-connection-drops.md index cfb5424c0..2d7183ebc 100644 --- a/internal/manual/chat/when-the-connection-drops.md +++ b/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 diff --git a/internal/provider/client.go b/internal/provider/client.go index a97ea9cdc..1c6feea8f 100644 --- a/internal/provider/client.go +++ b/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()}}, diff --git a/internal/provider/sse.go b/internal/provider/sse.go index a56fd8df1..ca1cf0128 100644 --- a/internal/provider/sse.go +++ b/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 diff --git a/internal/provider/sse_test.go b/internal/provider/sse_test.go index a2f486ab2..e0f7d3351 100644 --- a/internal/provider/sse_test.go +++ b/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) } diff --git a/internal/provider/streamforming_test.go b/internal/provider/streamforming_test.go index 47058ea9c..37d534ee3 100644 --- a/internal/provider/streamforming_test.go +++ b/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") diff --git a/internal/provider/streamguard.go b/internal/provider/streamguard.go index ce06bb7b4..c6f6b64fb 100644 --- a/internal/provider/streamguard.go +++ b/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" } diff --git a/internal/provider/streamintel_test.go b/internal/provider/streamintel_test.go index cc5f337ad..e478491de 100644 --- a/internal/provider/streamintel_test.go +++ b/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" { diff --git a/internal/provider/truncatedstream_test.go b/internal/provider/truncatedstream_test.go new file mode 100644 index 000000000..4cfb8df66 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/internal/session/loop.go b/internal/session/loop.go index 1dc25ecc5..cf0f414c9 100644 --- a/internal/session/loop.go +++ b/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)) diff --git a/internal/session/streamcut_test.go b/internal/session/streamcut_test.go index 03dd8f75c..40ee84c3c 100644 --- a/internal/session/streamcut_test.go +++ b/internal/session/streamcut_test.go @@ -2,6 +2,9 @@ package session import ( "context" + "io" + "net/http" + "net/http/httptest" "strings" "testing" @@ -110,6 +113,127 @@ func TestTheJunkNeverReachesTheTranscript(t *testing.T) { } } +func TestAnIncompleteReplyIsAskedAgainWithoutKeepingItsText(t *testing.T) { + const partial = "answer cut off before it was finished" + completer := &scriptedCompleter{steps: []step{ + cutStep(provider.CutTruncated, partial), + func(_ context.Context, _ []ai.Message) (*ai.Response, error) { + return textResponse("the complete answer"), nil + }, + }} + agent, _ := newTestAgent(t, completer, nil) + events, err := agent.Submit(context.Background(), "what happened?") + if err != nil { + t.Fatalf("submit: %v", err) + } + collected := collect(t, events) + + retry, retried := firstOfKind(collected, EventRetrying) + if !retried || !strings.Contains(retry.Text, "connection ended before the reply was finished") || !strings.Contains(retry.Text, "dropped") { + t.Fatalf("retry note = %+v, want the incomplete reply described as dropped", retry) + } + if _, failed := firstOfKind(collected, EventError); failed { + t.Fatalf("a recovered reply ended in an error; events were %v", kinds(collected)) + } + for _, message := range completer.request(1) { + for _, part := range message.Content { + if strings.Contains(part.Text, partial) { + t.Fatalf("the incomplete reply was sent again: %q", part.Text) + } + } + } + for _, message := range agent.snapshot() { + for _, part := range message.Content { + if strings.Contains(part.Text, partial) { + t.Fatalf("the incomplete reply was saved in the conversation: %q", part.Text) + } + } + } +} + +func TestATruncatedProviderReplyIsNotSettledByTheSession(t *testing.T) { + const partial = "Partial ans" + const complete = "The complete answer" + var attempts int + var requests [][]byte + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + body, err := io.ReadAll(request.Body) + if err != nil { + t.Errorf("read request: %v", err) + return + } + requests = append(requests, body) + attempts++ + writer.Header().Set("Content-Type", "text/event-stream") + if attempts == 1 { + _, _ = io.WriteString(writer, "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\""+partial+"\"}}]}\n\n") + return + } + _, _ = io.WriteString(writer, "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\""+complete+"\"},\"finish_reason\":\"stop\"}]}\n\n") + })) + defer server.Close() + + client, err := provider.NewClient(provider.Config{ + APIKey: "k", BaseURL: server.URL, Model: "test/model", HTTPClient: server.Client(), + }) + if err != nil { + t.Fatalf("NewClient: %v", err) + } + agent, _ := newTestAgent(t, client, nil) + events, err := agent.Submit(context.Background(), "what happened?") + if err != nil { + t.Fatalf("submit: %v", err) + } + collected := collect(t, events) + + if attempts != 2 || len(requests) != 2 { + t.Fatalf("provider attempts = %d, requests = %d, want one truncated call and one retry", attempts, len(requests)) + } + if strings.Contains(string(requests[1]), partial) { + t.Fatalf("the truncated text was sent in the retry request: %s", requests[1]) + } + if _, failed := firstOfKind(collected, EventError); failed { + t.Fatalf("a recovered reply ended in an error; events were %v", kinds(collected)) + } + var sawComplete bool + for _, event := range collected { + sawComplete = sawComplete || event.Kind == EventTextDelta && strings.Contains(event.Text, complete) + } + if !sawComplete { + t.Fatalf("the complete retry answer was not delivered; events were %v", kinds(collected)) + } + for _, message := range agent.snapshot() { + if strings.Contains(messageText(message), partial) { + t.Fatalf("the truncated reply was kept in the conversation: %q", messageText(message)) + } + } +} + +func TestRepeatedIncompleteRepliesSayWhatWasDropped(t *testing.T) { + completer := &scriptedCompleter{steps: []step{ + cutStep(provider.CutTruncated, "first partial"), + cutStep(provider.CutTruncated, "second partial"), + cutStep(provider.CutTruncated, "third partial"), + }} + agent, _ := newTestAgent(t, completer, nil) + events, err := agent.Submit(context.Background(), "go on") + if err != nil { + t.Fatalf("submit: %v", err) + } + collected := collect(t, events) + + failure, failed := firstOfKind(collected, EventError) + if !failed { + t.Fatalf("two incomplete replies did not end the turn; events were %v", kinds(collected)) + } + said := failure.Err.Error() + for _, want := range []string{"connection ended before the reply was finished", "partial reply was dropped", "/model"} { + if !strings.Contains(said, want) { + t.Fatalf("the sentence %q does not contain %q", said, want) + } + } +} + // TestAReplyThatComesApartTwiceEndsTheTurnInWordsThatHelp: the give-up sentence // names what happened and the two doors that actually open. func TestAReplyThatComesApartTwiceEndsTheTurnInWordsThatHelp(t *testing.T) {