Compare commits

...
2 Commits
Author SHA1 Message Date
weichen e86e74bc36 Preserve streamed tool delta identity
ci / docker-checkout-hostgateway (pull_request) Successful in 1s
ci / vet-and-test (pull_request) Successful in 52s
Why: Backend emitted tool deltas from a separate goroutine, racing with serialized run events and occasionally losing a UI delta.

Evidence: Backend race detector reproduced concurrent event writes; Agent-Native identity test RED→GREEN; full vet and tests pass.

Impact: runtime tool delta events now carry provider slot and call ID for single-owner downstream projection.

Refs: runtime/loop.go; runtime/event_vocabulary_test.go
2026-10-14 02:22:34 +00:00
weichen 94afd71f00 Reject incomplete streamed tool calls
ci / docker-checkout-hostgateway (pull_request) Successful in 0s
ci / vet-and-test (pull_request) Successful in 51s
Why: a model tool finish could execute partial JSON or an unterminated SSE stream, while post-Usage errors could lose their charge.

Evidence: missing-DONE, malformed-JSON, identity, non-object and post-Usage RED→GREEN tests; GOWORK=off go vet ./... and go test -count=1 ./... passed.

Impact: invalid tool terminals fail without execution or automatic retry and retain returned Usage; valid text and multi-part tool streams keep working.

