package runtime import ( "context" "encoding/json" "fmt" "strings" "sync" "time" "agent-native/runtime/hooks" "agent-native/runtime/llm" "agent-native/runtime/reminders" "agent-native/runtime/session" "agent-native/runtime/tools" "logging-system/logsys" ) const ( ExecSequential = "sequential" ExecParallel = "parallel" maxStopContinuations = 3 maxLengthContinuations = 3 maxRepeatReminders = 3 ) const repeatReminderText = "你已连续 3 次以相同参数调用同一工具。如果结果不符合预期,请更换参数或做法,不要再重复这一调用。" const lengthContinuePrompt = "上一段回复因为达到长度上限被截断。请从断点接着写,不要重复已经写出的内容。" // Event is a lifecycle notice for logs and UI. type Event struct { Type string Message llm.Message Tool string Text string // Detail carries tool arguments/results or reasoning deltas for // process-visualization consumers; log writers may ignore it. Detail string // PromptTokens / CompletionTokens / TotalTokens 在 type=usage 或 // type=message_end 时有意义,供调用方累计用量。 PromptTokens int CompletionTokens int TotalTokens int } // Config drives one agent session. type Config struct { Model string SystemPrompt string Tools *tools.Registry Stream llm.StreamFunc Session *session.Store // Trace 开始一个 span,返回的函数结束它。nil 表示不采集。 // 属性里不要放提示词正文。 Trace func(name string, attrs map[string]string) func() MaxTurns int ToolExecution string GetSteering func() []llm.Message GetFollowUp func() []llm.Message // OnEvent 观察循环事件。同一次 Run 内不会并发调用 // (与 BeforeTool、AfterTool 共用一把锁串行)。 OnEvent func(Event) // ThinkingLevel is forwarded on each provider request. ThinkingLevel string // Resolve, when set, overrides Model and ThinkingLevel at the start of each turn. Resolve func() (model, thinking string) // Log 可选。nil 时丢弃日志,不改变循环结果。 Log *logsys.Logger // ExecutionID 只用于日志字段,不参与控制流。 ExecutionID string // Budget 可选。任一上限命中即停止并返回 budget_exceeded 错误(StopError); // 零值字段不设限。截止时刻与外层 ctx 先到者生效。 Budget *Budget // History 是会话转录为空时的初始转录。以 system 开头时,不再前置 // SystemPrompt 和工具清单。不回写进 Session。 History []llm.Message // Stop 在模型不再调用工具、且本轮没有 follow-up 时询问是否续跑。 // nil 表示不询问。内核最多接受 3 次 Continue。 Stop func(last llm.Message, info StopInfo) StopDecision // BeforeTool 返回错误时不执行工具,但仍写一条 IsError 的 tool 结果。 // 同一次 Run 内与 OnEvent、AfterTool 串行,不会并发调用。 BeforeTool func(call llm.ToolCall) error // UnknownTool 处理注册表里没有的调用。nil 时仍返回 unknown tool。 UnknownTool func(call llm.ToolCall) tools.Result // AfterTool 只观察工具结果,不能改写。 // 同一次 Run 内与 OnEvent、BeforeTool 串行,不会并发调用。 AfterTool func(call llm.ToolCall, res llm.Message) // Reminders 在每次调用模型前提供提醒。正文包在 system-reminder 里, // 并入最后一条工具结果或用户消息,只进入本次请求,不写进 Session。 Reminders func(turn int) []string // RepeatGuard 为真时,同名且参数按键排序后相同的调用连续 3 次,注入一条提醒。 // 每个 run 最多 3 条,不中断循环。 RepeatGuard bool // ContinueOnLength 为真时,FinishReason=length 且没有工具调用就续写。 // 续写提示不写进 Session。次数由 MaxLengthContinuations 决定。 ContinueOnLength bool // MaxLengthContinuations 是 length 续写的次数上限。 // 0 表示默认 3 次;负数表示不限制。 MaxLengthContinuations int // ContextWindow 是模型窗口的字符估算。0 表示用保守默认值。 // 超过窗口减去 OutputReserve 与 ContextBuffer 时,清理最近若干条之前的旧工具结果。 ContextWindow int // OutputReserve 是留给模型输出的字符数。0 表示默认预留。 OutputReserve int // ContextBuffer 是窗口估算的余量。0 表示默认余量。 ContextBuffer int // KeepToolResults 是清理时保留的最近工具结果条数。0 表示 5。 KeepToolResults int // ResultsDir 非空时,被清理的旧工具结果完整内容落盘到该目录 // (占位写路径);空则占位指向会话记录(旧行为)。 ResultsDir string // Hooks 是进程内钩子。空总线不改变停止、提醒和拒绝次数。 Hooks *hooks.Bus // PermissionAsk 在钩子对一次工具调用表态 ask 时向宿主请求裁决。 // nil 时保持旧行为:ask 与 deny 一样按拒绝处理。 // 返回 Allow 时工具照常执行;宿主需要「本会话始终允许」时自行记录, // 内核不替宿主保存放行状态。 PermissionAsk func(ctx context.Context, call llm.ToolCall) AskVerdict } // AskVerdict 是宿主对一次 ask 决策的裁决。 type AskVerdict struct { Allow bool Reason string // 拒绝时写进配对 tool 结果的理由;放行时忽略 } // StopInfo 是 Stop 钩子看到的本 run 进度。 type StopInfo struct { Continuations int ToolCalls int } // StopDecision 决定模型停下之后是否再跑一轮。 type StopDecision struct { Continue bool Replacement string Reason string Notice string } // Budget 是循环层的执行预算;与 exec.Budget 字段同构,由适配层换算。 type Budget struct { MaxPromptTokens int MaxCompletionTokens int MaxToolCalls int Deadline time.Time } // Result is the final assistant text and the transcript produced by one run. type Result struct { Text string Messages []llm.Message StopReason llm.StopReason Usage llm.Usage Notice string Err error Retryable bool } // Run executes the turn loop until the model stops, follow-ups drain, or the turn cap is hit. func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) { if cfg.Stream == nil { return Result{StopReason: llm.StopError, Err: errString("stream func is required")}, nil } if cfg.MaxTurns <= 0 { cfg.MaxTurns = 12 } if cfg.ToolExecution == "" { cfg.ToolExecution = ExecSequential } serializeHostCallbacks(&cfg) endRun := startSpan(cfg, "invoke_agent", map[string]string{"model": cfg.Model}) defer endRun() var messages []llm.Message if cfg.Session != nil { loaded, err := cfg.Session.Load() if err != nil { return Result{}, err } messages = loaded } if err := recoverPending(ctx, cfg); err != nil { return Result{}, err } // 尾部配对兜底:无论进程死在 persist(assistant) 与 commitToolIntent // 之间的哪个窗口,补全尾部 assistant 的无结果调用,转录配对恒成立。 if err := repairOrphanTail(cfg); err != nil { return Result{}, err } if cfg.Session != nil { loaded, err := cfg.Session.Load() if err != nil { return Result{}, err } messages = loaded } if len(messages) == 0 && len(cfg.History) > 0 { messages = append([]llm.Message{}, cfg.History...) } if len(messages) == 0 || messages[0].Role != llm.RoleSystem { sys := llm.Message{Role: llm.RoleSystem, Content: systemText(cfg)} messages = append([]llm.Message{sys}, messages...) if err := persist(cfg, sys); err != nil { return Result{}, err } } if err := persist(cfg, user); err != nil { return Result{}, err } messages = append(messages, user) emit(cfg, Event{Type: "agent_start"}) cfg.Hooks.Emit(hooks.Call{Event: hooks.SessionStart}) if out := cfg.Hooks.Emit(hooks.Call{Event: hooks.UserPromptSubmit, LastText: user.Content}); out.Decision != "" { switch out.Decision { case hooks.Deny, hooks.Block: // 输入护栏拒绝:不进入循环,以理由直接收尾(原文已持久化留痕)。 why := out.Reason if why == "" { why = "输入被护栏拒绝" } return Result{Text: why, Messages: messages, StopReason: llm.StopEnd, Notice: why}, nil case hooks.Modify: // 改写后的输入追加为新的用户消息(原文保留,可审计)。 if out.Replacement != "" { note := llm.Message{Role: llm.RoleUser, Content: out.Replacement} if err := persist(cfg, note); err != nil { return Result{}, err } messages = append(messages, note) user = note } } } var last llm.Message var usage llm.Usage toolCalls := 0 stopContinues := 0 lengthContinues := 0 lengthNudge := false var repeats repeatState defer func() { cfg.Hooks.Emit(hooks.Call{Event: hooks.SessionEnd, LastText: last.Content, ToolCalls: toolCalls}) }() for turn := 0; turn < cfg.MaxTurns; turn++ { if ctx.Err() != nil { cfg.logger().Error("runtime turn aborted", ctx.Err()) return Result{Text: last.Content, Messages: messages, StopReason: llm.StopAborted, Err: ctx.Err()}, nil } if reason := budgetExceeded(cfg.Budget, usage, toolCalls, time.Now()); reason != "" { return budgetStop(cfg, last, messages, usage, reason), nil } emit(cfg, Event{Type: "turn_start"}) emit(cfg, Event{Type: "message_start"}) request := prepareRequest(cfg, messages, turn, &repeats, &lengthNudge) end, errEv := consume(ctx, cfg, request) if errEv != nil { cfg.logger().Error("runtime turn failed", errEv, "retryable", end.Retryable) return Result{Text: end.Message.Content, Messages: messages, StopReason: end.Message.StopReason, Err: errEv, Retryable: end.Retryable}, nil } last = end.Message usage.PromptTokens += end.Usage.PromptTokens usage.CompletionTokens += end.Usage.CompletionTokens usage.TotalTokens += end.Usage.TotalTokens usage.LastPromptTokens = end.Usage.PromptTokens usage.LastCompletionTokens = end.Usage.CompletionTokens usage.LastTotalTokens = end.Usage.TotalTokens if err := persist(cfg, last); err != nil { return Result{}, err } messages = append(messages, last) emit(cfg, Event{Type: "message_end", Message: last, PromptTokens: end.Usage.PromptTokens, CompletionTokens: end.Usage.CompletionTokens, TotalTokens: end.Usage.TotalTokens}) if last.StopReason == llm.StopToolCalls && len(last.ToolCalls) > 0 { if reason := budgetExceeded(cfg.Budget, usage, toolCalls, time.Now()); reason != "" { var err error messages, err = appendSkippedToolResults(cfg, last.ToolCalls, messages) if err != nil { return Result{}, err } return budgetStop(cfg, last, messages, usage, reason), nil } if cfg.Budget != nil && cfg.Budget.MaxToolCalls > 0 && toolCalls+len(last.ToolCalls) > cfg.Budget.MaxToolCalls { reason := fmt.Sprintf("budget_exceeded: tool calls %d+%d > max %d", toolCalls, len(last.ToolCalls), cfg.Budget.MaxToolCalls) var err error messages, err = appendSkippedToolResults(cfg, last.ToolCalls, messages) if err != nil { return Result{}, err } return budgetStop(cfg, last, messages, usage, reason), nil } toolCalls += len(last.ToolCalls) if cfg.RepeatGuard { repeats.observe(last.ToolCalls) } results := executeTools(ctx, cfg, last) for _, res := range results { if err := persist(cfg, res); err != nil { return Result{}, err } messages = append(messages, res) } if steered := drain(cfg.GetSteering); len(steered) > 0 { for _, m := range steered { if err := persist(cfg, m); err != nil { return Result{}, err } messages = append(messages, m) } } cfg.Hooks.Emit(hooks.Call{Event: hooks.PostToolBatch, Turn: turn, ToolCalls: len(last.ToolCalls)}) emit(cfg, Event{Type: "turn_end", Message: last}) continue } emit(cfg, Event{Type: "turn_end", Message: last}) if cfg.ContinueOnLength && end.FinishReason == "length" && len(last.ToolCalls) == 0 { lengthCap := cfg.MaxLengthContinuations if lengthCap == 0 { lengthCap = maxLengthContinuations } if lengthCap > 0 && lengthContinues >= lengthCap { errCap := errString("length continuation exceeded") cfg.logger().Error("runtime length continuation exceeded", errCap) return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError, Usage: usage, Err: errCap}, nil } lengthContinues++ lengthNudge = true continue } if follow := drain(cfg.GetFollowUp); len(follow) > 0 { for _, m := range follow { if err := persist(cfg, m); err != nil { return Result{}, err } messages = append(messages, m) } continue } if decision, hooked := stopDecision(cfg, last, stopContinues, toolCalls); hooked { if decision.Continue && stopContinues < maxStopContinuations { stopContinues++ if decision.Replacement != "" && len(messages) > 0 { // 替换文只以 user 消息追加(N3 WP1/N6:不再改写内存里的 // assistant——内存与转录保持一致;模型经该消息看到更正)。 note := llm.Message{Role: llm.RoleUser, Content: decision.Replacement} if err := persist(cfg, note); err != nil { return Result{}, err } messages = append(messages, note) } if decision.Reason != "" { reasonMsg := llm.Message{Role: llm.RoleUser, Content: decision.Reason} if err := persist(cfg, reasonMsg); err != nil { return Result{}, err } messages = append(messages, reasonMsg) } continue } return settle(cfg, last, messages, usage, decision.Notice), nil } return settle(cfg, last, messages, usage, ""), nil } errCap := errString("max turns exceeded") cfg.logger().Error("runtime turn cap hit", errCap) return Result{ Text: last.Content, Messages: messages, StopReason: llm.StopError, Usage: usage, Err: errCap, }, nil } func (cfg Config) logger() *logsys.Logger { if cfg.Log == nil { return logsys.Nop() } log := cfg.Log.Module("runtime") if cfg.ExecutionID == "" { return log } return log.With("execution_id", cfg.ExecutionID) } func systemText(cfg Config) string { var b strings.Builder b.WriteString(cfg.SystemPrompt) if cfg.Tools == nil { return b.String() } decls := cfg.Tools.Decls() if len(decls) == 0 { return b.String() } if b.Len() > 0 { b.WriteString("\n\n") } b.WriteString("Tools available in this workspace:") for _, d := range decls { b.WriteString("\n- ") b.WriteString(d.Name) b.WriteString(": ") b.WriteString(d.Description) } return b.String() } func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event, error) { endSpan := startSpan(cfg, "chat", map[string]string{"model": cfg.Model}) defer endSpan() var decls []llm.ToolDecl if cfg.Tools != nil { for _, d := range cfg.Tools.Decls() { decls = append(decls, llm.ToolDecl{Name: d.Name, Description: d.Description, Parameters: d.Parameters}) } } model := cfg.Model opts := llm.StreamOptions{ThinkingLevel: cfg.ThinkingLevel} if cfg.Resolve != nil { if m, th := cfg.Resolve(); m != "" || th != "" { if m != "" { model = m } if th != "" { opts.ThinkingLevel = th } } } var end llm.Event sawEnd := false for ev := range cfg.Stream(ctx, model, messages, decls, opts) { switch ev.Kind { case llm.KindTextDelta: emit(cfg, Event{Type: "text_delta", Text: ev.Text}) 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}) case llm.KindError: return ev, ev.Err case llm.KindMessageEnd: end = ev sawEnd = true } } if !sawEnd { return llm.Event{Message: llm.Message{StopReason: llm.StopError}}, errString("stream ended without a message") } return end, nil } const toolBudgetSkipContent = "not executed: tool budget exceeded" // appendSkippedToolResults 为未执行的 tool_call 各写一条错误结果,保持调用与结果成对。 // 不发工具事件:这些调用没有进入执行。 func appendSkippedToolResults(cfg Config, calls []llm.ToolCall, messages []llm.Message) ([]llm.Message, error) { for _, call := range calls { msg := llm.Message{ Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: toolBudgetSkipContent, IsError: true, } if err := persist(cfg, msg); err != nil { return nil, err } messages = append(messages, msg) } return messages, nil } // serializeHostCallbacks 让同一次 Run 的宿主回调不会并发进入。 func serializeHostCallbacks(cfg *Config) { var mu sync.Mutex if cfg.OnEvent != nil { inner := cfg.OnEvent cfg.OnEvent = func(ev Event) { mu.Lock() defer mu.Unlock() inner(ev) } } if cfg.BeforeTool != nil { inner := cfg.BeforeTool cfg.BeforeTool = func(call llm.ToolCall) error { mu.Lock() defer mu.Unlock() return inner(call) } } if cfg.AfterTool != nil { inner := cfg.AfterTool cfg.AfterTool = func(call llm.ToolCall, res llm.Message) { mu.Lock() defer mu.Unlock() inner(call, res) } } } func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message { if cfg.ToolExecution == ExecParallel && len(assistant.ToolCalls) > 1 { return executeParallel(ctx, cfg, assistant) } out := make([]llm.Message, 0, len(assistant.ToolCalls)) for _, call := range assistant.ToolCalls { if err := ctx.Err(); err != nil { out = append(out, cancelledToolMessage(call, err)) continue } out = append(out, runOne(ctx, cfg, call)) } return out } func executeParallel(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message { out := make([]llm.Message, len(assistant.ToolCalls)) var wg sync.WaitGroup for i, call := range assistant.ToolCalls { wg.Add(1) go func(i int, call llm.ToolCall) { defer wg.Done() out[i] = runOne(ctx, cfg, call) }(i, call) } wg.Wait() return out } func cancelledToolMessage(call llm.ToolCall, err error) llm.Message { return llm.Message{ Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: "not executed: " + err.Error(), IsError: true, } } func rejectIncompleteToolCall(call llm.ToolCall) string { if strings.TrimSpace(call.Name) == "" { return "工具调用未执行:名称不完整" } args := strings.TrimSpace(call.Arguments) if !json.Valid([]byte(args)) || !strings.HasPrefix(args, "{") { return "工具调用未执行:参数不是完整的 JSON 对象" } return "" } func stopDecision(cfg Config, last llm.Message, continuations, toolCalls int) (StopDecision, bool) { if cfg.Hooks != nil { out := cfg.Hooks.Emit(hooks.Call{ Event: hooks.Stop, LastText: last.Content, Continuations: continuations, ToolCalls: toolCalls, }) if out.Decision != "" { return StopDecision{ Continue: out.Decision == hooks.Block, Reason: out.Reason, Replacement: out.Replacement, Notice: out.Notice, }, true } } if cfg.Stop != nil { return cfg.Stop(last, StopInfo{Continuations: continuations, ToolCalls: toolCalls}), true } return StopDecision{}, false } func gateTool(ctx context.Context, cfg Config, call llm.ToolCall) (llm.ToolCall, llm.Message, bool) { if cfg.Hooks == nil { return call, llm.Message{}, false } out := cfg.Hooks.Emit(hooks.Call{Event: hooks.PreToolUse, Tool: call.Name, Args: call.Arguments}) switch out.Decision { case hooks.Ask: // ask 先问宿主;没有裁决端口(或宿主拒绝)时保持旧行为:按拒绝处理。 cfg.Hooks.Emit(hooks.Call{Event: hooks.PermissionRequest, Tool: call.Name, Args: call.Arguments}) why := out.Reason if cfg.PermissionAsk != nil { if verdict := cfg.PermissionAsk(ctx, call); verdict.Allow { return call, llm.Message{}, false } else if verdict.Reason != "" { why = verdict.Reason } } return call, rejectToolMessage(call, why), true case hooks.Deny, hooks.Block: return call, rejectToolMessage(call, out.Reason), true case hooks.Modify: if out.Args != "" { call.Arguments = out.Args } } return call, llm.Message{}, false } func rejectToolMessage(call llm.ToolCall, why string) llm.Message { if why == "" { why = "工具调用被拒绝" } return llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true} } func noteToolHook(cfg Config, call llm.ToolCall, msg llm.Message) { if cfg.Hooks == nil { return } ev := hooks.PostToolUse if msg.IsError { ev = hooks.PostToolUseFailure } cfg.Hooks.Emit(hooks.Call{Event: ev, Tool: call.Name, Args: call.Arguments, Result: msg.Content, IsError: msg.IsError}) } func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message { if err := ctx.Err(); err != nil { return cancelledToolMessage(call, err) } if why := rejectIncompleteToolCall(call); why != "" { msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true} emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments}) emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: why}) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) } return msg } emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments}) if gated, msg, blocked := gateTool(ctx, cfg, call); blocked { emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) noteToolHook(cfg, call, msg) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) } return msg } else { call = gated } if cfg.BeforeTool != nil { if err := cfg.BeforeTool(call); err != nil { msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true} emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) noteToolHook(cfg, call, msg) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) } return msg } } emit(cfg, Event{Type: "tool_execution_update", Tool: call.Name, Detail: call.Arguments}) if err := commitToolIntent(cfg, call); err != nil { msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true} emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) return msg } endSpan := startSpan(cfg, "execute_tool", map[string]string{"tool": call.Name}) defer endSpan() res := tools.Result{Content: "no tools configured", IsError: true} known := cfg.Tools != nil && cfg.Tools.Has(call.Name) switch { case known: res = cfg.Tools.Execute(ctx, call.Name, json.RawMessage(call.Arguments)) case cfg.UnknownTool != nil: res = cfg.UnknownTool(call) case cfg.Tools != nil: res = cfg.Tools.Execute(ctx, call.Name, json.RawMessage(call.Arguments)) } msg := llm.Message{ Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: res.Content, IsError: res.IsError, } emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: res.Content}) noteToolHook(cfg, call, msg) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) } return msg } func settle(cfg Config, last llm.Message, messages []llm.Message, usage llm.Usage, notice string) Result { emit(cfg, Event{Type: "agent_end", Message: last}) emit(cfg, Event{Type: "agent_settled", Message: last}) emitUsage(cfg, usage) return Result{Text: last.Content, Messages: messages, StopReason: last.StopReason, Usage: usage, Notice: notice} } func prepareRequest(cfg Config, messages []llm.Message, turn int, repeats *repeatState, lengthNudge *bool) []llm.Message { request := clearOldResults(cfg, persistedProjection(cfg, messages)) return reminders.Attach(request, turnReminderTexts(cfg, turn, repeats, lengthNudge)) } // persistedProjection 在有会话存储时只采用已经落盘的消息。 // steer 与 follow-up 必须先写入存储,下一轮请求才会看见它们。 func persistedProjection(cfg Config, messages []llm.Message) []llm.Message { if cfg.Session == nil { return messages } loaded, err := cfg.Session.Load() if err != nil { return messages } return loaded } const unknownSideEffect = "副作用未知,未重试" // crashInterrupted 是尾部兜底配对的正文(进程死在 persist(assistant) 与 // commitToolIntent 之间的窗口时使用)。 const crashInterrupted = "进程中断,该调用未执行" func recoverPending(ctx context.Context, cfg Config) error { if cfg.Session == nil { return nil } pending, err := cfg.Session.PendingIntents() if err != nil { return err } for _, in := range pending { current := tools.ReplayNever if cfg.Tools != nil { current = cfg.Tools.Replay(in.Name) } msg := llm.Message{ Role: llm.RoleTool, Name: in.Name, ToolCallID: in.CallID, Content: unknownSideEffect, IsError: true, } if in.Replay == tools.ReplaySafe && current == tools.ReplaySafe && cfg.Tools != nil { res := cfg.Tools.Execute(ctx, in.Name, json.RawMessage(in.Arguments)) msg.Content = res.Content msg.IsError = res.IsError } if err := persist(cfg, msg); err != nil { return err } } return nil } // repairOrphanTail 兜底配对:找到转录里最后一个带 tool_calls 的 assistant, // 凡没有对应 tool 结果的调用,补一条「进程中断,该调用未执行」的 IsError // 结果并落盘。与意图日志正交(按 CallID 判重,已配对的跳过;意图路径 // 已处理的执行中死亡不受影响)。历史深处、其后已有新消息的孤立调用不在 // 此处修补(追加只落在尾部,位置语义不破)——那种形态由毒化轮转覆盖。 func repairOrphanTail(cfg Config) error { if cfg.Session == nil { return nil } msgs, err := cfg.Session.Load() if err != nil { return err } tail := -1 for i := len(msgs) - 1; i >= 0; i-- { m := msgs[i] if m.Role == llm.RoleAssistant && len(m.ToolCalls) > 0 { tail = i break } if m.Role == llm.RoleTool { continue // 尾部连续 tool 结果属于同一条 assistant,继续上溯 } break } if tail < 0 { return nil } results := map[string]bool{} for _, m := range msgs { if m.Role == llm.RoleTool && m.ToolCallID != "" { results[m.ToolCallID] = true } } for _, call := range msgs[tail].ToolCalls { if results[call.ID] { continue } pair := llm.Message{ Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: crashInterrupted, IsError: true, } if err := persist(cfg, pair); err != nil { return err } results[call.ID] = true } return nil } func commitToolIntent(cfg Config, call llm.ToolCall) error { if cfg.Session == nil { return nil } replay := tools.ReplayNever if cfg.Tools != nil { replay = cfg.Tools.Replay(call.Name) } return cfg.Session.AppendToolIntent(session.ToolIntent{ CallID: call.ID, Name: call.Name, Arguments: call.Arguments, Replay: replay, }) } func startSpan(cfg Config, name string, attrs map[string]string) func() { if cfg.Trace == nil { return func() {} } end := cfg.Trace(name, attrs) if end == nil { return func() {} } return end } func clearOldResults(cfg Config, messages []llm.Message) []llm.Message { window := cfg.ContextWindow if window <= 0 { window = session.DefaultContextWindow } reserve := cfg.OutputReserve if reserve <= 0 { reserve = session.DefaultOutputReserve } buffer := cfg.ContextBuffer if buffer <= 0 { buffer = session.DefaultContextBuffer } if !session.NeedsClear(session.EstimateChars(messages), window, reserve, buffer) { return messages } return session.ClearOldToolResultsTo(messages, cfg.KeepToolResults, cfg.ResultsDir) } func turnReminderTexts(cfg Config, turn int, repeats *repeatState, lengthNudge *bool) []string { var extra []string if cfg.Reminders != nil { for _, text := range cfg.Reminders(turn) { if strings.TrimSpace(text) == "" { continue } extra = append(extra, text) } } if lengthNudge != nil && *lengthNudge { extra = append(extra, lengthContinuePrompt) *lengthNudge = false } if cfg.RepeatGuard && repeats != nil { due := repeats.streak/3 - repeats.reminders room := maxRepeatReminders - repeats.reminders if due > room { due = room } for i := 0; i < due; i++ { extra = append(extra, repeatReminderText) repeats.reminders++ } } return extra } type repeatState struct { sig string streak int reminders int } func (s *repeatState) observe(calls []llm.ToolCall) { for _, call := range calls { sig := call.Name + "\n" + canonicalArgs(call.Arguments) if sig == s.sig { s.streak++ continue } s.sig = sig s.streak = 1 } } func canonicalArgs(raw string) string { var v any if err := json.Unmarshal([]byte(raw), &v); err != nil { return raw } b, err := json.Marshal(v) if err != nil { return raw } return string(b) } func persist(cfg Config, msg llm.Message) error { if cfg.Session == nil { return nil } return cfg.Session.Append(msg) } func drain(fn func() []llm.Message) []llm.Message { if fn == nil { return nil } return fn() } func emitUsage(cfg Config, usage llm.Usage) { emit(cfg, Event{Type: "usage", PromptTokens: usage.PromptTokens, CompletionTokens: usage.CompletionTokens, TotalTokens: usage.TotalTokens}) } func emit(cfg Config, ev Event) { if cfg.OnEvent != nil { cfg.OnEvent(ev) } } // budgetExceeded 返回命中的预算说明;空串表示预算仍充足(或未设预算)。 // 只在「即将消耗更多预算」的节点调用(轮次顶端 / 工具批执行前)。 func budgetExceeded(b *Budget, usage llm.Usage, toolCalls int, now time.Time) string { if b == nil { return "" } if !b.Deadline.IsZero() && now.After(b.Deadline) { return fmt.Sprintf("budget_exceeded: deadline %s passed", b.Deadline.Format(time.RFC3339)) } if b.MaxPromptTokens > 0 && usage.LastPromptTokens > b.MaxPromptTokens { return fmt.Sprintf("budget_exceeded: prompt tokens %d > max %d", usage.LastPromptTokens, b.MaxPromptTokens) } if b.MaxCompletionTokens > 0 && usage.CompletionTokens > b.MaxCompletionTokens { return fmt.Sprintf("budget_exceeded: completion tokens %d > max %d", usage.CompletionTokens, b.MaxCompletionTokens) } if b.MaxToolCalls > 0 && toolCalls > b.MaxToolCalls { return fmt.Sprintf("budget_exceeded: tool calls %d > max %d", toolCalls, b.MaxToolCalls) } return "" } // budgetStop 以预算超限收尾:发事件、留日志、按 StopError 返回(不可重试)。 func budgetStop(cfg Config, last llm.Message, messages []llm.Message, usage llm.Usage, reason string) Result { emit(cfg, Event{Type: "budget_exceeded", Text: reason}) err := errString(reason) cfg.logger().Error("runtime budget exceeded", err) return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError, Usage: usage, Err: err} } type errString string func (e errString) Error() string { return string(e) }