// Package agent 实现 Agent 循环(蓝本:pigo runtime/loop.go + ccb query.ts): // Run 是薄驱动(轮次上限保护),每轮 step() 返回显式转移原因(借鉴 ccb // query/transitions.ts 的 Terminal/Continue 枚举),终止路径可枚举、可测试。 package agent import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "os" "path/filepath" "strings" "sync" "time" "git.gopher.ink/mirrors2/AutoZero/execx" "git.gopher.ink/mirrors2/AutoZero/imagecache" "git.gopher.ink/mirrors2/AutoZero/llm" "git.gopher.ink/mirrors2/AutoZero/session" "git.gopher.ink/mirrors2/AutoZero/skills" "git.gopher.ink/mirrors2/AutoZero/textx" "git.gopher.ink/mirrors2/AutoZero/tools" ) // DefaultMaxTurns 是单条消息内模型往返次数的默认上限,与 config 的钳制上限同值。 // 到顶**不是报错**:agent 会强制收尾并交代"还剩什么没做"(见 wrapUpNote),所以这个数 // 字只管"跑偏的模型最多能白转多久",不会把干到一半的任务打断。 // // 原来按「一次多文件改造」的规模定的是 120,但真实取证里一个任务单回合就能走到 // 101 步(会话 20261012-120622),再保守下去就是人被"到顶收尾"卡住。仍要留上限: // 无人值守入口(cron / -p / 渠道)没人能按停止。可用 config 的 max_turns 调小。 const DefaultMaxTurns = 1024 // 一次模型调用的重试预算,按**时间**算而不是按次数算:坏流量是成簇来的,一簇几十秒到 // 几分钟(20261012 那份会话 13 个失败回合里 8 个落在 22:00–01:00;10-15 夜里 23:55 与 // 00:04 两条连续失败是同一簇),而“2 次、间隔 400ms”根本熬不过去,症状就是干到一半 // 停下来等人打「继续」。上限仍然要有:网关真死了,无限重试就是无限烧钱。 // // 成 var 而不是 const:用例要把预算压到毫秒级,否则一条“始终失败”的用例白等 10 分钟。 var ( streamRetryFirst = time.Second streamRetryCeil = 20 * time.Second streamRetryBudget = 10 * time.Minute // streamRetrySwitchAfter 是同一个模型连试几次还没答上就换备选模型。取 3:一簇 // 抖动通常头几次就过去;连到第 3 次说明更像"这个模型/这条上游在抽风",换人比 // 继续耗着值钱(未配置 fallback_model 时这条整个跳过)。 streamRetrySwitchAfter = 3 ) // sleepCtx 睡到点;期间 ctx 被取消就立刻返回 false——叫停之后不该再等完这一段。 func sleepCtx(ctx context.Context, d time.Duration) bool { t := time.NewTimer(d) defer t.Stop() select { case <-ctx.Done(): return false case <-t.C: return true } } // ToolEvent 描述一次工具调用的生命周期,供上层观察者(如 WebUI)实时展示。 // 执行前回调一次 start(Done=false,Result 为空),执行后回调一次完成事件 // (Done=true,Result 为结果预览)。ID 用于把两次回调配对。 type ToolEvent struct { ID string // tool_call id Name string Input string // 原始 JSON 参数字符串 Result string // 完成时的结果预览(start 时为空) Done bool // SideEffect 表示这个工具会改变现实世界(write/edit/bash/agent…)。 // 前端据此在回合失败时禁用"自动整轮重试"——副作用已经发生,自动重跑 // 可能把它做第二遍。 SideEffect bool // 本地 trace:start 帧带 StartedAt,done 帧带 DurationMs(毫秒), // 便于前端在消息落盘前就显示耗时。 StartedAt int64 // unix ms DurationMs int64 // Images 是工具结果里图片的「可服务文件名」(已写入图片缓存), // 供前端在工具行内联渲染(如截图);无图片则为空。 Images []string } // SteerMessage 是一条"回合中插话":用户在当前回合还没结束时补充的输入。 // Images 是图片缓存里的文件名(不是像素);与普通用户消息一样随轮次落盘。 type SteerMessage struct { Text string Images []string } // ApprovalVerdict 是交互审批的裁决。 type ApprovalVerdict int const ( ApprovalDeny ApprovalVerdict = iota // 拒绝本次调用 ApprovalOnce // 仅本次放行 ApprovalAlways // 放行并记住该工具(会话内信任) ) // 审批档位(Agent.Mode)。零值 "" 等同 ModeAsk(= 历史行为)。 const ( ModeAsk = "ask" // 变更前确认:副作用工具都要问 ModeAcceptEdits = "accept-edits" // 自动编辑:改文件自动放行,执行类仍要问 ModeBypass = "bypass" // 完全访问:全部放行 ) // 危险动作策略(Agent.DangerousPolicy)。零值 "" 等同 DangerousAsk(fail-closed)。 const ( DangerousAsk = "ask" // 危险动作强制确认(无交互通道时拒绝) DangerousAllow = "allow" // 关闭危险拦截,恢复全放行(供自动化) ) // Options 控制 Agent 行为。 type Options struct { ApproveAll bool // 跳过副作用工具确认(--yes / 无人值守入口) Stdin io.Reader Stdout io.Writer Stderr io.Writer OnTool func(ToolEvent) // 可空;非空时每次工具执行前后各回调一次 OnReasoning func(string) // 可空;非空时把模型的推理文本增量回传(展示用) // OnAbandon 可空;一次模型尝试失败、马上要重跑同一步时回调一次(带原因)。 // 那半截字已经通过 Stdout/OnReasoning 流给界面了,重跑会把同一段话再流一遍—— // 宿主据此把屏幕上那段作废。会话里不受影响:落盘的是 provider 返回值。 // 在 agent 工作 goroutine 上调用,实现方必须非阻塞。 OnAbandon func(reason string, dropBytes int) // OnCompact 可空;自动压缩完成后回调一次结果(UI 借此把压缩落成 // 用户可见的一等事件)。在 agent 工作 goroutine 上调用,实现方必须 // 非阻塞(TUI 投递进带缓冲的 channel)。 OnCompact func(CompactionResult) // OnNotice 可空;一句"不是模型说的、也不是工具说的"的系统说明要落进界面时回调 // (自动压缩、checkpoint 留档……)。failed 让界面把它渲染成警示而不是普通灰字。 // 与 OnCompact 同一条约束:在 agent 工作 goroutine 上调用,实现方必须非阻塞。 OnNotice func(text string, failed bool) // OnTurnEnd 可空;Run 的唯一出口 finishTurn 落完账之后回调一次,带本回合的 // 结构化收尾报告(Err 是本轮返回值,nil = 正常收尾)。工作区级的收尾(进展 // 沉淀、复盘、checkpoint 留档)挂这一处才能四端全覆盖——分散到各入口必然漏 // (复盘钩子当初就是这么挪的)。同步调用,实现方自己保证会结束; // 子 agent(Depth>0)不回调。 OnTurnEnd func(TurnReport) // Steer 可空;每个 step 边界(工具结果已回填、下一次模型调用之前)以及 // "模型本来要收尾但又有新输入"时调用,返回本轮要插入的用户消息(按顺序)。 // 宿主用它实现回合中插话:消息作为普通 user 消息落盘,模型下一步就能看到。 // 必须非阻塞;返回空表示没有插话。 Steer func() []SteerMessage // OnSteer 可空;一条插话真正写进会话后回调一次(UI 据此把气泡放到正确位置)。 OnSteer func(SteerMessage) // Occupy 可空;Run 在碰任何东西(连补齐悬挂 tool_use 都不做)之前调用一次, // 返回 error 就代表这一轮不该开始——错误原样交给宿主。只有返回 nil 才会配对 // 调用 Release。子 agent(Depth>0)不回调:占用窗口是父级拿的,子级在同一 // 个目录里干活,自己撞自己。 Occupy func() error // Release 可空;本次 Run 彻底收尾后调用一次(在 OnTurnEnd 之后——checkpoint // 的 git 操作也在占用窗口内)。用 defer,中断与报错都还得回来。 Release func() // Approval 可空的交互审批回调:返回裁决替代 stdin 询问。阻塞期间 // 由入口自行决定如何呈现(TUI 弹确认行);nil 时走 stdin [y/N]。 // // 注意它**没有 ctx、也没有超时**:不返回就会一直阻塞那条工具 goroutine。 // 所以无人值守的调用方必须自己保证不会挂住(例如 WebUI 在没人订阅时直接 // 返回拒绝,TUI 有 10 分钟兜底)。 Approval func(ApprovalRequest) ApprovalVerdict // Ask 可空的提问回调(ask_user 工具到人这一端的通道)。与 Approval 同样的 // 阻塞契约,同样的**没有 ctx、没有超时**——接入方必须自己保证会结束。 // // nil = 这个入口没有到人的交互通道:ask_user 按 Agent.AskDegradation 退化 // (默认把问题作为本轮结尾发出去,headless/cron 则自动采纳推荐项)。 // 所以渠道与定时任务**不必**注入任何东西就已经"能问且不挂死"。 Ask func(AskRequest) AskAnswer } // ApprovalRequest 是一次审批询问。 type ApprovalRequest struct { Tool string Preview string // Danger 表示这次询问来自**不可逆动作门**(DangerousPolicy=ask 拦下的 // rm -rf / git push / dd 之类),而不是常规的副作用工具确认。 // // 调用方据此区别对待:无人值守场景下普通工具可以放行,危险动作必须拒绝 // ——两者走的是同一个回调,没有这个标识就分不出来,只能要么全放行(危险) // 要么全拒绝(后台任务什么也做不了)。 Danger bool } // Agent 把 Provider、工具表和会话组装成可运行的循环。 type Agent struct { Provider llm.Provider Tools *tools.Registry Session *session.Session System string Opt Options trusted map[string]bool // 已批准的工具名 // ApprovalPolicy 工具审批策略:never | dangerous-only | always。 // ""(零值)按 dangerous-only 门控(只问副作用工具); // always 额外询问非副作用工具;never 由入口直接 ApproveAll。 ApprovalPolicy string // Mode 是审批档位(按工具类别预批准): // ""(零值) ≡ 旧 dangerous-only:所有副作用工具都要问 // ModeAsk 同零值(显式表达「变更前确认」) // ModeAcceptEdits 自动放行「改文件」类(write/edit),执行类仍需确认 // ModeBypass 全部放行(等价 ApproveAll) Mode string // DangerousPolicy 决定不可逆动作(rm -rf / git push / dd …)是否强制确认。 // 零值 "" 与 DangerousAsk 同为启用;DangerousAllow 关闭拦截。 // 该门**先于档位、ApproveAll 与信任缓存**,故 bypass / --yes 也拦得住。 DangerousPolicy string // ---- ask_user(向人提问)的策略 ---- // AskDegradation 是没有交互通道(Opt.Ask==nil)时的落点: // "" / AskDegradeDeferred 把问题渲染成本轮最终回复,回合结束(有人在场, // 只是暂时没面板:TUI、REPL、聊天渠道、WebUI) // AskDegradeUnattended 自动采纳推荐项继续(-p headless、定时任务) // 默认取 deferred 而不是 unattended:屏幕后面通常有人,把问题问出来比替他 // 做决定好;只有真正没人值守的入口该显式改成 unattended。 AskDegradation string // AskGraceSeconds 是宽限期:这段时间里界面不显示倒计时(用户正在打字时旁边 // 跳数字很扰)。0 = 立刻显示倒计时。 AskGraceSeconds int // AskTimeoutSeconds 是自动结束的总时限;<=0 = 永不自动结束(一直等)。 AskTimeoutSeconds int // AskAutoContinue 决定倒计时到点之后怎么办:true = 采纳推荐项继续, // false = 按 Deferred 收尾(问题留在界面上,等人回答)。 // 默认值来自配置(true),所以由入口显式赋值——忘了接的后果是"到点不替 // 用户决定",方向上是安全的。 AskAutoContinue bool // 本回合的提问账(见 ask.go):asked/askAuto 供复盘,askDegraded 用于 // 「连续两次没人可答就停用」的硬护栏,pendingAsk 是 Deferred 的收尾文本。 // askMu 保护它们——工具本身串行执行,但 step 收尾与渲染分处两个调用点。 askMu sync.Mutex asked int askAuto int askDegraded int pendingAsk string // Secrets 是"绝不能出现在工具结果里"的字面量(本进程配置中的 API key 等)。 // 由 main 从配置注入;凭据读取拦截挡不住绕过名字匹配的读法(变量间接、脚本里 // open()),这一层保证读到之后密钥也进不了 transcript 与外发通道。见 redactSecrets。 Secrets []string // Thinking 是本轮请求的思考/推理深度(零值 = off,不发新字段)。 Thinking llm.ThinkingLevel // SystemSuffix 可空的每轮动态后缀:在 step() 里追加到 System 之后发给模型。 // 用于注入不进 transcript、因而**躲过上下文压缩**的内容(如当前计划, // 见 PlanSection)。为空时行为与以前完全一致。 SystemSuffix func() string // ExtraSections 可空;宿主提供的「每轮注入、不进 transcript」的额外动态段 // (AutoGo 里的计划块与记忆段经它接入,躲过上下文压缩)。经 DynamicSections // 与框架自有段(放弃分支说明、子树指令)拼在一起。每次 step 重算,必须无副作用。 ExtraSections func() string // failStreak 记录每个工具的连续失败次数(用于 [STUCK] 换思路提示), // failMu 保护它:execConcurrent 会在多个 goroutine 里调 execResult。 failStreak map[string]int failMu sync.Mutex // 重复调用检测(见 anomaly.go):数的是"同名同参",盖的是**成功却空转**那一类, // 与只数失败的 failStreak 互补。repeatMu 同样是因为并发批次会多 goroutine 进来。 repeatCounts map[string]int anomalyNotes int repeatMu sync.Mutex // 上下文预算(零值走 applyDefaults 的默认)。 ContextWindow int // 模型上下文窗口(tokens,粗估) CompactAt float64 // 窗口占用率达到该比例触发摘要压缩 MinCompactTokens int // 低于该 token 数不压缩(保护短对话) MaxToolResultChars int // 单条工具结果超过该字符数在视图层截断 MaxImages int // 工作集中最多保留的图片数(驱逐更旧的截图) // StreamRetryBudget 是一次模型调用"没答上"时的重试时间预算(config 的 // retry_budget_seconds):0 = 用内置默认(10 分钟),负数 = 不限(只被停止/取消 // 打断)。无人值守入口想更快放弃或更肯等,就调它——子 agent 继承父级。 StreamRetryBudget time.Duration ImageBudgetBytes int64 // 工作集图片 base64 总字节预算(防网关 413) // ResultDir 非空时,视图层截断的超长工具结果会落盘到该目录,提示里带上 // 可读路径(模型可用 read 取回全文)。空则退回"重跑工具"提示。 // 注意:该目录必须已在 tools.SetAllowedRoots 内,否则 read 读不到。 ResultDir string // ImgCache 是图片缓存(可空)。工具结果里的截图经它落盘得到「可服务文件名」, // 写入 ToolEvent.Images,供前端在工具行内联渲染(边跑边显示,不必等整轮答完)。 // 同时回填 result.Images[*].Filename,让 collectToolImages 跳过、会话重进仍可加载。 ImgCache *imagecache.Cache // 使用统计(当前会话累计,不含子 agent)。 usedIn, usedOut int // curStep 是模型调用序号(每轮 step 自增),用于给 trace 的 Step 字段赋值。 curStep int // outputContinues 是本回合因"输出上限切断"而自动续写的次数(见 continue.go)。 // 按回合清零:它既是复盘里的一个指标,也是续写预算的计数器。 outputContinues int // 子 agent 委派控制。Depth 是当前层级(顶层 0,子 agent = 父+1); // MaxDepth 是允许的最大委派层数——**零值 0 表示禁止委派**(fail-closed)。 // MaxDepth=1 即"顶层可派发、子 agent 不能再派发"(历史行为)。 Depth int MaxDepth int SubAgentConcurrency int // 单次 agent 调用内并行子任务的上限 // nextSynthetic / syntheticTurn:本次 Run 追加的用户消息是否由系统生成 // (后台通知)。RunSynthetic 置位一次,Run 开头消费并记到 syntheticTurn, // 供 beginTurn / 复盘跳过人工提问计数。 nextSynthetic bool syntheticTurn bool // 副作用门控:stepHadSideEffect 一旦为真,本 step 的模型调用不再重试 //(提前启动工具/工具执行中途失败时,重跑会重复副作用);turnHadSideEffect // 跨 step 保留,回合失败时据此告诉宿主"别自动整轮重试"。 stepHadSideEffect bool turnHadSideEffect bool // MaxTurns 是单条消息内模型往返次数上限(<=0 时用 DefaultMaxTurns)。 // 超限后不是抛错,而是禁掉工具再要一轮收尾(见 wrapUp)。 MaxTurns int // SelfAssess 开启回合收尾的模型自评(config retro_selfassess,默认关): // 每个有工具调用的回合结束后异步发一次小调用做归因(retro.SelfAssess), // 不阻塞回合返回。复盘的硬信号记录不受此开关影响,始终落盘。 SelfAssess bool // 模型预设(config fallback_model / multimodal_model,空 = 不启用): // FallbackModel 在调用失败时粘性回落;MultimodalModel 承接含图回合 // (回合结束恢复原模型)。路由实现见 model_routing.go。 FallbackModel string MultimodalModel string // spillNS 是超长结果落盘文件名的前缀。并行子 agent 的 tool_call ID 常相同 // (call_1 等),不加前缀会互相复用同名文件读到别人的内容。 spillNS string // 压缩后的工作集缓存:Session 是完整 transcript(JSONL), // contextView = compacted + Session[compactFrom:],新消息自动接上。 // compactMu 保护 compacted/compactFrom:Run 循环的自动压缩与手动 // CompactNow(/compact 命令)可能并发访问同一对字段。 compactMu sync.Mutex compacted []llm.Message compactFrom int // 压缩交接文档:checkpoint 是结构化摘要,checkpointCovered 是它覆盖到原 // transcript 的第几条,checkpointHash 是该边界消息的指纹(失效保护)。 // 三者与 compacted/compactFrom 一起受 compactMu 保护。 checkpoint string checkpointCovered int checkpointHash string // 真实 token 观测:上一次成功请求的 prompt tokens 与当时的视图规模。 // 压缩判据优先用它(见 observedTokens),不再只靠 bytes/3。 usageMu sync.Mutex lastInputTokens int lastInputMsgs int lastSystemBytes int // abandonedNotes 是"用户已经回退掉的那条分支"的摘要文本,经 AbandonedSection // 注入 system 后缀:模型因此知道这条路试过、被放弃了(连同它查出来的事实), // 而不是整段失忆。它不进 transcript、不参与压缩、重载会话后自然消失。 // // 用**自己那把锁**:这段经 sysNow 注入,而 sysNow 在 contextView/CompactNow 里 // 是持着 compactMu 跑的——复用 compactMu 会自己等自己(真死过一次,测试里抓到的)。 abandonedMu sync.Mutex abandonedNotes []string // compactGaveUp 记着"本回合压缩试过但没成功"。不记的话每个 step 都会再撞一次 // 摘要请求——一次 30 步的回合就是 30 次注定失败的调用,慢、烧钱,而且用户什么 // 都看不到。Run 开头清零:下一回合再给一次机会(网关可能已经恢复)。 compactGaveUp bool // Skills 是可用技能列表(SKILL.md);skillFilter 非空时工具表白名单生效。 Skills []skills.Skill skillFilter map[string]bool // 读取文件时发现的嵌套 AGENTS.md(子树约定)。经 DynamicSections 每轮注入, // 不进 transcript,因此躲过上下文压缩;按绝对路径去重,并限制总量。 instructionsMu sync.Mutex discoveredInstructions []discoveredInstruction discoveredSeen map[string]bool discoveredBytes int // TurnMarker 是会话留档层的回合钩子:每开始一个回合报一次「第几个用户 // prompt」,留档条目据此归档,按 prompt 粒度回退才有意义。声明成内联小接口 // 而不是具体类型,agent 层就不必依赖留档包。 TurnMarker interface{ BeginTurn(int) } } // ErrAfterSideEffects 表示本回合已经执行过副作用工具,失败不能自动整轮重试。 // 宿主(WebUI 前端)据此把"自动重试"降级成"手动重试"。 var ErrAfterSideEffects = errors.New("turn already executed side-effecting tools; retry manually") // Run 执行一个完整 turn:持续消费模型输出与工具调用,直到模型给出最终回复。 // userMsg 为空表示续跑(历史已在会话中)。 func (a *Agent) Run(ctx context.Context, userMsg string) error { synthetic := a.nextSynthetic a.nextSynthetic = false a.syntheticTurn = synthetic // 续跑(userMsg 为空)时,未完结的回合可能已经跑过副作用;新回合从零算。 if userMsg == "" { a.turnHadSideEffect = a.historyHasSideEffects() } else { a.turnHadSideEffect = false } defer func() { a.syntheticTurn = false }() if a.trusted == nil { a.trusted = map[string]bool{} } a.applyDefaults() // 工作区互斥:整个回合(含收尾的 checkpoint)都待在占用窗口里,抢不到就不开始。 // 登记动作挂在 Run 上而不是各入口里——分散到入口必然漏一个(终端就是漏的那个: // 一个终端会话和一个 WebUI 会话并行改同一个项目,互相覆盖之外还会撞 git 的 // index.lock)。宿主只在顶层回合调它,子 agent 与父级共享同一个窗口。 if a.Opt.Occupy != nil && a.Depth == 0 { if err := a.Opt.Occupy(); err != nil { return err } if a.Opt.Release != nil { defer a.Opt.Release() } } // 提问账按回合清零(见 ask.go:连续退化护栏只在本次任务内生效)。 a.resetAskTurn() a.outputContinues = 0 a.compactGaveUp = false // 压缩失败的"本回合放弃"标记按回合清零 a.resetAnomalyCounters() // 上一次运行可能死在工具执行中途(Ctrl-C、进程被杀、机器重启),留下没有 // tool_result 配对的 tool_use。必须在写入本回合的新消息**之前**补齐: // Anthropic 要求 tool_result 紧挨着含 tool_use 的那条 assistant,中间插了 // user 消息就修不回来了——届时这条会话每次请求都 400,问什么都不回。 // (幂等:已配对的调用不动。) a.synthMissingToolResults() a.beginTurn(userMsg) goalIdx := len(a.Session.Messages) // 本回合新消息的起始下标(user 消息即将落在这里) started := time.Now() if userMsg != "" { if err := a.Session.Append(llm.Message{Role: llm.RoleUser, Content: userMsg, Synthetic: synthetic}); err != nil { return err } } // 含图回合路由到多模态模型(回合结束恢复原模型)。 // 续跑(userMsg 为空)时触发回合的用户消息已在历史末尾,扫描起点回退一条。 imgStart := goalIdx if userMsg == "" && imgStart > 0 { imgStart-- } hasImages := false for _, m := range a.Session.Messages[imgStart:] { if len(m.Images) > 0 { hasImages = true break } } if restore := a.routeMultimodal(hasImages); restore != nil { defer restore() } for turn := 0; turn < a.MaxTurns; turn++ { term, err := a.step(ctx) if err != nil { return a.finishTurn(goalIdx, started, ctx, err, false) } // 插话只在 step 边界落盘:工具结果已回填、下一次请求之前。模型本来说完了 // 也不算收尾——有新输入就继续这一回合(与"回合结束再发"的旧排队语义兼容, // 只是提前到了最近的边界)。 delivered, serr := a.deliverSteers() if serr != nil { return a.finishTurn(goalIdx, started, ctx, serr, false) } if term == transitionCompleted { if delivered == 0 { return a.finishTurn(goalIdx, started, ctx, nil, false) } continue } // continueNextTurn:工具结果已回填,进入下一轮 } // 步数用尽:不再直接抛错(那会让用户只拿到一句 reached max turns,完全不知道 // 做到哪了、还剩什么),改为禁掉工具再要最后一轮收尾说明,把上限从「中断」 // 降级成「体面收尾」。工作不会丢——消息早就逐条落盘,用户回一句「继续」即 // 可接着跑(续跑会重新拿到一份完整的步数预算)。 return a.finishTurn(goalIdx, started, ctx, a.wrapUp(ctx), true) } // RunSynthetic 与 Run 相同,但把本次追加的用户消息标记为系统生成(后台通知等): // 展示上按系统说明处理,统计与 /rewind 提问号不计入人工输入。 func (a *Agent) RunSynthetic(ctx context.Context, text string) error { a.nextSynthetic = true return a.Run(ctx, text) } // SetModel 切换当前 Provider 的模型(运行时生效,作用于下一次调用)。 func (a *Agent) SetModel(model string) error { if model == "" { return fmt.Errorf("model name is empty") } if s, ok := a.Provider.(llm.ModelSwitcher); ok { s.SetModel(model) return nil } return fmt.Errorf("provider does not support live model switching") } // ListModels 枚举当前 Provider 可用模型。 func (a *Agent) ListModels(ctx context.Context) ([]string, error) { if l, ok := a.Provider.(llm.ModelLister); ok { return l.ListModels(ctx) } return nil, fmt.Errorf("provider does not support listing models") } // TurnReport 是一次顶层回合收尾的结构化报告,OnTurnEnd 的入参。 // 复盘、进展沉淀、checkpoint 留档等宿主侧收尾经它取数。 type TurnReport struct { Err error ByTurnCap bool // true = 因 MaxTurns 收尾(err 为 nil 但不是模型自己说完) Steps int // 本回合模型调用次数 Ms int64 GoalIndex int // 本回合第一条用户消息在 transcript 里的下标(复盘取目标用) Started time.Time } // 回合结束记录的原因码,写入 session.TurnEnd.Reason。 // **取值必须与 AutoGo retro 包保持一致**:会话文件格式跨两端兼容。 const ( OutcomeCompleted = "completed" OutcomeFailed = "failed" OutcomeStopped = "stopped" OutcomeMaxTurns = "max_turns" ) // finishTurn 是 Run 的唯一终止出口:先落会话文件里的回合结束记录(best effort, // 不改变返回值),再原样返回 err。 // // byTurnCap 用来区分"模型自己说完了"和"步数上限收尾"——两者的 err 都是 nil, // 光看错误分不出来,而"回合到底为什么停"恰恰是事后取证最想知道的那件事 // (实报:一条会话里 8 次发言有 4 次的上一条是工具结果,从会话文件看不出是被谁切的)。 func (a *Agent) finishTurn(goalIdx int, started time.Time, ctx context.Context, err error, byTurnCap bool) error { a.recordTurnEnd(started, ctx, err, byTurnCap) // 工作区级收尾挂在同一个出口:四端(TUI/REPL/WebUI/渠道/cron)都经这里, // 分散到各入口就会有的做了有的没做。子 agent 不回调——它的会话是 ephemeral, // 留档与进展归父级那一次回合。 if a.Opt.OnTurnEnd != nil && a.Depth == 0 { a.Opt.OnTurnEnd(TurnReport{ Err: err, ByTurnCap: byTurnCap, Steps: a.curStep, Ms: time.Since(started).Milliseconds(), GoalIndex: goalIdx, Started: started, }) } // 已经动过现实世界:告诉宿主这是一个"可手动重试、不可自动整轮重试"的失败。 // 用户主动停止不算失败,不包这一层——否则 errors.Is(err, context.Canceled) // 会被破坏,界面会把"停止"误报成错误。 if err != nil && a.turnHadSideEffect && !errors.Is(err, ErrAfterSideEffects) && !errors.Is(err, context.Canceled) { err = fmt.Errorf("%w: %w", ErrAfterSideEffects, err) } return err } // Notify 把一句系统说明交给宿主(界面时间线上的一行)。宿主没接就什么都不做—— // 压缩那条路径自己有 stderr 日志,这里不再重复写,否则同一条信息在日志里出现两遍。 // 入口层(工作区收尾、留档)也用它:这些话是"发生过的实事",不该只进日志文件。 func (a *Agent) Notify(text string, failed bool) { if a.Opt.OnNotice != nil { a.Opt.OnNotice(text, failed) } } // deliverSteers 把宿主排队的插话按顺序落成普通 user 消息,返回实际插入的条数。 // 一条新输入等于一次新的输出预算:续写计数与"压缩失败本回合放弃"都要重置, // 否则上一段被长度上限截断过,插话后的第一段会少一次续写机会。 func (a *Agent) deliverSteers() (int, error) { if a.Opt.Steer == nil { return 0, nil } delivered := 0 for _, m := range a.Opt.Steer() { if strings.TrimSpace(m.Text) == "" && len(m.Images) == 0 { continue } msg := llm.Message{Role: llm.RoleUser, Content: m.Text} for _, name := range m.Images { msg.Images = append(msg.Images, llm.Image{Filename: name}) } if err := a.Session.Append(msg); err != nil { return delivered, err } delivered++ a.outputContinues = 0 a.compactGaveUp = false if a.Opt.OnSteer != nil { a.Opt.OnSteer(m) } } return delivered, nil } // reportCompact 把一次压缩报出去:结构化结果给要算数字的宿主,同一句人话给 // 只渲染文本的宿主(TUI 的系统块、WebUI 时间线上的一行)。措辞只有一份。 func (a *Agent) reportCompact(res CompactionResult) { if a.Opt.OnCompact != nil { a.Opt.OnCompact(res) } a.Notify(res.Text(), res.Err != "") } // recordTurnEnd 往会话文件追加一行回合结束记录。 func (a *Agent) recordTurnEnd(started time.Time, ctx context.Context, err error, byTurnCap bool) { if a.Session == nil || a.Depth > 0 { return // 子 agent 的会话是 ephemeral,记了只是噪声 } reason := OutcomeCompleted switch { case byTurnCap: reason = OutcomeMaxTurns case err != nil && (errors.Is(err, context.Canceled) || errors.Is(context.Cause(ctx), context.Canceled) || ctx.Err() != nil): reason = OutcomeStopped case err != nil: reason = OutcomeFailed } note := "" if err != nil { // 按字截,不按字节截:错误信息里通常有中文,字节切片会切出半个字。 note = session.TruncateRunes(err.Error(), 160) } a.Session.AppendTurnEnd(session.TurnEnd{ Reason: reason, Error: note, Step: a.curStep, Ms: time.Since(started).Milliseconds(), }) } // turnTransition 是 step 的返回转移。错误一律通过 error 返回;出错时转移值 // 为 transitionFailed,Run 只需区分 completed / 其他。 type turnTransition string const ( transitionCompleted turnTransition = "completed" // 模型给出最终回复,无工具调用 transitionNextTurn turnTransition = "next_turn" // 工具结果已回填,进入下一轮 transitionFailed turnTransition = "failed" // 出错退出(原因经 error 返回) ) // sysNow 返回本轮实际下发的 system 提示:System + SystemSuffix(若设置)。 // 每轮重算,供 token 估算与子 agent 的冻结快照使用,保证两者看到的长度一致。 // 注意:发请求时这两段是**分开传**的(见 step),动态段不进入缓存断点。 func (a *Agent) sysNow() string { return a.System + a.systemSuffix() } // systemSuffix 返回每轮动态段;nil 时为空串。 func (a *Agent) systemSuffix() string { if a.SystemSuffix == nil { return "" } return a.SystemSuffix() } // DynamicSections 组合所有「每轮注入、不进 transcript」的动态段。作为 SystemSuffix // 挂上后,这些段既躲过上下文压缩,也能立刻反映本轮的修改(如刚写入的记忆)。 // 段为空时返回空串,不占 token。 func (a *Agent) DynamicSections() string { extra := "" if a.ExtraSections != nil { extra = a.ExtraSections() } return extra + a.AbandonedSection() + a.InstructionsSection() } // discoveredInstruction 是一份读取时发现的子树 AGENTS.md。 type discoveredInstruction struct { path string text string } const ( maxDiscoveredFiles = 8 maxDiscoveredBytes = 64 * 1024 ) // AddInstructions 记下一份嵌套 AGENTS.md。工具层在 read 成功时调用;按绝对路径 // 去重并限制总量——这些内容每轮都会进 system,不能无限膨胀。并发安全: // 并行的只读工具可能同时发现不同子树。 func (a *Agent) AddInstructions(path, text string) { text = strings.TrimSpace(text) if path == "" || text == "" { return } a.instructionsMu.Lock() defer a.instructionsMu.Unlock() if a.discoveredSeen == nil { a.discoveredSeen = map[string]bool{} } if a.discoveredSeen[path] { return } if len(a.discoveredInstructions) >= maxDiscoveredFiles || a.discoveredBytes+len(text) > maxDiscoveredBytes { return } a.discoveredSeen[path] = true a.discoveredBytes += len(text) a.discoveredInstructions = append(a.discoveredInstructions, discoveredInstruction{path: path, text: text}) } // InstructionsSection 渲染读取时发现的子树指令;没有则返回空串。 func (a *Agent) InstructionsSection() string { a.instructionsMu.Lock() defer a.instructionsMu.Unlock() if len(a.discoveredInstructions) == 0 { return "" } cwd, _ := os.Getwd() var b strings.Builder b.WriteString("\n\n# Project instructions (discovered for the subtree you read)\n") for _, item := range a.discoveredInstructions { name := item.path if rel, err := filepath.Rel(cwd, item.path); err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) { name = rel } b.WriteString("\n## " + name + "\n" + item.text + "\n") } return b.String() } // step 执行一轮:上下文视图 → 流式调模型 → 停止分派 → 工具执行回填。 func (a *Agent) step(ctx context.Context) (turnTransition, error) { msgs, err := a.contextView(ctx) if err != nil { return transitionFailed, fmt.Errorf("context: %w", err) } a.curStep++ a.stepHadSideEffect = false started := time.Now() // 提前执行只读工具:provider 在某个工具参数完整时回调,onDelta 里把它 // 丢到独立 goroutine。每次模型尝试都有自己的一组(重试时旧的一律取消)。 earlyCtx, cancelEarly := context.WithCancel(ctx) defer func() { cancelEarly() }() early := map[string]chan earlyOutcome{} resetEarly := func() { cancelEarly() earlyCtx, cancelEarly = context.WithCancel(ctx) early = map[string]chan earlyOutcome{} } streamed := false // partial 攒的是**已经流到屏幕上的字**。provider 出错时返回的是零值 Message // (半截内容不在里面),所以只能自己攒——否则一次中断就把用户已经看到的 // 那段话整个丢掉(ZCode 的 cancelled-stream-persistence 治的就是这个)。 var partialText, partialReasoning strings.Builder call := func() (llm.Message, error) { partialText.Reset() partialReasoning.Reset() return a.Provider.Stream(ctx, a.System, msgs, a.toolDefs(), llm.StreamOptions{Thinking: a.Thinking, SystemDynamic: a.systemSuffix()}, func(d llm.StreamDelta) { if d.ToolCall != nil { a.startEarlyTool(earlyCtx, *d.ToolCall, early) } if d.Text != "" { streamed = true partialText.WriteString(d.Text) fmt.Fprint(a.Opt.Stdout, d.Text) } if d.Reasoning != "" { partialReasoning.WriteString(d.Reasoning) if a.Opt.OnReasoning != nil { a.Opt.OnReasoning(d.Reasoning) } } }) } final, err := call() // 确定性失败(400、鉴权、余额)当场换人——重试救不回来。偶发的坏流反过来: // 先在**同一个模型**上重试几轮,真熬不过去再切备选(下面那条环),否则一次抖动 // 就把整轮从用户选的模型上拽走。 // 副作用门控:一旦本 step 已经跑过会改变现实的工具,就绝不再重发同一个 // 请求——重发等于让模型可能把同一笔写/命令做第二遍。正常路径下工具在 // 调用成功之后才执行,所以这里通常是 false;它是为"工具提前启动"与 // 工具执行中途失败留的硬约束。 if err != nil && !a.stepHadSideEffect && !llm.IsTransientStream(err) && a.fallbackTo(ctx, err, streamed) { a.abandonStream("已切到备选模型重跑本步", &partialText, &partialReasoning) resetEarly() final, err = call() // 已切到备选模型,重试本步 } // 交付才停:坏流不重试,等于把一次网关抖动变成一次人工中断。provider 现在对 // "整轮空 / 流被截断 / 流卡死"都报错而不是返回一条空消息——因为下面那句"没有 // tool_calls ⇒ 模型说完了"会把它们当成**成功收尾**(取证见 llm/completion.go); // 但"报错了就结束回合"仍然是要人打「继续」。所以按时间预算退避重发本步。 budget := a.StreamRetryBudget if budget == 0 { budget = streamRetryBudget } unlimited := budget < 0 deadline := time.Now().Add(budget) wait := streamRetryFirst for attempt := 1; llm.IsTransientStream(err) && !a.stepHadSideEffect; attempt++ { if ctx.Err() != nil { break // 用户已经喊停 } if a.requestLooksOversized(err, msgs) { break // 上下文大到可疑:该走压缩,不是继续重发同一个大请求(下面那条自救) } if !unlimited && time.Now().Add(wait).After(deadline) { break // 预算见底:把失败如实交出去,不再多打一次请求 } if attempt == 1 { span := budget.String() if unlimited { span = "不限时长" } a.Notify(fmt.Sprintf("模型没答上(%v),正在重发这一步,最多再试 %s", err, span), true) } else if attempt%5 == 0 { a.Notify(fmt.Sprintf("仍在重试:第 %d 次,本步已耗 %s", attempt, time.Since(started).Round(time.Second)), true) } fmt.Fprintf(a.Opt.Stderr, "[stream] %v,第 %d 次重试(退避 %s,预算 %s)\n", err, attempt, wait, budget) if !sleepCtx(ctx, wait) { break // 叫停:保留原错误交给上层按"停止"分派,不伪装成模型失败 } wait *= 2 if wait > streamRetryCeil { wait = streamRetryCeil } if attempt >= streamRetrySwitchAfter && a.fallbackTo(ctx, err, streamed) { a.Notify(fmt.Sprintf("连试 %d 次没答上,切到备选模型 %s 继续", attempt, a.FallbackModel), true) } a.abandonStream("上一次回复没流完,重跑本步", &partialText, &partialReasoning) streamed = false resetEarly() final, err = call() } // 请求过大自救:网关的 token 计数和估算必然有偏差,真撞墙时先压一次再重发。 // 一次不够就按 0.7 / 0.5 / 0.35 收缩目标再压——每次重发前都重取视图,不能拿 // 压缩前的那一份原样再撞一次墙。最多三次,压完还失败就如实交出原错误。 if err != nil && !a.stepHadSideEffect && a.requestLooksOversized(err, msgs) { for _, shrink := range []float64{0.7, 0.5, 0.35} { if cerr := a.forceCompactForSize(ctx, msgs, shrink); cerr != nil { break } if fresh, verr := a.contextView(ctx); verr == nil { msgs = fresh } streamed = false a.abandonStream("上下文已压缩,重发本步", &partialText, &partialReasoning) resetEarly() final, err = call() if err == nil || !a.requestLooksOversized(err, msgs) { break } } } fmt.Fprintln(a.Opt.Stdout) if err != nil { // 中断时把**已经流出来的那半截**落盘。丢掉它的代价是双份的:用户屏幕上 // 看到的内容在会话里不存在(刷新/重进就没了),模型下一轮也看不到自己 // 说到哪了,只能从头再说一遍。 // // 只落正文与思考,**绝不落 tool_calls**:半截的工具参数是没写完的 JSON, // 落了就会有一条永远配不上对的 tool_use(Anthropic 会拒绝整条会话)。 a.persistPartial(partialText.String(), partialReasoning.String(), started) return transitionFailed, err } // 本地 trace:仅供度量与展示,不会进 API 报文(见 llm.Message 注释)。 final.StartedAt = started.UnixMilli() final.DurationMs = time.Since(started).Milliseconds() final.Step = a.curStep a.stampIdentity(&final) if final.Usage != nil { a.usedIn += final.Usage.InputTokens a.usedOut += final.Usage.OutputTokens // 缓存明细值得单独露出来:它证明 system+tools 前缀的 cache_control 真的 // 命中了(命中时 in 只报未缓存那一段,看不到 cache 会误判成"上下文很小")。 cache := "" if u := final.Usage; u.CacheReadTokens > 0 || u.CacheCreationTokens > 0 || u.ReasoningTokens > 0 { cache = fmt.Sprintf(" | cache %d/%d write, reasoning %d", u.CacheReadTokens, u.CacheCreationTokens, u.ReasoningTokens) } fmt.Fprintf(a.Opt.Stderr, "[tokens] +%d in / +%d out%s | total %d in / %d out\n", final.Usage.InputTokens, final.Usage.OutputTokens, cache, a.usedIn, a.usedOut) // 记下真实 prompt 规模:下一轮压缩判据以它为基准(+新增消息估算)。 a.noteRequestUsage(final.Usage, msgs) } // 输出上限切断 → 自动续写(见 continue.go)。放在落盘之前:几段输出要合并成 // **一条** assistant 消息,会话里才不会出现两条相邻 assistant(那会让"哪条是 // 本回合最终回复"和回合号都多一套特例)。 if final.OutputLimited && len(final.ToolCalls) == 0 { final = a.continueOutput(ctx, final, msgs) } if err := a.Session.Append(final); err != nil { return transitionFailed, err } if len(final.ToolCalls) == 0 { // 防御:provider 提前通知过工具、最终消息却没有 tool_calls。取消提前任务, // 结果一律丢弃(只读工具,无副作用)。 cancelEarly() return transitionCompleted, nil // 模型给出最终回复,turn 结束 } // 消息已落盘,现在才收提前执行的结果:tool_result 必须排在 tool_use 之后。 // 失败尝试走不到这里——那些结果随 ctx 取消一起丢弃(全是只读工具),暂存的 // 工具事件也一并不发,界面不会出现"没真正发生"的工具行。 earlyRaw := make(map[string]earlyOutcome, len(early)) for id, ch := range early { earlyRaw[id] = <-ch } earlyResults := make(map[string]toolOutcome, len(final.ToolCalls)) for _, tc := range final.ToolCalls { e, ok := earlyRaw[tc.ID] if !ok { continue } for _, ev := range e.events { a.emitTool(ev) } earlyResults[tc.ID] = e.outcome } // 本轮挂起了 tool_use:无论 step 以何路径退出,都要保证 tool_result // 配对(借鉴 ccb query.ts:149),否则续跑时 API 报文不配对。 pendingTools := true defer func() { if pendingTools { a.synthMissingToolResults() } }() if err := a.execToolsWithEarly(ctx, final.ToolCalls, earlyResults); err != nil { return transitionFailed, err } pendingTools = false // Deferred 收尾(见 ask.go):ask_user 在没有人可答的入口把问题登记成「本轮 // 以这个问题结束」。不能只往 tool_result 里写一句"请把问题转达给用户"—— // 模型未必听话,更常见的是它自己猜一个方向继续跑,那正是这个功能要省掉的浪费。 // 这里直接合成一条 assistant 消息并结束回合:tool_result 已在上面配对完整, // 所以报文仍然满足 Anthropic 的 tool_use/tool_result 相邻约束。 if askText := a.takePendingAsk(); askText != "" { fmt.Fprintln(a.Opt.Stdout, askText) if err := a.Session.Append(llm.Message{ Role: llm.RoleAssistant, Content: askText, Step: a.curStep, }); err != nil { return transitionFailed, err } return transitionCompleted, nil } return transitionNextTurn, nil } // InvalidateContext 丢掉内存里的压缩摘要,让下一次 contextView 从完整历史重建。 // // 会话被截短(回退 / 编辑重发)之后必须调用:摘要描述的是"截断之前"的历史, // 继续拿它当上下文,等于让模型基于一份已经不存在的对话作答——比不压缩更糟。 func (a *Agent) InvalidateContext() { a.compactMu.Lock() a.compacted, a.compactFrom = nil, 0 a.compactMu.Unlock() } // beginTurn 把回合号交给留档层:回合号 = 第几个用户 prompt,留档条目据此归档, // `a2go rewind 3` 这种按 prompt 粒度回退才有意义。续跑(userMsg 为空)沿用上一个 // 编号——它是同一个 prompt 的继续,不是新的一次改动批次。 func (a *Agent) beginTurn(userMsg string) { if a.TurnMarker == nil { return } turns := 0 for _, m := range a.Session.Messages { if m.Role == llm.RoleUser && !m.Synthetic { turns++ } } switch { case userMsg != "" && !a.syntheticTurn: turns++ case turns == 0: turns = 1 } a.TurnMarker.BeginTurn(turns) } // stampIdentity 给 assistant 消息盖上「协议格式 + 模型」的归因戳。 // provider 没实现 Identifier 时静默跳过(自定义 provider 的测试替身常见)。 func (a *Agent) stampIdentity(m *llm.Message) { if id, ok := a.Provider.(llm.Identifier); ok { m.Format, m.Model = id.Identity() } } // applyDefaults 补齐零值选项的默认值。 func (a *Agent) applyDefaults() { if a.ContextWindow <= 0 { a.ContextWindow = 128000 } if a.CompactAt <= 0 || a.CompactAt > 1 { a.CompactAt = 0.85 } if a.MinCompactTokens <= 0 { a.MinCompactTokens = 10000 } if a.MaxImages <= 0 { a.MaxImages = 6 } if a.ImageBudgetBytes <= 0 { a.ImageBudgetBytes = defaultImageBudgetBytes } if a.MaxToolResultChars <= 0 { a.MaxToolResultChars = 20000 } // 子 agent 并行上限:默认 5(参考 picoclaw SubTurn),并钳到工具并发上限。 // 注意**不给 MaxDepth 默认值**——零值即禁止委派(fail-closed),由入口显式配置。 if a.SubAgentConcurrency <= 0 { a.SubAgentConcurrency = 5 } if a.SubAgentConcurrency > maxToolConcurrency { a.SubAgentConcurrency = maxToolConcurrency } if a.MaxTurns <= 0 { a.MaxTurns = DefaultMaxTurns } } // CanSpawnSubAgent 报告当前 agent 是否还能派发子 agent(深度未达上限)。 func (a *Agent) CanSpawnSubAgent() bool { return a.Depth < a.MaxDepth } // startEarlyTool 在流式阶段提前执行一个只读工具。只接受 ReadOnly、非副作用、 // ConcurrencySafe、且不需要审批/不命中危险门的调用;其余等消息完整后走正常编排。 // 提前执行的结果会在 assistant 消息落盘后才回填,保证 tool_result 仍排在 tool_use 之后。 func (a *Agent) startEarlyTool(ctx context.Context, tc llm.ToolCall, early map[string]chan earlyOutcome) { if tc.ID == "" || tc.Name == "" { return } if _, exists := early[tc.ID]; exists { return } t, ok := a.Tools.Get(tc.Name) if !ok || !t.ReadOnly || t.SideEffect || !t.ConcurrencySafe { return } if a.needsApproval(tc.Name, t) { return } if _, danger := detectDanger(tc); danger { return } ch := make(chan earlyOutcome, 1) early[tc.ID] = ch go func() { var events []ToolEvent out := a.execResultWith(ctx, tc, func(ev ToolEvent) { events = append(events, ev) }) ch <- earlyOutcome{outcome: out, events: events} }() } // execTools 执行一批工具调用并按原顺序回填结果。 // 编排策略借鉴 ccb toolOrchestration.ts:106:连续的 ConcurrencySafe 调用并发 // 执行(上限 10),其余(写操作、bash)严格串行;回填顺序始终与调用顺序一致。 // // 串行路径(所有副作用工具都在这里)每完成一个就立刻落盘,不等整批结束: // 副作用已经发生之后,结果只留在内存里的窗口越短,崩溃/自动重试造成重复执行 // 的机会越小。 func (a *Agent) execTools(ctx context.Context, calls []llm.ToolCall) error { return a.execToolsWithEarly(ctx, calls, nil) } // execToolsWithEarly 是带提前执行结果(流式阶段已经跑完的只读调用)的完整版本。 func (a *Agent) execToolsWithEarly(ctx context.Context, calls []llm.ToolCall, early map[string]toolOutcome) error { for i := 0; i < len(calls); { // 已经在流式阶段跑完的调用:直接按位置回填,绝不重跑。 if res, ok := early[calls[i].ID]; ok { if err := a.appendToolResult(calls[i], res); err != nil { return err } i++ continue } if n := a.safeRunLenExcept(calls, i, early); n > 1 { batch := make([]toolOutcome, n) a.execConcurrent(ctx, calls[i:i+n], batch) for j := range batch { if err := a.appendToolResult(calls[i+j], batch[j]); err != nil { return err } } i += n continue } res := a.execResult(ctx, calls[i]) if err := a.appendToolResult(calls[i], res); err != nil { return err } i++ } return nil } // appendToolResult 把一次工具结果按 trace 字段写成 tool 消息并落盘。 func (a *Agent) appendToolResult(tc llm.ToolCall, res toolOutcome) error { fmt.Fprintf(a.Opt.Stderr, "[tool] %s(%s) -> %s\n", tc.Name, compact(tc.Input), preview(res.res.Text)) msg := llm.Message{ Role: llm.RoleTool, Content: res.res.Text, ToolCallID: tc.ID, // 本地 trace:工具耗时与该调用所属的模型轮次(不出网)。 StartedAt: res.startedAt, DurationMs: res.durMs, Step: a.curStep, // 结构化的失败类别 / 工具名 / 退出码 / 截断标记:消费方(stats、retro、 // 前端)从此读字段,不再解析失败块的文字。 Tool: tc.Name, Fail: res.kind, Exit: res.res.Exit, Truncated: res.res.Truncated, } if len(res.res.Images) > 0 { msg.Images = res.res.Images } return a.Session.Append(msg) } // safeRunLen 返回从 calls[i] 开始连续 ConcurrencySafe 的调用数(0 表示 calls[i] // 本身不可并发)。契约强制:有副作用的工具永远不进并发批次——即使被错误地 // 标注了 ConcurrencySafe(零值 fail-closed 的运行时兜底)。 func (a *Agent) safeRunLen(calls []llm.ToolCall, i int) int { return a.safeRunLenExcept(calls, i, nil) } // safeRunLenExcept 与 safeRunLen 相同,但遇到已经提前完成的调用就断开批次: // 那些调用只回填、不重跑。 func (a *Agent) safeRunLenExcept(calls []llm.ToolCall, i int, early map[string]toolOutcome) int { n := 0 for ; i+n < len(calls); n++ { if early != nil { if _, done := early[calls[i+n].ID]; done { break } } t, ok := a.Tools.Get(calls[i+n].Name) if !ok || !t.ConcurrencySafe || t.SideEffect { break } } return n } const maxToolConcurrency = 10 // execConcurrent 并发执行一段 ConcurrencySafe 调用,结果按索引写回。 func (a *Agent) execConcurrent(ctx context.Context, calls []llm.ToolCall, out []toolOutcome) { for _, tc := range calls { fmt.Fprintf(a.Opt.Stderr, "[tool] %s(%s) (concurrent)\n", tc.Name, compact(tc.Input)) } sem := make(chan struct{}, maxToolConcurrency) var wg sync.WaitGroup for i, tc := range calls { wg.Add(1) go func(i int, tc llm.ToolCall) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() out[i] = a.execResult(ctx, tc) }(i, tc) } wg.Wait() } // earlyOutcome 是提前执行的只读工具结果 + 暂存的事件:assistant 消息落盘后 // 才把 start/done 发给界面,失败的尝试不会留下幽灵工具行。 type earlyOutcome struct { outcome toolOutcome events []ToolEvent } // toolOutcome 是一次工具执行的结果 + 本地 trace(耗时/起始时刻/失败类别)。 // 单独包一层,避免把 agent 的度量字段塞进 tools.Result 污染工具契约。 type toolOutcome struct { res tools.Result startedAt int64 // unix ms durMs int64 // kind 是 classifyFailure 归一出的失败类别(空 = 成功)。随消息落盘, // stats/retro 从此不必解析 "[TOOL FAILED] …(kind)" 文本前缀。 kind string } // execResult 执行单个工具并返回结果(错误也作为结果回给模型,不中断批次)。 // 串行与并发两条路径都经此,故工具事件回调与计时只需挂在这里。 func (a *Agent) execResult(ctx context.Context, tc llm.ToolCall) toolOutcome { return a.execResultWith(ctx, tc, a.emitTool) } // execResultWith 是 execResult 的可替换事件出口版本:提前执行的工具先把 // start/done 事件缓存起来,等 assistant 消息落盘后再一次性发出,避免失败的 // 流式尝试在界面上留下不存在的工具行。 func (a *Agent) execResultWith(ctx context.Context, tc llm.ToolCall, emit func(ToolEvent)) toolOutcome { started := time.Now() startedAt := started.UnixMilli() if emit != nil { emit(ToolEvent{ID: tc.ID, Name: tc.Name, Input: tc.Input, StartedAt: startedAt, SideEffect: a.sideEffectTool(tc.Name)}) } result, err := a.execOne(ctx, tc) kind := "" // 自有密钥脱敏:路径/命令匹配能被绕过(变量间接、脚本里 open()、分块 base64), // 但绕过读到的密钥值不该进 transcript——transcript 会发给模型、写进会话文件、 // 出现在 bot 聊天与 WebUI 里。放在失败格式化之前,两条分支都覆盖。 if result.Text != "" { result.Text = a.redactSecrets(result.Text) } // 归一:工具也可以通过 Result.Err 报告失败(如 bash 的非零退出码), // 这样它能保留已有 stdout 的同时表明自己失败了。 if err == nil { err = result.Err } if err != nil { // 结构化失败块 + 连续失败提示,替代原先裸的 "\nerror: ..."。 // 判定必须带上下文:真实耗时(提示里不许出现编造的数字)、调用入参 // (pkill 自匹配只有看命令才认得出)、以及是不是用户停止——它和超时报的是 // 同一个 "signal: killed",归错类就等于指着一条没问题的命令说"它太慢了"。 streak := a.noteToolFailure(tc.Name) block, k := formatToolFailure(failInfo{ Tool: tc.Name, Input: tc.Input, Elapsed: time.Since(started), Cancelled: errors.Is(ctx.Err(), context.Canceled), }, err, result.Text, streak) result.Text, kind = block, k } else { a.noteToolSuccess(tc.Name) } // 重复调用提醒(同名同参,成功空转那一类,见 anomaly.go)。放在失败格式化之后: // 两条提示可以共存("这次又失败了"和"你已经在原地转了三圈"是不同的信息), // 但顺序反过来会被失败块的 TrimRight 吃掉。 if n := a.noteToolRepeat(tc.Name, tc.Input); n > 0 { result.Text = strings.TrimRight(result.Text, "\n") + "\n" + repeatNote(tc.Name, n) } out := toolOutcome{res: result, startedAt: startedAt, durMs: time.Since(started).Milliseconds(), kind: kind} // 工具结果里的截图:落盘到图片缓存,拿到可服务文件名,供前端工具行内联渲染。 // 文件名同时回填 result.Images,使会话落盘后 Filename 已就绪(collectToolImages // 会跳过已有 Filename 的项,重进对话也能直接加载)。 var toolImages []string if a.ImgCache != nil && len(result.Images) > 0 { for i := range result.Images { img := &result.Images[i] if img.Path == "" || img.Filename != "" { if img.Filename != "" { toolImages = append(toolImages, img.Filename) } continue } data, err := os.ReadFile(img.Path) if err != nil { continue } ext := filepath.Ext(img.Path) if ext == "" { ext = ".png" } name, err := a.ImgCache.Store(data, ext) if err != nil { continue } img.Filename = name toolImages = append(toolImages, name) } } if emit != nil { emit(ToolEvent{ ID: tc.ID, Name: tc.Name, Input: tc.Input, Result: result.Text, Done: true, SideEffect: a.sideEffectTool(tc.Name), StartedAt: startedAt, DurationMs: out.durMs, Images: toolImages, }) } return out } // emitTool 安全地触发工具事件回调。 func (a *Agent) emitTool(ev ToolEvent) { if a.Opt.OnTool != nil { a.Opt.OnTool(ev) } } // execOne 执行单个工具,副作用工具先过信任门控;skill 白名单生效时 // 拒绝白名单之外的调用。 func (a *Agent) execOne(ctx context.Context, tc llm.ToolCall) (tools.Result, error) { if err := ctx.Err(); err != nil { // 用户已经停止这一回合:这一批里还没开跑的工具必须一个都不碰现实世界。 // 不能指望各工具自己看 ctx——read/write/edit/ls/grep 完全忽略 ctx, // 停止之后照样会落盘(此前"停止"只拦得住网络和 exec 类工具)。 return tools.Result{}, err } t, ok := a.Tools.Get(tc.Name) if !ok { return tools.Result{}, fmt.Errorf("unknown tool %q", tc.Name) } if a.skillFilter != nil && !a.skillFilter[tc.Name] { return tools.Result{}, fmt.Errorf("tool %q is not allowed by the active skill (allowed: %v)", tc.Name, a.skillFilter) } // ask_user 直接执行:它不改现实世界,只是等一句话,所以不过审批门 // (ApprovalPolicy=always 下非副作用工具也会被问一次,那会变成「为了问一个 // 问题,先批准问这个问题」)。放在危险门**之前**:危险判定看的是这条调用要跑 // 什么命令,而提问没有命令(commandOf 只认 bash);今天不会被误判,但这个顺序 // 保证将来判定扩到别的工具时,「删掉多少文件」这类选项描述也不会被当成动作。 if t.EffectiveKind() == tools.KindAsk { return a.runTool(ctx, t, tc) } // 危险动作门。两种语义要分清: // - ModeBypass(WebUI「完全访问」,前端文案就是"全部自动放行,不再确认"): // 人已经把话说到这份上了,这里既不该问也不该暗中拒绝——放行。这一档此前 // 既不装审批回调又留着这道门,结果是 push/rm 在有人盯着屏幕时被静默拒掉, // 档位越开放人越没有表态的机会(实测日志:`allow tool "bash" ? [y/N]` 后面 // 直接 deny,因为 WebUI 的 stdin 是 nil)。 // - 其余档位(ask / accept-edits / 零值):**先于信任缓存与 --yes**。顺序放在 // needsApproval 里或其后,「本会话总是允许 bash」就会把 `rm -rf` 一起放行, // 功能归零。没人可问(-p / cron / 渠道)时 fail-closed 拒绝。 // - 例外:凭据读取(alwaysAskRules)任何档位都要问——"完全访问"授权的是干活, // 不是把 ~/.ssh 抄进回复,那是提示注入的兑现方式。 // 判定本身是启发式减速带,不是安全边界(见 dangerous.go)。 hit, danger := detectDanger(tc) if danger && a.DangerousPolicy != DangerousAllow && (hit.alwaysAsk() || a.Mode != ModeBypass) { ok, err := a.requestApproval(tc, hit) if err != nil { return tools.Result{}, err } if !ok { a.logDangerDenied(tc, hit) return tools.Result{}, dangerDeniedErr(tc, hit) } return a.runTool(ctx, t, tc) } // 常规审批门控:策略与档位共同决定是否询问(见 needsApproval)。 if a.needsApproval(tc.Name, t) { ok, err := a.requestApproval(tc, dangerHit{}) if err != nil { return tools.Result{}, err } if !ok { return tools.Result{}, fmt.Errorf("user denied tool %q", tc.Name) } } return a.runTool(ctx, t, tc) } // sideEffectTool 报告该工具是否会改变现实世界(ask_user 不算:它只等一句话)。 func (a *Agent) sideEffectTool(name string) bool { t, ok := a.Tools.Get(name) return ok && t.SideEffect && t.EffectiveKind() != tools.KindAsk } // historyHasSideEffects 报告当前未完结的回合(最后一条真实用户消息之后)是否 // 已经有副作用工具的结果。续跑时据此继续禁止自动整轮重试。 func (a *Agent) historyHasSideEffects() bool { if a.Session == nil { return false } for i := len(a.Session.Messages) - 1; i >= 0; i-- { m := a.Session.Messages[i] if m.Role == llm.RoleUser && !m.Synthetic { return false } if m.Role == llm.RoleTool && a.sideEffectTool(m.Tool) { return true } } return false } // runTool 是工具真正碰现实世界的唯一入口。副作用工具在这里先立"已执行"标记, // 再开始跑:哪怕它中途失败或被进程杀掉,本轮也不会再自动重试。 func (a *Agent) runTool(ctx context.Context, t *tools.Tool, tc llm.ToolCall) (tools.Result, error) { if t.SideEffect && t.EffectiveKind() != tools.KindAsk { a.stepHadSideEffect = true a.turnHadSideEffect = true } return t.Run(ctx, tc.Input) } // dangerDeniedErr 构造**给模型看**的拒绝文本:只说被拦下了、以及不许绕。 // // 旧文案末尾带着 "Set `dangerous=allow` (or AUTOGO_DANGEROUS=allow) to disable // this guard"——那是给人看的运维提示,却顺着 transcript 发给了模型,而模型手里 // 有 bash 和 write:等于把关掉安全门的办法摆在它面前(实测有会话连吃 4 次拦截, // 每次读到这句)。现在分两条通道:模型只收到拒绝,逃生口提示进日志 // (见 logDangerDenied)。 func dangerDeniedErr(tc llm.ToolCall, hit dangerHit) error { return fmt.Errorf("dangerous action blocked (%s): %s. This needs an explicit human decision and "+ "there is no approver in this session. Do not work around it — no config edit, no env var, no "+ "alternate tool or shell trick. Tell the user what you were about to do and stop there", hit.Rule, hit.Reason) } // logDangerDenied 把"确需自动化怎么办"写给**人**看的日志,不进 transcript。 func (a *Agent) logDangerDenied(tc llm.ToolCall, hit dangerHit) { if a.Opt.Stderr == nil { return } preview, cut := execx.Truncate([]byte(tc.Input), 300) if cut { preview = append(preview, "…"...) } fmt.Fprintf(a.Opt.Stderr, "[danger] 已拦下 %s(规则 %s:%s):%s\n"+ "[danger] 无人值守确需放行时设 dangerous=allow 或 AUTOGO_DANGEROUS=allow——"+ "这会关闭不可逆动作门,请知情\n", tc.Name, hit.Rule, hit.Reason, preview) } // needsApproval 判定一次工具调用是否需要征得用户同意。 // // 档位(Mode)× 类别(ToolKind)的映射: // - ""/ask 所有副作用工具都要问 // - accept-edits 只放行「改文件」(write/edit),执行/网络/未分类仍要问 // - bypass 全部放行 // // 更优先的短路:ApproveAll(无人值守入口)与会话内信任缓存。 // 非副作用工具仅当旧策略为 always 且挂了交互回调时才问(保留历史语义)。 func (a *Agent) needsApproval(name string, t *tools.Tool) bool { if a.Opt.ApproveAll || a.trusted[name] { return false } if a.Mode == ModeBypass { return false } if !t.SideEffect { return a.ApprovalPolicy == "always" && a.Opt.Approval != nil } if a.Mode == ModeAcceptEdits { return t.EffectiveKind() != tools.KindWrite } return true // ModeAsk 与零值 ""(≡ 旧的 dangerous-only) } // requestApproval 走交互回调(可缓存信任)或 stdin 询问。返回是否放行。 // hit 非空表示这是危险动作:preview 会带上人类可读原因(服务端预缩进 JSON, // 这样 TUI 与前端「解析失败原样回退」时仍是可读的多行文本),且**不写信任缓存** // ——「本会话总是允许 bash」不该等于「以后 rm -rf 也不用问」。 func (a *Agent) requestApproval(tc llm.ToolCall, hit dangerHit) (bool, error) { if a.Opt.Approval != nil { switch a.Opt.Approval(ApprovalRequest{ Tool: tc.Name, Preview: approvalPreview(tc, hit), Danger: hit.Rule != "", }) { case ApprovalAlways: if hit.Rule == "" { // 危险动作不缓存 if a.trusted == nil { a.trusted = map[string]bool{} } a.trusted[tc.Name] = true // 会话内记忆授权(借鉴 pigo trust) } case ApprovalOnce: default: return false, nil } return true, nil } return a.confirm(fmt.Sprintf("allow tool %q ?", tc.Name)) } // approvalPreview 组装审批面板里展示的内容。非危险调用保持历史行为(原始入参), // 危险调用在最前面加一行醒目原因。 func approvalPreview(tc llm.ToolCall, hit dangerHit) string { if hit.Rule == "" { return tc.Input } pretty := tc.Input var buf bytes.Buffer if err := json.Indent(&buf, []byte(tc.Input), "", " "); err == nil { pretty = buf.String() } return "⚠ " + hit.Reason + "\n" + pretty } // confirm 通过 stdin 询问用户。无 stdin(无人值守入口)时视为拒绝。 func (a *Agent) confirm(prompt string) (bool, error) { if a.Opt.Stderr != nil { fmt.Fprintf(a.Opt.Stderr, "%s [y/N] ", prompt) } if a.Opt.Stdin == nil { return false, nil } var ans string if _, err := fmt.Fscanln(a.Opt.Stdin, &ans); err != nil { return false, nil } ans = strings.ToLower(strings.TrimSpace(ans)) return ans == "y" || ans == "yes", nil } // synthMissingToolResults 为最后一个含 tool_use 的 assistant 消息中尚未得到 // tool_result 的调用合成错误结果(幂等:已配对的调用不动)。 func (a *Agent) synthMissingToolResults() { msgs := a.Session.Messages last := -1 for i := len(msgs) - 1; i >= 0; i-- { if msgs[i].Role == llm.RoleAssistant && len(msgs[i].ToolCalls) > 0 { last = i break } } if last < 0 { return } answered := map[string]bool{} for _, m := range msgs[last+1:] { if m.Role == llm.RoleTool { answered[m.ToolCallID] = true } } for _, tc := range msgs[last].ToolCalls { if answered[tc.ID] { continue } err := a.Session.Append(llm.Message{ Role: llm.RoleTool, Content: "error: tool execution was interrupted before a result was produced", ToolCallID: tc.ID, }) if err != nil { fmt.Fprintf(a.Opt.Stderr, "[warn] failed to persist synthesized tool_result: %v\n", err) } } } // preview 截短工具结果用于终端展示。 func preview(result string) string { result = strings.TrimSpace(result) if i := strings.IndexByte(result, '\n'); i >= 0 { result = result[:i] + "…" } if len(result) > 60 { // 工具输出里中文与 emoji 都常见:60 是"给人看的字数",按字节切会劈开字符。 result = textx.ClipRunes(result, 60) + "…" } return result } // compact 截短工具参数用于终端展示。 func compact(input string) string { var m map[string]any if err := json.Unmarshal([]byte(input), &m); err == nil { parts := []string{} for k, v := range m { s := fmt.Sprint(v) if len(s) > 60 { s = textx.ClipRunes(s, 60) + "…" } parts = append(parts, k+"="+s) } input = strings.Join(parts, ", ") } if len(input) > 120 { input = textx.ClipRunes(input, 120) + "…" } return input } // wrapUpNote 是收尾回合前置在回复正文前的一行标记。写成正文的一部分而不是 // 单独的告警通道,是为了让它同时满足两件事:用户一眼知道这轮被截断了;它也会 // 作为上下文留在会话里,下一句「继续」时模型知道上一轮结束在哪里。 const wrapUpNote = "⏸ 已达本轮步数上限(%d)——以下是基于当前进度的收尾说明,回复「继续」可接着做。" // wrapUpInstruction 是收尾回合临时追加到 system 末尾的指令。 // 挂在 system 而不是往会话里塞一条 user 消息:后者会作为真实用户输入落盘, // 污染历史(用户从没说过这句话)。 const wrapUpInstruction = ` ## 本轮步数已用尽 你已经用完本轮的步数预算,无法再调用任何工具。请**不要**声称完成了未验证的事情。 基于已有进度直接给出收尾说明: 1. 已经完成了什么(具体到文件/命令); 2. 还剩什么没做、下一步该做什么; 3. 如果有任何未验证的假设,明确指出。` // wrapUp 在步数用尽后做一次收尾:禁用工具再要一轮,让用户拿到可用的结论, // 而不是一句 "reached max turns"。收尾失败(网络/取消)仍然如实返回错误。 func (a *Agent) wrapUp(ctx context.Context) error { msgs, err := a.contextView(ctx) if err != nil { return fmt.Errorf("wrap-up context: %w", err) } a.curStep++ started := time.Now() // 保留 tools 定义(只禁用调用):与上一步相同的工具数组,缓存前缀才不被打碎。 final, err := a.Provider.Stream(ctx, a.System, msgs, a.toolDefs(), llm.StreamOptions{Thinking: a.Thinking, SystemDynamic: a.systemSuffix() + wrapUpInstruction, DisableTools: true}, func(d llm.StreamDelta) { if d.Text != "" { fmt.Fprint(a.Opt.Stdout, d.Text) } if d.Reasoning != "" && a.Opt.OnReasoning != nil { a.Opt.OnReasoning(d.Reasoning) } }) fmt.Fprintln(a.Opt.Stdout) if err != nil { return err } final.StartedAt = started.UnixMilli() final.DurationMs = time.Since(started).Milliseconds() final.Step = a.curStep a.stampIdentity(&final) // 去掉模型可能仍然发出的工具调用:本轮已明确禁用工具,留着会破坏 // tool_use / tool_result 配对。 final.ToolCalls = nil final.Content = fmt.Sprintf(wrapUpNote, a.MaxTurns) + "\n\n" + final.Content if final.Usage != nil { a.usedIn += final.Usage.InputTokens a.usedOut += final.Usage.OutputTokens } return a.Session.Append(final) }