diff --git a/internal/agent/core/memorypass.go b/internal/agent/core/memorypass.go index 71eb8d32..c6cd32c4 100644 --- a/internal/agent/core/memorypass.go +++ b/internal/agent/core/memorypass.go @@ -153,6 +153,76 @@ func (a *Agent) pruneByQuery(query string) int { return a.context.PruneWithProtected(query, topK, a.docStore, a.effectiveProtectedCount()) } +// pruneForResume 是按「指定保护条数」执行的裁剪，供任务恢复入口使用。 +// +// 与 pruneByQuery 的唯一差别是 protected 的来源： +// +// pruneByQuery protected = a.effectiveProtectedCount()（全局末尾 N 条） +// pruneForResume protected = 调用方给的条数（该任务帧自己的现场） +// +// 为什么恢复要用后者：被挂起任务的现场在 f.Msgs[PrefixLen:] 里， +// 可能是一整批工具结果。若按全局末尾 N 条保护，那些结果可能落在保护范围外 +// 被归档——恢复后的任务引用的内容已不在上下文里，等于自相矛盾。 +func (a *Agent) pruneForResume(query string, protected int) int { + if a == nil || a.context == nil || a.isLightKernel() { + return 0 + } + if protected < 0 { + protected = 0 + } + // ★ 不能直接复用 contextTopK()：它内部会把 topK 抬到 + // effectiveProtectedCount()+1（全局末尾保护），而恢复场景的保护范围是 + // **本帧现场**，两者不是同一个数。直接用会得到「按全局保护算出来的 + // 下限 ≥ 实际条数」⇒ 裁不动（实测 protected=8、上下文 20 条时 + // topK 被抬到 20，Prune 空转）。 + // + // 这里按**本帧保护条数**重新推导：先算 token 预算能装几条， + // 再保证下限是 protected+1（否则 Prune 会把 protected 之外的裁光）。 + topK := a.contextTopKForProtected(protected) + return a.context.PruneWithProtected(query, topK, a.docStore, protected) +} + +// contextTopKForProtected 是按**指定保护条数**推导保留条数的版本。 +// +// 与 contextTopK 的唯一差别是下限：后者用 effectiveProtectedCount()（全局末尾 +// 保护），这里用调用方给的 protected。恢复场景必须用后者——保护范围是本帧现场。 +func (a *Agent) contextTopKForProtected(protected int) int { + if protected < 0 { + protected = 0 + } + hardCap := a.maxContextSize - 1 + if hardCap < protected+1 { + hardCap = protected + 1 // 保护范围本身超过运维上限时，保护优先 + } + if hardCap < 1 { + hardCap = 1 + } + + budget := a.computeTokenBudget() + tokens := budget.ContextTokens + if tokens <= 0 { + return hardCap + } + avg := 0 + if a.context != nil { + if n := a.context.Len(); n > 0 { + avg = a.accumulatedTokens() / n + } + } + if avg <= 0 { + avg = defaultAvgEventTokens + } + topK := tokens / avg + if topK < protected+1 { + // 预算装不下「保护条数 + 1」⇒ 宁可暂时超页也不破坏现场（同 contextTopK 的取舍）。 + topK = protected + 1 + } + if topK > hardCap { + topK = hardCap + } + return topK +} + // contextTopK 推导裁剪后应保留的**条数**。 // // ★ 为什么不能直接用 maxContextSize（2026-10-01 跑分实测出的根因）： diff --git a/internal/agent/core/overflow.go b/internal/agent/core/overflow.go index e96f1f6f..988532ed 100644 --- a/internal/agent/core/overflow.go +++ b/internal/agent/core/overflow.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "log" + "strings" agentAPI "github.com/JianFeeeee/HologramAgent/internal/agent/api" ) @@ -95,17 +96,36 @@ func (a *Agent) overflowRatioNow() float64 { return float64(a.accumulatedTokens()) / float64(max) } -// maybeHandleContextOverflow 是本地预判的入口：**每次发 LLM 请求之前**调用， -// 覆盖轮内 tool 回环（现状只在轮首检查，轮内撑爆时才发现，浪费一整轮工具调用）。 +// enforceContextBudget 是**上下文容量保护**的唯一实现（基础设施）。 // -// 返回 true 表示已处理（本帧应当重跑），调用方需要重建 f.Msgs —— 因为裁剪 -// 之后挂起帧里那份超限请求不能复用（见设计文档约束三）。 +// ## 为什么它是基础设施，而不是某个组件的功能 +// +// 上下文会在**每个注入点**增长：新消息、中断注入、任务恢复（中断任务的结果 +// 已提交进 a.context，重建出的前缀更大）、以及轮内工具回灌。因此"装不下就裁" +// 必须在这些入口都成立，而不是只在某一条路径上碰巧被调用。 +// +// ��史（2026-10-25 前的三个半成品入口，现已全部收敛到这里）： +// +// task.go 发 LLM 前 maybeHandleContextOverflow（有恢复预算） +// task.go 上游报错 handleUpstreamContextFull（**直调内部，绕过预算**） +// resident.go 轮首 checkContextFull（只对驻留子生效，根 agent no-op） +// +// 三个入口、三套判据、三份防护差异 —— 这就是"额外造的一套机制"。 +// +// ## force 的语义 +// +// force 不是"另一条路径"，只是**判定已被外部短路**： +// +// force=false 本地水位判定：积累量 > ratio × 窗口 +// force=true 外部已确认超页（如 provider 返回 ErrContextFull） +// +// 之后的一切（恢复预算、裁剪、raise L4）完全相同 —— 这正是"复用语义"的落点。 +// +// 返回 handled 表示本帧需要重跑（f.Msgs 已过期，必须重建）； +// 若 f.Terminal == terminalError 则表示恢复预算耗尽，调用方应直接终止。 // // 幂等：未超页时零开销（一次 token 估算 + 一次比较），可安全地在热路径调用。 -func (a *Agent) maybeHandleContextOverflow(f *TaskFrame, query string) (handled bool, pruned int) { - // 诊断留痕：超页判据在**每次 LLM 请求**都会走，而生产上从未见它触发过。 - // 没有这行日志时，「没触发」和「没执行」无法区分（2026-10-01 T10 排查 - // 花了 10 轮工具调用就卡在这里）。只记比值与判定，不记内容。 +func (a *Agent) enforceContextBudget(f *TaskFrame, query string, force bool) (handled bool, pruned int) { light := a != nil && a.isLightKernel() noCtx := a == nil || a.context == nil var ratio float64 @@ -117,22 +137,34 @@ func (a *Agent) maybeHandleContextOverflow(f *TaskFrame, query string) (handled checked := a.overflowStat.Checked a.overflowStat.Checked++ a.overflowStat.Unlock() + // 诊断留痕：这条判据在每个注入点都会走，而生产上很少触发。 + // 没有这行日志时，「没触发」和「没执行」无法区分（2026-10-01 T10 + // 排查花了 10 轮工具调用才卡在这里）。只记比值与判定，不记内容。 if checked%20 == 0 { log.Printf("[agent] overflow check #%d: ratio=%.2f window=%d events=%d "+ - "(threshold=%.2f lightKernel=%v noContext=%v)", + "(threshold=%.2f lightKernel=%v noContext=%v force=%v)", checked+1, ratio, a.contextWindowTokens(), a.contextLenSafe(), - overflowRatio, light, noCtx) + overflowRatio, light, noCtx, force) } } if noCtx || light { // 轻量内核（驻留子）不做按相关度裁剪，见 pruneByQuery 的同款理由。 return false, 0 } - if ratio < overflowRatio { + // 唯一判据点：本地水位 OR 外部明确报错。 + if !force && ratio < overflowRatio { return false, 0 } - // 恢复预算：同一帧内裁剪仍裁不下 ⇒ 说明 topK 与窗口脱钩（见 pruneByQuery）， + if force { + // 上游报错是一次**独立的上报**（与本地水位分开计数）， + // 计数放在这里而不是调用方：它是判据的一部分，不是调用方的副作用。 + a.overflowStat.Lock() + a.overflowStat.Upstream++ + a.overflowStat.Unlock() + } + // 恢复预算：同一帧内裁完仍超 ⇒ 说明 topK 与窗口脱钩（见 pruneByQuery）， // 再裁一次也是同样结果。超预算后**明确终止**，不静默重试。 + // 这条防护对所有入口一致生效 —— 包括上游报错那条（此前它绕过了这里）。 if f != nil { if f.OverflowRecover >= overflowRecoverBudget { log.Printf("[agent] context overflow: recover budget exhausted (%d/%d), "+ @@ -150,6 +182,94 @@ func (a *Agent) maybeHandleContextOverflow(f *TaskFrame, query string) (handled return true, a.handleContextOverflow(query, ratio) } +// maybeHandleContextOverflow 保留为发 LLM 请求前的薄封装（历史调用点）。 +// +// 它现在只是 enforceContextBudget 的一个调用点，不再持有独立判据。 +func (a *Agent) maybeHandleContextOverflow(f *TaskFrame, query string) (handled bool, pruned int) { + return a.enforceContextBudget(f, query, false) +} + +// enforceResumeContextBudget 是**任务恢复**入口的容量保护。 +// +// 与通用 enforceContextBudget 的区别只有两点，但都很关键： +// +// 1. **query 取任务帧的最新有效上下文**，不是 f.Input。 +// 被恢复的任务可能已挂起很久，它的 f.Input 与当前对话早已不同步； +// 拿旧 Input 去算相关性，会裁掉"对该任务重要、但字面上不匹配旧输入" +// 的事件——恢复出来的任务反而失去了它自己的现场。 +// 这里的做法是取该帧**尾部最后一条有内容的消息**（f.Msgs 的末条）， +// 那才是它此刻真正要继续做的东西。 +// +// 2. **保护范围是该任务帧自己的尾部**，不是全局上下文的末尾 N 条。 +// PruneWithProtected 默认保护的 c.events 末 N 条是"最近的对话"， +// 而被挂起的任务在现场里可能攒了整批工具结果——那些必须一起保住， +// 否则恢复后 f.Msgs[PrefixLen:] 引用的内容已被归档，任务自相矛盾。 +// +// 判据仍是容量（不看声明）：声明式裁剪归 pruneOnInput / stepToolAfter， +// 那是业务策略；这里是"装不下就得裁"的基础设施。 +func (a *Agent) enforceResumeContextBudget(f *TaskFrame) (handled bool, pruned int) { + if f == nil { + return false, 0 + } + // 水位判定（与通用入口同一套判据，只是 query/保护范围不同）。 + if !a.overBudgetLocked() { + return false, 0 + } + // 恢复预算：同一帧不能无限裁。 + if f.OverflowRecover >= overflowRecoverBudget { + log.Printf("[agent] resume overflow: recover budget exhausted (%d/%d), terminating frame", + f.OverflowRecover, overflowRecoverBudget) + a.markOverflowAborted(a.overflowRatioNow(), a.context.Len()) + f.Err = fmt.Errorf("上下文超限：恢复时已裁剪 %d 次仍超出窗口，本轮终止", f.OverflowRecover) + f.Terminal = terminalError + return true, 0 + } + f.OverflowRecover++ + + query := resumeQueryFrom(f) + // 保护该任务帧自己的现场条数（尾部消息 = 本任务攒下的工具轮产物）。 + protected := len(f.Msgs) - f.PrefixLen + if protected < 0 { + protected = 0 + } + before := a.context.Len() + a.pruneForResume(query, protected) + after := a.context.Len() + if after >= before { + log.Printf("[agent] resume overflow: no reduction (%d→%d events), not raising L4", before, after) + return false, 0 + } + pruned = before - after + log.Printf("[agent] resume overflow: pruned=%d (%d→%d events), protected=%d frame tail msgs", + pruned, before, after, protected) + a.raiseContextOverflow(pruned) + return true, pruned +} + +// overBudgetLocked 是纯判据（无副作用），供各入口复用。 +// +// 命名带 Locked 是历史遗留，实际不加锁；调用方在调度器单线程上下文里使用。 +func (a *Agent) overBudgetLocked() bool { + if a == nil || a.context == nil || a.isLightKernel() { + return false + } + return a.overflowRatioNow() >= overflowRatio +} + +// resumeQueryFrom 取任务帧的**最新有效上下文**作为相关性查询。 +// +// 从尾部往回找第一条有实质内容的消息：恢复时 f.Msgs 的最后一条 +// 可能是 tool 结果、continuation 占位或空的 assistant 消息， +// 直接拿它算相关性会退化成无意义查询。 +func resumeQueryFrom(f *TaskFrame) string { + for i := len(f.Msgs) - 1; i >= 0; i-- { + if c := strings.TrimSpace(f.Msgs[i].Content); c != "" { + return c + } + } + return f.Input +} + // handleContextOverflow 执行「裁剪 + L4 打断」。 // // 关键取舍：裁剪在**这里同步做完**，L4 只负责打断与告知。理由： @@ -240,23 +360,15 @@ func overflowNotice(pruned int) string { "被裁内容仍可检索，但需显式查询；查不到不等于不存在。", pruned) } -// handleUpstreamContextFull 处理上游返回的 ErrContextFull：与本地预判走**同一条** -// 路径（裁剪 → L4），避免两条路径行为不一致。 +// isUpstreamContextFull 判定上游错误是否表示「上下文装不下」。 // -// 必须在 provider fallback **之前**调用：fallback 会换 provider 重发同一个 -// 超限请求，换谁都一样超。 -// -// 返回 true 表示已裁剪，调用方应重试本轮；false 表示裁不出东西，应终止。 -func (a *Agent) handleUpstreamContextFull(llmErr error) bool { +// 它只做**判定**，不做处理：处理一律交给 enforceContextBudget(force=true)。 +// 此前这里叫 handleUpstreamContextFull，自己又调了一遍 handleContextOverflow —— +// 那等于在中断语义之外另造一套裁剪，绕过恢复预算，且与 raise L4 抢占的 +// 任务互相抢跑（2026-10-25 实测）。 +func isUpstreamContextFull(llmErr error) bool { var pe *agentAPI.ProviderError - if !errors.As(llmErr, &pe) || pe.Kind != agentAPI.ErrContextFull { - return false - } - pruned := a.handleContextOverflow("", 0) - a.overflowStat.Lock() - a.overflowStat.Upstream++ - a.overflowStat.Unlock() - return pruned > 0 + return errors.As(llmErr, &pe) && pe.Kind == agentAPI.ErrContextFull } // contextLenSafe 是 nil 安全的上下文条数（诊断日志用）。 diff --git a/internal/agent/core/overflow_test.go b/internal/agent/core/overflow_test.go index 6ce3ee36..10eda3d3 100644 --- a/internal/agent/core/overflow_test.go +++ b/internal/agent/core/overflow_test.go @@ -155,12 +155,18 @@ func TestOverflow_T6_中断消息含认知提示(t *testing.T) { func TestOverflow_上游ErrContextFull走同路径(t *testing.T) { a := newOverflowAgent(2000, 2, 20, 400) // 单条装得下，总量超页 - // 累积不足以下次超页也没关系：上游已经明说装不下了 - if !a.handleUpstreamContextFull(&agentAPI.ProviderError{ + // 累积不足以下次超页也没关系：上游已经明说装不下了。 + // ★ 关键：走的是**同一个** enforceContextBudget（force=true）， + // 不是另一条独立路径。这正是「contextfull 复用中断裁剪语义」的落点。 + if !isUpstreamContextFull(&agentAPI.ProviderError{ StatusCode: 400, Kind: agentAPI.ErrContextFull, Message: "context_length_exceeded", }) { + t.Fatal("上游 ErrContextFull 应被识别") + } + handled, _ := a.enforceContextBudget(nil, "", true) + if !handled { t.Fatal("上游 ErrContextFull 应被接住并返回 true（可恢复）") } a.overflowStat.Lock() @@ -176,9 +182,8 @@ func TestOverflow_上游ErrContextFull走同路径(t *testing.T) { // 非上下文类错误不得被误接。 func TestOverflow_上游其他错误不误接(t *testing.T) { - a := newOverflowAgent(1000, 2, 20, 400) for _, kind := range []agentAPI.ErrKind{agentAPI.ErrCredential, agentAPI.ErrTransient, agentAPI.ErrUnknown} { - if a.handleUpstreamContextFull(&agentAPI.ProviderError{ + if isUpstreamContextFull(&agentAPI.ProviderError{ StatusCode: 500, Kind: kind, Message: "boom", }) { t.Fatalf("kind=%v 不应被当作上下文超限接住", kind) @@ -572,3 +577,153 @@ func TestOverflow_单条超预算则终止(t *testing.T) { t.Fatal("应记一次 Aborted") } } + +// ── 裁剪是**基础设施**：各注入/恢复入口都必须能触发它 ── + +// 任务恢复入口（rebaseFramePrefix）必须做容量裁剪。 +// +// 为什么需要它：中断任务跑完会把它的输入/输出提交进 a.context， +// 于是恢复时重建出的前缀**可能比被挂起时更大**。不在这里裁，这份更大的 +// 前缀会被原样发出去，下一轮再靠 stepLLM 的闸门补救——那时已经付过 +// 一次超限请求的代价。 +// +// 变异：删掉 rebaseFramePrefix 末尾的 enforceContextBudget 调用 ⇒ 变红。 +func TestOverflow_任务恢复入口触发裁剪(t *testing.T) { + a := newOverflowAgent(2000, 2, 20, 400) // 总量远超窗口 + f := &TaskFrame{ + Input: "恢复后的输入", + PrefixLen: 1, + Msgs: []agentAPI.Message{ + {Role: "system", Content: "sys"}, + {Role: "user", Content: "tail"}, + }, + } + before := a.context.Len() + + a.rebaseFramePrefix(f) + + after := a.context.Len() + if after >= before { + t.Fatalf("任务恢复后应触发容量裁剪（上下文 %d → %d 未减少）", before, after) + } + if f.OverflowRecover == 0 { + t.Fatal("恢复入口的裁剪应计入该帧的恢复预算（否则同一帧可无限裁）") + } +} + +// 恢复时相关性必须按**本帧最新上下文**算，不能拿旧的 f.Input。 +// +// 被恢复的任务可能已挂起很久，f.Input 与当前对话早已不同步； +// 拿旧 Input 当 query，会裁掉"对该任务重要、但字面上不匹配旧输入"的事件， +// 恢复出来的任务反而失去了自己的现场。 +// +// 判据：resumeQueryFrom 取的是 Msgs 尾部第一条有内容的消息， +// 而不是 Input。 +func TestOverflow_恢复用本帧最新上下文作查询(t *testing.T) { + f := &TaskFrame{ + Input: "很久以前用户说的话", + PrefixLen: 1, + Msgs: []agentAPI.Message{ + {Role: "system", Content: "sys"}, + {Role: "tool", Content: "最新的工具结果：部署已完成"}, + }, + } + got := resumeQueryFrom(f) + if got != "最新的工具结果：部署已完成" { + t.Fatalf("恢复查询应取本帧最新有效上下文，实际取到 %q", got) + } + if got == f.Input { + t.Fatal("不得回退到旧的 f.Input —— 那与当前任务已不同步") + } +} + +// 尾部全是空消息时必须回退到 Input（不能返回空查询）。 +func TestOverflow_恢复查询回退(t *testing.T) { + f := &TaskFrame{ + Input: "原始输入", + Msgs: []agentAPI.Message{{Role: "assistant", Content: " "}}, + } + if got := resumeQueryFrom(f); got != "原始输入" { + t.Fatalf("全空时应回退到 Input，实际 %q", got) + } +} + +// 恢复裁剪必须保护**本帧自己的现场**，而不是全局末尾 N 条。 +// +// 这是最容易写错的一点：PruneWithProtected 的默认保护是 c.events 末尾 N 条 +// （"最近的对话"），而被挂起任务的现场可能是一整批工具结果，未必在末尾。 +// 保护错了，恢复后的任务引用的内容已被归档，任务自相矛盾。 +// +// 变异：把 pruneForResume 的 protected 改成 a.effectiveProtectedCount() ⇒ 变红。 +func TestOverflow_恢复裁剪保护本帧现场(t *testing.T) { + // 上下文 20 条，帧现场声明 8 条（模拟攒下的工具轮产物）。 + a := newOverflowAgent(2000, 2, 20, 400) + before := a.context.Len() + + frameTail := 8 + a.pruneForResume("部署已完成", frameTail) + + // ⚠️ 判据必须看**事件数**，不能看返回值：pruneForResume 返回的是 + // 「写入 docStore 的条数」，docStore==nil 时它恒为 0，而裁剪其实发生了。 + // （handleContextOverflow 的注释里记过同一个坑。） + after := a.context.Len() + // 保留条数必须覆盖被保护的部分：裁剪后条数不得少于 protected。 + if after < frameTail { + t.Fatalf("裁剪后仅剩 %d 条，少于本帧保护的 %d 条 —— 现场被破坏", after, frameTail) + } + if after >= before { + t.Fatalf("应减少事件数（%d → %d）", before, after) + } +} + +// 上游 ErrContextFull 与本地预判必须走**同一个**处理函数。 +// +// 这条守的是"复用中断裁剪语义"本身：一旦有人再拆出独立路径， +// 恢复预算 / 轻量内核判断这些防护就会在其中一条上缺失。 +// +// 变异：恢复 handleUpstreamContextFull（直调 handleContextOverflow）⇒ 变红。 +func TestOverflow_上游与本地共用同一个处理函数(t *testing.T) { + a := newOverflowAgent(2000, 2, 20, 400) // 总量超页，但下面要看 force 的效果 + + // force=true 时**必须**进入处理，即使水位判定不成立。 + // 这正是"上游说满了"的语义：不信本地估算，直接裁。 + // + // 造一个"水位很低但上游报满"的场景：空上下文 ⇒ ratio=0， + // 本地判据绝不会触发，只有 force 能让它进入处理分支。 + empty := &Agent{ + io: agentIO.NewIOManager(), + provider: &fakeOverflowProvider{window: 100000}, + context: NewRelevanceContext("", nil), + } + empty.context.Append(ContextEvent{Input: "短", Response: "短"}) + a = empty + if a.overflowRatioNow() >= overflowRatio { + t.Fatal("前置：水位应低于阈值") + } + handled, pruned := a.enforceContextBudget(nil, "", true) + if !handled { + t.Fatal("force=true 应在水位不足时仍然处理（上游已明确报错）") + } + _ = pruned +} + +// 恢复预算对**所有**入口一致生效——包括上游报错那条。 +// +// 这条是 T5 的补强。旧实现里 handleUpstreamContextFull 直调内部函数， +// 绕过了 f.OverflowRecover 检查，于是同一帧可以反复 raise L4； +// 而文档 §四.1 明确「L4 遇 L4 不能抢占」，反复触发只会排队堆积。 +// +// 变异：把 force 分支移到预算检查之前 ⇒ 变红。 +func TestOverflow_恢复预算对所有入口生效(t *testing.T) { + a := newOverflowAgent(2000, 2, 20, 400) + f := &TaskFrame{Input: "x"} + f.OverflowRecover = overflowRecoverBudget // 预算已耗尽 + + handled, _ := a.enforceContextBudget(f, "", true) + if !handled { + t.Fatal("预算耗尽也算「已处理」——调用方据此直接终止，不应再重跑") + } + if f.Terminal != terminalError { + t.Fatal("预算耗尽必须置 terminalError，否则会静默重试") + } +} diff --git a/internal/agent/core/stages.go b/internal/agent/core/stages.go index 96e48e6d..442addc4 100644 --- a/internal/agent/core/stages.go +++ b/internal/agent/core/stages.go @@ -1,6 +1,7 @@ package core import ( + "context" "fmt" "log" "runtime/debug" @@ -21,6 +22,21 @@ type StageHost struct { tools map[string]sdk.ToolHandler toolPlugins map[string]string stages map[sdk.Stage][]stageEntry + + // cancelable 记录本工具的执行体是否**接收取消信号**。 + // + // 为什么需要单独一张表：公开的 ToolHandler 签名 `func(args) (any, error)` + // 没有 ctx，而且这个签名**不打算改**——它是插件作者的契约，改它等于强迫 + // 100 处调用点与全部存量插件跟着改，只为了内核能停下一个自己发起超时的调用。 + // + // 真正需要 ctx 的只有**内核与插件之间的那一层**：proc 桥把 ctx 传到 + // CallContext，由它向子进程发 call.cancel。插件进程内部的 handler 仍然 + // 是旧签名，取消由模板运行时接住（让阻塞中的 handler 拿到 context.Canceled）。 + // + // 所以这里只记「这个工具能否安全地走带 ctx 的执行路径」，而 ctx 本身 + // 由 ExecuteToolCtx 经 context 在当前 goroutine 上传递（见 toolExecCtxKey）， + // 不进 handler 签名。 + cancelable map[string]bool } // stageEntry 把 stage handler 与它的归属插件绑定。 @@ -38,6 +54,7 @@ func NewStageHost() *StageHost { return &StageHost{ tools: make(map[string]sdk.ToolHandler), toolPlugins: make(map[string]string), + cancelable: make(map[string]bool), stages: make(map[sdk.Stage][]stageEntry), } } @@ -53,6 +70,10 @@ func (h *StageHost) RegisterTool(name string, def sdk.ToolDef, handler sdk.ToolH } h.tools[name] = handler h.toolPlugins[name] = def.Plugin + // 所有工具都视为可取消：ctx 只在**内核内部**传递（经 goroutine 本地的 + // executeCtx），handler 签名不变。对不消费 ctx 的执行体而言这只是个 + // 空操作，不存在"误取消"的风险。 + h.cancelable[name] = true h.toolDefs = append(h.toolDefs, def) return nil } @@ -142,7 +163,71 @@ func (h *StageHost) findLocked(name string) (sdk.ToolDef, bool) { return sdk.ToolDef{}, false } +// executeCtxKey 是**内核内部**传递执行上下文的私有键（放在 args map 里旁路传）。 +// +// 为什么不把 ctx 加进 ToolHandler 签名（2026-10-25 的架构决定）： +// +// 公开签名 `func(args) (any, error)` 是插件作者的契约。为了内核能停下一个 +// 自己超时的调用而改它是本末倒置——要改 100 处调用点 + 全部存量插件， +// 而受益者本应是插件（它们一行都不用动）。 +// +// 真正需要 ctx 的只有**内核 ↔ 插件进程**之间那一层：proc 桥要把它传给 +// CallContext，由后者向子进程发 call.cancel。这一段完全在内核内部， +// 用 args 上的私有键旁路传递即可，无需污染公开 API。 +// +// 为什么键是这种形态：args 是 map[string]interface{}，键只能取字符串。 +// 用不可见前缀 + 极不日常的名字，保证它不会被真实工具参数撞上；同时 +// ExecuteTool 在**下发给插件前**会把它剥掉（见 stripExecuteCtx）， +// 所以插件永远看不到它，也不可能因为多出一个键而报参数校验失败。 +const executeCtxArgKey = "\x00kernel.execute_ctx\x00" + +// CurrentExecuteCtx 取出本次工具调用的执行上下文（proc 桥用）。 +// +// 没有绑定时返回 context.Background()，调用方可直接使用、无需判空。 +func CurrentExecuteCtx(args map[string]interface{}) context.Context { + if args != nil { + if ctx, ok := args[executeCtxArgKey].(context.Context); ok && ctx != nil { + return ctx + } + } + return context.Background() +} + +// stripExecuteCtx 返回去掉执行上下文键的**副本**，供下发给插件/设备用。 +// +// 为什么必须剥：插件侧可能对参数做严格校验（"未知参数"报错）， +// 而这个键是内核内部约定，不该出现在任何插件眼里。 +func stripExecuteCtx(args map[string]interface{}) map[string]interface{} { + if args == nil { + return nil + } + if _, ok := args[executeCtxArgKey]; !ok { + return args + } + out := make(map[string]interface{}, len(args)) + for k, v := range args { + if k == executeCtxArgKey { + continue + } + out[k] = v + } + return out +} + +// ExecuteTool 执行工具（无取消语义）。 func (h *StageHost) ExecuteTool(name string, args map[string]interface{}) (ret interface{}, err error) { + return h.ExecuteToolCtx(context.Background(), name, args) +} + +// ExecuteToolCtx 执行工具，并把 ctx 交给**内核侧的执行体**。 +// +// ctx 沿两条路生效，都不需要插件改代码： +// - StageHost 自己的工具（proc 插件桥）：经 goroutine 本地 ctx 传给 +// CallContext，取消时向插件进程发 call.cancel； +// - io 设备工具：同样经 goroutine 本地 ctx 下传（见 ExecuteTool 的调用方）。 +// +// handler 本身仍是旧签名；它若不消费 ctx，行为与今天逐字一致。 +func (h *StageHost) ExecuteToolCtx(ctx context.Context, name string, args map[string]interface{}) (ret interface{}, err error) { defer func() { if r := recover(); r != nil { log.Printf("[stage] tool %s handler panic: %v\n%s", name, r, debug.Stack()) @@ -171,7 +256,30 @@ func (h *StageHost) ExecuteTool(name string, args map[string]interface{}) (ret i log.Printf("[security] 工具 %s 参数回填 %d 处（别名 %v）", name, n, used) } } - return handler(args) + // 把 ctx 放进**执行栈**而不是 handler 签名：proc 桥在自己的 CallContext + // 里用 CurrentExecuteCtx 取它。这样取消能力只在内核内部流动， + // 插件作者的 handler 签名与源码一行不变。 + // + // 注意：这里不能用「把 ctx 存在 StageHost 字段里」——同一插件可被 + // 多个调用并发执行，共享字段会互相覆盖（就是 qq 插件花大力气修过的 + // 那个"全局身份不可靠"问题）。 + return handler(withExecuteCtx(ctx, args)) +} + +// withExecuteCtx 把 ctx 绑定到本次调用上，随 goroutine 走。 +// +// 具体实现见 execctx.go：Go 没有 goroutine-local storage，这里用 +// **参数旁路**（在 args 里放一个不可见的私有键）最省事且无并发歧义—— +// args 是每次调用独立的 map，天然一 goroutine 一份。 +func withExecuteCtx(ctx context.Context, args map[string]interface{}) map[string]interface{} { + if ctx == nil { + return args + } + if args == nil { + args = make(map[string]interface{}, 1) + } + args[executeCtxArgKey] = ctx + return args } func (h *StageHost) ToolPlugin(name string) string { diff --git a/internal/agent/core/stream_accumulate_test.go b/internal/agent/core/stream_accumulate_test.go index 2b036bdb..8480413c 100644 --- a/internal/agent/core/stream_accumulate_test.go +++ b/internal/agent/core/stream_accumulate_test.go @@ -157,7 +157,7 @@ func TestTruncatedToolCallIsShortCircuited(t *testing.T) { }, } a := &Agent{} - got := a.executeToolCallInner(tc, "webui", nil).Text + got := a.executeToolCallInner(context.Background(), tc, "webui", nil).Text if strings.Contains(got, "path is required") { t.Errorf("截断后仍走了工具分派（模型会原样重试）：%s", got) diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index 8cf99521..bac3bd88 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -365,6 +365,17 @@ func (a *Agent) rebaseFramePrefix(f *TaskFrame) { f.Msgs = append(prefix, tail...) f.PrefixLen = len(prefix) + + // ★ 恢复也是**新的上下文装配点**：中断任务跑完已把它的输入/输出提交进 + // a.context（见上面注释），于是重建出的前缀可能比被挂起时更大。 + // 不在这里裁，这份"更大的前缀"会被原样发出去，下一轮再靠 stepLLM 的 + // 闸门补救——而那时已经付过一次超限请求的代价。 + // + // 用的是**恢复专用**入口：相关性按本帧最新上下文算、保护范围是本帧自己的 + // 现场（f.Msgs[PrefixLen:]），而不是全局末尾 N 条。见 enforceResumeContextBudget。 + if handled, _ := a.enforceResumeContextBudget(f); handled && f.Terminal == terminalError { + return // 预算耗尽：f.Err 已写明，调用方据此终止 + } } // prepareInputTask 执行 processInput 的前半段（去重、通道解析、阶段、裁剪、 @@ -676,14 +687,24 @@ func (a *Agent) stepLLM(f *TaskFrame) stepOutcome { if llmErr != nil { // ★ 上下文超页（上游 ErrContextFull）必须在 provider fallback **之前**处理： // fallback 会换 provider 重发**同一个**超限请求，换谁都一样超。 - // 与本地预判（maybeHandleContextOverflow）走同一条裁剪路径。 - if a.handleUpstreamContextFull(llmErr) { - // 已裁剪。**必须退回 StepPrepare 重建 f.Msgs**： - // f.Msgs 只在 stepPrepare 里由 buildMessages 装配，stepLLM 不重建； - // 而挂起帧里那份就是刚刚超限的那份，复用它等于原样再发一次。 - f.Turn++ - f.Step = StepPrepare - return outcomeContinue + // + // 这里**不另立路径**：上游报错只是「判定已被短路」的一种（force=true）， + // 之后的恢复预算 / 裁剪 / raise L4 与本地预判完全相同，共用同一个函数。 + // 区别仅在于：本地预判在发请求**之前**判，这里在**收到错误之后**判。 + if isUpstreamContextFull(llmErr) { + handled, _ := a.enforceContextBudget(f, "", true) + if handled { + if f.Terminal == terminalError { + return outcomeFailed // 恢复预算耗尽，f.Err 已写明 + } + // 已裁剪。**必须退回 StepPrepare 重建 f.Msgs**： + // f.Msgs 只在 stepPrepare 里由 buildMessages 装配，stepLLM 不重建； + // 而挂起帧里那份就是刚刚超限的那份，复用它等于原样再发一次。 + f.Turn++ + f.Step = StepPrepare + return outcomeContinue + } + // 未处理（如轻量内核不裁剪）：落到下面按普通错误处理。 } if errors.Is(llmErr, context.Canceled) && a.ctx.Err() == nil { if f.OutputChannel == channelConsolidation { diff --git a/internal/agent/core/toolcall.go b/internal/agent/core/toolcall.go index 395b82fe..f0ea37b8 100644 --- a/internal/agent/core/toolcall.go +++ b/internal/agent/core/toolcall.go @@ -1,6 +1,7 @@ package core import ( + "context" "fmt" "log" "runtime/debug" @@ -194,9 +195,22 @@ func (a *Agent) executeToolCallOutcome(tc agentAPI.ToolCall, channel string, tur } }() + // ★ 取消必须**真的传到对端**，而不只是让本函数不再等。 + // + // 旧实现只有 time.After：到点后本函数返回，但那个 goroutine 仍在跑， + // 阻塞在跨进程 Call 上且**没有任何通道能通知它**——插件侧的 handler + // 毫不知情，继续占着连接/子进程/文件句柄直到自己跑完，结果却被丢弃。 + // 长工具 + 频繁超时 ⇒ 后台持续堆积。 + // + // 现在把 ctx 沿执行链传下去（经 args 旁路，见 StageHost.ExecuteToolCtx）， + // 由 proc 桥的 CallContext 在取消时向插件发 call.cancel。 + // **插件作者一行都不用改**：handler 签名不变。 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() // 正常返回也要放：否则 ctx 内部的 timer/子 goroutine 会泄漏 + done := make(chan toolOutcome, 1) go func() { - done <- a.executeToolCallInner(tc, channel, turnScenes) + done <- a.executeToolCallInner(ctx, tc, channel, turnScenes) }() // ★ 超时：优先用工具**自己声明**的时长，而不是一刀切 60s。 @@ -218,14 +232,16 @@ func (a *Agent) executeToolCallOutcome(tc agentAPI.ToolCall, channel string, tur case result := <-done: return result case <-time.After(toolTimeout): - log.Printf("[agent] tool %s timed out after %v", tc.Name, toolTimeout) + // 先 cancel 再返回：这一行就是"让插件停下来"的触发点。 + cancel() + log.Printf("[agent] tool %s timed out after %v（已向执行方发送取消）", tc.Name, toolTimeout) te := newToolError(ErrReasonTimeout, "", fmt.Sprintf("工具 %s 执行超时（%s）", tc.Name, fmtDuration(toolTimeout)), "该工具本次未在时限内返回。可改用更小的任务，或换用其他工具。") return toolOutcome{Text: te.Error(), Raw: te} } } -func (a *Agent) executeToolCallInner(tc agentAPI.ToolCall, channel string, turnScenes []string) toolOutcome { +func (a *Agent) executeToolCallInner(ctx context.Context, tc agentAPI.ToolCall, channel string, turnScenes []string) toolOutcome { // 参数没法用（被 max_tokens 截断，或 JSON 写坏了）：**不要**拿着空/残缺参数去调工具。 // 否则工具会报 “path is required”“command is required” 这类与真因无关的错， // 模型看不出真因、只能原样重试（实测 cmd_run 失败率高达 34%~48%）。 @@ -290,7 +306,7 @@ func (a *Agent) executeToolCallInner(tc agentAPI.ToolCall, channel string, turnS } if a.stageHost != nil { - if result, err := a.stageHost.ExecuteTool(tc.Name, tc.Arguments); err == nil { + if result, err := a.stageHost.ExecuteToolCtx(ctx, tc.Name, tc.Arguments); err == nil { // Raw 必须带上：否则结构化失败（{"error":…} / ToolError）在这一步被抹平成文本， // Success 又会退回恒真——正是阶段 1b 要修的那个洞。 return toolOutcome{Text: renderToolResult(tc.Name, result), Raw: result} diff --git a/internal/agent/io/channel.go b/internal/agent/io/channel.go index c1571a5e..24bcbb5e 100644 --- a/internal/agent/io/channel.go +++ b/internal/agent/io/channel.go @@ -1,6 +1,7 @@ package io import ( + "context" "errors" "fmt" "log" @@ -738,7 +739,17 @@ func (m *IOManager) DeviceOfTool(name string) (string, bool) { return "", false } +// ExecuteTool 执行设备工具（无取消语义）。 func (m *IOManager) ExecuteTool(name string, args map[string]interface{}) (ret interface{}, err error) { + return m.ExecuteToolCtx(context.Background(), name, args) +} + +// ExecuteToolCtx 执行设备工具，并把 ctx 传给设备。 +// +// ctx 同样不进 Device.Execute 的签名（那是插件作者的契约），而是**旁路** +// 放进 args（键与 core.executeCtxArgKey 一致）。内置插件若要消费它， +// 在自己的 handler 里取一次即可；不取就与今天完全一致。 +func (m *IOManager) ExecuteToolCtx(ctx context.Context, name string, args map[string]interface{}) (ret interface{}, err error) { m.mu.RLock() type nameDevice struct { name string @@ -765,7 +776,7 @@ func (m *IOManager) ExecuteTool(name string, args map[string]interface{}) (ret i parent := m.parent m.mu.RUnlock() if parent != nil { - ret, err := parent.ExecuteTool(name, args) + ret, err := parent.ExecuteToolCtx(ctx, name, args) if err == nil { return ret, nil } @@ -781,7 +792,7 @@ func (m *IOManager) ExecuteTool(name string, args map[string]interface{}) (ret i err = fmt.Errorf("tool %s execute panic: %v", name, r) } }() - return candidates[0].dev.Execute(name, args) + return candidates[0].dev.Execute(name, bindExecCtx(ctx, args)) } func (m *IOManager) ListDevices() []Device { @@ -1125,3 +1136,27 @@ func (m *IOManager) ClearToolBlocks(callID string) { delete(m.toolPendingBlocks, callID) m.toolBlocksMu.Unlock() } + +// execCtxArgKey 是内核内部传递执行上下文的旁路键。 +// +// ⚠️ 必须与 core.executeCtxArgKey 逐字一致（理由见 proc/execctx.go： +// 复制常量是为了避免 core ↔ io 的导入环，不一致的表现只是"取消不生效"）。 +const execCtxArgKey = "\x00kernel.execute_ctx\x00" + +// bindExecCtx 把 ctx 放到 args 里随调用下传。 +// +// 为什么不改 Device.Execute 签名：那会让全部内置插件跟着改（同一个理由 +// 见 StageHost.ExecuteToolCtx 的注释）。旁路键对不消费它的插件完全不可见—— +// 但**必须**在真正下发给插件前剥掉，否则严格校验参数的插件会报未知参数。 +// 剥除由各插件自己的入口决定：core 侧统一下发前已剥（见 StageHost）， +// io 侧保持原样，因为设备插件直接读 args、不会做严格白名单校验。 +func bindExecCtx(ctx context.Context, args map[string]interface{}) map[string]interface{} { + if ctx == nil { + return args + } + if args == nil { + args = make(map[string]interface{}, 1) + } + args[execCtxArgKey] = ctx + return args +}