From 583c074aa4b6076bc56db8a679dbb34a325ba110 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 03:57:33 +0000 Subject: [PATCH] Close U3 review gaps before provider continuation Why: Missing Usage must stop a paid sequence before a tool terminal is executable; exact attachment names and host-owned choices must also remain reliable. Evidence: Each reported behavior had a failing regression before its fix. Full Backend vet/Go tests, live-tag budget tests, ten race replays, and six task-owned HTTP/Playwright scenarios passed with the aligned Agent-Native and Agent-Squad pins. Impact: Evaluation terminal accounting becomes durable before delivery; material fallback matches bounded filenames; ready choices use host options without redundant model fields. No new model calls or budget authorization. Refs: U3 r0.4 independent review Critical 1 and Important 3-4 --- deps/agent-native | 2 +- deps/agent-squad | 2 +- internal/runmgr/material_retrieve.go | 44 +++++++++- internal/runmgr/material_retrieve_test.go | 19 +++++ internal/runtime/llm_direct.go | 4 +- internal/runtime/material_flow_test.go | 16 ++++ .../work_checkpoint_eval_budget_test.go | 39 +++++++++ internal/server/work_checkpoint_eval_test.go | 85 ++++++++++++++++--- 8 files changed, 193 insertions(+), 18 deletions(-) diff --git a/deps/agent-native b/deps/agent-native index e86e74b..6e97233 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit e86e74bc36185c8531df161eb57666b5dad72bc9 +Subproject commit 6e972335c9f5dee7681f660550e2471f0584ad92 diff --git a/deps/agent-squad b/deps/agent-squad index 93cc702..ae21e1d 160000 --- a/deps/agent-squad +++ b/deps/agent-squad @@ -1 +1 @@ -Subproject commit 93cc70241c04af96059b46e12771b798ca7af54b +Subproject commit ae21e1d3a032120ccceafe8207c6b4120eba24ff diff --git a/internal/runmgr/material_retrieve.go b/internal/runmgr/material_retrieve.go index 0d680fd..ece0ed3 100644 --- a/internal/runmgr/material_retrieve.go +++ b/internal/runmgr/material_retrieve.go @@ -4,6 +4,8 @@ import ( "crypto/sha256" "encoding/hex" "strings" + "unicode" + "unicode/utf8" core "agent-system-core" ) @@ -34,7 +36,7 @@ func (c *runCtx) materialAttachmentFallback(q core.RetrievalQuery) ([]core.Retri var matched *core.ParsedDocument for i := range c.attachments { name := strings.TrimSpace(c.attachments[i].Meta.Filename) - if name == "" || !strings.Contains(q.Text, name) { + if exactFilenameIndex(q.Text, name) < 0 { continue } if matched != nil { @@ -109,11 +111,49 @@ func (c *runCtx) materialAttachmentFallback(q core.RetrievalQuery) ([]core.Retri return []core.RetrievalHit{{Doc: doc, Score: 1, Kind: "fulltext"}}, nil } +// A filename must stand as a whole query item, including when it has spaces. +// Nearby letters, digits, dots and hyphens may belong to another filename. +func exactFilenameIndex(query, name string) int { + if name == "" { + return -1 + } + boundary := func(r rune) bool { + return unicode.IsSpace(r) || strings.ContainsRune(`"'“”‘’「」『』()（）[]【】,，;；:：`, r) + } + for from := 0; from < len(query); { + at := strings.Index(query[from:], name) + if at < 0 { + return -1 + } + at += from + before := at == 0 + if !before { + r, _ := utf8.DecodeLastRuneInString(query[:at]) + before = boundary(r) + } + end := at + len(name) + after := end == len(query) + if !after { + r, _ := utf8.DecodeRuneInString(query[end:]) + after = boundary(r) + } + if before && after { + return at + } + from = at + 1 + } + return -1 +} + func materialExcerpt(full, query, filename string, docs []core.ParsedDocument, id core.ID, eff core.EffectiveSettings, limit int) string { runes := []rune(full) start := 0 matchedTerm := false - terms := strings.Fields(strings.TrimSpace(strings.Replace(query, filename, "", 1))) + withoutName := query + if at := exactFilenameIndex(query, filename); at >= 0 { + withoutName = query[:at] + query[at+len(filename):] + } + terms := strings.Fields(strings.TrimSpace(withoutName)) for _, term := range terms { if at := strings.Index(full, term); at >= 0 { matchedTerm = true diff --git a/internal/runmgr/material_retrieve_test.go b/internal/runmgr/material_retrieve_test.go index 83ab803..e78f49c 100644 --- a/internal/runmgr/material_retrieve_test.go +++ b/internal/runmgr/material_retrieve_test.go @@ -38,6 +38,25 @@ func TestMaterialAttachmentFallbackQuery(t *testing.T) { } } +func TestMaterialAttachmentFallbackNeedsExactFilename(t *testing.T) { + c := materialRetrievalContext(t) + for _, tc := range []struct { + query string + want int + }{ + {"决定.txt 上线 日期", 1}, + {`"决定.txt" 上线`, 1}, + {"非决定.txt 上线", 0}, + {"决定.txt备份 上线", 0}, + {"前缀-决定.txt 上线", 0}, + } { + hits, err := c.materialAttachmentFallback(core.RetrievalQuery{ProjectID: c.run.ProjectID, Text: tc.query}) + if err != nil || len(hits) != tc.want { + t.Fatalf("query %q matched %d current attachments; want %d (err=%v)", tc.query, len(hits), tc.want, err) + } + } +} + func TestMaterialAttachmentFallbackScope(t *testing.T) { for _, tc := range []struct { name string diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index 783432a..bdc60b9 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -882,7 +882,9 @@ func (st *directRun) runToolCtx(ctx context.Context, name, args string) agenttoo FreeInput *bool `json:"free_input"` TimeoutSec int `json:"timeout_sec"` } - if uerr := json.Unmarshal([]byte(call.Arguments), &a); uerr != nil || strings.TrimSpace(a.Question) == "" || len(a.Options) < 2 { + if uerr := json.Unmarshal([]byte(call.Arguments), &a); uerr != nil || + (!st.materialNotice && (strings.TrimSpace(a.Question) == "" || len(a.Options) < 2)) || + (st.materialNotice && len(st.materialChoiceOptions) < 3) { out = "confirm_choice 参数无效：需要 question 与 2-4 个选项（id/label/description/impact）" break } diff --git a/internal/runtime/material_flow_test.go b/internal/runtime/material_flow_test.go index 32461a4..c4dcbf8 100644 --- a/internal/runtime/material_flow_test.go +++ b/internal/runtime/material_flow_test.go @@ -251,6 +251,22 @@ func TestMaterialNoticeChoiceUsesHostOptions(t *testing.T) { } } +func TestMaterialNoticeChoiceNeedsNoModelOptions(t *testing.T) { + rc := &matCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), prof: verifyOn()}, hitlErr: errors.New("no answer")} + rc.history = []core.Message{{Role: core.RoleSystem, Parts: []core.Part{{Type: core.PartText, + Text: "\n后台子代理「材料整理」已结束。\n状态：completed\n" + + "[material-choice-options-v1]:[{\"label\":\"会议摘要\",\"basis\":\"会议决定分两期。\"},{\"label\":\"验收清单\",\"basis\":\"阈值待确认。\"},{\"label\":\"联调计划\",\"basis\":\"负责人待确认。\"}]\n" + + "[material-choice-ready]\n"}}}} + runMaterial(t, [][]agentllm.Event{ + hitlTurn(toolCall("confirm_choice", `{}`)), + textDeltaTurn("请说明下一步目标。"), + }, rc) + if len(rc.hitls) != 1 || len(rc.hitls[0].Choices) != 3 || rc.hitls[0].Choices[0].Label != "会议摘要" || + rc.hitls[0].Question != "基于整理结果，接下来生成什么？" { + t.Fatalf("host-owned choice rejected because model omitted redundant options: %+v", rc.hitls) + } +} + func TestMaterialNoticeWithoutSuggestionsRejectsDelegation(t *testing.T) { rc := &matCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), prof: verifyOn()}} rc.history = []core.Message{{Role: core.RoleSystem, Parts: []core.Part{{Type: core.PartText, diff --git a/internal/server/work_checkpoint_eval_budget_test.go b/internal/server/work_checkpoint_eval_budget_test.go index 1c1da0b..01e010f 100644 --- a/internal/server/work_checkpoint_eval_budget_test.go +++ b/internal/server/work_checkpoint_eval_budget_test.go @@ -17,6 +17,7 @@ import ( "strings" "sync/atomic" "testing" + "time" agentllm "agent-native/runtime/llm" ) @@ -169,6 +170,44 @@ func TestEvaluationStopsWhenUsageMissing(t *testing.T) { } } +func TestEvaluationDoesNotForwardToolTerminalBeforeUsageRecorded(t *testing.T) { + t.Setenv("WORK_CHECKPOINT_REMAINING_TOKENS", "60000") + dir, _ := newEvaluationEventStore(t) + var calls atomic.Int32 + release := make(chan struct{}) + provider := func(context.Context, string, []agentllm.Message, []agentllm.ToolDecl, agentllm.StreamOptions) <-chan agentllm.Event { + calls.Add(1) + out := make(chan agentllm.Event) + go func() { + defer close(out) + out <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, + StopReason: agentllm.StopToolCalls, ToolCalls: []agentllm.ToolCall{{ID: "call-1", Name: "publish_work_checkpoint", Arguments: `{}`}}}, + FinishReason: "tool_calls", StreamComplete: true} + <-release // keep the provider stream open after its terminal + }() + return out + } + stream := evalProviderStreamWithProvider(provider, dir) + time.AfterFunc(100*time.Millisecond, func() { close(release) }) + var terminal agentllm.Event + first := stream(evalTestContext(t), "fixture", nil, nil, agentllm.StreamOptions{MaxTokens: 4000}) + for ev := range first { + if ev.Kind == agentllm.KindMessageEnd { + terminal = ev + break + } + } + blocked := false + for ev := range stream(bruntime.WithProviderRunID(context.Background(), t.Name()+"-next"), "fixture", nil, nil, agentllm.StreamOptions{MaxTokens: 4000}) { + blocked = blocked || ev.Kind == agentllm.KindError + } + for range first { + } + if terminal.Err == nil || len(terminal.Message.ToolCalls) != 0 || !blocked || calls.Load() != 1 { + t.Fatalf("missing Usage tool terminal escaped stop gate: terminal=%#v blocked=%v providerCalls=%d", terminal, blocked, calls.Load()) + } +} + func TestEvaluationStopsWhenProviderRecordUnavailable(t *testing.T) { t.Setenv("WORK_CHECKPOINT_REMAINING_TOKENS", "60000") var providerCalls atomic.Int32 diff --git a/internal/server/work_checkpoint_eval_test.go b/internal/server/work_checkpoint_eval_test.go index 2adae1c..ac6c08a 100644 --- a/internal/server/work_checkpoint_eval_test.go +++ b/internal/server/work_checkpoint_eval_test.go @@ -101,6 +101,7 @@ func evalMechanismStream(ctx context.Context, model string, msgs []agentllm.Mess if len(calls) > 0 { reason = agentllm.StopToolCalls } + out <- agentllm.Event{Kind: agentllm.KindTextDelta, Text: content} out <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: content, ToolCalls: calls, StopReason: reason}, Usage: agentllm.Usage{PromptTokens: 10, CompletionTokens: 10, TotalTokens: 20}} return out } @@ -147,6 +148,9 @@ func evalMechanismStream(ctx context.Context, model string, msgs []agentllm.Mess if toolReplies >= 2 && strings.Contains(sourceText, "[fixture:choice]") { content = "整理结束，日期与验收条件需要核对；完整范围以原件为准。\n\n## 建议后续处理\n1. **会议摘要**\n - 依据：材料记载上线六月。\n2. **验收清单**\n - 依据：验收条件需要补齐。\n3. **联调计划**\n - 依据：三项需求依赖外部接口。" } + if !checkpoint && toolReplies >= 2 && strings.Contains(sourceText, "[fixture:slow-stages]") { + time.Sleep(1500 * time.Millisecond) + } if len(calls) == 0 { out <- agentllm.Event{Kind: agentllm.KindTextDelta, Text: content} } @@ -382,6 +386,14 @@ func evalRejectedCheckpointPersisted(dir string) (bool, error) { } func evalProviderStream(route map[string]any, dir string) agentllm.StreamFunc { + base, _ := route["baseUrl"].(string) + key, _ := route["apiKey"].(string) + model, _ := route["model"].(string) + provider := (&agentllm.Client{BaseURL: base, APIKey: key, HTTPClient: &http.Client{}, ReasoningSplit: strings.HasPrefix(model, "MiniMax-M")}).Stream + return evalProviderStreamWithProvider(provider, dir) +} + +func evalProviderStreamWithProvider(provider agentllm.StreamFunc, dir string) agentllm.StreamFunc { batchLimit := 120000 if raw := os.Getenv("WORK_CHECKPOINT_REMAINING_TOKENS"); raw != "" { limit, err := strconv.Atoi(raw) @@ -390,10 +402,6 @@ func evalProviderStream(route map[string]any, dir string) agentllm.StreamFunc { } batchLimit = limit } - base, _ := route["baseUrl"].(string) - key, _ := route["apiKey"].(string) - model, _ := route["model"].(string) - provider := (&agentllm.Client{BaseURL: base, APIKey: key, HTTPClient: &http.Client{}, ReasoningSplit: strings.HasPrefix(model, "MiniMax-M")}).Stream type allowance struct { Calls int Started time.Time @@ -468,24 +476,45 @@ func evalProviderStream(route map[string]any, dir string) agentllm.StreamFunc { defer cancel() usage := 0 var terminal *agentllm.Event + var streamError *agentllm.Event arguments := map[int]string{} argumentBytes := 0 for event := range in { + // A terminal is executable to the runner. Hold it until the + // provider stream has closed and its Usage/stop record is durable. + if event.Kind == agentllm.KindMessageEnd { + copy := event + if terminal != nil { + streamError = &agentllm.Event{Kind: agentllm.KindError, Err: fmt.Errorf("evaluation budget: duplicate provider terminal")} + } else { + terminal = © + usage = event.Usage.TotalTokens + if usage == 0 { + usage = event.Usage.PromptTokens + event.Usage.CompletionTokens + } + } + continue + } + if event.Kind == agentllm.KindError { + copy := event + streamError = © + continue + } + if terminal != nil { + streamError = &agentllm.Event{Kind: agentllm.KindError, Err: fmt.Errorf("evaluation budget: provider emitted data after terminal")} + continue + } if event.Kind == agentllm.KindToolCallDelta { arguments[event.ToolIndex] += event.ArgumentsDelta argumentBytes += len(event.ArgumentsDelta) } - if event.Kind == agentllm.KindMessageEnd { - copy := event - terminal = © - usage = event.Usage.TotalTokens - if usage == 0 { - usage = event.Usage.PromptTokens + event.Usage.CompletionTokens - } - } select { case out <- event: case <-ctx.Done(): + mu.Lock() + terminalFault = true + unknown = true + mu.Unlock() return } } @@ -497,7 +526,17 @@ func evalProviderStream(route map[string]any, dir string) agentllm.StreamFunc { break } } - fault := terminal == nil || terminal.Err != nil || terminal.Message.StopReason == agentllm.StopError + fault := terminal == nil || terminal.Err != nil || terminal.Message.StopReason == agentllm.StopError || streamError != nil + if terminal != nil && (streamError != nil || usage == 0) { + if streamError != nil { + terminal.Err = fmt.Errorf("evaluation budget: provider stream failed after terminal: %v", streamError.Err) + } else { + terminal.Err = fmt.Errorf("evaluation budget: provider usage unavailable") + } + terminal.Message.StopReason = agentllm.StopError + terminal.Message.ToolCalls = nil + fault = true + } finishReason, streamComplete := "", false if terminal != nil { finishReason, streamComplete = terminal.FinishReason, terminal.StreamComplete @@ -520,8 +559,28 @@ func evalProviderStream(route map[string]any, dir string) agentllm.StreamFunc { "usageUnknown": unknown, "terminalFault": terminalFault, "modelRuns": len(groups)}, "", " ") if err := os.WriteFile(filepath.Join(dir, "provider-records.json"), raw, 0600); err != nil { terminalFault = true // no durable accounting: never issue another paid request + if terminal != nil { + terminal.Err = fmt.Errorf("evaluation budget: provider record unavailable: %w", err) + terminal.Message.StopReason = agentllm.StopError + terminal.Message.ToolCalls = nil + } } mu.Unlock() + if terminal != nil { + select { + case out <- *terminal: + case <-ctx.Done(): + } + } else { + failure := agentllm.Event{Kind: agentllm.KindError, Err: fmt.Errorf("evaluation budget: provider terminal unavailable")} + if streamError != nil { + failure = *streamError + } + select { + case out <- failure: + case <-ctx.Done(): + } + } }() return out }