From 8c5612a0a369006a33b31919134234924235607b Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 7 Oct 2026 17:17:56 +0000 Subject: [PATCH] Split usage accounting end to end, retry by default, and take the WP6 pins (N3 WP6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: subagent runs recorded zero tokens, cache hits were invisible in run records, the usage event skipped terminal paths, and stream retry existed only on the direct-chat path (audit N20 residual). Evidence: go test -count=1 ./... green; TestRetryAttemptsMapping (setting 0=off→-1, N→N), TestDirectRunUsageSplitOnRecord (Last* and cache land in the run usage), TestSubRunAccountsTokens (subagent accounting), TestAddUsageAccumulatesSplitFields (handle chain); pins carry Agent-Native#35/#37/#38/#39, Agent-Squad#41, Core Usage fields. Impact: run.Usage carries Last*/cache on every path; subagent runs are accounted; runtime.llmRetries maps onto the library default (setting 0 disables via -1); the verifier wraps retry+fallback explicitly; retry events surface as process parts. Refs: Agent-Native#35-#39, Agent-Squad#41, N3 plan WP6 --- deps/agent-native | 2 +- deps/agent-squad | 2 +- deps/agent-system-core | 2 +- internal/runmgr/runmgr.go | 23 ++++++++++ internal/runmgr/subagent_n2_test.go | 15 ++++++ internal/runtime/llm.go | 15 +++++- internal/runtime/llm_direct.go | 9 ++++ internal/runtime/llm_test.go | 10 ++++ internal/runtime/squad_runner.go | 6 ++- internal/runtime/sub_loop.go | 23 ++++++++-- internal/runtime/usage_n3_test.go | 71 +++++++++++++++++++++++++++++ internal/runtime/verify.go | 5 +- 12 files changed, 171 insertions(+), 12 deletions(-) create mode 100644 internal/runtime/usage_n3_test.go diff --git a/deps/agent-native b/deps/agent-native index 0e03cc2..1ac21c6 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit 0e03cc2144471661a4c6423fb1dd8779f6454c2a +Subproject commit 1ac21c68d4e2c2728d92d6db67a30ff06e0066d3 diff --git a/deps/agent-squad b/deps/agent-squad index bf7fe84..c462bc4 160000 --- a/deps/agent-squad +++ b/deps/agent-squad @@ -1 +1 @@ -Subproject commit bf7fe8429bdfa7ae1cbf021de11111b9e581fa2f +Subproject commit c462bc4ae5bbd96e6caeb09c2ad24bf87ed153ab diff --git a/deps/agent-system-core b/deps/agent-system-core index 0b5c774..1b7e2c0 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit 0b5c7747a3165bf0d0f2219204b7b70ec3ed7645 +Subproject commit 1b7e2c0f4793766032810d11cee2a68615ce9df6 diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 020580d..c9fe329 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -101,6 +101,10 @@ type runHandle struct { started time.Time promptTokens int completionTokens int + lastPromptTokens int + lastCompletionTokens int + cacheReadTokens int + cacheWriteTokens int } type hitlAnswer struct { @@ -236,10 +240,15 @@ func (c *runCtx) ConversationSandbox() string { } // AddUsage 累加本 run 的模型用量，finishRun 写入 run.Usage。 +// 计费拆分（N3 WP6）：Last* 覆盖（最近一次请求）、Cache* 累计。 func (c *runCtx) AddUsage(u core.Usage) { c.mgr.mu.Lock() c.h.promptTokens += u.PromptTokens c.h.completionTokens += u.CompletionTokens + c.h.lastPromptTokens = u.LastPromptTokens + c.h.lastCompletionTokens = u.LastCompletionTokens + c.h.cacheReadTokens += u.CacheReadTokens + c.h.cacheWriteTokens += u.CacheWriteTokens c.mgr.mu.Unlock() } func (c *runCtx) Subtask() string { return c.subTask } @@ -1080,6 +1089,10 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, usage.PromptTokens = rc.h.promptTokens usage.CompletionTokens = rc.h.completionTokens usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens + usage.LastPromptTokens = rc.h.lastPromptTokens + usage.LastCompletionTokens = rc.h.lastCompletionTokens + usage.CacheReadTokens = rc.h.cacheReadTokens + usage.CacheWriteTokens = rc.h.cacheWriteTokens m.mu.Unlock() if m.shuttingDown.Load() && status == core.RunCancelled { status = core.RunInterrupted @@ -1395,7 +1408,17 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool return runner.Execute(rc) }() + // 子代理记账（N3 WP6）：token 用量随子 run 入账（此前只有 DurationMs）。 usage := core.Usage{DurationMs: float64(time.Since(rc.h.started).Milliseconds())} + m.mu.Lock() + usage.PromptTokens = rc.h.promptTokens + usage.CompletionTokens = rc.h.completionTokens + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens + usage.LastPromptTokens = rc.h.lastPromptTokens + usage.LastCompletionTokens = rc.h.lastCompletionTokens + usage.CacheReadTokens = rc.h.cacheReadTokens + usage.CacheWriteTokens = rc.h.cacheWriteTokens + m.mu.Unlock() status := core.RunCompleted summary := "完成" payload := core.AgentResultPayload{AgentRun: rc.run.ID, Status: status, Summary: summary} diff --git a/internal/runmgr/subagent_n2_test.go b/internal/runmgr/subagent_n2_test.go index 5720480..83fa2a0 100644 --- a/internal/runmgr/subagent_n2_test.go +++ b/internal/runmgr/subagent_n2_test.go @@ -207,3 +207,18 @@ func TestResumeSkipsCompletedMarker(t *testing.T) { t.Fatalf("completed 终态不得续跑：before=%v after=%v", before, ids) } } + +// N3 WP6：子代理记账链——AddUsage 累计拆分字段，finishChild 合并进 child.Usage。 +func TestAddUsageAccumulatesSplitFields(t *testing.T) { + c, _ := newSpawnEnv(t, &gateRunner{}) + c.h = &runHandle{} + c.AddUsage(core.Usage{PromptTokens: 100, CompletionTokens: 10, LastPromptTokens: 100, LastCompletionTokens: 10, CacheReadTokens: 50, CacheWriteTokens: 4}) + c.AddUsage(core.Usage{PromptTokens: 60, CompletionTokens: 6, LastPromptTokens: 60, LastCompletionTokens: 6, CacheReadTokens: 20, CacheWriteTokens: 2}) + c.mgr.mu.Lock() + defer c.mgr.mu.Unlock() + if c.h.promptTokens != 160 || c.h.completionTokens != 16 || + c.h.lastPromptTokens != 60 || c.h.lastCompletionTokens != 6 || + c.h.cacheReadTokens != 70 || c.h.cacheWriteTokens != 6 { + t.Fatalf("记账链：%+v", c.h) + } +} diff --git a/internal/runtime/llm.go b/internal/runtime/llm.go index fed67fd..d08bff7 100644 --- a/internal/runtime/llm.go +++ b/internal/runtime/llm.go @@ -205,9 +205,10 @@ func (l *LLMRunner) Execute(rc RunCtx) error { stream := l.Stream if stream == nil { + // 重试经库默认（WP6：runtime.llmRetries → Config.RetryAttempts 映射）， + // 此处不再显式包装（防双重）。 client := openLLMClient(route.Base, route.Key) - retries := int(eff.GetNumber("runtime.llmRetries", 2)) - stream = agentllm.WithRetry(client.Stream, retries, nil) + stream = client.Stream } // 知识注入：当前消息包含 @knowledge 引用时检索注入（最小实现：显式引用）。 @@ -404,6 +405,16 @@ func outputTokenCap(budget core.RunBudget, eff core.EffectiveSettings, key strin // 组合根经 runCtx 注入，runmgr 的 storeMemory 实现之）。 type memoryStore = agentmemory.Store +// retryAttemptsOf 把 runtime.llmRetries 的设置语义（0=不重试，N=N 次） +// 映射到库 Config.RetryAttempts（0=默认 2，负值=关，正数=次数）。 +func retryAttemptsOf(eff core.EffectiveSettings) int { + n := int(eff.GetNumber("runtime.llmRetries", 2)) + if n <= 0 { + return -1 + } + return n +} + // agentNameOf 是 OTel gen_ai.agent.name 的取值（画像 ID，空回退 agent-system）。 func agentNameOf(prof core.AgentProfile) string { if strings.TrimSpace(prof.ID) != "" { diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index 145554b..ffcc6a2 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -211,6 +211,7 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin Trace: l.Trace, AgentName: agentNameOf(rc.Profile()), Memory: st.memoryStore, + RetryAttempts: retryAttemptsOf(eff), UnknownTool: func(call agentllm.ToolCall) agenttools.Result { return st.runTool(call.Name, call.Arguments) }, @@ -275,6 +276,9 @@ func (st *directRun) noteToolDelta(ev agentllm.Event) { func (st *directRun) onEvent(ev nruntime.Event) { switch ev.Type { + case "auto_retry_start": + // 重试可观测（N3 WP6）：沿用 process part 形态。 + st.finalParts = append(st.finalParts, core.Part{Type: core.PartText, Text: "模型在输出前失败，正在重试（" + ev.Text + "）", Process: true}) case "turn_start": st.responding = false st.acting = false @@ -314,6 +318,11 @@ func (st *directRun) onEvent(ev nruntime.Event) { st.usageTotal.PromptTokens += ev.PromptTokens st.usageTotal.CompletionTokens += ev.CompletionTokens st.usageTotal.TotalTokens += ev.TotalTokens + // 计费拆分（N3 WP6）：上一请求覆盖、缓存命中累计。 + st.usageTotal.LastPromptTokens = ev.PromptTokens + st.usageTotal.LastCompletionTokens = ev.CompletionTokens + st.usageTotal.CacheReadTokens += ev.CacheReadTokens + st.usageTotal.CacheWriteTokens += ev.CacheWriteTokens st.queued = append([]agentllm.ToolCall(nil), ev.Message.ToolCalls...) hasCalls := len(ev.Message.ToolCalls) > 0 if st.turnThink.Len() > 0 { diff --git a/internal/runtime/llm_test.go b/internal/runtime/llm_test.go index a29305a..699f709 100644 --- a/internal/runtime/llm_test.go +++ b/internal/runtime/llm_test.go @@ -43,6 +43,7 @@ type fakeRunCtx struct { bgErr error sandbox string // 非空时 ConversationSandbox 返回之（N3 WP2 转录测试） memEntries []memEntry // 非空时 StoreMemory 返回之（N3 WP5 记忆注入测试） + usageTotal core.Usage // AddUsage 累计（N3 WP6 记账断言用） bgCalls int } @@ -62,6 +63,15 @@ func (f *fakeRunCtx) Emit(typ core.StreamEventType, payload any) error { return nil } func (f *fakeRunCtx) SetAssistantParts(parts []core.Part) error { f.assistant = parts; return nil } +func (f *fakeRunCtx) AddUsage(u core.Usage) { + f.usageTotal.PromptTokens += u.PromptTokens + f.usageTotal.CompletionTokens += u.CompletionTokens + f.usageTotal.TotalTokens += u.TotalTokens + f.usageTotal.LastPromptTokens = u.LastPromptTokens + f.usageTotal.LastCompletionTokens = u.LastCompletionTokens + f.usageTotal.CacheReadTokens += u.CacheReadTokens + f.usageTotal.CacheWriteTokens += u.CacheWriteTokens +} func (f *fakeRunCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) { return core.HITLResponse{}, fmt.Errorf("not implemented") } diff --git a/internal/runtime/squad_runner.go b/internal/runtime/squad_runner.go index 67044fb..5abb248 100644 --- a/internal/runtime/squad_runner.go +++ b/internal/runtime/squad_runner.go @@ -53,7 +53,11 @@ func (s *SquadRunner) Execute(rc RunCtx) error { toolParts = append(toolParts, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: ev.Tool, Status: "ok", Summary: truncateRun(ev.Detail, 160)}}) case "usage": if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok { - sink.AddUsage(core.Usage{PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens}) + sink.AddUsage(core.Usage{ + PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens, + LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens, + CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens, + }) } } } diff --git a/internal/runtime/sub_loop.go b/internal/runtime/sub_loop.go index 566b334..1d34ef8 100644 --- a/internal/runtime/sub_loop.go +++ b/internal/runtime/sub_loop.go @@ -111,7 +111,12 @@ func (l *SubLLMRunner) Execute(rc RunCtx) error { PermissionAsk: PermissionAskOf(rc), Trace: l.Trace, AgentName: agentNameOf(rc.Profile()), + RetryAttempts: retryAttemptsOf(rc.Settings()), }, user) + // 子代理记账（N3 WP6）：token 用量入账（此前只有 DurationMs）。 + if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok { + sink.AddUsage(st.usageTotal) + } if len(st.parts) == 0 { st.parts = append(st.parts, core.Part{Type: core.PartText, Text: "（子代理未返回内容）"}) } @@ -169,11 +174,12 @@ type subSlot struct { } type subRun struct { - rc RunCtx - parts []core.Part - queued []agentllm.ToolCall - inflight *subSlot - turnText strings.Builder + rc RunCtx + parts []core.Part + queued []agentllm.ToolCall + inflight *subSlot + turnText strings.Builder + usageTotal core.Usage } func (st *subRun) onEvent(ev nruntime.Event) { @@ -184,6 +190,13 @@ func (st *subRun) onEvent(ev nruntime.Event) { st.turnText.WriteString(ev.Text) _ = st.rc.Emit(core.EvAgentProgress, core.AgentProgressPayload{AgentRun: st.rc.Run().ID, Note: truncateRun(st.turnText.String(), 80)}) case "message_end": + st.usageTotal.PromptTokens += ev.PromptTokens + st.usageTotal.CompletionTokens += ev.CompletionTokens + st.usageTotal.TotalTokens += ev.TotalTokens + st.usageTotal.LastPromptTokens = ev.PromptTokens + st.usageTotal.LastCompletionTokens = ev.CompletionTokens + st.usageTotal.CacheReadTokens += ev.CacheReadTokens + st.usageTotal.CacheWriteTokens += ev.CacheWriteTokens st.queued = append([]agentllm.ToolCall(nil), ev.Message.ToolCalls...) hasCalls := len(ev.Message.ToolCalls) > 0 if body := strings.TrimSpace(st.turnText.String()); body != "" { diff --git a/internal/runtime/usage_n3_test.go b/internal/runtime/usage_n3_test.go new file mode 100644 index 0000000..686abed --- /dev/null +++ b/internal/runtime/usage_n3_test.go @@ -0,0 +1,71 @@ +package runtime + +import ( + "context" + "testing" + + agentllm "agent-native/runtime/llm" + core "agent-system-core" +) + +// ---- N3 WP6：计费拆分与重试收口（backend 侧）---- + +// retryAttemptsOf 映射：设置 0=不重试→-1；N→N；默认 2→2。 +func TestRetryAttemptsMapping(t *testing.T) { + if got := retryAttemptsOf(core.EffectiveSettings{}); got != 2 { + t.Fatalf("默认应 2，得到 %d", got) + } + if got := retryAttemptsOf(core.EffectiveSettings{"runtime": map[string]any{"llmRetries": float64(0)}}); got != -1 { + t.Fatalf("设置 0 应映射为关闭（-1），得到 %d", got) + } + if got := retryAttemptsOf(core.EffectiveSettings{"runtime": map[string]any{"llmRetries": float64(5)}}); got != 5 { + t.Fatalf("设置 5 应映射为 5，得到 %d", got) + } +} + +// 直连 run 的用量拆分：message_end 到达后 Last* 覆盖、cache 累计入账。 +func TestDirectRunUsageSplitOnRecord(t *testing.T) { + rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}} + stream := func(ctx context.Context, model string, msgs []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + ch := make(chan agentllm.Event, 1) + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, + Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "x", StopReason: agentllm.StopEnd}, + Usage: agentllm.Usage{PromptTokens: 120, CompletionTokens: 12, TotalTokens: 132, CacheReadTokens: 80, CacheWriteTokens: 6}, + } + close(ch) + return ch + } + r := &LLMRunner{Stream: stream} + if err := r.Execute(rc); err != nil { + t.Fatalf("Execute: %v", err) + } + u := rc.usageTotal + if u.PromptTokens != 120 || u.CompletionTokens != 12 || + u.LastPromptTokens != 120 || u.LastCompletionTokens != 12 || + u.CacheReadTokens != 80 || u.CacheWriteTokens != 6 { + t.Fatalf("用量拆分未入账：%+v", u) + } +} + +// 子代理记账：subRun 的 message_end 累计 token 并经 AddUsage 入账。 +func TestSubRunAccountsTokens(t *testing.T) { + rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}} + stream := func(ctx context.Context, model string, msgs []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + ch := make(chan agentllm.Event, 1) + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, + Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "子结果", StopReason: agentllm.StopEnd}, + Usage: agentllm.Usage{PromptTokens: 66, CompletionTokens: 6, TotalTokens: 72, CacheReadTokens: 30}, + } + close(ch) + return ch + } + r := &SubLLMRunner{Stream: stream} + if err := r.Execute(rc); err != nil { + t.Fatalf("Execute: %v", err) + } + u := rc.usageTotal + if u.PromptTokens != 66 || u.CompletionTokens != 6 || + u.LastPromptTokens != 66 || u.LastCompletionTokens != 6 || u.CacheReadTokens != 30 { + t.Fatalf("子代理用量未入账：%+v", u) + } +} diff --git a/internal/runtime/verify.go b/internal/runtime/verify.go index 43edcca..37df912 100644 --- a/internal/runtime/verify.go +++ b/internal/runtime/verify.go @@ -33,7 +33,10 @@ func (l *LLMRunner) runVerifier(rc RunCtx, body string, evidence []toolEvidence, return gates.Verdict{}, fmt.Errorf("验证器调用失败") } client := openLLMClient(route.Base, route.Key) - stream = client.Stream + // 验证器不经内核循环：显式接重试（WP6 同映射）与 5xx 回退链。 + stream = agentllm.WithFallback( + agentllm.WithRetry(client.Stream, retryAttemptsOf(rc.Settings()), nil), + ChatFallbacks(rc.Settings()), nil) } model := route.Model if model == "" {