diff --git a/internal/agent/core/process.go b/internal/agent/core/process.go index dc59975..23c261b 100644 --- a/internal/agent/core/process.go +++ b/internal/agent/core/process.go @@ -7,12 +7,10 @@ import ( "fmt" "log" "strings" - "time" agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" "gitcode.com/JianFeeeee/HomeAgent/internal/events" sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" - pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" ) // continuationPlaceholder 是工具轮之后补的 user 占位内容。 @@ -99,420 +97,6 @@ func dropContinuationPlaceholders(msgs []agentAPI.Message) []agentAPI.Message { return out } -func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) { - a.mu.Lock() - defer a.mu.Unlock() - - if a.provider == nil { - return "", nil, nil, fmt.Errorf("agent: no LLM provider configured") - } - - budget := ComputeTokenBudget(a.provider, a.systemPrompt) - - memContext := a.buildMemoryContext(input, budget.MemoryTokens) - sysPrompt := a.buildSystemPrompt(memContext, input) - tools := a.buildToolDefs() - - msgs := a.buildMessages(sysPrompt, input, budget.ContextTokens) - // 工具提醒（interrupt）：以 system 角色注入，不让模型误认为用户发言 - if a.interruptInput { - last := msgs[len(msgs)-1] - last.Role = "system" - last.Content = "[中断消息] " + last.Content - msgs[len(msgs)-1] = last - a.interruptInput = false - } - if blocks, ok := stageCtx.Extra["media_blocks"].([]agentAPI.ContentBlock); ok && len(blocks) > 0 { - if len(msgs) > 0 { - msgs[len(msgs)-1].Blocks = blocks - } - } - - log.Printf("[agent] tool call loop start, max_ctx=%d target=%d fixed=%d mem=%d ctx=%d %d tools, %d events, personality=%t, docs=%d", - budget.MaxContext, budget.TargetUsage, budget.FixedTokens, budget.MemoryTokens, budget.ContextTokens, - len(tools), a.context.Len(), - a.personality != nil && a.personality.Content != "", - a.docStoreSize()) - - if a.runStage(sdk.StagePreAction, stageCtx) { - return *stageCtx.Response, toolsUsed, toolResults, nil - } - if len(stageCtx.ContextMsgs) > 0 { - for _, m := range stageCtx.ContextMsgs { - role, _ := m["role"].(string) - content, _ := m["content"].(string) - if role != "" { - msgs = append(msgs, agentAPI.Message{Role: role, Content: content}) - } - } - } - - // lastBatchReplyOnly 记录上一批工具调用是否全部是输出通道发送。 - lastBatchReplyOnly := false - - for turn := 0; ; turn++ { - for _, interrupt := range a.drainInterrupts() { - msgs = append(msgs, agentAPI.Message{ - Role: "system", - Content: "[中断消息] " + interrupt, - }) - } - - // zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验), - // 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。 - // 注意:仅当尾部确为工具轮产物(assistant/tool)时才补位;首轮 system 上下文结尾不补, - // 否则会错误覆盖实际用户输入(如 injectSourceContext 追加的 system 说明)。 - // - // 补位前先移除前面轮次插入的同类占位，保证占位**不随轮次线性累积**—— - // 占位是核心插的传输层附加物，不是用户发言，不该在 prompt 里叠成 N 条。 - // - // 文案分情况：上一批全是 output_send__* 时不能说“继续”，详见 - // replyDeliveredPlaceholder 的说明。 - msgs = dropContinuationPlaceholders(msgs) - if last := msgs[len(msgs)-1]; last.Role == "assistant" || last.Role == "tool" { - msgs = append(msgs, agentAPI.Message{ - Role: "user", - Content: continuationFor(lastBatchReplyOnly), - }) - } - - req := &agentAPI.CompletionRequest{ - Messages: msgs, - MaxTokens: 4096, - Tools: tools, - ToolChoice: "auto", - DisableThinking: !a.thinkingEnabled, - } - - var providers []agentAPI.Provider - if a.providerManager != nil { - // 精确模型名走 byModel 路由；AUTO/空走优先级链 - var allProviders []agentAPI.Provider - if req.Model != "" && !strings.EqualFold(req.Model, "AUTO") { - allProviders = a.providerManager.ResolveForModel(req.Model) - } else { - allProviders = a.providerManager.OrderedProviders() - } - providers = make([]agentAPI.Provider, 0, len(allProviders)) - for _, p := range allProviders { - if a.providerManager.IsAvailable(p.Name()) { - providers = append(providers, p) - } - } - } - if len(providers) == 0 { - providers = []agentAPI.Provider{a.provider} - } - var resp *agentAPI.CompletionResponse - var llmErr error - - for pi, fbProvider := range providers { - if pi > 0 { - log.Printf("[agent] LLM fallback: trying provider %q (fallback #%d/%d)", - fbProvider.Name(), pi, len(providers)-1) - } - - // 同源瞬时错误重试：网关瞬断（502/503/504/429/网络抖动）通常秒级恢复， - // 直接跳下一个 provider（或直接报错）会丢掉本可成功的请求。 - // 凭证错误（401/403）与用户中断不重试。 - const maxAttempts = 2 - for attempt := 1; attempt <= maxAttempts; attempt++ { - if attempt > 1 { - log.Printf("[agent] provider %q transient failure, retry %d/%d in 2s: %v", - fbProvider.Name(), attempt, maxAttempts, llmErr) - select { - case <-time.After(2 * time.Second): - case <-a.ctx.Done(): - llmErr = a.ctx.Err() - } - if llmErr == nil || errors.Is(llmErr, context.Canceled) || errors.Is(llmErr, context.DeadlineExceeded) { - break - } - } - - fCtx, fCancel := context.WithCancel(a.ctx) - a.llmMu.Lock() - a.cancelLLM = fCancel - a.llmMu.Unlock() - - resp, llmErr = chatStreamWithFallback(fCtx, fbProvider, req, a) - - a.llmMu.Lock() - a.cancelLLM = nil - a.llmMu.Unlock() - fCancel() - - if llmErr == nil { - a.providerManager.ResetAvailability(fbProvider.Name()) - if fbProvider != a.provider { - a.provider = fbProvider - log.Printf("[agent] switched active provider to %q after fallback", - fbProvider.Name()) - } - break - } - - // 用户中断：立即终止，不重试也不换 provider - if errors.Is(llmErr, context.Canceled) { - break - } - // 凭证错误：重试无意义，跳出重试循环进入 provider 标记/切换 - var pe *agentAPI.ProviderError - if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) { - break - } - // 其余错误（含 5xx/429/网络）：还有重试机会则继续，否则跳出 - } - - if llmErr == nil { - break - } - if errors.Is(llmErr, context.Canceled) { - break - } - var pe *agentAPI.ProviderError - if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) { - a.providerManager.ReportStatus(fbProvider.Name(), pe.StatusCode) - log.Printf("[agent] provider %q marked unavailable (HTTP %d)", fbProvider.Name(), pe.StatusCode) - } else { - a.providerManager.MarkUnavailable(fbProvider.Name()) - } - log.Printf("[agent] provider %q failed: %v", fbProvider.Name(), llmErr) - } - - if llmErr != nil { - if errors.Is(llmErr, context.Canceled) && a.ctx.Err() == nil { - if a.currentOutputChannel == "_consolidation_" { - return "", toolsUsed, toolResults, fmt.Errorf("interrupted by user input") - } - continue - } - return "", toolsUsed, toolResults, fmt.Errorf("all %d providers failed, last error: %w", - len(providers), llmErr) - } - - stageCtx.LLMText = resp.Content - stageCtx.ReasoningContent = resp.ReasoningContent - stageCtx.TokenUsage = map[string]int{ - "prompt_tokens": resp.TokenUsage.Prompt, - "completion_tokens": resp.TokenUsage.Completion, - "total_tokens": resp.TokenUsage.Total, - } - stageCtx.ToolCalls = convertToolCalls(resp.ToolCalls) - for i := range stageCtx.ToolCalls { - if stageCtx.ToolCalls[i].Plugin == "" { - stageCtx.ToolCalls[i].Plugin = a.resolveToolPlugin(stageCtx.ToolCalls[i].Name) - } - } - if a.runStage(sdk.StagePostAction, stageCtx) { - return *stageCtx.Response, toolsUsed, toolResults, nil - } - resp.Content = stageCtx.LLMText - resp.ToolCalls = convertBackToolCalls(stageCtx.ToolCalls) - - chainPayload := map[string]interface{}{ - "content": resp.Content, - "reasoning": resp.ReasoningContent, - "tool_calls": resp.ToolCalls, - "phase": "intermediate", - "turn": turn, - } - if resp.TokenUsage.Total > 0 { - chainPayload["usage"] = map[string]int{ - "prompt": resp.TokenUsage.Prompt, - "completion": resp.TokenUsage.Completion, - "total": resp.TokenUsage.Total, - } - } - a.publishEvent(events.EventAgentLLMChain, chainPayload) - - if resp.ReasoningContent != "" { - a.publishEvent(events.EventReasoning, map[string]interface{}{ - "content": resp.ReasoningContent, - "channel": a.currentOutputChannel, - }) - } - - if len(resp.ToolCalls) == 0 { - return resp.Content, toolsUsed, toolResults, nil - } - - // 本批是否全部是输出通道发送（=模型刚交付了给用户的回复）。 - // 必须在执行前判定：执行过程中的中断/拒绝分支会 continue/break， - // 放在循环里统计会漏。 - replyOnly := true - for _, tc := range resp.ToolCalls { - if !isOutputDeliveryTool(tc.Name) { - replyOnly = false - break - } - } - - contentOnce := true - for _, tc := range resp.ToolCalls { - if len(a.interceptCh) > 0 { - for _, interrupt := range a.drainInterrupts() { - msgs = append(msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt}) - } - a.publishEvent(events.EventToolCall, map[string]interface{}{ - "tool": tc.Name, - "plugin": a.resolveToolPlugin(tc.Name), - "args": tc.Arguments, - "status": "interrupted", - "reason": "user interrupt before execution", - "channel": a.currentOutputChannel, - }) - break - } - - toolsUsed = append(toolsUsed, tc.Name) - pluginName := a.resolveToolPlugin(tc.Name) - log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID) - if tc.RawArguments != "" { - log.Printf("[agent] tool %s raw_arguments: %s", tc.Name, truncateStr(tc.RawArguments, 300)) - } - - sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Plugin: pluginName, Arguments: tc.Arguments} - stageCtx.ToolCalls = []sdk.ToolCall{sdkTC} - stageCtx.ToolResults = nil - if a.runStage(sdk.StageBeforeToolcall, stageCtx) { - result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name) - msgs = append(msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}}) - msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}) - a.publishEvent(events.EventToolCall, map[string]interface{}{ - "tool": tc.Name, - "plugin": pluginName, - "args": tc.Arguments, - "result": result, - "status": "denied", - "channel": a.currentOutputChannel, - }) - continue - } - tc.Arguments = stageCtx.ToolCalls[0].Arguments - - if pluginName != "" && !a.pluginHealth.isHealthy(pluginName) { - result := fmt.Sprintf("插件 %s 处于崩溃状态，已跳过执行，等待自动恢复重载", pluginName) - log.Printf("[agent] skip tool %s: plugin %s unhealthy", tc.Name, pluginName) - msgs = append(msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}}) - msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}) - continue - } - - result := a.executeToolCall(tc) - toolResults = append(toolResults, ToolResultItem{Name: tc.Name, Output: result}) - log.Printf("[agent] tool %s result: %s", tc.Name, truncateStr(result, 100)) - - stageCtx.ToolResults = []sdk.ToolResult{{CallID: tc.ID, Name: tc.Name, Plugin: pluginName, Success: true, Result: result}} - a.runStage(sdk.StageAfterToolcall, stageCtx) - if len(stageCtx.ToolResults) > 0 { - if r, ok := stageCtx.ToolResults[0].Result.(string); ok { - result = r - } - } - // ContextPolicy: prune 工具调用后执行上下文裁剪（§13.8） - if def := a.stageHost.ToolDef(tc.Name); def != nil && def.ContextPolicy == "prune" { - if a.context != nil { - topK := a.maxContextSize - 1 - if topK < 1 { - topK = 1 - } - // 查询向量取**清洗后**的有效内容，否则噪声（ANSI/base64/JSON - // 包装）会把相关性打分带偏，裁掉本该保留的事件。 - a.context.Prune(a.toolOutputForQuery(tc.Name, result), topK, a.docStore) - } - } - - msgContent := "" - if contentOnce { - msgContent = resp.Content - contentOnce = false - } - msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: msgContent, ReasoningContent: resp.ReasoningContent, ToolCalls: []agentAPI.ToolCall{tc}}) - - // 多模态工具结果：插件通过 SDK.SetToolBlocks 注入 image_url/audio_url block。 - // - // 媒体不挂在 tool message 上，而是另起一条紧随其后的 user message—— - // 这也是插件文案一直在说的「注入后续对话」。 - // 为何不能挂 tool message：同一张图、同一模型、三轮实测—— - // 图在 user message → 3/3 读到 - // 图在 tool message → 0/3（模型答「没能读到这张图」） - // tool 纯文本 + 后接 user → 3/3 读到 - // tool message 那轮 prompt_tokens 反而更高（7967 vs 7089），base64 确实 - // 进了上游，但 role=tool 上的多模态 content 数组不被当作可视内容。 - // - // 主模型不支持该模态时更不能直接塞：网关会把 image_url 静默剥离后仍 - // 返回 200，模型回答「我没有看到图片」而内核以为注入成功。改走回退链。 - toolMsg := agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result} - var mediaMsg *agentAPI.Message - if rawBlocks := a.io.ConsumeToolBlocks(); len(rawBlocks) > 0 { - var blocks []agentAPI.ContentBlock - for _, b := range rawBlocks { - if cb, ok := b.(pubsdk.ContentBlock); ok { - // 跨包类型拷贝（pubsdk.ContentBlock → agentAPI.ContentBlock） - block := agentAPI.ContentBlock{Type: cb.Type, Text: cb.Text} - if cb.ImageURL != nil { - block.ImageURL = &agentAPI.ImageURL{URL: cb.ImageURL.URL, Detail: cb.ImageURL.Detail} - } - if cb.AudioURL != nil { - block.AudioURL = &agentAPI.AudioURL{URL: cb.AudioURL.URL} - } - blocks = append(blocks, block) - } - } - if len(blocks) > 0 { - // 先落进 CAS：无论下面走直视还是回退转写，媒体本体都该进记忆。 - // 不存的后果是 ToolResultItem.Output 只剩那句 - // "[已将图片注入后续对话] /tmp/x.png"，文件一删线索就断了。 - a.stageMediaDigests(a.captureBlockMedia(blocks, tc.Name)...) - - if native, fallbackText := a.prepareToolBlocks(blocks); len(native) > 0 { - // 能直视：另起一条 user message 承载媒体，并补一句来源说明， - // 否则模型会把它当成用户新发的图而不是工具拉回来的。 - mediaBlocks := append([]agentAPI.ContentBlock{{ - Type: "text", - Text: fmt.Sprintf("[以下是 %s 注入的媒体内容]", tc.Name), - }}, native...) - mediaMsg = &agentAPI.Message{Role: "user", Blocks: mediaBlocks} - } else if fallbackText != "" { - // 回退链已把媒体转写成文字：并进 tool message 的纯文本 content， - // 不再另起消息（文字在 tool message 里本来就能被读到）。 - toolMsg.Content = result + "\n\n" + fallbackText - result = toolMsg.Content - if len(toolResults) > 0 { - toolResults[len(toolResults)-1].Output = result - } - } - } - } - msgs = append(msgs, toolMsg) - if mediaMsg != nil { - // 必须紧跟在 toolMsg 之后：中间插入其他消息会让 tool_call_id 配对断开。 - msgs = append(msgs, *mediaMsg) - } - - a.publishEvent(events.EventToolCall, map[string]interface{}{ - "tool": tc.Name, - "plugin": pluginName, - "args": tc.Arguments, - "result": result, - "status": "ok", - "channel": a.currentOutputChannel, - }) - - if len(a.interceptCh) > 0 { - for _, interrupt := range a.drainInterrupts() { - msgs = append(msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt}) - } - break - } - } - - // 供下一轮顶部选择补位文案。 - lastBatchReplyOnly = replyOnly - } -} - // chatStreamWithFallback 优先流式调用 provider，失败时回退非流式 Chat()。 // // 流式路径：ChatStream 拿到 chunk channel，逐块累积 content/reasoning_content， diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go new file mode 100644 index 0000000..ec4c40d --- /dev/null +++ b/internal/agent/core/task.go @@ -0,0 +1,613 @@ +package core + +// 任务状态机（M1：行为等价的纯重构）。 +// +// 背景与设计见 docs/zh/input-scheduler-design.md。 +// +// M1 只做一件事：把原先「一个 425 行的 process() 大函数」拆成 +// **显式 step 游标 + TaskFrame**。目的不是加能力，而是让「现场」变成数据—— +// 之后 M3 才能把帧存进 suspendPool 并在安全点恢复。 +// +// 行为等价的判据：既有全部 agent 测试通过，且 R3/X3（见设计文档 §11）通过。 +// +// 本文件**不引入**优先级、抢占、队列与并发；那些在 M2 起逐层加上。 +// +// Step 与安全点（设计文档 §4.2）： +// step 与 step 之间是安全点；StepToolExec（工具执行）与 StepPrepare 中的 +// ONNX/落盘片段是**临界区**，执行中不可抢占。 +// +// 与设计文档的差异：文档里的 StepBeforeOutput / StepAfterOutput / StepCommit / +// StepFinish 属于 emitResponse 与 processInput 层，M1 不动它们（M6 再迁入帧）。 + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "time" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + "gitcode.com/JianFeeeee/HomeAgent/internal/events" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" + pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" +) + +// Step 是任务状态机的游标。 +type Step int + +const ( + // StepPrepare 构建消息与工具表、应用中断标记、跑 pre_action 阶段。 + StepPrepare Step = iota + // StepLLM 轮次顶部（中断/占位）+ LLM 调用（含 provider 回退与重试）+ post_action。 + StepLLM + // StepToolBegin 取本批下一个工具，跑 before_toolcall；被拒/插件不健康则跳过。 + StepToolBegin + // StepToolExec 执行工具。**临界区**：副作用不可回滚，执行中不是安全点。 + StepToolExec + // StepToolAfter after_toolcall 阶段、上下文裁剪、消息与事件组装、批后中断检查。 + StepToolAfter + // StepTurnEnd 收尾本批并进入下一轮。 + StepTurnEnd +) + +// stepOutcome 是一次 step 执行的结果。 +type stepOutcome int + +const ( + // outcomeContinue 继续执行下一个 step（游标可能停在原地以表达"重跑本 step"）。 + outcomeContinue stepOutcome = iota + // outcomeDone 任务成功结束，响应在 frame.Response。 + outcomeDone + // outcomeFailed 任务失败结束，错误在 frame.Err。 + outcomeFailed +) + +// TaskFrame 承载一个任务在安全点之间必须存活的所有状态。 +// +// 不变量（设计文档 §8.1 I3）：帧是**纯数据**；不得持有任何锁或资源跨越安全点。 +type TaskFrame struct { + Input string + StageCtx *sdk.StageContext + + // 跨轮次状态 + Msgs []agentAPI.Message + Tools []interface{} + ToolsUsed []string + ToolResults []ToolResultItem + Turn int + LastBatchReplyOnly bool + + // 当前工具批 + PendingTools []agentAPI.ToolCall + ToolIdx int + ReplyOnly bool + ContentOnce bool + CurTool agentAPI.ToolCall + CurToolPlugin string + CurResult string + Resp *agentAPI.CompletionResponse + + // 游标与终态 + Step Step + Response string + Err error +} + +func (a *Agent) newTaskFrame(input string, stageCtx *sdk.StageContext) *TaskFrame { + return &TaskFrame{Input: input, StageCtx: stageCtx, Step: StepPrepare} +} + +// process 驱动状态机直到任务结束，返回与原实现完全相同的四元组。 +// +// 保留该签名是为了让 M1 成为纯内部重构：所有调用方（processInput / +// processConsolidation / 测试）无需改动。 +func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) { + a.mu.Lock() + defer a.mu.Unlock() + + if a.provider == nil { + return "", nil, nil, fmt.Errorf("agent: no LLM provider configured") + } + + f := a.newTaskFrame(input, stageCtx) + // 步数上限只是防"转移缺失导致死循环"的护栏；正常任务远达不到。 + const maxSteps = 1 << 20 + for i := 0; i < maxSteps; i++ { + switch a.step(f) { + case outcomeDone: + return f.Response, f.ToolsUsed, f.ToolResults, nil + case outcomeFailed: + return "", f.ToolsUsed, f.ToolResults, f.Err + } + } + return "", f.ToolsUsed, f.ToolResults, + fmt.Errorf("agent: task step budget exhausted（状态机未收敛，疑似转移缺失）") +} + +// step 执行恰好一个 step。 +func (a *Agent) step(f *TaskFrame) stepOutcome { + switch f.Step { + case StepPrepare: + return a.stepPrepare(f) + case StepLLM: + return a.stepLLM(f) + case StepToolBegin: + return a.stepToolBegin(f) + case StepToolExec: + return a.stepToolExec(f) + case StepToolAfter: + return a.stepToolAfter(f) + case StepTurnEnd: + return a.stepTurnEnd(f) + default: + f.Err = fmt.Errorf("agent: unknown task step %d", f.Step) + return outcomeFailed + } +} + +// stepPrepare 构建本轮任务的初始帧。 +func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome { + budget := ComputeTokenBudget(a.provider, a.systemPrompt) + + memContext := a.buildMemoryContext(f.Input, budget.MemoryTokens) + sysPrompt := a.buildSystemPrompt(memContext, f.Input) + f.Tools = a.buildToolDefs() + + f.Msgs = a.buildMessages(sysPrompt, f.Input, budget.ContextTokens) + // 工具提醒（interrupt）：以 system 角色注入，不让模型误认为用户发言 + if a.interruptInput { + last := f.Msgs[len(f.Msgs)-1] + last.Role = "system" + last.Content = "[中断消息] " + last.Content + f.Msgs[len(f.Msgs)-1] = last + a.interruptInput = false + } + if blocks, ok := f.StageCtx.Extra["media_blocks"].([]agentAPI.ContentBlock); ok && len(blocks) > 0 { + if len(f.Msgs) > 0 { + f.Msgs[len(f.Msgs)-1].Blocks = blocks + } + } + + log.Printf("[agent] tool call loop start, max_ctx=%d target=%d fixed=%d mem=%d ctx=%d %d tools, %d events, personality=%t, docs=%d", + budget.MaxContext, budget.TargetUsage, budget.FixedTokens, budget.MemoryTokens, budget.ContextTokens, + len(f.Tools), a.context.Len(), + a.personality != nil && a.personality.Content != "", + a.docStoreSize()) + + if a.runStage(sdk.StagePreAction, f.StageCtx) { + f.Response = *f.StageCtx.Response + return outcomeDone + } + if len(f.StageCtx.ContextMsgs) > 0 { + for _, m := range f.StageCtx.ContextMsgs { + role, _ := m["role"].(string) + content, _ := m["content"].(string) + if role != "" { + f.Msgs = append(f.Msgs, agentAPI.Message{Role: role, Content: content}) + } + } + } + + f.Step = StepLLM + return outcomeContinue +} + +// stepLLM 是轮次顶部与 LLM 调用。 +// +// 取消（context.Canceled 且 agent 未退出）时**留在本 step 并 Turn++**——等价于 +// 原实现的 `continue`：重新排空中断、补占位、重新请求。抢占挂起将在 M3 从这里接管。 +func (a *Agent) stepLLM(f *TaskFrame) stepOutcome { + for _, interrupt := range a.drainInterrupts() { + f.Msgs = append(f.Msgs, agentAPI.Message{ + Role: "system", + Content: "[中断消息] " + interrupt, + }) + } + + // zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验), + // 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。 + f.Msgs = dropContinuationPlaceholders(f.Msgs) + if last := f.Msgs[len(f.Msgs)-1]; last.Role == "assistant" || last.Role == "tool" { + f.Msgs = append(f.Msgs, agentAPI.Message{ + Role: "user", + Content: continuationFor(f.LastBatchReplyOnly), + }) + } + + req := &agentAPI.CompletionRequest{ + Messages: f.Msgs, + MaxTokens: 4096, + Tools: f.Tools, + ToolChoice: "auto", + DisableThinking: !a.thinkingEnabled, + } + + providers := a.resolveProviders(req) + resp, llmErr := a.callLLMWithFallback(req, providers) + + if llmErr != nil { + if errors.Is(llmErr, context.Canceled) && a.ctx.Err() == nil { + if a.currentOutputChannel == "_consolidation_" { + f.Err = fmt.Errorf("interrupted by user input") + return outcomeFailed + } + f.Turn++ + return outcomeContinue // 重跑 StepLLM + } + f.Err = fmt.Errorf("all %d providers failed, last error: %w", len(providers), llmErr) + return outcomeFailed + } + + f.StageCtx.LLMText = resp.Content + f.StageCtx.ReasoningContent = resp.ReasoningContent + f.StageCtx.TokenUsage = map[string]int{ + "prompt_tokens": resp.TokenUsage.Prompt, + "completion_tokens": resp.TokenUsage.Completion, + "total_tokens": resp.TokenUsage.Total, + } + f.StageCtx.ToolCalls = convertToolCalls(resp.ToolCalls) + for i := range f.StageCtx.ToolCalls { + if f.StageCtx.ToolCalls[i].Plugin == "" { + f.StageCtx.ToolCalls[i].Plugin = a.resolveToolPlugin(f.StageCtx.ToolCalls[i].Name) + } + } + if a.runStage(sdk.StagePostAction, f.StageCtx) { + f.Response = *f.StageCtx.Response + return outcomeDone + } + resp.Content = f.StageCtx.LLMText + resp.ToolCalls = convertBackToolCalls(f.StageCtx.ToolCalls) + + chainPayload := map[string]interface{}{ + "content": resp.Content, + "reasoning": resp.ReasoningContent, + "tool_calls": resp.ToolCalls, + "phase": "intermediate", + "turn": f.Turn, + } + if resp.TokenUsage.Total > 0 { + chainPayload["usage"] = map[string]int{ + "prompt": resp.TokenUsage.Prompt, + "completion": resp.TokenUsage.Completion, + "total": resp.TokenUsage.Total, + } + } + a.publishEvent(events.EventAgentLLMChain, chainPayload) + + if resp.ReasoningContent != "" { + a.publishEvent(events.EventReasoning, map[string]interface{}{ + "content": resp.ReasoningContent, + "channel": a.currentOutputChannel, + }) + } + + if len(resp.ToolCalls) == 0 { + f.Response = resp.Content + return outcomeDone + } + + // 本批是否全部是输出通道发送（=模型刚交付了给用户的回复）。 + // 必须在执行前判定：执行过程中的中断/拒绝分支会 continue/break，放在循环里统计会漏。 + f.Resp = resp + f.ReplyOnly = true + for _, tc := range resp.ToolCalls { + if !isOutputDeliveryTool(tc.Name) { + f.ReplyOnly = false + break + } + } + f.ContentOnce = true + f.PendingTools = resp.ToolCalls + f.ToolIdx = 0 + f.Step = StepToolBegin + return outcomeContinue +} + +// stepToolBegin 取本批下一个工具；批已耗尽或发生中断则进入收尾。 +func (a *Agent) stepToolBegin(f *TaskFrame) stepOutcome { + if f.ToolIdx >= len(f.PendingTools) { + f.Step = StepTurnEnd + return outcomeContinue + } + tc := f.PendingTools[f.ToolIdx] + + if len(a.interceptCh) > 0 { + for _, interrupt := range a.drainInterrupts() { + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt}) + } + a.publishEvent(events.EventToolCall, map[string]interface{}{ + "tool": tc.Name, + "plugin": a.resolveToolPlugin(tc.Name), + "args": tc.Arguments, + "status": "interrupted", + "reason": "user interrupt before execution", + "channel": a.currentOutputChannel, + }) + f.Step = StepTurnEnd + return outcomeContinue + } + + f.ToolsUsed = append(f.ToolsUsed, tc.Name) + pluginName := a.resolveToolPlugin(tc.Name) + log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID) + if tc.RawArguments != "" { + log.Printf("[agent] tool %s raw_arguments: %s", tc.Name, truncateStr(tc.RawArguments, 300)) + } + + sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Plugin: pluginName, Arguments: tc.Arguments} + f.StageCtx.ToolCalls = []sdk.ToolCall{sdkTC} + f.StageCtx.ToolResults = nil + if a.runStage(sdk.StageBeforeToolcall, f.StageCtx) { + result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name) + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}}) + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}) + a.publishEvent(events.EventToolCall, map[string]interface{}{ + "tool": tc.Name, + "plugin": pluginName, + "args": tc.Arguments, + "result": result, + "status": "denied", + "channel": a.currentOutputChannel, + }) + f.ToolIdx++ + return outcomeContinue + } + tc.Arguments = f.StageCtx.ToolCalls[0].Arguments + + if pluginName != "" && !a.pluginHealth.isHealthy(pluginName) { + result := fmt.Sprintf("插件 %s 处于崩溃状态，已跳过执行，等待自动恢复重载", pluginName) + log.Printf("[agent] skip tool %s: plugin %s unhealthy", tc.Name, pluginName) + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}}) + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}) + f.ToolIdx++ + return outcomeContinue + } + + f.CurTool = tc + f.CurToolPlugin = pluginName + f.Step = StepToolExec + return outcomeContinue +} + +// stepToolExec 执行工具。**临界区**：见设计文档 §4.3。 +func (a *Agent) stepToolExec(f *TaskFrame) stepOutcome { + result := a.executeToolCall(f.CurTool) + f.CurResult = result + f.ToolResults = append(f.ToolResults, ToolResultItem{Name: f.CurTool.Name, Output: result}) + log.Printf("[agent] tool %s result: %s", f.CurTool.Name, truncateStr(result, 100)) + + f.StageCtx.ToolResults = []sdk.ToolResult{{ + CallID: f.CurTool.ID, Name: f.CurTool.Name, Plugin: f.CurToolPlugin, + Success: true, Result: result, + }} + f.Step = StepToolAfter + return outcomeContinue +} + +// stepToolAfter 是工具执行后的全部后处理（阶段、裁剪、消息与事件）。 +func (a *Agent) stepToolAfter(f *TaskFrame) stepOutcome { + tc := f.CurTool + pluginName := f.CurToolPlugin + result := f.CurResult + + a.runStage(sdk.StageAfterToolcall, f.StageCtx) + if len(f.StageCtx.ToolResults) > 0 { + if r, ok := f.StageCtx.ToolResults[0].Result.(string); ok { + result = r + } + } + // ContextPolicy: prune 工具调用后执行上下文裁剪（§13.8） + if def := a.stageHost.ToolDef(tc.Name); def != nil && def.ContextPolicy == "prune" { + if a.context != nil { + topK := a.maxContextSize - 1 + if topK < 1 { + topK = 1 + } + // 查询向量取**清洗后**的有效内容，否则噪声（ANSI/base64/JSON 包装） + // 会把相关性打分带偏，裁掉本该保留的事件。 + a.context.Prune(a.toolOutputForQuery(tc.Name, result), topK, a.docStore) + } + } + + msgContent := "" + if f.ContentOnce { + msgContent = f.Resp.Content + f.ContentOnce = false + } + f.Msgs = append(f.Msgs, agentAPI.Message{ + Role: "assistant", Content: msgContent, + ReasoningContent: f.Resp.ReasoningContent, + ToolCalls: []agentAPI.ToolCall{tc}, + }) + + // 多模态工具结果：插件通过 SDK.SetToolBlocks 注入 image_url/audio_url block。 + // + // 媒体不挂在 tool message 上，而是另起一条紧随其后的 user message—— + // 这也是插件文案一直在说的「注入后续对话」。 + // 为何不能挂 tool message：同一张图、同一模型、三轮实测—— + // 图在 user message → 3/3 读到 + // 图在 tool message → 0/3（模型答「没能读到这张图」） + // tool 纯文本 + 后接 user → 3/3 读到 + // tool message 那轮 prompt_tokens 反而更高（7967 vs 7089），base64 确实 + // 进了上游，但 role=tool 上的多模态 content 数组不被当作可视内容。 + // + // 主模型不支持该模态时更不能直接塞：网关会把 image_url 静默剥离后仍 + // 返回 200，模型回答「我没有看到图片」而内核以为注入成功。改走回退链。 + toolMsg := agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result} + var mediaMsg *agentAPI.Message + if rawBlocks := a.io.ConsumeToolBlocks(); len(rawBlocks) > 0 { + var blocks []agentAPI.ContentBlock + for _, b := range rawBlocks { + if cb, ok := b.(pubsdk.ContentBlock); ok { + // 跨包类型拷贝（pubsdk.ContentBlock → agentAPI.ContentBlock） + block := agentAPI.ContentBlock{Type: cb.Type, Text: cb.Text} + if cb.ImageURL != nil { + block.ImageURL = &agentAPI.ImageURL{URL: cb.ImageURL.URL, Detail: cb.ImageURL.Detail} + } + if cb.AudioURL != nil { + block.AudioURL = &agentAPI.AudioURL{URL: cb.AudioURL.URL} + } + blocks = append(blocks, block) + } + } + if len(blocks) > 0 { + // 先落进 CAS：无论下面走直视还是回退转写，媒体本体都该进记忆。 + a.stageMediaDigests(a.captureBlockMedia(blocks, tc.Name)...) + + if native, fallbackText := a.prepareToolBlocks(blocks); len(native) > 0 { + // 能直视：另起一条 user message 承载媒体，并补一句来源说明。 + mediaBlocks := append([]agentAPI.ContentBlock{{ + Type: "text", + Text: fmt.Sprintf("[以下是 %s 注入的媒体内容]", tc.Name), + }}, native...) + mediaMsg = &agentAPI.Message{Role: "user", Blocks: mediaBlocks} + } else if fallbackText != "" { + toolMsg.Content = result + "\n\n" + fallbackText + result = toolMsg.Content + if len(f.ToolResults) > 0 { + f.ToolResults[len(f.ToolResults)-1].Output = result + } + } + } + } + f.Msgs = append(f.Msgs, toolMsg) + if mediaMsg != nil { + // 必须紧跟在 toolMsg 之后：中间插入其他消息会让 tool_call_id 配对断开。 + f.Msgs = append(f.Msgs, *mediaMsg) + } + + a.publishEvent(events.EventToolCall, map[string]interface{}{ + "tool": tc.Name, + "plugin": pluginName, + "args": tc.Arguments, + "result": result, + "status": "ok", + "channel": a.currentOutputChannel, + }) + + f.ToolIdx++ + if len(a.interceptCh) > 0 { + for _, interrupt := range a.drainInterrupts() { + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt}) + } + f.Step = StepTurnEnd + return outcomeContinue + } + f.Step = StepToolBegin + return outcomeContinue +} + +// stepTurnEnd 收尾本批并进入下一轮。 +func (a *Agent) stepTurnEnd(f *TaskFrame) stepOutcome { + // 供下一轮顶部选择补位文案。 + f.LastBatchReplyOnly = f.ReplyOnly + f.Turn++ + f.Step = StepLLM + return outcomeContinue +} + +// resolveProviders 按请求模型解析候选 provider（保持原语义）。 +func (a *Agent) resolveProviders(req *agentAPI.CompletionRequest) []agentAPI.Provider { + var providers []agentAPI.Provider + if a.providerManager != nil { + var allProviders []agentAPI.Provider + if req.Model != "" && !strings.EqualFold(req.Model, "AUTO") { + allProviders = a.providerManager.ResolveForModel(req.Model) + } else { + allProviders = a.providerManager.OrderedProviders() + } + providers = make([]agentAPI.Provider, 0, len(allProviders)) + for _, p := range allProviders { + if a.providerManager.IsAvailable(p.Name()) { + providers = append(providers, p) + } + } + } + if len(providers) == 0 { + providers = []agentAPI.Provider{a.provider} + } + return providers +} + +// callLLMWithFallback 在候选 provider 间回退，并把同源瞬时错误重试一次。 +// 逐行等价于原 process() 内的双层循环。 +func (a *Agent) callLLMWithFallback(req *agentAPI.CompletionRequest, providers []agentAPI.Provider) (*agentAPI.CompletionResponse, error) { + var resp *agentAPI.CompletionResponse + var llmErr error + + for pi, fbProvider := range providers { + if pi > 0 { + log.Printf("[agent] LLM fallback: trying provider %q (fallback #%d/%d)", + fbProvider.Name(), pi, len(providers)-1) + } + + // 同源瞬时错误重试：网关瞬断（502/503/504/429/网络抖动）通常秒级恢复， + // 直接跳下一个 provider（或直接报错）会丢掉本可成功的请求。 + // 凭证错误（401/403）与用户中断不重试。 + const maxAttempts = 2 + for attempt := 1; attempt <= maxAttempts; attempt++ { + if attempt > 1 { + log.Printf("[agent] provider %q transient failure, retry %d/%d in 2s: %v", + fbProvider.Name(), attempt, maxAttempts, llmErr) + select { + case <-time.After(2 * time.Second): + case <-a.ctx.Done(): + llmErr = a.ctx.Err() + } + if llmErr == nil || errors.Is(llmErr, context.Canceled) || errors.Is(llmErr, context.DeadlineExceeded) { + break + } + } + + fCtx, fCancel := context.WithCancel(a.ctx) + a.llmMu.Lock() + a.cancelLLM = fCancel + a.llmMu.Unlock() + + resp, llmErr = chatStreamWithFallback(fCtx, fbProvider, req, a) + + a.llmMu.Lock() + a.cancelLLM = nil + a.llmMu.Unlock() + fCancel() + + if llmErr == nil { + a.providerManager.ResetAvailability(fbProvider.Name()) + if fbProvider != a.provider { + a.provider = fbProvider + log.Printf("[agent] switched active provider to %q after fallback", fbProvider.Name()) + } + break + } + + // 用户中断：立即终止，不重试也不换 provider + if errors.Is(llmErr, context.Canceled) { + break + } + // 凭证错误：重试无意义，跳出重试循环进入 provider 标记/切换 + var pe *agentAPI.ProviderError + if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) { + break + } + // 其余错误（含 5xx/429/网络）：还有重试机会则继续，否则跳出 + } + + if llmErr == nil { + break + } + if errors.Is(llmErr, context.Canceled) { + break + } + var pe *agentAPI.ProviderError + if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) { + a.providerManager.ReportStatus(fbProvider.Name(), pe.StatusCode) + log.Printf("[agent] provider %q marked unavailable (HTTP %d)", fbProvider.Name(), pe.StatusCode) + } else { + a.providerManager.MarkUnavailable(fbProvider.Name()) + } + log.Printf("[agent] provider %q failed: %v", fbProvider.Name(), llmErr) + } + + return resp, llmErr +} diff --git a/internal/agent/core/task_test.go b/internal/agent/core/task_test.go new file mode 100644 index 0000000..b36870b --- /dev/null +++ b/internal/agent/core/task_test.go @@ -0,0 +1,209 @@ +package core + +// M1 验收测试：状态机与 TaskFrame 的**行为等价性**。 +// +// 设计依据 docs/zh/input-scheduler-design.md §11（R3 / X3 的 M1 形态）： +// M1 不引入抢占，因此 R3 退化为「经状态机跑出的结果与脚本预期一致」； +// X3 在 M1 退化为「驱动循环必然以一次终态返回结束（不空转、不超步数）」。 +// +// 抢占/挂起/恢复/优先级在 M3 起才有测试。 + +import ( + "context" + "errors" + "testing" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// scriptProvider 按脚本依次返回 CompletionResponse。 +// +// ChatStream 故意返回错误：驱动 chatStreamWithFallback 走非流式回退， +// 这样脚本就是「第 N 次调用返回第 N 个响应」，不依赖流式分片语义。 +type scriptProvider struct { + script []*agentAPI.CompletionResponse + idx int + reqs []*agentAPI.CompletionRequest +} + +func (s *scriptProvider) Name() string { return "script" } +func (s *scriptProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) { + s.reqs = append(s.reqs, req) + if s.idx >= len(s.script) { + return &agentAPI.CompletionResponse{Content: ""}, nil + } + r := s.script[s.idx] + s.idx++ + return r, nil +} +func (s *scriptProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) { + return nil, errors.New("script provider: streaming disabled") +} +func (s *scriptProvider) MaxContextTokens() int { return 8192 } + +func newTaskTestAgent(t *testing.T, sp agentAPI.Provider, sh *StageHost) *Agent { + t.Helper() + return New(AgentConfig{ + ID: "task-test", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + StageHost: sh, + IO: agentIO.NewIOManager(), + }) +} + +func tc(id, name string) agentAPI.ToolCall { + return agentAPI.ToolCall{ID: id, Name: name, Arguments: map[string]interface{}{"q": id}} +} + +// R3（M1 形态）：一次工具轮 + 一次收尾轮，结果与工具调用计数必须正确。 +func TestTaskFrame_R3_ToolRoundTrip(t *testing.T) { + sh := NewStageHost() + var got []string + sh.RegisterTool("t_echo", sdk.ToolDef{Name: "t_echo", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) { + got = append(got, args["q"].(string)) + return "OUT", nil + }) + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{ + {Content: "让我调用工具", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_echo")}}, + {Content: "最终答复"}, + }} + a := newTaskTestAgent(t, sp, sh) + + stageCtx := a.stageCtxFromInput("你好", "", "") + resp, toolsUsed, toolResults, err := a.process("你好", stageCtx) + if err != nil { + t.Fatalf("process 返回错误: %v", err) + } + if resp != "最终答复" { + t.Fatalf("响应=%q，期望 %q", resp, "最终答复") + } + if len(toolsUsed) != 1 || toolsUsed[0] != "t_echo" { + t.Fatalf("toolsUsed=%v，期望恰好一次 t_echo", toolsUsed) + } + if len(toolResults) != 1 || toolResults[0].Name != "t_echo" || toolResults[0].Output != "OUT" { + t.Fatalf("toolResults=%+v，期望一条 t_echo/OUT", toolResults) + } + if len(got) != 1 || got[0] != "c1" { + t.Fatalf("工具实参=%v，期望恰好执行一次且参数来自脚本", got) + } + if len(sp.reqs) != 2 { + t.Fatalf("LLM 调用次数=%d，期望 2（工具轮 + 收尾轮）", len(sp.reqs)) + } + + // 第二轮请求必须携带 assistant(tool_call) + tool 结果两条消息。 + msgs := sp.reqs[1].Messages + var hasAssistantCall, hasToolResult bool + for _, m := range msgs { + if m.Role == "assistant" && len(m.ToolCalls) == 1 && m.ToolCalls[0].ID == "c1" { + hasAssistantCall = true + } + if m.Role == "tool" && m.ToolCallID == "c1" && m.Content == "OUT" { + hasToolResult = true + } + } + if !hasAssistantCall || !hasToolResult { + t.Fatalf("第二轮请求缺少工具调用配对：assistant=%v tool=%v", hasAssistantCall, hasToolResult) + } +} + +// X3（M1 形态）：多轮脚本必须在有限步内以一次终态返回结束。 +func TestTaskFrame_X3_TerminatesWithinBudget(t *testing.T) { + sh := NewStageHost() + sh.RegisterTool("t_noop", sdk.ToolDef{Name: "t_noop", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) { + return "ok", nil + }) + + // 3 个工具轮 + 收尾轮：状态机会在 StepLLM/StepToolBegin/.../StepTurnEnd 间往返 4 次。 + var script []*agentAPI.CompletionResponse + for i := 0; i < 3; i++ { + script = append(script, &agentAPI.CompletionResponse{ + Content: "round", + ToolCalls: []agentAPI.ToolCall{tc("c"+string(rune('a'+i)), "t_noop")}, + }) + } + script = append(script, &agentAPI.CompletionResponse{Content: "done"}) + + sp := &scriptProvider{script: script} + a := newTaskTestAgent(t, sp, sh) + + resp, toolsUsed, toolResults, err := a.process("跑三轮", a.stageCtxFromInput("跑三轮", "", "")) + if err != nil { + t.Fatalf("process 返回错误: %v", err) + } + if resp != "done" { + t.Fatalf("响应=%q，期望 done", resp) + } + if len(toolsUsed) != 3 || len(toolResults) != 3 { + t.Fatalf("toolsUsed=%d toolResults=%d，期望各 3", len(toolsUsed), len(toolResults)) + } + // 步数护栏未触发（触发了会是 "step budget exhausted" 错误）。 + if len(sp.reqs) != 4 { + t.Fatalf("LLM 调用次数=%d，期望 4", len(sp.reqs)) + } +} + +// 中断（interceptCh）在工具批中途到达时：本批**剩余工具被放弃**，直接进入下一轮。 +// +// 这是原实现的 `break` 语义（process.go 旧版工具循环尾部），必须保持。 +func TestTaskFrame_InterruptAbandonsRemainingBatch(t *testing.T) { + sh := NewStageHost() + var a *Agent + sh.RegisterTool("t_first", sdk.ToolDef{Name: "t_first", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) { + // 工具执行期间产生一次中断（模拟插件在工具里注入打断）。 + a.interceptCh <- &agentIO.InputEvent{Source: "t", Payload: map[string]interface{}{"content": "新的用户输入"}} + return "first-out", nil + }) + sh.RegisterTool("t_second", sdk.ToolDef{Name: "t_second", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) { + t.Fatal("批内第二个工具不应被执行：中断必须放弃剩余批次") + return nil, nil + }) + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{ + {Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_first"), tc("c2", "t_second")}}, + {Content: "处理完中断后的答复"}, + }} + a = newTaskTestAgent(t, sp, sh) + + resp, toolsUsed, toolResults, err := a.process("开始", a.stageCtxFromInput("开始", "", "")) + if err != nil { + t.Fatalf("process 返回错误: %v", err) + } + if resp != "处理完中断后的答复" { + t.Fatalf("响应=%q", resp) + } + if len(toolsUsed) != 1 || toolsUsed[0] != "t_first" { + t.Fatalf("toolsUsed=%v，期望只有 t_first", toolsUsed) + } + if len(toolResults) != 1 || toolResults[0].Output != "first-out" { + t.Fatalf("toolResults=%+v，期望只有 t_first 的结果", toolResults) + } + + // 第二轮请求里必须出现被打断内容（system 角色、[中断消息] 前缀）。 + found := false + for _, m := range sp.reqs[1].Messages { + if m.Role == "system" && len(m.Content) > 0 && m.Content[0:len("[中断消息]")] == "[中断消息]" { + found = true + } + } + if !found { + t.Fatalf("第二轮请求缺少 [中断消息] system 消息") + } +} + +// 状态机对未知 step 必须失败退出而不是空转。 +func TestTaskFrame_UnknownStepFails(t *testing.T) { + sp := &scriptProvider{} + a := newTaskTestAgent(t, sp, NewStageHost()) + f := a.newTaskFrame("x", a.stageCtxFromInput("x", "", "")) + f.Step = Step(999) + if out := a.step(f); out != outcomeFailed { + t.Fatalf("未知 step 应返回 outcomeFailed，实际 %v", out) + } + if f.Err == nil { + t.Fatal("未知 step 必须带错误信息") + } +}