From 6a1dcddb48d71844c6b66045c6dc1fff5fd53e93 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 07:50:39 +0000 Subject: [PATCH] Record content-free stream terminal observations Why: The r0.4 trial could prove that the parser missed [DONE] but could not distinguish clean EOF, malformed frames, read failure, or provider error without retaining raw SSE. Evidence: RED receipts fast-slow-u3-r05-task1-terminal-red.json and task1-http-error-red.json; GREEN targeted receipt and full vet/test/race receipt fast-slow-u3-r05-task1-agent-native-full.json. Impact: Add bounded scalar diagnostics and successful HTTP status to terminal events while preserving the existing incomplete-tool rejection, returned Usage, and retry behavior. Refs: U3 r0.5 approved design and plan; Agent-Native PR #46 --- runtime/llm/client.go | 20 +++++---- runtime/llm/client_test.go | 65 ++++++++++++++++++++++++++++ runtime/llm/sse.go | 89 +++++++++++++++++++++++++++++++------- runtime/llm/sse_test.go | 80 ++++++++++++++++++++++++++++++++++ runtime/llm/types.go | 23 ++++++++-- 5 files changed, 249 insertions(+), 28 deletions(-) create mode 100644 runtime/llm/client_test.go diff --git a/runtime/llm/client.go b/runtime/llm/client.go index 8771124..630debf 100644 --- a/runtime/llm/client.go +++ b/runtime/llm/client.go @@ -100,10 +100,10 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool resp, err := hc.Do(req) if err != nil { if ctx.Err() != nil { - emitAbort(out, "", ctx.Err()) + emitAbort(out, "", ctx.Err(), &TerminalObservation{ReadOutcome: "cancelled"}) return } - emitError(out, "", true, err.Error()) + emitError(out, "", true, err.Error(), &TerminalObservation{ReadOutcome: "transport_error"}) return } defer resp.Body.Close() @@ -111,12 +111,13 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool slurp, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) msg := fmt.Sprintf("provider status %d: %s", resp.StatusCode, strings.TrimSpace(string(slurp))) ev := Event{ - Kind: KindError, - HTTPStatus: resp.StatusCode, - Retryable: StatusRetryable(resp.StatusCode) && !ContextOverflow(msg), - ErrorMessage: msg, - Err: &StreamError{Message: msg, Retryable: StatusRetryable(resp.StatusCode) && !ContextOverflow(msg)}, - Message: Message{Role: RoleAssistant, StopReason: StopError}, + Kind: KindError, + HTTPStatus: resp.StatusCode, + TerminalObservation: &TerminalObservation{ReadOutcome: "provider_error"}, + Retryable: StatusRetryable(resp.StatusCode) && !ContextOverflow(msg), + ErrorMessage: msg, + Err: &StreamError{Message: msg, Retryable: StatusRetryable(resp.StatusCode) && !ContextOverflow(msg)}, + Message: Message{Role: RoleAssistant, StopReason: StopError}, } if d, ok := ParseRetryAfter(resp.Header.Get("Retry-After"), time.Now()); ok { ev.HasRetryAfter = true @@ -126,6 +127,9 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool return } for ev := range ParseSSE(ctx, resp.Body) { + if ev.Kind == KindMessageEnd || ev.Kind == KindError { + ev.HTTPStatus = resp.StatusCode + } out <- ev } } diff --git a/runtime/llm/client_test.go b/runtime/llm/client_test.go new file mode 100644 index 0000000..6a426ec --- /dev/null +++ b/runtime/llm/client_test.go @@ -0,0 +1,65 @@ +package llm + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" +) + +type terminalRoundTripFunc func(*http.Request) (*http.Response, error) + +func (f terminalRoundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func TestClientTerminalObservationHTTPStatus(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/chat/completions" { + t.Errorf("path = %q", r.URL.Path) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1,\"total_tokens\":3}}\n\ndata: [DONE]\n\n")) + })) + defer server.Close() + client := &Client{BaseURL: server.URL, APIKey: "fixture", HTTPClient: server.Client()} + var end Event + for ev := range client.Stream(context.Background(), "fixture", []Message{{Role: RoleUser, Content: "hello"}}, nil, StreamOptions{}) { + if ev.Kind == KindMessageEnd { + end = ev + } + } + if end.HTTPStatus != http.StatusOK || end.TerminalObservation == nil || + end.TerminalObservation.ReadOutcome != "done_marker" || !end.StreamComplete || end.Usage.TotalTokens != 3 { + t.Fatalf("successful HTTP terminal lost transport evidence: %#v", end) + } +} + +func TestClientTerminalObservationProviderAndTransportFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "PRIVATE_PROVIDER_BODY_R05", http.StatusTooManyRequests) + })) + defer server.Close() + tests := []struct { + name string + client *Client + status int + outcome string + }{ + {"provider-http", &Client{BaseURL: server.URL, APIKey: "fixture", HTTPClient: server.Client()}, http.StatusTooManyRequests, "provider_error"}, + {"transport", &Client{BaseURL: "http://test.invalid", APIKey: "fixture", HTTPClient: &http.Client{Transport: terminalRoundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("PRIVATE_TRANSPORT_ERROR_R05") + })}}, 0, "transport_error"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var end Event + for ev := range tc.client.Stream(context.Background(), "fixture", []Message{{Role: RoleUser, Content: "hello"}}, nil, StreamOptions{}) { + end = ev + } + if end.Kind != KindError || end.HTTPStatus != tc.status || end.TerminalObservation == nil || + end.TerminalObservation.ReadOutcome != tc.outcome { + t.Fatalf("provider/transport failure was not classified: %#v", end) + } + }) + } +} diff --git a/runtime/llm/sse.go b/runtime/llm/sse.go index 6edf9d2..ab2bff1 100644 --- a/runtime/llm/sse.go +++ b/runtime/llm/sse.go @@ -39,6 +39,11 @@ type usageWire struct { } `json:"prompt_tokens_details"` } +const ( + maxObservedToolFragments = 10000 + maxObservedToolArgumentBytes = 4*1024*1024 + 1 +) + // ParseSSE reads an OpenAI-compatible server-sent event body and emits // deltas plus one terminal message_end or error event. func ParseSSE(ctx context.Context, r io.Reader) <-chan Event { @@ -59,15 +64,36 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { finished := "" sawDone := false sawUsage := false + readOutcome := "clean_eof" + lastFrameJSONComplete := false + toolFragments := 0 + toolArgumentBytes := 0 + observation := func() *TerminalObservation { + argumentsValid := len(order) > 0 + for _, idx := range order { + var args map[string]json.RawMessage + if err := json.Unmarshal([]byte(calls[idx].Arguments), &args); err != nil || args == nil { + argumentsValid = false + break + } + } + return &TerminalObservation{ + ReadOutcome: readOutcome, LastFrameJSONComplete: lastFrameJSONComplete, + UsagePresent: sawUsage, ToolFragments: toolFragments, + ToolArgumentBytes: toolArgumentBytes, ToolArgumentsValid: argumentsValid, + } + } sendTerminalFailure := func(why string) { message := Message{Role: RoleAssistant, Content: content.String(), StopReason: StopError} if reasoning.Len() > 0 { message.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}} } failed := terminalStreamFailure(message, usage, finished, sawDone, why) + failed.TerminalObservation = observation() select { case <-ctx.Done(): - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) case out <- failed: } } @@ -80,30 +106,36 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { break } if ctx.Err() != nil { - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) return } if data == "[DONE]" { sawDone = true + readOutcome = "done_marker" break } var chunk sseChunk if err := json.Unmarshal([]byte(data), &chunk); err != nil { + readOutcome = "invalid_frame" + lastFrameJSONComplete = false if sawUsage { sendTerminalFailure("malformed stream chunk after reported usage") return } - emitError(out, content.String(), false, "malformed stream chunk: "+err.Error()) + emitError(out, content.String(), false, "malformed stream chunk: "+err.Error(), observation()) return } + lastFrameJSONComplete = true if chunk.Error != nil && chunk.Error.Message != "" { + readOutcome = "provider_error" msg := chunk.Error.Message if sawUsage { sendTerminalFailure("provider error after reported usage: " + msg) return } retryable := CapacityLimited(msg) && !ContextOverflow(msg) - emitError(out, content.String(), retryable, msg) + emitError(out, content.String(), retryable, msg, observation()) return } if chunk.Usage != nil { @@ -124,7 +156,8 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { content.WriteString(choice.Delta.Content) select { case <-ctx.Done(): - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) return case out <- Event{Kind: KindTextDelta, Text: choice.Delta.Content}: } @@ -139,12 +172,23 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { reasoning.WriteString(choice.Delta.Reasoning) select { case <-ctx.Done(): - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) return case out <- Event{Kind: KindReasoningDelta, ReasoningDelta: choice.Delta.Reasoning}: } } for _, tc := range choice.Delta.ToolCalls { + if toolFragments < maxObservedToolFragments { + toolFragments++ + } + if n := len(tc.Function.Arguments); n > 0 && toolArgumentBytes < maxObservedToolArgumentBytes { + if n > maxObservedToolArgumentBytes-toolArgumentBytes { + toolArgumentBytes = maxObservedToolArgumentBytes + } else { + toolArgumentBytes += n + } + } cur, ok := calls[tc.Index] if !ok { cur = &ToolCall{} @@ -162,7 +206,8 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { } select { case <-ctx.Done(): - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) return case out <- Event{ Kind: KindToolCallDelta, @@ -175,21 +220,24 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { } } if err := sc.Err(); err != nil { + readOutcome = "read_error" if ctx.Err() != nil { - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) return } if sawUsage { sendTerminalFailure("stream read failed after reported usage") return } - emitError(out, content.String(), true, err.Error()) + emitError(out, content.String(), true, err.Error(), observation()) return } // 取消必须压过正常结束。select 在两个分支都就绪时随机挑选， // 竞态下会把 message_end 发给已经取消的调用方。 if ctx.Err() != nil { - emitAbort(out, content.String(), ctx.Err()) + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) return } // A finish reason alone does not prove the tool arguments arrived intact. @@ -237,8 +285,9 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { } select { case <-ctx.Done(): - emitAbort(out, content.String(), ctx.Err()) - case out <- Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished, StreamComplete: sawDone}: + readOutcome = "cancelled" + emitAbort(out, content.String(), ctx.Err(), observation()) + case out <- Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished, StreamComplete: sawDone, TerminalObservation: observation()}: } } @@ -337,8 +386,8 @@ func mapFinish(reason string, hasTools bool) StopReason { } } -func emitError(out chan<- Event, partial string, retryable bool, message string) { - out <- Event{ +func emitError(out chan<- Event, partial string, retryable bool, message string, observed ...*TerminalObservation) { + event := Event{ Kind: KindError, Retryable: retryable, ErrorMessage: message, @@ -349,14 +398,18 @@ func emitError(out chan<- Event, partial string, retryable bool, message string) StopReason: StopError, }, } + if len(observed) > 0 { + event.TerminalObservation = observed[0] + } + out <- event } -func emitAbort(out chan<- Event, partial string, err error) { +func emitAbort(out chan<- Event, partial string, err error, observed ...*TerminalObservation) { msg := "aborted" if err != nil { msg = err.Error() } - out <- Event{ + event := Event{ Kind: KindError, Retryable: false, ErrorMessage: msg, @@ -367,6 +420,10 @@ func emitAbort(out chan<- Event, partial string, err error) { StopReason: StopAborted, }, } + if len(observed) > 0 { + event.TerminalObservation = observed[0] + } + out <- event } // StreamError is a provider or transport failure encoded on the event stream. diff --git a/runtime/llm/sse_test.go b/runtime/llm/sse_test.go index 3f79d8c..1da4afb 100644 --- a/runtime/llm/sse_test.go +++ b/runtime/llm/sse_test.go @@ -2,6 +2,7 @@ package llm import ( "context" + "encoding/json" "io" "net/http" "net/http/httptest" @@ -87,6 +88,10 @@ func TestParseSSEToolCallIncrements(t *testing.T) { if end.Message.StopReason != StopToolCalls || len(end.Message.ToolCalls) != 1 { t.Fatalf("end %#v", end.Message) } + if end.TerminalObservation == nil || end.TerminalObservation.ReadOutcome != "done_marker" || + !end.TerminalObservation.ToolArgumentsValid || end.TerminalObservation.ToolFragments != 3 { + t.Fatalf("valid multi-fragment terminal was not classified: %#v", end.TerminalObservation) + } tc := end.Message.ToolCalls[0] if tc.ID != "call_1" || tc.Name != "read_file" || tc.Arguments != `{"path":"a.md"}` { t.Fatalf("tool call %#v", tc) @@ -168,6 +173,77 @@ func TestParseSSEToolFinishWithoutDone(t *testing.T) { } } +func TestTerminalObservationCleanEOFWithoutDone(t *testing.T) { + body := strings.Join([]string{ + `data: {"choices":[{"delta":{"content":"PRIVATE_PROMPT_R05","tool_calls":[{"index":0,"id":"call_1","function":{"name":"PRIVATE_TOOL_R05","arguments":"{}"}}]}}]}`, + `data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`, + "", + }, "\n\n") + var end Event + for ev := range ParseSSE(context.Background(), strings.NewReader(body)) { + if ev.Kind == KindMessageEnd { + end = ev + } + } + if end.Err == nil || end.StreamComplete || len(end.Message.ToolCalls) != 0 || end.Usage.TotalTokens != 17 { + t.Fatalf("unfinished tool stream was released or unmetered: %#v", end) + } + meta := end.TerminalObservation + if meta == nil || meta.ReadOutcome != "clean_eof" || !meta.LastFrameJSONComplete || + !meta.UsagePresent || meta.ToolFragments != 1 || meta.ToolArgumentBytes != 2 || !meta.ToolArgumentsValid { + t.Fatalf("missing content-free clean EOF evidence: %#v", meta) + } + raw, err := json.Marshal(meta) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(raw), "PRIVATE_PROMPT_R05") || strings.Contains(string(raw), "PRIVATE_TOOL_R05") { + t.Fatalf("terminal observation retained model content: %s", raw) + } +} + +func TestTerminalObservationMalformedAndReadError(t *testing.T) { + usageAndTool := `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}` + "\n\n" + tests := []struct { + name string + ctx context.Context + reader io.Reader + outcome string + usage int + frameValid bool + }{ + {"invalid-frame", context.Background(), strings.NewReader(usageAndTool + "data: {not-json}\n\n"), "invalid_frame", 17, false}, + {"read-error", context.Background(), &errorAfterDataReader{data: strings.NewReader(usageAndTool)}, "read_error", 17, true}, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + tests = append(tests, struct { + name string + ctx context.Context + reader io.Reader + outcome string + usage int + frameValid bool + }{"cancelled", ctx, strings.NewReader(usageAndTool), "cancelled", 0, false}) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var end Event + for ev := range ParseSSE(tc.ctx, tc.reader) { + if ev.Kind == KindMessageEnd || ev.Kind == KindError { + end = ev + } + } + if end.Err == nil || len(end.Message.ToolCalls) != 0 || end.Usage.TotalTokens != tc.usage { + t.Fatalf("failure lost Usage or released tools: %#v", end) + } + meta := end.TerminalObservation + if meta == nil || meta.ReadOutcome != tc.outcome || meta.LastFrameJSONComplete != tc.frameValid { + t.Fatalf("wrong terminal classification: %#v", meta) + } + }) + } +} + func TestParseSSEInvalidToolArguments(t *testing.T) { body := strings.Join([]string{ `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"publish_work_checkpoint","arguments":"{\"findings\":"}}]}}]}`, @@ -186,6 +262,10 @@ func TestParseSSEInvalidToolArguments(t *testing.T) { end.Usage.TotalTokens != 17 { t.Fatalf("unsafe tool call with invalid JSON arguments: %#v", end) } + if end.TerminalObservation == nil || end.TerminalObservation.ToolArgumentsValid || + end.TerminalObservation.ReadOutcome != "done_marker" { + t.Fatalf("invalid arguments incorrectly classified: %#v", end.TerminalObservation) + } } func TestParseSSERejectsMissingToolIdentity(t *testing.T) { diff --git a/runtime/llm/types.go b/runtime/llm/types.go index dbe1993..4c72deb 100644 --- a/runtime/llm/types.go +++ b/runtime/llm/types.go @@ -98,6 +98,18 @@ type Usage struct { LastTotalTokens int `json:"last_total_tokens,omitempty"` } +// TerminalObservation records how a stream ended without retaining provider +// content, tool identity, arguments, request headers, or error text. Counts +// saturate at fixed limits; this evidence never authorizes a tool call. +type TerminalObservation struct { + ReadOutcome string `json:"readOutcome"` + LastFrameJSONComplete bool `json:"lastFrameJSONComplete"` + UsagePresent bool `json:"usagePresent"` + ToolFragments int `json:"toolFragments"` + ToolArgumentBytes int `json:"toolArgumentBytes"` + ToolArgumentsValid bool `json:"toolArgumentsValid"` +} + // Event is one streamed update. Failures are KindError plus a final assistant // message whose StopReason is error or aborted. Stream implementations do not panic. type Event struct { @@ -116,10 +128,13 @@ type Event struct { // StreamComplete records whether the OpenAI-compatible SSE stream supplied // its explicit [DONE] marker. Tool calls require this before execution. StreamComplete bool - Err error - Retryable bool - ErrorMessage string - // HTTPStatus 是提供方的 HTTP 状态。0 表示这条错误不是 HTTP 响应。 + // TerminalObservation is present only on a terminal/error event. It does + // not change StreamComplete or the tool-execution safety gate. + TerminalObservation *TerminalObservation + Err error + Retryable bool + ErrorMessage string + // HTTPStatus 是提供方的 HTTP 状态；0 表示尚无 HTTP 响应。 HTTPStatus int // RetryAfter is the provider's requested wait. HasRetryAfter distinguishes // a real zero from a missing or unparseable header.