From 7565248f6188c31390d28a5053881ab16be8692b Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 13 Sep 2026 00:34:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20M4=20=E4=B8=B4=E7=95=8C?= =?UTF-8?q?=E5=8C=BA=E6=98=BE=E5=BC=8F=E5=8C=96=20+=20=E6=B8=85=E6=8E=89?= =?UTF-8?q?=E8=A2=AB=20pendingInterrupts=20=E5=8F=96=E4=BB=A3=E7=9A=84=20i?= =?UTF-8?q?nterceptCh?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 设计依据 docs/zh/input-scheduler-design.md §4.3、§11.1（P5/P6）。 - 临界区语义显式化：让位检查**只在 step 之间**做，执行中的 step （工具 RPC / ONNX / CAS 落盘）天然不可抢占；_consolidation_ 整任务 经 inCriticalSection() 判为不可抢占（它直接改图库） - 删除 interceptCh 与 drainInterrupts：M3b 起中断一律走 pendingInterrupts， 旧的「同行注入 + 三处 drain + 批次放弃」已无写入者，属死代码 - 新增 scheduler_critical_test.go 3 项： P5/P6 工具执行中 arm 了让位信号也不得挂起、必须等工具返回后的安全点； _consolidation_ 判为临界区；无抢占时同批工具必须全部执行（新语义回归） - 验收：agent 全量 + -race；全仓 build/vet 通过 --- internal/agent/core/agent.go | 4 - internal/agent/core/eventloop.go | 27 ---- .../agent/core/scheduler_critical_test.go | 151 ++++++++++++++++++ internal/agent/core/task.go | 30 ---- internal/agent/core/task_test.go | 48 ------ 5 files changed, 151 insertions(+), 109 deletions(-) create mode 100644 internal/agent/core/scheduler_critical_test.go diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index fca8d90..172ddc9 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -110,9 +110,6 @@ type Agent struct { // childSeq 给完成的任务排个序，用于有界淘汰。 childSeq int64 - // 高优先级打断通道：interceptLoop 注入，process() 在工具循环轮次间非阻塞读取 - interceptCh chan *agentIO.InputEvent - // 输入调度器：就绪队列、任务抽象与快照（见 scheduler.go）。 // M2 起取代 eventLoop 的隐式 channel 排队。 sched *scheduler @@ -312,7 +309,6 @@ func New(cfg AgentConfig) *Agent { eventBus: cfg.EventBus, selfInputCh: make(chan selfInputMsg, 64), childTasks: make(map[string]*childTaskState), - interceptCh: make(chan *agentIO.InputEvent, 64), sched: newScheduler(256), priorityLookup: cfg.PriorityLookup, pluginHealth: newPluginHealthTracker(), diff --git a/internal/agent/core/eventloop.go b/internal/agent/core/eventloop.go index 082c5a7..aa03fb2 100644 --- a/internal/agent/core/eventloop.go +++ b/internal/agent/core/eventloop.go @@ -321,33 +321,6 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) { a.runStage(sdk.StageAfterOutput, stageCtx) } -func (a *Agent) drainInterrupts() []string { - var out []string - for { - select { - case evt := <-a.interceptCh: - if evt == nil { - continue - } - text, _ := evt.Payload["content"].(string) - if text == "" { - continue - } - source := evt.Source - if source == "" { - source = "unknown" - } - channel := evt.OutputChannel - if channel == "" { - channel = source - } - out = append(out, fmt.Sprintf("[打断消息][来源:%s][输出通道:%s] %s", source, channel, text)) - default: - return out - } - } -} - // pruneOnInput 按声明的上下文策略裁剪上下文，返回归档的事件数。 // // 默认**不裁剪**：ContextPolicy 必须在注入点（payload 的 context_policy） diff --git a/internal/agent/core/scheduler_critical_test.go b/internal/agent/core/scheduler_critical_test.go new file mode 100644 index 0000000..bbdfb62 --- /dev/null +++ b/internal/agent/core/scheduler_critical_test.go @@ -0,0 +1,151 @@ +package core + +// M4 验收测试：临界区语义显式化 + 抢占延迟到安全点 + 批次不再被中断放弃。 +// +// 设计依据 docs/zh/input-scheduler-design.md §4.3（临界区）、§11.1（P5/P6）。 +// +// 关键结构事实：让位检查**只在 step 之间**进行，因此任何正在执行的 step +// （工具 RPC、ONNX、CAS 落盘）天然不可抢占——中断只能等它返回。 + +import ( + "sync" + "testing" + "time" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// P5/P6：工具执行期间到达的高优先级中断不得立即抢占；必须等工具返回后的安全点。 +func TestPreempt_DeferredDuringToolExec(t *testing.T) { + sh := NewStageHost() + entered := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + sh.RegisterTool("t_slow", sdk.ToolDef{Name: "t_slow", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) { + once.Do(func() { close(entered) }) + <-release + return "slow-done", nil + }) + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{ + {Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_slow")}}, // 低优先级任务调用慢工具 + {Content: "low-done"}, // 恢复后收尾 + {Content: "intr-done"}, // 中断任务 + }} + a := New(AgentConfig{ + ID: "crit", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + IO: agentIO.NewIOManager(), + StageHost: sh, + }) + + if _, _ = enqueueTask(t, a, LevelBackground, "qq", "低优先级任务"); true { + } + lt, _, _ := a.sched.nextRef() + done := make(chan struct{}) + go func() { a.executeNewTask(lt); close(done) }() + + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("慢工具未被调用") + } + + // 工具执行中注入 L4 中断。 + intrEvt, _ := textEvent("cli", "紧急打断") + intrEvt.Payload["interrupt"] = true + if !a.sched.requestPreempt(intrEvt, LevelCritical) { + t.Fatal("L4 应 arm 让位信号") + } + // 关键断言：信号已 arm，但任务仍在工具里 —— 绝不能挂起。 + if !a.sched.preemptGrantedFor(LevelBackground) { + t.Fatal("让位信号应已 arm") + } + if a.DumpScheduler().Running == nil { + t.Fatal("工具执行中不得挂起（StepToolExec 是临界区）") + } + if len(a.DumpScheduler().SuspendPool) != 0 { + t.Fatal("工具执行中 suspendPool 应为空") + } + + // 放行工具 → 工具返回后的安全点才挂起。 + close(release) + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("工具返回后未挂起") + } + + snap := a.DumpScheduler() + if len(snap.SuspendPool) != 1 { + t.Fatalf("工具返回后 suspendPool=%d，期望 1", len(snap.SuspendPool)) + } + if snap.SuspendPool[0].Frame.Step != StepToolAfter { + t.Fatalf("应在工具执行后的安全点挂起（StepToolAfter），实际 %v", snap.SuspendPool[0].Frame.Step) + } + if len(snap.PendingInterrupts) != 1 { + t.Fatalf("中断请求不得丢失，pendingInterrupts=%d", len(snap.PendingInterrupts)) + } +} + +// _consolidation_ 整任务视为不可抢占（它直接改图库）。 +func TestCriticalSection_ConsolidationMarked(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + + a.currentOutputChannel = "cli" + if a.inCriticalSection() { + t.Fatal("普通通道不应被判为临界区") + } + a.currentOutputChannel = channelConsolidation + if !a.inCriticalSection() { + t.Fatal("记忆整理必须是不可抢占临界区") + } +} + +// 新语义：没有抢占时，同批的多个工具必须全部执行——不再有「中断放弃剩余批」。 +func TestBatch_NotAbandonedWithoutPreemption(t *testing.T) { + sh := NewStageHost() + var mu sync.Mutex + var ran []string + reg := func(name string) { + sh.RegisterTool(name, sdk.ToolDef{Name: name, Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) { + mu.Lock() + ran = append(ran, name) + mu.Unlock() + return name + "-out", nil + }) + } + reg("t_a") + reg("t_b") + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{ + {Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_a"), tc("c2", "t_b")}}, + {Content: "全部完成"}, + }} + a := New(AgentConfig{ + ID: "batch", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + IO: agentIO.NewIOManager(), + StageHost: sh, + }) + + if _, _ = enqueueTask(t, a, LevelBackground, "cli", "跑两个工具"); true { + } + tt, _, _ := a.sched.nextRef() + a.executeNewTask(tt) + + if len(ran) != 2 || ran[0] != "t_a" || ran[1] != "t_b" { + t.Fatalf("同批工具应全部按序执行，实际 %v", ran) + } + snap := a.DumpScheduler() + if len(snap.SuspendPool) != 0 || len(snap.PendingInterrupts) != 0 { + t.Fatalf("无抢占时不应有挂起或待处理中断：%+v", snap) + } + if snap.Stats.Executed != 1 { + t.Fatalf("Executed=%d，期望 1", snap.Stats.Executed) + } +} diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index 4ae32a8..7019f57 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -455,13 +455,6 @@ func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome { // 取消（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) @@ -569,22 +562,6 @@ func (a *Agent) stepToolBegin(f *TaskFrame) stepOutcome { } 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) @@ -744,13 +721,6 @@ func (a *Agent) stepToolAfter(f *TaskFrame) stepOutcome { }) 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 } diff --git a/internal/agent/core/task_test.go b/internal/agent/core/task_test.go index 24c8711..6cdbc4e 100644 --- a/internal/agent/core/task_test.go +++ b/internal/agent/core/task_test.go @@ -152,54 +152,6 @@ func TestTaskFrame_X3_TerminatesWithinBudget(t *testing.T) { } } -// 中断（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{}