diff --git a/docs/zh/input-scheduler-design.md b/docs/zh/input-scheduler-design.md index cc1edbd..d9116e9 100644 --- a/docs/zh/input-scheduler-design.md +++ b/docs/zh/input-scheduler-design.md @@ -203,14 +203,35 @@ CriticalSection:step 标记 nonPreemptible = true,或任务进入声明区 | `_consolidation_` 中 | 降级为排队 | `_consolidation_` 是后台临界区 → B 入 `pendingInterrupts`,临界区结束后求值 | | `a.interceptCh` 满 | 降级为排队 | 不存在该队列;`pendingInterrupts` 有界丢弃 | -### 5.3 中断任务与被打断任务的关系(**待确认 D1**) +### 5.3 中断任务与被打断任务的关系(**已定:D1 = 方案 B**) -两选一: +> 用户明确: +> *“中断打断时,上个任务到达以来的所有上下文现场被保护(含 toolcall), +> 然后中断在**上个任务前的那个完整状态**上开始运行。中断运行结束,再把被挂起的 +> 任务与其上下文现场**加载回中断任务之上**,并继续运行。”* -- **方案 A(推荐)**:中断任务以**运行任务帧的只读前缀**为上下文(模型能看到"当前进行到哪一步"),但其产出**不合并回**被打断任务;被打断任务恢复后,对中断内容一无所知(除了可选的一行机器生成的提示)。 -- **方案 B**:中断任务只看到**任务开始前的持久状态**(提交点在 `S_COMMIT`,所以 Context 层天然停在 pre-task 状态)。代价:模型失去"进行中"的上下文,"别搜了改成 X"这类指令难以处理。 +因此语义是: -> D1 默认取 **方案 A**:可用性更好,且不违反"帧由调度器独占"。 +1. **被挂起任务的现场 = 它自到达以来累积的全部上下文(含 toolcall 结果)**, + 原样保存在 `TaskFrame` 里。 +2. **中断任务从「上一个任务之前的完整状态」开始运行**——它**看不到**被打断 + 任务的任何部分进展。等价于:中断任务就是一个普通新任务,正常走 `S_PREPARE` + (重建 system prompt + timeline + 自己的输入)。 +3. **中断结束后,把被挂起任务与其现场加载回「中断任务之上」再继续**: + 中断已提交的那段上下文留在**下面**(成为重建前缀的一部分),本任务自己的 + 现场接回**其上**。 + +实现对应(`internal/agent/core/task.go`): + +- `TaskFrame.PrefixLen` 记录 prepare 段构建的**基础前缀**长度 + (system + timeline + 用户输入);其后的 Stage 上下文与工具轮产物都是“自己的现场”。 +- `rebaseFramePrefix(f)`:恢复时重建基础前缀(因中断结束已把它的输入/输出提交进 + `a.context`,重建出的 timeline 已含中断效果),再把 `f.Msgs[PrefixLen:]` 原样接回; + 并补回 prepare 段的両处尾部改写(`IsInterrupt` 的 `[中断消息]` 标记、输入多模态块)。 +- 调用点:`resumeTask` 在 `runTaskSteps` **之前**调用它。 + +> 代价(已知且接受):中断看不到“进行到哪一步”,所以“别搜了改成 X”这类指令 +> 只能靠它自己重新理解;换来的是中断起点总是一个**一致的完整状态**。 --- @@ -232,7 +253,15 @@ running.state = done_for_now ### 6.2 恢复 -从 `suspendPool` 取出后,从 `frame.step` 继续执行;`frame.msgs` 原样复用。 +从 `suspendPool` 取出后: + +1. **重建基础前缀**(`rebaseFramePrefix`)—— 此时中断任务已结束并提交, + 重建出的 timeline 包含中断的输入/输出,即“现场加载回中断任务之上”; +2. 把本任务自己的尾部(Stage 上下文 + 工具轮产物 + 占位)原样接回; +3. 从 `frame.Step` 继续执行。 + +被丢弃的只有那次**不完整的 LLM 请求**(幂等),已执行的工具与已累积的 +`toolResults` 全部保留。 ### 6.3 嵌套 @@ -404,7 +433,7 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st | 编号 | 问题 | 默认取值 | |---|---|---| -| **D1** | 中断任务的上下文:方案 A(可见运行任务的只读前缀)还是方案 B(只见 pre-task 持久状态) | **A** | +| **D1** | 中断任务的上下文 | **方案 B(已定)**:中断从上一个任务之前的完整状态开始;恢复时把被挂起任务的现场加载回中断之上 | | **D2** | 阻塞 step 处置:v1 全部声明为临界区(调度器可被阻塞)还是引入异步 step | **v1 = 临界区**;异步 step 留到 v2 | | **D3** | `pendingInterrupts` 与 `readyQueue` 是否合一 | **保持分离**(中断请求带 interrupt 语义,取出时以中断语义启动);但**共用同一个排序键** | | **D4** | readyQueue 满时:阻塞发送方 or 返回错误 | **阻塞发送方 + 计数**(与现状一致,避免丢用户输入) | diff --git a/internal/agent/core/eventloop.go b/internal/agent/core/eventloop.go index ccfefba..fd1b5f3 100644 --- a/internal/agent/core/eventloop.go +++ b/internal/agent/core/eventloop.go @@ -105,13 +105,13 @@ func selfEvent(msg selfInputMsg) *agentIO.InputEvent { } func (a *Agent) handleSelfInput(msg selfInputMsg) { - _, _ = a.runInputTask(selfEvent(msg), nil) + _, _ = a.runInputTask(selfEvent(msg)) } func (a *Agent) handleInput(evt *agentIO.InputEvent) { switch evt.Type { case "text", "image", "audio": - _, _ = a.runInputTask(evt, nil) + _, _ = a.runInputTask(evt) case "event": log.Printf("[agent] event from %s: %v", evt.Source, evt.Payload) diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index 0e81c0a..f02d11a 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -24,7 +24,6 @@ import ( "sync/atomic" "time" - agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" "gitcode.com/JianFeeeee/HomeAgent/internal/events" sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" @@ -107,10 +106,6 @@ type Task struct { Event *agentIO.InputEvent // Kind == TaskKindInput Self selfInputMsg // Kind == TaskKindSelf - // SeedMsgs 是抢占式中断任务的只读前缀(D1=A):由被打断的任务在挂起时 - // 附上,使中断任务看得见「进行到哪一步」,但其产出不合并回原任务。 - SeedMsgs []agentAPI.Message - // PreemptCount 是本任务被抢占的次数,用于饥饿防护: // effectiveLevel = min(L4, Level + min(PreemptCount, 2))。 PreemptCount int @@ -425,23 +420,8 @@ func (s *scheduler) suspend(t *Task, f *TaskFrame) { s.running = nil } - // D1=A:把被抢占任务的只读前缀交给造成本次抢占的中断任务。 - // 选最高优先级的待处理中断;若它已有前缀(嵌套抢占)则不覆盖。 - if s.preemptArmed { - var victim *Task - for _, it := range s.pendingInterrupts { - if it.Level < s.preemptLevel { - continue - } - if victim == nil || taskBefore(victim, it) { - victim = it - } - } - if victim != nil && len(victim.SeedMsgs) == 0 { - victim.SeedMsgs = append([]agentAPI.Message(nil), f.Msgs...) - } - } - + // D1=B:中断任务在上一个任务之前的完整状态上开始运行, + // 因此这里**不**把被打断任务的任何内容交给它。 s.preemptArmed = false s.preemptLevel = 0 } @@ -645,9 +625,9 @@ func (a *Agent) executeNewTask(t *Task) { }() switch t.Kind { case TaskKindInput: - f, out = a.runInputTask(t.Event, t.SeedMsgs) + f, out = a.runInputTask(t.Event) case TaskKindSelf: - f, out = a.runInputTask(selfEvent(t.Self), nil) + f, out = a.runInputTask(selfEvent(t.Self)) } }() @@ -664,6 +644,10 @@ func (a *Agent) executeNewTask(t *Task) { // resumeTask 从保存的现场继续一个被抢占的任务。 // // 关键:不重建帧、不重跑 prepare 段——否则会重复提交上下文与事件。 +// resumeTask 从保存的现场继续一个被抢占的任务。 +// +// 关键:不重跑 prepare 段(否则会重复提交上下文与事件),而是先把基础前缀 +// 重建到「中断任务之上」,再把本任务自己的现场接回去(见 rebaseFramePrefix)。 func (a *Agent) resumeTask(t *Task, f *TaskFrame) { a.sched.mu.Lock() a.sched.stats.Resumed++ @@ -671,6 +655,7 @@ func (a *Agent) resumeTask(t *Task, f *TaskFrame) { a.publishEvent(events.EventScheduler, map[string]interface{}{ "action": "resume", "task": t.ID, "level": int(t.Level), }) + a.rebaseFramePrefix(f) defer func() { if r := recover(); r != nil { log.Printf("[agent] resume task#%d panic recovered: %v\n%s", diff --git a/internal/agent/core/scheduler_preempt_test.go b/internal/agent/core/scheduler_preempt_test.go index 1f4e9ef..ed87614 100644 --- a/internal/agent/core/scheduler_preempt_test.go +++ b/internal/agent/core/scheduler_preempt_test.go @@ -9,6 +9,7 @@ package core import ( "context" + "strings" "sync" "testing" "time" @@ -148,16 +149,21 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { if k != nextPending || it.Level != LevelCritical { t.Fatalf("应取到 pending 中断,kind=%v level=%v", k, it.Level) } - // D1=A:抢占式中断任务继承被打断任务的只读前缀。 - if len(it.SeedMsgs) == 0 { - t.Fatal("抢占式中断任务必须继承只读前缀(D1=A)") + // D1=B:中断任务在**上一个任务之前的完整状态**上开始运行,不继承本任务的现场。 + // 因此它能看到的唯一输入就是它自己携带的内容。 + if it.Event == nil { + t.Fatal("中断任务必须携带自己的输入事件") + } + if got, _ := it.Event.Payload["content"].(string); got != "紧急打断" { + t.Fatalf("中断任务输入=%q,期望 紧急打断", got) } a.executeNewTask(it) if len(a.DumpScheduler().PendingInterrupts) != 0 { t.Fatal("中断任务执行后 pendingInterrupts 应清空") } - // R1:恢复被抢占任务;msgs 与被抢占前逐字节一致(长度不变),并从 S_LLM 重发。 + // R1:恢复被抢占任务。基础前缀被重建到「中断任务之上」(含中断已提交的上下文), + // 本任务自己的现场接回其后,然后从 S_LLM 重发。 rt, rf, k2 := a.sched.nextRef() if k2 != nextSuspended || rt != lowTask { t.Fatalf("应恢复被抢占任务,kind=%v", k2) @@ -165,11 +171,17 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { if rf.Step != StepLLM { t.Fatalf("恢复游标=%v,期望 StepLLM", rf.Step) } - if len(rf.Msgs) != msgsBefore { - t.Fatalf("恢复后 msgs 长度=%d,期望 %d(不得被中断污染)", len(rf.Msgs), msgsBefore) - } a.resumeTask(rt, rf) + // 「加载回中断之上」的判据:恢复后的消息序列里必须出现中断任务的上下文。 + if !msgsContain(rf.Msgs, "紧急打断") { + t.Fatal("恢复后的任务应看见中断任务的上下文(现场未加载回中断之上)") + } + // 本任务自己的现场(工具轮产物)仍然在。 + if len(rf.Msgs) < msgsBefore { + t.Fatalf("恢复后消息数=%d,不应少于被抢占前的 %d", len(rf.Msgs), msgsBefore) + } + if sp.callCount() != 3 { t.Fatalf("LLM 总调用=%d,期望 3(丢弃 1 + 中断 1 + 恢复 1)", sp.callCount()) } @@ -273,25 +285,48 @@ func TestPreempt_IdleInterruptIsQueuedNotLost(t *testing.T) { } } -// 带媒体/中断标记的输入走新任务路径时不得污染下一个任务的尾部消息。 -func TestPreempt_SeedPathDoesNotLeakInterruptFlag(t *testing.T) { +// msgsContain 报告消息序列里是否出现过某段文本(用于“现场是否合回”的断言)。 +func msgsContain(msgs []agentAPI.Message, sub string) bool { + for _, m := range msgs { + if strings.Contains(m.Content, sub) { + return true + } + } + return false +} + +// 恢复时的重建必须把 prepare 段对尾部消息的两处改写补回: +// 中断标记(IsInterrupt)与多模态块(InputBlocks)。 +func TestPreempt_ResumeRebaseRestoresTailDecorations(t *testing.T) { sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "ok"}}} a := newPreemptAgent(t, sp) + block := agentAPI.ContentBlock{Type: "text", Text: "图"} f := a.newTaskFrame("打断文本", a.stageCtxFromInput("打断文本", "", "")) - f.SeedMsgs = []agentAPI.Message{{Role: "system", Content: "S"}, {Role: "user", Content: "U"}} - a.interruptInput = true - if out := a.stepPrepare(f); out != outcomeContinue { - t.Fatalf("seed 路径应继续,实际 %v", out) + f.IsInterrupt = true + f.InputBlocks = []agentAPI.ContentBlock{block} + // 模拟 prepare 后的形状:基础前缀 + 一段“本任务自己的现场” + f.Msgs = []agentAPI.Message{ + {Role: "system", Content: "S"}, + {Role: "user", Content: "[中断消息] 打断文本"}, + {Role: "assistant", Content: "进行中"}, } - if a.interruptInput { - t.Fatal("seed 路径必须消费 interruptInput,否则下一个任务尾部会被误改") + f.PrefixLen = 2 + + a.rebaseFramePrefix(f) + + if f.PrefixLen <= 0 || f.PrefixLen >= len(f.Msgs) { + t.Fatalf("重建后 PrefixLen=%d,消息数=%d,前缀应短于总数", f.PrefixLen, len(f.Msgs)) } - last := f.Msgs[len(f.Msgs)-1] - if last.Role != "user" || last.Content != "打断文本" { - t.Fatalf("seed 路径尾部应为中断输入本身,实际 %+v", last) + // 尾部现场(assistant 进行中)必须还在最后。 + if last := f.Msgs[len(f.Msgs)-1]; last.Content != "进行中" { + t.Fatalf("本任务现场应接回最后,实际 %+v", last) } - if f.Step != StepLLM { - t.Fatalf("seed 路径应直接进入 StepLLM,实际 %v", f.Step) + prefixLast := f.Msgs[f.PrefixLen-1] + if prefixLast.Role != "system" || !strings.HasPrefix(prefixLast.Content, "[中断消息]") { + t.Fatalf("中断标记未补回:%+v", prefixLast) + } + if len(prefixLast.Blocks) != 1 || prefixLast.Blocks[0].Text != "图" { + t.Fatalf("多模态块未补回:%+v", prefixLast.Blocks) } } diff --git a/internal/agent/core/scheduler_starvation_test.go b/internal/agent/core/scheduler_starvation_test.go index bc1f563..e414454 100644 --- a/internal/agent/core/scheduler_starvation_test.go +++ b/internal/agent/core/scheduler_starvation_test.go @@ -111,7 +111,7 @@ func TestStarvation_SelectionUsesEffectiveLevel(t *testing.T) { // 提升不改变调度器自身的排序稳定性:同为有效级时按入队时刻。 a := &Task{ID: 3, Level: LevelBackground, PreemptCount: 1, EnqueuedAt: base.Add(2 * time.Second)} // 有效 L2 - b := &Task{ID: 4, Level: LevelMessage, EnqueuedAt: base.Add(time.Second)} // L2,更早 + b := &Task{ID: 4, Level: LevelMessage, EnqueuedAt: base.Add(time.Second)} // L2,更早 if !taskBefore(b, a) { t.Fatal("同有效级时应先到先服务") } diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index 1621968..f41493b 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -128,9 +128,11 @@ type TaskFrame struct { Terminal taskTerminal Level Level - // SeedMsgs 非空时,stepPrepare 不重建 system prompt / 记忆上下文, - // 而是以它为前缀继续(D1=A:抢占式中断任务继承被打断任务的**只读前缀**)。 - SeedMsgs []agentAPI.Message + // PrefixLen 是 stepPrepare 构建的**基础前缀**长度(system + timeline + 用户输入)。 + // 恢复时用它把「本任务自己的现场」接回重建后的前缀之上(见 rebaseFramePrefix)。 + PrefixLen int + // InputBlocks 是本轮输入携带的多模态块;重建前缀时要重新挂回。 + InputBlocks []agentAPI.ContentBlock } func (a *Agent) newTaskFrame(input string, stageCtx *sdk.StageContext) *TaskFrame { @@ -202,7 +204,7 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri // 而 finish 段(上下文提交与回执)只在任务真正结束时执行一次。 // // M3a 还没有抢占,因此 outcomeSuspended 只会由 M3b 的抢占检查产生。 -func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) (*TaskFrame, stepOutcome) { +func (a *Agent) runInputTask(evt *agentIO.InputEvent) (*TaskFrame, stepOutcome) { // 临界区标记由调度器 goroutine 维护,任务结束(含挂起)即清。 // interceptLoop 读它来决定“能不能取消”,因此必须是原子的。 defer a.sched.setCritical(false) @@ -212,7 +214,6 @@ func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) ( case terminalSkipped, terminalStageShortCircuit, terminalConsolidation: return nil, outcomeDone } - f.SeedMsgs = seed out := a.runTaskSteps(f) if out == outcomeSuspended { @@ -223,6 +224,43 @@ func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) ( return f, out } +// rebaseFramePrefix 把被挂起任务的上下文现场「加载回中断任务之上」。 +// +// 语义(用户明确): +// - 中断打断时,被挂起任务自到达以来累积的全部现场(含 toolcall)被保护; +// - 中断在上一个任务之前的**完整状态**上开始运行(所以中断看不到本任务的部分进展); +// - 中断结束后,把被挂起任务与其现场加载回中断任务**之上**再继续—— +// 即中断已提交的那段上下文留在下面(前缀),本任务自己的现场落回其上。 +// +// 实现:重建基础前缀(system + timeline + 用户输入);由于中断结束时已把它的 +// 输入/输出提交进 a.context,重建出的 timeline 已含中断的效果;再把本任务 +// 自己的尾部(Stage 上下文 + 工具轮产物 + 占位)原样接回。 +func (a *Agent) rebaseFramePrefix(f *TaskFrame) { + if f == nil || f.PrefixLen <= 0 || f.PrefixLen > len(f.Msgs) { + return + } + tail := append([]agentAPI.Message(nil), f.Msgs[f.PrefixLen:]...) + + budget := ComputeTokenBudget(a.provider, a.systemPrompt) + memContext := a.buildMemoryContext(f.Input, budget.MemoryTokens) + sysPrompt := a.buildSystemPrompt(memContext, f.Input) + prefix := a.buildMessages(sysPrompt, f.Input, budget.ContextTokens) + + // 重建会丢掉 prepare 段对尾部消息的两处改写,这里等价地补回。 + if f.IsInterrupt && len(prefix) > 0 { + last := prefix[len(prefix)-1] + last.Role = "system" + last.Content = "[中断消息] " + last.Content + prefix[len(prefix)-1] = last + } + if len(f.InputBlocks) > 0 && len(prefix) > 0 { + prefix[len(prefix)-1].Blocks = f.InputBlocks + } + + f.Msgs = append(prefix, tail...) + f.PrefixLen = len(prefix) +} + // prepareInputTask 执行 processInput 的前半段(去重、通道解析、阶段、裁剪、 // 输入事件落上下文)。返回终态不为 terminalNone 时调用方不得进入 run 段。 func (a *Agent) prepareInputTask(evt *agentIO.InputEvent) (*TaskFrame, taskTerminal) { @@ -416,18 +454,6 @@ func (a *Agent) step(f *TaskFrame) stepOutcome { // stepPrepare 构建本轮任务的初始帧。 func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome { - // 抢占式中断任务:继承被打断任务的只读前缀(D1=A),不重建上下文。 - if len(f.SeedMsgs) > 0 { - f.Tools = a.buildToolDefs() - f.Msgs = append([]agentAPI.Message(nil), f.SeedMsgs...) - f.Msgs = append(f.Msgs, agentAPI.Message{Role: "user", Content: f.Input}) - // 前缀路径不重写尾部消息(那是「同行注入」的旧形态); - // 必须消费掉标志位,否则下一个普通任务的尾部会被误改。 - a.interruptInput = false - f.Step = StepLLM - return outcomeContinue - } - budget := ComputeTokenBudget(a.provider, a.systemPrompt) memContext := a.buildMemoryContext(f.Input, budget.MemoryTokens) @@ -447,7 +473,11 @@ func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome { if len(f.Msgs) > 0 { f.Msgs[len(f.Msgs)-1].Blocks = blocks } + f.InputBlocks = blocks } + // 基础前缀到此为止(system + timeline + 用户输入);其后的 Stage 上下文 + // 与工具轮产物都属于“本任务自己的现场”,恢复时要接回重建后的前缀之上。 + f.PrefixLen = len(f.Msgs) 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, diff --git a/internal/agent/core/task_lifecycle_test.go b/internal/agent/core/task_lifecycle_test.go index 743952b..a550914 100644 --- a/internal/agent/core/task_lifecycle_test.go +++ b/internal/agent/core/task_lifecycle_test.go @@ -51,7 +51,7 @@ func TestLifecycle_NormalCommitsOnceAndReplies(t *testing.T) { a := newLifecycleAgent(t, sp, bus, NewStageHost()) evt, respCh := textEvent("cli", "你好") - if _, out := a.runInputTask(evt, nil); out != outcomeDone { + if _, out := a.runInputTask(evt); out != outcomeDone { t.Fatalf("runInputTask=%v,期望 outcomeDone", out) } @@ -90,13 +90,13 @@ func TestLifecycle_DuplicateSkippedHasTerminal(t *testing.T) { a := newLifecycleAgent(t, sp, bus, NewStageHost()) e1, _ := textEvent("webui", "同样的消息") - if _, out := a.runInputTask(e1, nil); out != outcomeDone { + if _, out := a.runInputTask(e1); out != outcomeDone { t.Fatalf("首次输入=%v,期望 outcomeDone", out) } after1, outputs1 := a.context.Len(), outputs e2, ch2 := textEvent("webui", "同样的消息") - if _, out := a.runInputTask(e2, nil); out != outcomeDone { + if _, out := a.runInputTask(e2); out != outcomeDone { t.Fatalf("去重输入应正常返回(不挂起),实际 %v", out) } if a.context.Len() != after1 { @@ -130,7 +130,7 @@ func TestLifecycle_OnInputShortCircuit(t *testing.T) { a := newLifecycleAgent(t, sp, events.NewBus(), sh) evt, respCh := textEvent("cli", "任意") - if _, out := a.runInputTask(evt, nil); out != outcomeDone { + if _, out := a.runInputTask(evt); out != outcomeDone { t.Fatalf("短路任务=%v,期望 outcomeDone", out) } select { @@ -159,7 +159,7 @@ func TestLifecycle_ErrorPathTerminal(t *testing.T) { a := newLifecycleAgent(t, sp, bus, NewStageHost()) evt, respCh := textEvent("cli", "会失败") - if _, out := a.runInputTask(evt, nil); out != outcomeFailed { + if _, out := a.runInputTask(evt); out != outcomeFailed { t.Fatalf("runInputTask=%v,期望 outcomeFailed", out) } select { @@ -196,7 +196,7 @@ func TestLifecycle_ConsolidationRouted(t *testing.T) { evt, respCh := textEvent("system", "整理任务") evt.OutputChannel = channelConsolidation - if _, out := a.runInputTask(evt, nil); out != outcomeDone { + if _, out := a.runInputTask(evt); out != outcomeDone { t.Fatalf("consolidation=%v,期望 outcomeDone", out) } if len(respCh) != 0 { diff --git a/internal/agent/core/task_terminal_test.go b/internal/agent/core/task_terminal_test.go index a74fa76..4de34f9 100644 --- a/internal/agent/core/task_terminal_test.go +++ b/internal/agent/core/task_terminal_test.go @@ -87,7 +87,7 @@ func TestTerminal_EmptyInputGetsSkippedReply(t *testing.T) { OutputChannel: "cli", ResponseCh: ch, } - if _, out := a.runInputTask(evt, nil); out != outcomeDone { + if _, out := a.runInputTask(evt); out != outcomeDone { t.Fatalf("空输入应正常返回,实际 %v", out) } if len(ch) != 1 { @@ -111,7 +111,7 @@ func TestTerminal_NoTimeoutSyncCallerDoesNotHang(t *testing.T) { // 第一次成功 e1, ch1 := textEvent("cli", "重复内容") - if _, out := a.runInputTask(e1, nil); out != outcomeDone { + if _, out := a.runInputTask(e1); out != outcomeDone { t.Fatalf("首次=%v", out) } if len(ch1) != 1 { @@ -120,7 +120,7 @@ func TestTerminal_NoTimeoutSyncCallerDoesNotHang(t *testing.T) { // 第二次(去重命中):模拟同步调用方阻塞等待——必须在 1s 内拿到终态。 e2, ch2 := textEvent("cli", "重复内容") - go a.runInputTask(e2, nil) + go a.runInputTask(e2) select { case r := <-ch2: if skipped, _ := r.Payload["skipped"].(bool); !skipped {