Refs: runtime/llm/sse.go; runtime/loop.go
2026-10-14 02:09:44 +00:00
6 changed files with 313 additions and 27 deletions
+20 -1
View File
@@ -39,7 +39,7 @@ func vocabScript() llm.StreamFunc {
step++
ch <- llm.Event{Kind: llm.KindReasoningDelta, ReasoningDelta: "思考"}
ch <- llm.Event{Kind: llm.KindTextDelta, Text: "正文"}
ch <- llm.Event{Kind: llm.KindToolCallDelta, ToolName: "list_dir", ArgumentsDelta: `{"path":"."}`}
ch <- llm.Event{Kind: llm.KindToolCallDelta, ToolIndex: 2, ToolID: "c1", ToolName: "list_dir", ArgumentsDelta: `{"path":"."}`}
ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{
Role: llm.RoleAssistant, StopReason: llm.StopToolCalls,
ToolCalls: []llm.ToolCall{{ID: "c1", Name: "list_dir", Arguments: `{"path":"."}`}},
@@ -54,6 +54,25 @@ func vocabScript() llm.StreamFunc {
}
}
func TestLoopPreservesToolCallDeltaIdentity(t *testing.T) {
reg := tools.WorkspaceTools(t.TempDir())
var got Event
_, err := Run(context.Background(), Config{
Model: "fake", Tools: reg, Stream: vocabScript(), MaxTurns: 4,
OnEvent: func(ev Event) {
if ev.Type == "tool_call_delta" {
got = ev
}
},
}, llm.Message{Role: llm.RoleUser, Content: "run"})
if err != nil {
t.Fatal(err)
}
if got.ToolIndex != 2 || got.CallID != "c1" || got.Tool != "list_dir" || got.Detail != `{"path":"."}` {
t.Fatalf("tool delta lost its provider identity: %#v", got)
}
}
func TestLoopEmitsProcessVocabulary(t *testing.T) {
reg := tools.WorkspaceTools(t.TempDir())
var observed []string
+66 -9
View File
@@ -58,6 +58,19 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
var usage Usage
finished := ""
sawDone := false
sawUsage := false
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)
select {
case <-ctx.Done():
emitAbort(out, content.String(), ctx.Err())
case out <- failed:
}
}
sc := bufio.NewScanner(r)
sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
reader := &sseEventReader{sc: sc}
@@ -76,17 +89,26 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
}
var chunk sseChunk
if err := json.Unmarshal([]byte(data), &chunk); err != nil {
if sawUsage {
sendTerminalFailure("malformed stream chunk after reported usage")
return
}
emitError(out, content.String(), false, "malformed stream chunk: "+err.Error())
return
}
if chunk.Error != nil && chunk.Error.Message != "" {
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)
return
}
if chunk.Usage != nil {
usage = chunk.Usage.Usage
sawUsage = true
if chunk.Usage.Details.Cached > 0 {
usage.CacheReadTokens = chunk.Usage.Details.Cached
}
@@ -157,6 +179,10 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
emitAbort(out, content.String(), ctx.Err())
return
}
if sawUsage {
sendTerminalFailure("stream read failed after reported usage")
return
}
emitError(out, content.String(), true, err.Error())
return
}
@@ -166,9 +192,34 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
emitAbort(out, content.String(), ctx.Err())
return
}
// length、空结束原因,或没有 [DONE] 的中断,都不把半截工具调用交给执行层。
if !toolFinishExecutable(finished, sawDone) {
order = nil
// A finish reason alone does not prove the tool arguments arrived intact.
// Preserve returned usage when rejecting a partial terminal, so the host
// can charge this call and stop rather than treating it as a free retry.
if len(order) > 0 {
problem := ""
switch {
case !sawDone:
problem = "tool call stream ended before [DONE]"
case !toolFinishExecutable(finished, sawDone):
problem = "tool call ended without a supported finish reason"
default:
for _, idx := range order {
call := calls[idx]
var args map[string]json.RawMessage
if call.ID == "" || call.Name == "" {
problem = "tool call identity is incomplete"
break
}
if err := json.Unmarshal([]byte(call.Arguments), &args); err != nil || args == nil {
problem = "tool call arguments are not a JSON object"
break
}
}
}
if problem != "" {
sendTerminalFailure(problem)
return
}
}
msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)}
if reasoning.Len() > 0 {
@@ -183,10 +234,17 @@ 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}:
case out <- Event{Kind: KindMessageEnd, Message: msg, Usage: usage, FinishReason: finished, StreamComplete: sawDone}:
}
}
func terminalStreamFailure(message Message, usage Usage, reason string, complete bool, why string) Event {
err := &StreamError{Message: why, Retryable: false}
return Event{Kind: KindMessageEnd, Message: message, Usage: usage,
FinishReason: reason, StreamComplete: complete, Err: err,
ErrorMessage: why, Retryable: false}
}
// sseEventReader 按空行切事件,并把同一事件里的多行 data: 用换行拼回。
// 已经是完整 JSON 的 data 行单独成事件,这样单行流不用空行也能解析。
type sseEventReader struct {
@@ -241,17 +299,16 @@ func (r *sseEventReader) next() (string, bool) {
// toolFinishExecutable 只在供应商明确以工具调用或正常 stop 结束时执行。
// length、缺失的结束原因,以及没有 [DONE] 的断开,都视为未完成。
func toolFinishExecutable(reason string, sawDone bool) bool {
if !sawDone {
return false
}
switch reason {
case "tool_calls", "function_call":
return true
case "stop":
return true
case "length":
return false
case "":
return false
default:
return sawDone && reason != "error" && reason != "content_filter"
return false
}
}
+155 -6
View File
@@ -93,7 +93,7 @@ func TestParseSSEToolCallIncrements(t *testing.T) {
}
}
func TestParseSSELengthDropsUnfinishedToolCall(t *testing.T) {
func TestParseSSELengthFailsUnfinishedToolCall(t *testing.T) {
body := strings.Join([]string{
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"read_file","arguments":"{\"path\":"}}]}}]}`,
`data: {"choices":[{"delta":{},"finish_reason":"length"}]}`,
@@ -106,15 +106,15 @@ func TestParseSSELengthDropsUnfinishedToolCall(t *testing.T) {
end = ev
}
}
if end.Message.StopReason != StopEnd || len(end.Message.ToolCalls) != 0 {
t.Fatalf("truncated tool call must not be executable: %#v", end.Message)
if end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
t.Fatalf("truncated tool call must fail without execution: %#v", end)
}
if end.FinishReason != "length" {
t.Fatalf("finish %q", end.FinishReason)
}
}
func TestParseSSEUncleanEndDropsToolCall(t *testing.T) {
func TestParseSSEUncleanEndFailsToolCall(t *testing.T) {
body := `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"read_file","arguments":"{\"path\":"}}]}}]}`
var end Event
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
@@ -122,8 +122,97 @@ func TestParseSSEUncleanEndDropsToolCall(t *testing.T) {
end = ev
}
}
if end.Kind != KindMessageEnd || end.Message.StopReason != StopEnd || len(end.Message.ToolCalls) != 0 {
t.Fatalf("unclean stream must not execute tools: %#v", end)
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
t.Fatalf("unclean stream must fail without executing tools: %#v", end)
}
}
func TestParseSSEToolFinishWithoutDone(t *testing.T) {
body := strings.Join([]string{
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","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.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
end.Usage.TotalTokens != 17 {
t.Fatalf("unsafe tool call from unfinished stream: %#v", end)
}
}
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\":"}}]}}]}`,
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
`data: [DONE]`,
"",
}, "\n\n")
var end Event
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
if ev.Kind == KindMessageEnd {
end = ev
}
}
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
end.Usage.TotalTokens != 17 {
t.Fatalf("unsafe tool call with invalid JSON arguments: %#v", end)
}
}
func TestParseSSERejectsMissingToolIdentity(t *testing.T) {
for _, tc := range []struct{ name, firstChunk string }{
{"missing-id", `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"echo","arguments":"{}"}}]}}]}`},
{"missing-name", `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"arguments":"{}"}}]}}]}`},
} {
t.Run(tc.name, func(t *testing.T) {
body := strings.Join([]string{
tc.firstChunk,
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
`data: [DONE]`,
"",
}, "\n\n")
var end Event
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
if ev.Kind == KindMessageEnd {
end = ev
}
}
if end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
t.Fatalf("missing tool identity reached execution: %#v", end)
}
})
}
}
func TestParseSSERejectsNonObjectToolArguments(t *testing.T) {
for _, tc := range []struct{ name, arguments string }{
{"array", "[]"},
{"null", "null"},
} {
t.Run(tc.name, func(t *testing.T) {
body := strings.Join([]string{
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"` + tc.arguments + `"}}]}}]}`,
`data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
`data: [DONE]`,
"",
}, "\n\n")
var end Event
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
if ev.Kind == KindMessageEnd {
end = ev
}
}
if end.Message.StopReason != StopError || len(end.Message.ToolCalls) != 0 || end.Err == nil {
t.Fatalf("non-object tool arguments reached execution: %#v", end)
}
})
}
}
@@ -149,6 +238,66 @@ func TestParseSSEMalformedIsError(t *testing.T) {
}
}
func TestParseSSEMalformedChunkAfterUsageKeepsUsage(t *testing.T) {
body := strings.Join([]string{
`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}}`,
`data: {not-json}`,
"",
}, "\n\n")
var end Event
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
end = ev
}
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
end.Usage.TotalTokens != 17 {
t.Fatalf("malformed chunk dropped already returned usage: %#v", end)
}
}
func TestParseSSEProviderErrorAfterUsageKeepsUsage(t *testing.T) {
body := strings.Join([]string{
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]}}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
`data: {"error":{"message":"server is at capacity"}}`,
"",
}, "\n\n")
var end Event
for ev := range ParseSSE(context.Background(), strings.NewReader(body)) {
end = ev
}
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
end.Usage.TotalTokens != 17 {
t.Fatalf("provider error after usage retried or lost its charge: %#v", end)
}
}
type errorAfterDataReader struct{ data *strings.Reader }
func (r *errorAfterDataReader) Read(p []byte) (int, error) {
n, err := r.data.Read(p)
if err == io.EOF {
return 0, io.ErrUnexpectedEOF
}
return n, err
}
func TestParseSSEReaderErrorAfterUsageKeepsUsage(t *testing.T) {
body := strings.Join([]string{
`data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"echo","arguments":"{}"}}]}}],"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}`,
"",
}, "\n\n")
var end Event
for ev := range ParseSSE(context.Background(), &errorAfterDataReader{data: strings.NewReader(body)}) {
end = ev
}
if end.Kind != KindMessageEnd || end.Message.StopReason != StopError ||
len(end.Message.ToolCalls) != 0 || end.Err == nil || end.Retryable ||
end.Usage.TotalTokens != 17 {
t.Fatalf("reader error after usage retried or lost its charge: %#v", end)
}
}
func TestParseSSECancel(t *testing.T) {
pr, pw := io.Pipe()
ctx, cancel := context.WithCancel(context.Background())
+6 -3
View File
@@ -113,9 +113,12 @@ type Event struct {
// FinishReason 是供应商原始结束原因(stop、length、tool_calls 等)。
// 空表示这条事件不携带结束原因。StopReason 的映射不受此字段影响。
FinishReason string
Err error
Retryable bool
ErrorMessage string
// 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 响应。
HTTPStatus int
// RetryAfter is the provider's requested wait. HasRetryAfter distinguishes
+17 -8
View File
@@ -38,8 +38,10 @@ type Event struct {
Message llm.Message
Tool string
Text string
// ToolIndex preserves the provider's call slot for tool_call_delta.
ToolIndex int
// CallID 在 tool_execution_start / tool_execution_end 时携带工具调用的
// ID:并行执行下同回合多个调用并发进出,消费方按 ID 配对而非按顺序。
// ID;tool_call_delta 也携带同一 ID。消费方按 ID 配对而非按顺序。
CallID string
// Detail carries tool arguments/results or reasoning deltas for
// process-visualization consumers; log writers may ignore it.
@@ -72,8 +74,8 @@ type Config struct {
// SerialTools 是并行执行下的顺序敏感工具名集合:这些调用按模型给出的
// 原顺序在调用方 goroutine 逐个执行(互斥且保序),其余调用并发执行。
SerialTools map[string]bool
GetSteering func() []llm.Message
GetFollowUp func() []llm.Message
GetSteering func() []llm.Message
GetFollowUp func() []llm.Message
// OnEvent 观察循环事件。同一次 Run 内不会并发调用
// (与 BeforeTool、AfterTool 共用一把锁串行)。
OnEvent func(Event)
@@ -227,9 +229,9 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
}
cfg.traceCtx = newTraceContext()
endRun := startSpan(cfg, "invoke_agent "+agentName(cfg), map[string]string{
"gen_ai.system": providerName(cfg),
"gen_ai.agent.name": agentName(cfg),
"gen_ai.request.model": cfg.Model,
"gen_ai.system": providerName(cfg),
"gen_ai.agent.name": agentName(cfg),
"gen_ai.request.model": cfg.Model,
}, true)
defer endRun()
var messages []llm.Message
@@ -354,6 +356,12 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
usage.LastPromptTokens = end.Usage.PromptTokens
usage.LastCompletionTokens = end.Usage.CompletionTokens
usage.LastTotalTokens = end.Usage.TotalTokens
if end.Err != nil {
cfg.logger().Error("runtime terminal stream failed", end.Err)
emitUsage(cfg, usage)
return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError,
Usage: usage, Err: end.Err, Retryable: false}, nil
}
if err := persist(cfg, last); err != nil {
return Result{}, err
}
@@ -529,7 +537,8 @@ func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event
case llm.KindReasoningDelta:
emit(cfg, Event{Type: "reasoning_delta", Text: ev.ReasoningDelta})
case llm.KindToolCallDelta:
emit(cfg, Event{Type: "tool_call_delta", Tool: ev.ToolName, Detail: ev.ArgumentsDelta})
emit(cfg, Event{Type: "tool_call_delta", ToolIndex: ev.ToolIndex,
CallID: ev.ToolID, Tool: ev.ToolName, Detail: ev.ArgumentsDelta})
case llm.KindError:
return ev, ev.Err
case llm.KindMessageEnd:
@@ -1156,7 +1165,7 @@ func drain(fn func() []llm.Message) []llm.Message {
func emitUsage(cfg Config, usage llm.Usage) {
emit(cfg, Event{
Type: "usage",
Type: "usage",
PromptTokens: usage.PromptTokens, CompletionTokens: usage.CompletionTokens, TotalTokens: usage.TotalTokens,
LastPromptTokens: usage.LastPromptTokens, LastCompletionTokens: usage.LastCompletionTokens, LastTotalTokens: usage.LastTotalTokens,
CacheReadTokens: usage.CacheReadTokens, CacheWriteTokens: usage.CacheWriteTokens,
+49
View File
@@ -2,6 +2,7 @@ package runtime
import (
"context"
"errors"
"path/filepath"
"strings"
"sync"
@@ -12,6 +13,54 @@ import (
"agent-native/runtime/tools"
)
func TestRunTerminalToolFailureKeepsUsage(t *testing.T) {
hits, calls := 0, 0
reg := tools.NewRegistry(t.TempDir())
reg.Register(hitTool{hits: &hits})
stream := func(context.Context, string, []llm.Message, []llm.ToolDecl, llm.StreamOptions) <-chan llm.Event {
calls++
out := make(chan llm.Event, 1)
out <- llm.Event{
Kind: llm.KindMessageEnd,
Message: llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopError},
Usage: llm.Usage{PromptTokens: 12, CompletionTokens: 5, TotalTokens: 17},
Err: errors.New("incomplete tool call"),
}
close(out)
return out
}
res, err := Run(context.Background(), Config{Model: "fixture", Tools: reg, Stream: stream},
llm.Message{Role: llm.RoleUser, Content: "organize materials"})
if err != nil {
t.Fatal(err)
}
if res.Err == nil || res.StopReason != llm.StopError || res.Usage.TotalTokens != 17 ||
hits != 0 || calls != 1 {
t.Fatalf("terminal failure retried, executed a tool, or lost usage: result=%#v hits=%d calls=%d", res, hits, calls)
}
}
func TestRunTerminalErrorCannotMasqueradeAsSuccess(t *testing.T) {
stream := func(context.Context, string, []llm.Message, []llm.ToolDecl, llm.StreamOptions) <-chan llm.Event {
out := make(chan llm.Event, 1)
out <- llm.Event{Kind: llm.KindMessageEnd,
Message: llm.Message{Role: llm.RoleAssistant, Content: "incomplete", StopReason: llm.StopEnd},
Usage: llm.Usage{PromptTokens: 12, CompletionTokens: 5, TotalTokens: 17},
Err: errors.New("terminal failed"),
}
close(out)
return out
}
res, err := Run(context.Background(), Config{Model: "fixture", Stream: stream},
llm.Message{Role: llm.RoleUser, Content: "go"})
if err != nil {
t.Fatal(err)
}
if res.Err == nil || res.StopReason != llm.StopError || res.Usage.TotalTokens != 17 {
t.Fatalf("error-bearing terminal was accepted as success: %#v", res)
}
}
func script(turns ...llm.Message) llm.StreamFunc {
var mu sync.Mutex
i := 0