Merge pull request 'Split usage accounting end to end, retry by default, and take the WP6 pins (N3 WP6)' (#96) from feature/n3-usage-retry-wp6 into main
ci / vet-and-test (push) Successful in 41s

Split usage accounting end to end, retry by default, and take the WP6 pins (N3 WP6)
This commit was merged in pull request #96.
This commit is contained in:
2026-10-07 17:19:27 +00:00
12 changed files with 171 additions and 12 deletions
+23
View File
@@ -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}
+15
View File
@@ -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)
}
}
+13 -2
View File
@@ -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) != "" {
+9
View File
@@ -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 {
+10
View File
@@ -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")
}
+5 -1
View File
@@ -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,
})
}
}
}
+18 -5
View File
@@ -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 != "" {
+71
View File
@@ -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)
}
}
+4 -1
View File
@@ -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 == "" {