From 86a702b3b01063960ded4b9b7df45ee355935709 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 13 Sep 2026 06:30:23 +0800 Subject: [PATCH] =?UTF-8?q?fix(scheduler)!:=20=E4=B8=AD=E6=96=AD=E6=A0=88?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=EF=BC=88=E5=B5=8C=E5=A5=97=E6=8A=A2=E5=8D=A0?= =?UTF-8?q?=20LIFO=EF=BC=89=EF=BC=8C=E5=B9=B6=E4=BF=AE=E6=8E=89=E6=8A=A2?= =?UTF-8?q?=E5=8D=A0=E7=A9=BA=E8=BD=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户指正：存在**中断被中断**的场景，所以被打断的现场要进**中断栈**。 我此前把 suspendPool 明确写成“不是栈、按优先级取”，是错的。 改动： - suspendPool 改名 suspendStack，恢复纪律改为**严格 LIFO（只比栈顶）**； 栈内不做优先级重排——嵌套抢占天然使栈自底向上基础级递增， 且“后被打断的先恢复”才是栈语义。取出即弹栈。 - 修掉一个由此暴露的真 bug（抢占空转）：一次抢占生效后，被挂起的原任务 会因饥饿防护提升有效级，与抢占者同级；此时若按“先到先服务”，原任务 （入队更早）会被立刻选回，抢占者永远排不到 —— 抢占等于没发生。 现在**同级时 pendingInterrupts 优先于其它两类**，保证抢占必然生效。 - 状态 DTO：SuspendPool/suspend_pool → SuspendStack/suspend_stack - 设计稿：§2 用语更正（它**就是**中断栈）、§4.1 选择函数（候选只含栈顶 + pending 同级优先，并说明为何必需）、§6.2/§6.3/§9/§11 用例同步 测试新增 scheduler_stack_test.go 3 项： - 嵌套 L1→L2→L3，恢复严格 LIFO（B 先于 A） - 只比栈顶：人为构造“栈底 L3、栈顶 L2”，必须取栈顶（区分两种实现） - 嵌套下的深度上限 验收：agent 全量 + -race；全仓 build/vet 通过 --- docs/zh/input-scheduler-design.md | 45 ++-- internal/agent/core/scheduler.go | 68 +++++-- .../agent/core/scheduler_critical_test.go | 14 +- internal/agent/core/scheduler_e2e_test.go | 6 +- internal/agent/core/scheduler_preempt_test.go | 14 +- internal/agent/core/scheduler_stack_test.go | 192 ++++++++++++++++++ internal/agent/core/task.go | 2 +- internal/sdk/status.go | 6 +- 8 files changed, 287 insertions(+), 60 deletions(-) create mode 100644 internal/agent/core/scheduler_stack_test.go diff --git a/docs/zh/input-scheduler-design.md b/docs/zh/input-scheduler-design.md index d9116e9..c88d185 100644 --- a/docs/zh/input-scheduler-design.md +++ b/docs/zh/input-scheduler-design.md @@ -75,9 +75,11 @@ Step（枚举，顺序执行，步与步之间是安全点）： |---|---|---| | `readyQueue` | 排队输入形成的新任务 | 按 `(level desc, enqueueAt asc)` 排序 | | `pendingInterrupts` | **因优先级不足或临界区而未能抢占**的中断请求 | 同上排序；被取出时以中断语义启动 | -| `suspendPool` | 被打断、保存了现场的任务 | 同上排序；被取出时从 `frame.step` 继续 | +| `suspendStack`（**中断栈**） | 被打断、保存了现场的任务，用于“中断被中断”的嵌套 | **LIFO**：只有栈顶参与选择；被取出时从 `frame.step` 继续 | -> 注意用词：`suspendPool` **不是栈**。当恢复规则是"按优先级取"而非 LIFO 时，把它叫"中断栈"会误导实现（写死 pop 语义）。本设计统一称 `suspendPool`。 +> 用词（已更正）：它**就是中断栈**。用户明确存在「中断被中断」的场景，被打断的现场必须压栈； +> 因此恢复纪律是**严格 LIFO（只比栈顶）**，栈内不做优先级重排。 +> 早期稿把它写成“不是栈、按优先级取”是错的。 --- @@ -129,12 +131,18 @@ incoming.level < running.level → 入 pendingInterrupts 任务结束、或运行任务到达安全点且存在待处理抢占请求时，执行： ``` -pick = argmin over (readyQueue ∪ pendingInterrupts ∪ suspendPool) - by (-effectiveLevel(t), t.enqueueAt) +candidates = readyQueue ∪ pendingInterrupts ∪ { suspendStack.top } // 栈只出栈顶 +pick = argmin over candidates by (-effectiveLevel(t), kind, t.enqueueAt) ``` +- **中断栈只把栈顶**放进候选（严格 LIFO）——栈内更老的任务即使因饥饿防护提升了 + 有效级，也不得越过栈顶；“后被打断的先恢复”才是栈语义。 - 高有效级先； -- 同级**先到先服务**（`enqueueAt` 为首次入队时刻；挂起任务保留其原始入队时刻，因此倾向"先完成旧任务"，天然抑制饥饿）。 +- **同级时 `pendingInterrupts` 优先于其它两类**。 + 为何必需：抢占生效后，被挂起的原任务会因饥饿防护提升有效级，于是与抢占者同级； + 若同级按“先到先服务”，原任务（入队更早）会被立刻选回，抢占者永远排不到—— + 抢占变成空转。 +- 同级同类：先到先服务（`enqueueAt` 为首次入队时刻；挂起任务保留其原始入队时刻）。 ### 4.2 安全点（可切换点） @@ -168,7 +176,7 @@ CriticalSection：step 标记 nonPreemptible = true，或任务进入声明区 - `readyQueue` 有界（默认 256，可配）。 - 满时：**阻塞发送方**（与现状 `inputCh` 一致，避免静默丢用户输入），但必须**计数并打日志**。 - `pendingInterrupts` 有界（默认 64）；满时**丢弃最老的 pending 中断并计数**（中断是提示性输入，宁可丢旧保新）。 -- `suspendPool` 深度上限默认 4（见 §7.3）。 +- 中断栈深度上限默认 4（见 §6.3）。 --- @@ -198,7 +206,7 @@ CriticalSection：step 标记 nonPreemptible = true，或任务进入声明区 | 情形 | 现状 | 新模型 | |---|---|---| -| LLM 在跑，正常 | 真抢占（同轮 continue） | 真抢占：`S_LLM` 取消，任务 A 存 `suspendPool`，中断任务 B 从 `S_PREPARE` 启动 | +| LLM 在跑，正常 | 真抢占（同轮 continue） | 真抢占：`S_LLM` 取消，任务 A **压入中断栈**，中断任务 B 从 `S_PREPARE` 启动 | | LLM 没在跑 | 降级为排队 | B 入 `pendingInterrupts`（或直接成为 ready 任务），调度器立即选出 | | `_consolidation_` 中 | 降级为排队 | `_consolidation_` 是后台临界区 → B 入 `pendingInterrupts`，临界区结束后求值 | | `a.interceptCh` 满 | 降级为排队 | 不存在该队列；`pendingInterrupts` 有界丢弃 | @@ -242,7 +250,7 @@ CriticalSection：step 标记 nonPreemptible = true，或任务进入声明区 在安全点被抢占时： ``` -suspendPool.push(Task{frame: running.frame, state: suspended, +suspendStack.push(Task{frame: running.frame, state: suspended, step: running.frame.step, enqueueAt: running.enqueueAt}) running.state = done_for_now ``` @@ -253,7 +261,7 @@ running.state = done_for_now ### 6.2 恢复 -从 `suspendPool` 取出后： +从**中断栈栈顶**取出后： 1. **重建基础前缀**（`rebaseFramePrefix`）—— 此时中断任务已结束并提交， 重建出的 timeline 包含中断的输入/输出，即“现场加载回中断任务之上”； @@ -266,7 +274,8 @@ running.state = done_for_now ### 6.3 嵌套 - 允许中断任务自身被更高优先级抢占（嵌套）。 -- **`suspendPool` 深度上限 = 4**（与优先级档数一致，可配）。 +- **中断栈深度上限 = 4**（与优先级档数一致，可配）；嵌套时逐层压栈，恢复逐层弹出。 +- 栈自底向上的**基础级**天然递增（能被抢占者必然级别更高），因此栈顶通常就是最高级任务。 - 超限策略：**不继续下潜**——新的抢占请求转为 `pendingInterrupts`。超限丢弃/拒绝必须计数。 --- @@ -336,7 +345,7 @@ v1 采纳：**`S_TOOL_EXEC` / ONNX / CAS 属于临界区，调度器在这些 st | 失效 | 防御 | |---|---| | 饥饿（高优先级流反复抢占） | `preemptCount` 提升有效级：`effectiveLevel = min(4, baseLevel + min(preemptCount, 2))`；被抢占 +1 | -| 无界下潜 | `suspendPool` 深度上限 4，超限转 `pendingInterrupts` | +| 无界下潜 | 中断栈深度上限 4，超限转 `pendingInterrupts` | | 中断请求堆积 | `pendingInterrupts` 有界 64，满则丢最老并计数 | | 就绪队列满 | 阻塞发送方 + 计数（不静默丢） | | 同一任务反复被打断 | `preemptCount` 达阈值后有效级提升；另设**抢占冷却**：刚被抢占的任务在 `cooldown` 内不再被同级/更低级抢占 | @@ -363,16 +372,16 @@ v1 采纳：**`S_TOOL_EXEC` / ONNX / CAS 属于临界区，调度器在这些 st - **假 Provider**：实现 `agentAPI.Provider`，返回脚本化的 `tool_calls` 序列（支持"第 N 次调用时挂起直到放行"）。 - **假工具**：测试内 `StageHost.RegisterTool` 注册，可控制每次执行耗时、是否返回错误、是否触发中断注入。 - **同步栅栏**：测试通过 `scheduler.Inbox` 注入中断并用 `runtime.Gosched` + 显式 `waitFor(state)` 断言，不用 sleep 猜时序。 -- **快照断言**：`scheduler.Dump()` 返回 `{running, readyQueue, pendingInterrupts, suspendPool, counters}`，测试对纯数据断言。 +- **快照断言**：`scheduler.Dump()` 返回 `{running, readyQueue, pendingInterrupts, suspendStack, counters}`，测试对纯数据断言。 ### 11.1 优先级与抢占 | 编号 | 测试点 | 方式 | 预期结果 | |---|---|---|---| -| P1 | 高优先级抢占低优先级 | running=L2 在 `S_LLM`；注入 L3 中断 | L2 进 `suspendPool`（step=S_LLM）；L3 变 running；`preemptionCount==1` | +| P1 | 高优先级抢占低优先级 | running=L2 在 `S_LLM`；注入 L3 中断 | L2 压入中断栈（step=S_LLM）；L3 变 running；`preemptionCount==1` | | P2 | 相等优先级不抢占 | running=L2 在 `S_LLM`；注入 L2 | 不抢占；请求入 `pendingInterrupts`（或 readyQueue，按 D3）；running 不变 | | P3 | 低优先级不抢占 | running=L3；注入 L2 | 同上，不抢占 | -| P4 | 四级逐级抢占嵌套 | 依次注入 L4→L3→L2，均在 `S_LLM` | `suspendPool` 深度 3；running 为最新注入者；每层 step 均为 S_LLM | +| P4 | 四级逐级抢占嵌套 | 依次注入 L4→L3→L2，均在 `S_LLM` | 中断栈深度 3；running 为最新注入者；每层 step 均为 S_LLM | | P5 | 抢占后在安全点才生效 | running=L1 在 `S_TOOL_EXEC`；注入 L4 | 抢占**不立即生效**；工具返回后才保存/切换；`deferredPreemptions==1` | | P6 | 临界区不可抢占 | running=L1 声明临界区；注入 L4 | 同上；L4 请求留在 `pendingInterrupts`，临界区结束立即被选中 | @@ -399,9 +408,9 @@ v1 采纳：**`S_TOOL_EXEC` / ONNX / CAS 属于临界区，调度器在这些 st | 编号 | 测试点 | 方式 | 预期结果 | |---|---|---|---| -| Q1 | 选择函数排序 | 同时放入不同 level 与不同 `enqueueAt` 的三个集合成员 | 取值 = `(-effectiveLevel, enqueueAt)` 最小者；同级先到先服务 | +| Q1 | 选择函数排序 | 同时放入不同 level 与不同 `enqueueAt` 的三个集合成员 | 取值 = `(-effectiveLevel, kind, enqueueAt)` 最小者；同级 pending 优先，其次先到先服务 | | Q2 | 挂起任务优先恢复（同优先级） | A 挂起（早入队）+ B 就绪（晚入队），同级 | A 先被选中 | -| Q3 | 三类集合联动 | 结束 running 时 `pendingInterrupts` 与 `suspendPool` 同时非空且优先级不同 | 取优先级高者（与 §4.1 一致） | +| Q3 | 三类集合联动 | 结束 running 时 `pendingInterrupts` 与中断栈顶同时非空 | 高有效级者先；同级时 `pendingInterrupts` 优先（与 §4.1 一致） | | Q4 | 就绪队列背压 | readyQueue 满（256）后注入 | 发送方阻塞（或按 D4 返回错误）；计数 +1；不静默丢弃 | | Q5 | pending 队列溢出 | pendingInterrupts 满（64）后注入更多 | 丢最老的 + 计数；其余保持 | @@ -409,7 +418,7 @@ v1 采纳：**`S_TOOL_EXEC` / ONNX / CAS 属于临界区，调度器在这些 st | 编号 | 测试点 | 方式 | 预期结果 | |---|---|---|---| -| D1T | 下潜深度上限 | 连续注入 6 个逐级更高的中断 | `suspendPool` 深度 ≤ 4；超出部分在 `pendingInterrupts`；`depthRejections` 计数正确 | +| D1T | 下潜深度上限 | 连续注入 6 个逐级更高的中断 | 中断栈深度 ≤ 4；超出部分在 `pendingInterrupts`；`depthRejections` 计数正确 | | G1 | 饥饿防护（抢占提升） | 对同一 L1 任务连续抢占 5 次（同级/高级交替） | `effectiveLevel` 提升至 `min(4, 1+2)=3`；第 3 次后不再被 L1/L2 抢占 | | G2 | 冷却生效 | 同一任务刚被抢占后立刻再注入同级中断 | 冷却期内不抢占，请求入 pending | | K1 | panic 隔离 | 假工具 panic | 只有该任务变 `failed`；调度器存活；后续任务正常执行 | @@ -464,7 +473,7 @@ v1 采纳：**`S_TOOL_EXEC` / ONNX / CAS 属于临界区，调度器在这些 st | **M1** | **纯重构**：把 `process()` 拆成显式 step 状态机 + `TaskFrame`；仍由现有 `eventLoop` 驱动，无优先级/无抢占 | R3、X3 通过；既有全部 agent 测试通过（行为等价） | | **M2** | 调度器骨架：单 `schedulerLoop` + `readyQueue`，取代 `eventLoop` 的输入处理；无优先级（全部 L1，纯 FIFO） | Q1/Q4 通过；integration 测试通过 | | **M3a** | **前置重构（本次拆分引入）**：把一轮对话的所有权从 `processInput` 移到调度器——帧覆盖 `prepare → step… → finish`；同时移除 `process()` 整轮持有的 `a.mu`（挂起不能持锁） | 既有全部 agent 测试 + 既有 e2e 通过（行为等价）；`-race` 干净 | -| **M3b** | `interruptLoop` 重写 + 四级优先级 + 严格大于抢占 + `suspendPool`；只支持 `S_LLM` 抢占 | P1–P4、R1、R5、K1–K2 通过 | +| **M3b** | `interruptLoop` 重写 + 四级优先级 + 严格大于抢占 + 中断栈 LIFO；只支持 `S_LLM` 抢占 | P1–P4、R1、R5、K1–K2 通过；嵌套 LIFO 判据通过 | | **M4** | 临界区 + `S_TOOL_EXEC` 声明 + `pendingInterrupts` + 深度上限 | P5–P6、D1T、Q3、Q5 通过 | | **M5** | 饥饿防护（抢占计数提升 + 冷却） | G1–G2 通过 | | **M6** | 任务级 `responseCh` + 断链点统一为终态事件 | X1–X4 通过；`cli`/`clawhub` 不再挂起 | diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index f02d11a..cbab179 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -11,7 +11,7 @@ package core // - **每任务 panic 隔离**：panic 只使该任务失败，调度器本身存活（不变量 I6）。 // // M2 全部任务都是 LevelBackground（默认级），因此排序结果等价于 FIFO—— -// 与改造前的 channel 语义逐条一致。抢占、suspendPool、pendingInterrupts、 +// 与改造前的 channel 语义逐条一致。抢占、中断栈、pendingInterrupts、 // 任务级回执在 M3–M6 加入。 // // 并发模型（不变量 I2）：readyQueue/running/stats 只由 schedulerLoop 写， @@ -152,9 +152,10 @@ type SchedulerSnapshot struct { Running *Task Queue []*Task PendingInterrupts []*Task - SuspendPool []*suspendedTask - Stats SchedulerStats - MaxSuspendDepth int + // SuspendStack：中断栈（含嵌套抢占的多个现场），**栈顶**优先恢复。 + SuspendStack []*suspendedTask + Stats SchedulerStats + MaxSuspendDepth int } // schedulerStatus 把快照转成对外的状态 DTO（不暴露帧内容）。 @@ -166,7 +167,7 @@ func (a *Agent) schedulerStatus() sdk.SchedulerStatus { out := sdk.SchedulerStatus{ ReadyQueueDepth: len(snap.Queue), PendingInterrupts: len(snap.PendingInterrupts), - SuspendPool: len(snap.SuspendPool), + SuspendStack: len(snap.SuspendStack), MaxSuspendDepth: snap.MaxSuspendDepth, Enqueued: snap.Stats.Enqueued, Executed: snap.Stats.Executed, @@ -194,8 +195,9 @@ type scheduler struct { // pendingInterrupts：因优先级不足（或运行任务在临界区）而未立即抢占的中断请求。 // 与 readyQueue 分离：取出时以中断语义启动（设计文档 D3）。 pendingInterrupts []*Task - // suspendPool：被抢占后保存了现场、等待恢复的任务（**不是栈**，按优先级取）。 - suspendPool []*suspendedTask + // suspendStack：**中断栈**。被抢占后保存现场的任务压栈（LIFO）， + // 用于“中断被中断”的嵌套场景：只有**栈顶**参与恢复选择，栈内不做优先级重排。 + suspendStack []*suspendedTask // preemptArmed/preemptLevel：运行任务的“让位信号”。 // interruptLoop 只写这两个字段与 pendingInterrupts；帧永远只由调度器读写。 preemptArmed bool @@ -206,7 +208,7 @@ type scheduler struct { // wake 用于把空闲的调度器叫醒：pendingInterrupts 不是 channel， // 没有这个信号时“空闲时到达的中断”会一直等下一次输入（设计 §5.1 ③）。 wake chan struct{} - // maxSuspendDepth：suspendPool 深度上限（设计文档 §6.3，默认 4）。 + // maxSuspendDepth：中断栈深度上限（设计文档 §6.3，默认 4）。 maxSuspendDepth int } @@ -293,7 +295,32 @@ func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { var bestFrame *TaskFrame bestKind := nextNone consider := func(t *Task, k nextSelection, fr *TaskFrame) { - if bestTask == nil || taskBefore(t, bestTask) { + if bestTask == nil { + bestTask, bestKind, bestFrame = t, k, fr + return + } + lt, lb := effectiveLevel(t), effectiveLevel(bestTask) + if lt != lb { + if lt > lb { + bestTask, bestKind, bestFrame = t, k, fr + } + return + } + // 同级时 **pending 中断优先**。 + // + // 为何必需：一次抢占生效后，被挂起的原任务会因饥饿防护提升有效级， + // 于是与抢占者同级；若此时按“先到先服务”，原任务（入队更早）会被 + // 立刻选回，抢占者永远排不到——抢占变成空转。 + if k == nextPending && bestKind != nextPending { + bestTask, bestKind, bestFrame = t, k, fr + return + } + if bestKind == nextPending && k != nextPending { + return + } + // 同级同类：先到先服务（ID 兜底保证确定性）。 + if t.EnqueuedAt.Before(bestTask.EnqueuedAt) || + (t.EnqueuedAt.Equal(bestTask.EnqueuedAt) && t.ID < bestTask.ID) { bestTask, bestKind, bestFrame = t, k, fr } } @@ -303,8 +330,11 @@ func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { for _, t := range s.pendingInterrupts { consider(t, nextPending, nil) } - for _, st := range s.suspendPool { - consider(st.Task, nextSuspended, st.Frame) + // 中断栈：只比**栈顶**（严格 LIFO）。栈内不做优先级重排—— + // 嵌套抢占天然使栈自底向上优先级递增，且“后被打断的先恢复”才是栈语义。 + if n := len(s.suspendStack); n > 0 { + top := s.suspendStack[n-1] + consider(top.Task, nextSuspended, top.Frame) } if bestTask == nil { return nil, nil, nextNone @@ -316,12 +346,8 @@ func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { case nextPending: s.pendingInterrupts = removeTask(s.pendingInterrupts, bestTask) case nextSuspended: - for i, st := range s.suspendPool { - if st.Task == bestTask { - s.suspendPool = append(s.suspendPool[:i], s.suspendPool[i+1:]...) - break - } - } + // 只有栈顶可能被选中，故弹出即截断末位。 + s.suspendStack = s.suspendStack[:len(s.suspendStack)-1] } s.running = bestTask return bestTask, bestFrame, bestKind @@ -408,10 +434,10 @@ func (s *scheduler) clearPreempt() { func (s *scheduler) suspend(t *Task, f *TaskFrame) { s.mu.Lock() defer s.mu.Unlock() - if len(s.suspendPool) >= s.maxSuspendDepth { + if len(s.suspendStack) >= s.maxSuspendDepth { s.stats.Rejected++ } - s.suspendPool = append(s.suspendPool, &suspendedTask{Task: t, Frame: f}) + s.suspendStack = append(s.suspendStack, &suspendedTask{Task: t, Frame: f}) s.stats.Suspended++ // 饥饿防护：抢占计数 +1（提升有效级）并记录冷却起点。 t.PreemptCount++ @@ -430,7 +456,7 @@ func (s *scheduler) suspend(t *Task, f *TaskFrame) { func (s *scheduler) canSuspend() bool { s.mu.Lock() defer s.mu.Unlock() - return len(s.suspendPool) < s.maxSuspendDepth + return len(s.suspendStack) < s.maxSuspendDepth } // done 标记任务执行结束。 @@ -540,7 +566,7 @@ func (a *Agent) DumpScheduler() SchedulerSnapshot { snap := SchedulerSnapshot{Running: a.sched.running, Stats: a.sched.stats} snap.Queue = append(snap.Queue, a.sched.queue...) snap.PendingInterrupts = append(snap.PendingInterrupts, a.sched.pendingInterrupts...) - snap.SuspendPool = append(snap.SuspendPool, a.sched.suspendPool...) + snap.SuspendStack = append(snap.SuspendStack, a.sched.suspendStack...) snap.MaxSuspendDepth = a.sched.maxSuspendDepth return snap } diff --git a/internal/agent/core/scheduler_critical_test.go b/internal/agent/core/scheduler_critical_test.go index f4a82fd..28a0e30 100644 --- a/internal/agent/core/scheduler_critical_test.go +++ b/internal/agent/core/scheduler_critical_test.go @@ -67,8 +67,8 @@ func TestPreempt_DeferredDuringToolExec(t *testing.T) { if a.DumpScheduler().Running == nil { t.Fatal("工具执行中不得挂起（StepToolExec 是临界区）") } - if len(a.DumpScheduler().SuspendPool) != 0 { - t.Fatal("工具执行中 suspendPool 应为空") + if len(a.DumpScheduler().SuspendStack) != 0 { + t.Fatal("工具执行中 suspendStack 应为空") } // 放行工具 → 工具返回后的安全点才挂起。 @@ -80,11 +80,11 @@ func TestPreempt_DeferredDuringToolExec(t *testing.T) { } snap := a.DumpScheduler() - if len(snap.SuspendPool) != 1 { - t.Fatalf("工具返回后 suspendPool=%d，期望 1", len(snap.SuspendPool)) + if len(snap.SuspendStack) != 1 { + t.Fatalf("工具返回后 suspendStack=%d，期望 1", len(snap.SuspendStack)) } - if snap.SuspendPool[0].Frame.Step != StepToolAfter { - t.Fatalf("应在工具执行后的安全点挂起（StepToolAfter），实际 %v", snap.SuspendPool[0].Frame.Step) + if snap.SuspendStack[0].Frame.Step != StepToolAfter { + t.Fatalf("应在工具执行后的安全点挂起（StepToolAfter），实际 %v", snap.SuspendStack[0].Frame.Step) } if len(snap.PendingInterrupts) != 1 { t.Fatalf("中断请求不得丢失，pendingInterrupts=%d", len(snap.PendingInterrupts)) @@ -142,7 +142,7 @@ func TestBatch_NotAbandonedWithoutPreemption(t *testing.T) { t.Fatalf("同批工具应全部按序执行，实际 %v", ran) } snap := a.DumpScheduler() - if len(snap.SuspendPool) != 0 || len(snap.PendingInterrupts) != 0 { + if len(snap.SuspendStack) != 0 || len(snap.PendingInterrupts) != 0 { t.Fatalf("无抢占时不应有挂起或待处理中断：%+v", snap) } if snap.Stats.Executed != 1 { diff --git a/internal/agent/core/scheduler_e2e_test.go b/internal/agent/core/scheduler_e2e_test.go index 121791f..df0c462 100644 --- a/internal/agent/core/scheduler_e2e_test.go +++ b/internal/agent/core/scheduler_e2e_test.go @@ -42,14 +42,14 @@ func waitQuiescent(t *testing.T, a *Agent, wantExecuted uint64, timeout time.Dur for { snap := a.DumpScheduler() if snap.Running == nil && len(snap.Queue) == 0 && - len(snap.PendingInterrupts) == 0 && len(snap.SuspendPool) == 0 && + len(snap.PendingInterrupts) == 0 && len(snap.SuspendStack) == 0 && snap.Stats.Executed >= wantExecuted { return snap } if time.Now().After(deadline) { t.Fatalf("未在 %v 内排空：running=%v queue=%d pending=%d suspend=%d executed=%d", timeout, snap.Running != nil, len(snap.Queue), len(snap.PendingInterrupts), - len(snap.SuspendPool), snap.Stats.Executed) + len(snap.SuspendStack), snap.Stats.Executed) } time.Sleep(20 * time.Millisecond) } @@ -147,7 +147,7 @@ func TestObservability_SchedulerEventsAndStatus(t *testing.T) { // 状态快照（供状态页/诊断）：计数一致、三集合为空。 st := a.GetKernelStatus().Scheduler - if st.SuspendPool != 0 || st.PendingInterrupts != 0 || st.ReadyQueueDepth != 0 { + if st.SuspendStack != 0 || st.PendingInterrupts != 0 || st.ReadyQueueDepth != 0 { t.Fatalf("排空后状态非空：%+v", st) } if st.Suspended == 0 || st.Resumed == 0 { diff --git a/internal/agent/core/scheduler_preempt_test.go b/internal/agent/core/scheduler_preempt_test.go index ed87614..9531b1f 100644 --- a/internal/agent/core/scheduler_preempt_test.go +++ b/internal/agent/core/scheduler_preempt_test.go @@ -1,6 +1,6 @@ package core -// M3b 验收测试：四级优先级 + 严格大于抢占 + 现场保存/恢复 + suspendPool。 +// M3b 验收测试：四级优先级 + 严格大于抢占 + 现场保存/恢复 + suspendStack。 // // 设计依据 docs/zh/input-scheduler-design.md §11（P1–P4、R1、R5、D1T、Q3）。 // @@ -124,13 +124,13 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { if snap.Running != nil { t.Fatal("挂起后不应还有 running") } - if len(snap.SuspendPool) != 1 { - t.Fatalf("suspendPool=%d，期望 1", len(snap.SuspendPool)) + if len(snap.SuspendStack) != 1 { + t.Fatalf("suspendStack=%d，期望 1", len(snap.SuspendStack)) } if len(snap.PendingInterrupts) != 1 { t.Fatalf("pendingInterrupts=%d，期望 1", len(snap.PendingInterrupts)) } - sf := snap.SuspendPool[0].Frame + sf := snap.SuspendStack[0].Frame if sf.Terminal != terminalSuspended { t.Fatalf("挂起任务终态=%v，期望 terminalSuspended", sf.Terminal) } @@ -186,7 +186,7 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { t.Fatalf("LLM 总调用=%d，期望 3（丢弃 1 + 中断 1 + 恢复 1）", sp.callCount()) } snap = a.DumpScheduler() - if len(snap.SuspendPool) != 0 || snap.Running != nil { + if len(snap.SuspendStack) != 0 || snap.Running != nil { t.Fatalf("全部结束后应无挂起与运行任务：%+v", snap) } if snap.Stats.Executed != 2 { @@ -239,7 +239,7 @@ func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) { } } -// D1T：suspendPool 满时不再下潜（canSuspend=false），让位信号也不会 arm。 +// D1T：suspendStack 满时不再下潜（canSuspend=false），让位信号也不会 arm。 func TestPreempt_DepthCapBlocksSuspension(t *testing.T) { a := newPreemptAgent(t, newPreemptProvider()) @@ -259,7 +259,7 @@ func TestPreempt_DepthCapBlocksSuspension(t *testing.T) { if a.DumpScheduler().Stats.Rejected != before+1 { t.Fatal("超限挂起必须计数 Rejected") } - if len(a.DumpScheduler().SuspendPool) != a.sched.maxSuspendDepth+1 { + if len(a.DumpScheduler().SuspendStack) != a.sched.maxSuspendDepth+1 { t.Fatal("兜底路径必须保留帧而不是丢弃") } } diff --git a/internal/agent/core/scheduler_stack_test.go b/internal/agent/core/scheduler_stack_test.go new file mode 100644 index 0000000..3a3197d --- /dev/null +++ b/internal/agent/core/scheduler_stack_test.go @@ -0,0 +1,192 @@ +package core + +// 中断栈（嵌套抢占）验收测试。 +// +// 用户明确：存在**中断被中断**的场景，所以被打断的现场要压进**中断栈**。 +// 因此恢复纪律是**严格 LIFO（只比栈顶）**，而不是“全栈按优先级挑最优”。 +// +// 为什么这个区别成立：抢占判据是 adopted.level > effectiveLevel(running)， +// 所以嵌套时栈自底向上的**基础级**天然递增；但饥饿防护的“有效级提升”会让 +// 栈内某个更老的任务有效级超过栈顶，此时“只比栈顶”才保证嵌套语义不被破坏。 + +import ( + "context" + "sync" + "testing" + "time" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" +) + +// nestingProvider 第 1、2 次调用阻塞到 ctx 取消；第 3 次起按脚本返回。 +// 用序号精确对应 A（L1）→ B（L2）→ C（L3）→ 恢复 B → 恢复 A 的调用顺序。 +type nestingProvider struct { + mu sync.Mutex + calls int + entered chan int + script []string +} + +func newNestingProvider(script ...string) *nestingProvider { + return &nestingProvider{entered: make(chan int, 16), script: script} +} + +func (p *nestingProvider) Name() string { return "nesting" } +func (p *nestingProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) { + p.mu.Lock() + p.calls++ + n := p.calls + p.mu.Unlock() + + p.entered <- n + if n <= 2 { + <-ctx.Done() + return nil, ctx.Err() + } + p.mu.Lock() + defer p.mu.Unlock() + i := n - 3 + if i < len(p.script) { + return &agentAPI.CompletionResponse{Content: p.script[i]}, nil + } + return &agentAPI.CompletionResponse{Content: "?"}, nil +} +func (p *nestingProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) { + return nil, context.Canceled +} +func (p *nestingProvider) MaxContextTokens() int { return 8192 } + +func awaitEnter(t *testing.T, ch chan int, want int) { + t.Helper() + select { + case got := <-ch: + if got != want { + t.Fatalf("LLM 进入序号=%d，期望 %d", got, want) + } + case <-time.After(3 * time.Second): + t.Fatalf("等第 %d 次 LLM 调用超时", want) + } +} + +// 中断被中断：A(L1) → B(L2) → C(L3)，恢复必须按 LIFO（B 先，A 后）。 +func TestStack_NestedPreemptionResumesLIFO(t *testing.T) { + sp := newNestingProvider("c-done", "b-done", "a-done") + a := newPreemptAgent(t, sp) + + // A（L1）开始运行 + if _, _ = enqueueTask(t, a, LevelBackground, "qq", "任务A"); true { + } + at, _, _ := a.sched.nextRef() + doneA := make(chan struct{}) + go func() { a.executeNewTask(at); close(doneA) }() + awaitEnter(t, sp.entered, 1) + + // B（L2）抢占 A + bEvt, _ := textEvent("qq", "任务B") + if !a.sched.requestPreempt(bEvt, LevelMessage) { + t.Fatal("B(L2) 应抢占 A(L1)") + } + a.cancelCurrentLLM() + <-doneA + if n := len(a.DumpScheduler().SuspendStack); n != 1 { + t.Fatalf("第一次抢占后栈深=%d，期望 1", n) + } + + // B 开始运行 + bt, _, k := a.sched.nextRef() + if k != nextPending || bt.Level != LevelMessage { + t.Fatalf("应取到 B（pending），kind=%v level=%v", k, bt.Level) + } + doneB := make(chan struct{}) + go func() { a.executeNewTask(bt); close(doneB) }() + awaitEnter(t, sp.entered, 2) + + // C（L3）抢占 B —— 这就是“中断被中断” + cEvt, _ := textEvent("cli", "任务C") + if !a.sched.requestPreempt(cEvt, LevelInteractive) { + t.Fatal("C(L3) 应抢占 B(L2)") + } + a.cancelCurrentLLM() + <-doneB + + snap := a.DumpScheduler() + if len(snap.SuspendStack) != 2 { + t.Fatalf("嵌套后栈深=%d，期望 2", len(snap.SuspendStack)) + } + if snap.SuspendStack[0].Task.Level != LevelBackground { + t.Fatalf("栈底应为 A(L1)，实际 %v", snap.SuspendStack[0].Task.Level) + } + if snap.SuspendStack[1].Task.Level != LevelMessage { + t.Fatalf("栈顶应为 B(L2)，实际 %v", snap.SuspendStack[1].Task.Level) + } + + // C 运行完毕（第三次调用，不阻塞） + ct, _, k := a.sched.nextRef() + if k != nextPending || ct.Level != LevelInteractive { + t.Fatalf("应取到 C，kind=%v level=%v", k, ct.Level) + } + a.executeNewTask(ct) + + // LIFO：先恢复栈顶 B，再恢复 A + rt, rf, k := a.sched.nextRef() + if k != nextSuspended { + t.Fatalf("应恢复栈顶，kind=%v", k) + } + if rt.Level != LevelMessage { + t.Fatalf("应先恢复栈顶 B(L2)，实际 %v", rt.Level) + } + a.resumeTask(rt, rf) + + rt2, rf2, k2 := a.sched.nextRef() + if k2 != nextSuspended { + t.Fatalf("应继续恢复 A，kind=%v", k2) + } + if rt2.Level != LevelBackground { + t.Fatalf("最后应恢复 A(L1)，实际 %v", rt2.Level) + } + a.resumeTask(rt2, rf2) + + if n := len(a.DumpScheduler().SuspendStack); n != 0 { + t.Fatalf("全部恢复后栈应清空，实际 %d", n) + } +} + +// 只比栈顶：栈内更老的任务即使（因有效级提升）优先级更高，也不得越过栈顶。 +func TestStack_TopOnlyWinsOverHigherPrioritySuspended(t *testing.T) { + a := newPreemptAgent(t, &scriptProvider{}) + + // 人为构造“A(L3) 在栈底、B(L2) 在栈顶”。真实抢占不会产生这种顺序 + // （栈自底向上基础级递增），这里专门用来区分两种实现： + // · 只比栈顶 → 取 B + // · 全栈扫最优 → 取 A（L3 > L2） + a.sched.suspend(&Task{ID: 1, Level: LevelInteractive, EnqueuedAt: time.Now()}, + a.newTaskFrame("A", a.stageCtxFromInput("A", "", ""))) + a.sched.suspend(&Task{ID: 2, Level: LevelMessage, EnqueuedAt: time.Now()}, + a.newTaskFrame("B", a.stageCtxFromInput("B", "", ""))) + + rt, _, k := a.sched.nextRef() + if k != nextSuspended { + t.Fatalf("kind=%v，期望 nextSuspended", k) + } + if rt.ID != 2 { + t.Fatalf("应取栈顶 B(ID=2)，实际 ID=%d —— 说明在做全栈优先级扫描而非栈语义", rt.ID) + } + if n := len(a.DumpScheduler().SuspendStack); n != 1 { + t.Fatalf("取出栈顶后栈深=%d，期望 1", n) + } +} + +// 深度上限对嵌套同样成立：到顶后新的抢占请求不再下潜。 +func TestStack_DepthCapDuringNesting(t *testing.T) { + a := newPreemptAgent(t, &scriptProvider{}) + frame := func() *TaskFrame { return a.newTaskFrame("x", a.stageCtxFromInput("x", "", "")) } + for i := 0; i < a.sched.maxSuspendDepth; i++ { + a.sched.suspend(&Task{ID: uint64(i + 1), Level: Level(i + 1)}, frame()) + } + if a.sched.canSuspend() { + t.Fatal("栈已满，canSuspend 应为 false") + } + if n := len(a.DumpScheduler().SuspendStack); n != a.sched.maxSuspendDepth { + t.Fatalf("栈深=%d，期望上限 %d", n, a.sched.maxSuspendDepth) + } +} diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index f41493b..3f27d59 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -6,7 +6,7 @@ package core // // M1 只做一件事：把原先「一个 425 行的 process() 大函数」拆成 // **显式 step 游标 + TaskFrame**。目的不是加能力，而是让「现场」变成数据—— -// 之后 M3 才能把帧存进 suspendPool 并在安全点恢复。 +// 之后 M3 才能把帧存进 suspendStack 并在安全点恢复。 // // 行为等价的判据：既有全部 agent 测试通过，且 R3/X3（见设计文档 §11）通过。 // diff --git a/internal/sdk/status.go b/internal/sdk/status.go index 6caf231..eb71e4f 100644 --- a/internal/sdk/status.go +++ b/internal/sdk/status.go @@ -49,7 +49,7 @@ type KernelStatus struct { // Scheduler 是输入调度器的运行时快照（可观测性，设计文档 §11 O1/O2）。 // M2 起输入不再直接排队在 channel 上，而是经 readyQueue/pendingInterrupts/ - // suspendPool 三集合按优先级调度；这里把这些状态暴露出来。 + // suspendStack 三集合按优先级调度；这里把这些状态暴露出来。 Scheduler SchedulerStatus `json:"scheduler"` } @@ -57,10 +57,10 @@ type KernelStatus struct { type SchedulerStatus struct { // Running 是当前执行的任务（空表示空闲）。 Running *SchedulerTask `json:"running,omitempty"` - // ReadyQueueDepth / PendingInterrupts / SuspendPool 是三个集合的深度。 + // ReadyQueueDepth / PendingInterrupts / SuspendStack 是三个集合的深度。 ReadyQueueDepth int `json:"ready_queue_depth"` PendingInterrupts int `json:"pending_interrupts"` - SuspendPool int `json:"suspend_pool"` + SuspendStack int `json:"suspend_stack"` MaxSuspendDepth int `json:"max_suspend_depth"` Enqueued uint64 `json:"enqueued"`