diff --git a/docs/zh/input-scheduler-design.md b/docs/zh/input-scheduler-design.md index c88d185..974076e 100644 --- a/docs/zh/input-scheduler-design.md +++ b/docs/zh/input-scheduler-design.md @@ -1,4 +1,21 @@ -# 输入调度器设计(四级优先级 · 可抢占 · 现场保存) +# 输入调度器设计(四级中断优先级 · 两类别 · 可抢占 · 现场保存) + +> **模型更正(2026-09-13,据用户澄清重写 §2/§3/§4.1/§6.3/§9/§11/§12)** +> +> 本稿早期版本把「四级优先级」当成了**所有任务**的通用优先级,并按通道名 +> (qq→L2、cli→L3)由内核推断级别。那是错的。正确模型是**两类别 + 四级**: +> +> | | 中断输入(interrupt) | 排队输入(queued) | +> |---|---|---| +> | 注入 API | `InjectInterrupt*` | `InjectText*` / `InjectInputSync*` / 内核自循环 | +> | 级别 | L1–L4 | **无级别** | +> | 定位 | 需要及时处理 | 不需要及时处理 | +> | 可被谁打断 | 仅**严格更高级**的中断 | **任何**中断 | +> +> 级别(“这项工作有多不能等”)由插件在 `InjectOptions.Priority` 里声明 L1–L3; +> **L4 由内核独占**,只经 panic 与内核事件中断(selfip)产生。 +> 类别由**用哪个注入 API**决定,与通道名无关——QQ 走的是 `InjectInterruptTextOpts`, +> 所以它是**低级别中断(L1)**,不是排队输入。 > 分支:`feature/input-semantics` > 状态:**设计稿 v1**(待确认项见 §12,未确认处按 §12 的「默认取值」推进) @@ -33,7 +50,11 @@ ## 2. 术语与模型 ``` -Task = { id, level, origin, frame, state, enqueueAt, preemptCount, responseCh } +Task = { id, class, level, origin, frame, state, enqueueAt, preemptCount, responseCh } + class: queued | interrupt // **类别由注入 API 决定,与通道名无关** + queued —— 无级别;用于“不需及时处理”的场景;可被**任何**中断打断 + interrupt —— 带级别 L1..L4;仅被**严格更高级**的中断打断(被打断则压入中断栈) + level: 仅 interrupt 有意义(queued 恒无级别,effectiveLevel 视作 0) state: ready | running | suspended | done TaskFrame = { @@ -69,80 +90,103 @@ Step(枚举,顺序执行,步与步之间是安全点): S_FINISH 写 responseCh、发事件 ``` -三集合: +四个容器(**不是**“三集合并成一个比较器”): -| 集合 | 内容 | 语义 | +| 容器 | 内容 | 取出规则 | |---|---|---| -| `readyQueue` | 排队输入形成的新任务 | 按 `(level desc, enqueueAt asc)` 排序 | -| `pendingInterrupts` | **因优先级不足或临界区而未能抢占**的中断请求 | 同上排序;被取出时以中断语义启动 | -| `suspendStack`(**中断栈**) | 被打断、保存了现场的任务,用于“中断被中断”的嵌套 | **LIFO**:只有栈顶参与选择;被取出时从 `frame.step` 继续 | +| `interruptQueues[1..4]` | **中断队列**,每条队列一个级别 | 从 L4 到 L1 依次扫描;同级 FIFO | +| `immediate` | 刚抢占成功的那一条中断(**至多一个**) | 最先取出——抢占必须**立即生效** | +| `queue` | **排队输入**形成的新任务 | 纯 FIFO(无级别可比) | +| `suspendStack`(**中断栈**) | 被打断、保存了现场的任务 | **LIFO,只比栈顶**;栈内不做重排 | > 用词(已更正):它**就是中断栈**。用户明确存在「中断被中断」的场景,被打断的现场必须压栈; > 因此恢复纪律是**严格 LIFO(只比栈顶)**,栈内不做优先级重排。 > 早期稿把它写成“不是栈、按优先级取”是错的。 +> +> 早期稿还让 `immediate` 与别的容器共用同一个比较器,于是出现“抢占成功后, +> 抢占者与被挂起者同级 → 原任务被立刻选回 → 抢占空转”——为此打的 +> “同级 pending 优先”补丁已删除:抢占者根本不进队列。 --- -## 3. 优先级 +## 3. 优先级(只属于中断) -### 3.1 四级(内核预定义) +### 3.1 两类别 + 四级 + +**类别(`TaskClass`)由注入 API 决定,与通道名无关**: + +| 类别 | 注入入口 | 级别 | 可被谁打断 | +|---|---|---|---| +| `queued` 排队 | `InjectText*` / `InjectInputSync*` / `InjectInputMedia*` / 内核自循环(`selfInputCh`) | **无** | **任何**中断(L1 也能) | +| `interrupt` 中断 | `InjectInterrupt*` | L1–L4 | 仅**严格更高级**的中断 | + +**级别(`Level`)语义是“这项工作有多不能等”**: | Level | 名称 | 语义 | 典型来源 | |---|---|---|---| -| `L4` | CRITICAL | 紧急打断 | CLI/WebUI 的显式打断、系统告警、安全类中断 | -| `L3` | INTERACTIVE | 人机交互 | 用户在 CLI/WebUI 的直接对话 | -| `L2` | MESSAGE | 异步消息 | QQ/微信等入站消息、插件通知 | -| `L1` | BACKGROUND | 后台维护 | 心跳蒸馏/归档/合并/复审、`spawn_child` 子任务、consolidation、healthcheck | +| `L4` | CRITICAL | 内核紧急 | **内核独占**:panic 中断、内核事件中断(selfip) | +| `L3` | INTERACTIVE | 需及时处理 | 时钟/定时器到达、终端输出、交互输入 | +| `L2` | MESSAGE | 一般提醒 | 插件希望尽快看到、但不紧急的提示 | +| `L1` | BACKGROUND | 完全可等 | 异步消息(QQ/微信)、批量通知 | -- **默认级 = `L1`**:未显式声明一律最低级("显式才是特权",避免新插件默认获得抢占权)。 -- **取值域仅这四档**,不引入任意整数,避免"9 级比 4 级大但没人知道怎么排"。 +- **`queued` 没有级别**:它本就是“不需及时处理”的那一类, + 所以“可被任何中断打断”不是漏洞而是定义(`effectiveLevel(queued) == 0`)。 +- **默认级 = `L1`**:未声明一律最低级(“显式才是特权”,新插件不会默认拿到抢占权)。 -### 3.2 优先级从哪来(内核内部属性,**不做成配置项**) +### 3.2 级别从哪来 -优先级是**内核内部属性**:内核预定义四级,并按**内核自己的规则**为任务与中断定级。 +| 来源 | 可达级别 | 入口 | +|---|---|---| +| 插件声明 | L1–L3 | `InjectOptions.Priority`(空/非法 → L1;声明 L4 被夹到 L3) | +| 内核 | L4(唯一来源) | `(*Agent).raiseKernelInterrupt`(panic / selfip) | - ❌ **不是运维可调项**。不引入 `core.agent.priority.` 这类配置键, - 也不把 `PriorityLookup` 做成可注入的策略表——那等于把内核的调度内部属性 - 外化成配置,与“谁能打断谁”的内部语义相反。 -- ❌ 也不暴露给插件声明(这个方向曾写入 v2 计划,已删除)。 -- ✅ v1 的内部缺省规则(仅为实现缺省值,语义上是内核自己的事): - `cli`/`webui`/`http` → `L3`;`system` / `_consolidation_` → `L1`;其余 → `L1`(默认级)。 -- 定级规则可随内核演进调整,但**始终不对外暴露**。 - -> 具体“哪类工作算哪一级”的完整规则由内核定义;本稿只固定四级语义与定级位置 -> (`(*Agent).taskLevel`),不承诺配置面。 + 也不把 `PriorityLookup` 做成可注入的策略表。 +- ✅ 插件**可以声明**自己中断的级别(这不是“把内核内部属性外化”, + 而是调用方声明它自己那件事有多不能等),但**内核独占 L4**: + `clampPluginLevel` 把越权声明夹到 L3,`L4` 在插件可达路径上不存在。 +- 定级规则可随内核演进调整,但插件可声明域**始终不含 L4**。 ### 3.3 抢占判据 -``` -incoming.level > running.level → 请求抢占 -incoming.level == running.level → 入 readyQueue(或 pendingInterrupts,见 §5),FIFO -incoming.level < running.level → 入 pendingInterrupts +```go +effectiveLevel(queued) == 0 +canPreempt(incoming, running) = incoming.Class == TaskInterrupt + && effectiveLevel(incoming) > effectiveLevel(running) ``` -**严格大于才抢占**;相等一律排队——这条保证确定性,也是"较低无法打断较高"的字面实现。 +因为 `queued` 的有效级恒为 0,这一个比较同时覆盖两条规则: ---- +``` +running 是排队任务 → 任何中断(≥L1)都抢占 +running 是中断 Li → 只有 Lj > Li 的中断抢占(严格大于) +incoming 是排队输入 → 永不抢占 +``` + +**严格大于才抢占**;相等一律入队——这条保证确定性,也是“较低无法打断较高”的字面实现。 ## 4. 调度规则 -### 4.1 选择函数(统一三集合) +### 4.1 选择函数(四容器 · 固定次序) 任务结束、或运行任务到达安全点且存在待处理抢占请求时,执行: ``` -candidates = readyQueue ∪ pendingInterrupts ∪ { suspendStack.top } // 栈只出栈顶 -pick = argmin over candidates by (-effectiveLevel(t), kind, t.enqueueAt) +1. immediate 非空 → 取它(刚抢占成功的中断,抢占必须立即生效) +2. 中断队列非空 → 取 L4→L1 中最高级非空队列的队头(同级 FIFO) +3. 中断栈非空(与 2 比高) → 栈顶有效级 ≥ 队头级别 ? 弹栈顶 : 取队头 +4. queue 非空 → 取队头(纯 FIFO) +5. 都没有 → 空闲(阻塞等新输入 / 新中断) ``` -- **中断栈只把栈顶**放进候选(严格 LIFO)——栈内更老的任务即使因饥饿防护提升了 - 有效级,也不得越过栈顶;“后被打断的先恢复”才是栈语义。 -- 高有效级先; -- **同级时 `pendingInterrupts` 优先于其它两类**。 - 为何必需:抢占生效后,被挂起的原任务会因饥饿防护提升有效级,于是与抢占者同级; - 若同级按“先到先服务”,原任务(入队更早)会被立刻选回,抢占者永远排不到—— - 抢占变成空转。 -- 同级同类:先到先服务(`enqueueAt` 为首次入队时刻;挂起任务保留其原始入队时刻)。 +- **中断栈只把栈顶**放进比较(严格 LIFO)——栈内更老的任务即使因饥饿防护 + 提升了有效级,也不得越过栈顶;“后被打断的先恢复”才是栈语义。 +- 第 3 步就是用户给的规则:“先判断中断队列是否为空,同时判断中断栈中任务的 + 优先级,哪个优先级高取出哪个”。栈顶是 `queued`(有效级 0)时,任何中断都赢。 +- 第 1 步的存在,使“抢占者与被抢占者同级”这个比较**根本不会发生**: + 抢占者不经队列。这是删除早期“同级 pending 优先”补丁后的正确形态。 +- 排队任务只在中断与挂起现场都处理完后才执行——这正对应“排队输入用于 + 不需要及时处理的场景”。 ### 4.2 安全点(可切换点) @@ -169,14 +213,14 @@ CriticalSection:step 标记 nonPreemptible = true,或任务进入声明区 | 媒体 CAS 落盘 | 同上 | | 显式声明的 `_consolidation_` 类任务 | 记忆一致性 | -- 临界区期间到达的抢占请求**不丢失**:进入 `pendingInterrupts`,在临界区结束后的第一个安全点重新求值。 +- 临界区期间到达的抢占请求**不丢失**:按级别进入中断队列,在临界区结束后的第一个安全点重新求值。 ### 4.4 背压(v1 统一为一种) - `readyQueue` 有界(默认 256,可配)。 - 满时:**阻塞发送方**(与现状 `inputCh` 一致,避免静默丢用户输入),但必须**计数并打日志**。 -- `pendingInterrupts` 有界(默认 64);满时**丢弃最老的 pending 中断并计数**(中断是提示性输入,宁可丢旧保新)。 -- 中断栈深度上限默认 4(见 §6.3)。 +- 中断队列合计有界(默认同 `maxQueue`);满时**丢弃最低级别里最老的一条并计数**(中断是提示性输入,宁可丢旧保新)。 +- 中断栈帧数上界是**结构推论 = 4**(见 §6.3),不是配置项。 --- @@ -187,29 +231,29 @@ CriticalSection:step 标记 nonPreemptible = true,或任务进入声明区 `interruptLoop` 只做三件事,**绝不触碰任何 TaskFrame**: ``` -① 从 io.interruptCh 收中断 → 定级 -② 决策: - incoming.level > running.level 且 running 不在临界区 - → 置 preemptionRequest = {incoming, requestedAt},并调用 running 当前 step 的 cancel(若可取消) - incoming.level > running.level 但 running 在临界区 - → 入 pendingInterrupts - incoming.level <= running.level - → 入 pendingInterrupts -③ 唤醒调度器(向 schedulerInbox 投一个 wake 信号) +① 从 io.interruptCh 收中断 → 定级(读 payload["priority"],插件声明 L1..L3) +② 决策(scheduler.registerInterrupt 内): + canPreempt(incoming, running) 且 running 不在临界区 + → 置让位信号 + 把 incoming 放进 immediate 槽,并返回 true(调用方据此 + 取消当前可取消的 step,即 LLM 流式) + 否则 + → 按级别进入对应的中断队列 +③ 唤醒调度器(scheduler.wake,cap 1) ``` -共享面仅两处:`preemptionRequest`(原子指针)与 `running.stepCancel`(原子读)。**帧的保存与恢复只能由调度器做。** +共享面仅三处:让位信号(`preemptArmed`/`preemptLevel`)、中断队列、`critical` 原子标志。 +**帧的保存与恢复只能由调度器做。** ### 5.2 三种情形的统一 现状的三条降级路径在新模型里不再需要特殊分支: -| 情形 | 现状 | 新模型 | -|---|---|---| +| 情形 | 旧模型 | 新模型 | +|---|---|---|---| | LLM 在跑,正常 | 真抢占(同轮 continue) | 真抢占:`S_LLM` 取消,任务 A **压入中断栈**,中断任务 B 从 `S_PREPARE` 启动 | -| LLM 没在跑 | 降级为排队 | B 入 `pendingInterrupts`(或直接成为 ready 任务),调度器立即选出 | -| `_consolidation_` 中 | 降级为排队 | `_consolidation_` 是后台临界区 → B 入 `pendingInterrupts`,临界区结束后求值 | -| `a.interceptCh` 满 | 降级为排队 | 不存在该队列;`pendingInterrupts` 有界丢弃 | +| LLM 没在跑 | 降级为排队 | B 按其级别入中断队列(空闲时即被 `wake` 唤醒并选出) | +| `_consolidation_` 中 | 降级为排队 | `_consolidation_` 是后台**临界区**(且它是排队任务)→ B 入中断队列,临界区结束后求值 | +| `a.interceptCh` 满 | 降级为排队 | 不存在该队列;中断队列有界,满则丢最低级别里最老的一条 | ### 5.3 中断任务与被打断任务的关系(**已定:D1 = 方案 B**) @@ -273,12 +317,15 @@ running.state = done_for_now ### 6.3 嵌套 -- 允许中断任务自身被更高优先级抢占(嵌套)。 -- **中断栈深度上限 = 4**(与优先级档数一致,可配);嵌套时逐层压栈,恢复逐层弹出。 +- 允许中断任务自身被更高级中断抢占(嵌套)。 +- **中断栈帧数上界 = 4,是结构推论而不是配置项**: + 链条 = `排队(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4 运行中)`, + 被挂起 4 帧;L4 之上没有更高级别,链到此为止。 + (插件可达级别只到 L3,所以插件链最多挂起 3 帧 + 底层排队任务; + 第 4 帧只能由内核 L4 制造。) - 栈自底向上的**基础级**天然递增(能被抢占者必然级别更高),因此栈顶通常就是最高级任务。 -- 超限策略:**不继续下潜**——新的抢占请求转为 `pendingInterrupts`。超限丢弃/拒绝必须计数。 - ---- +- 超限在正确模型下不可达:`susp` 处只做**防御性计数**(`Rejected++`), + **不降级、不丢弃帧**——帧丢了会丢副作用记录。早期稿写的“超限转 pendingInterrupts”已删除。 ## 7. 回执路由(任务级) @@ -333,9 +380,13 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st - 代价:这些临界区期间**中断只能排队,不能抢占**。换言之,**中断的有效窗口 = `S_LLM`**(与今天的实际行为相同,但现在是显式声明而非隐式结果)。 - 演进(v2):把 `S_TOOL_EXEC` 改成异步 step(临时 goroutine + 完成事件),并给插件协议加 `tool.cancel`。此路径在文档保留,不在 v1 实现。 -### 8.3 panic 隔离 +### 8.3 panic 隔离与 panic 中断 - `runOneStep` 外包 `recover`:panic → 当前任务标记 `failed`,**调度器继续**。 +- panic 同时**产生一条内核 L4 中断**(`reportTaskPanic` → `raiseKernelInterrupt`): + 内核把自己发生了 panic 这件事作为最高级中断通知给调度器,让 agent 能知情/善后。 +- 递归保护是**结构性**的:若 panic 的任务本身就是 L4 内核中断,不再产生新的 L4—— + 否则同一个 panic 会自我放大成中断风暴。 - 取代现有 `eventLoop`/`interceptLoop` 的 `recover → sleep 1s → go loop()` 无退避重启(`eventloop.go:19-22,38-42`)。 --- @@ -344,9 +395,9 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st | 失效 | 防御 | |---|---| -| 饥饿(高优先级流反复抢占) | `preemptCount` 提升有效级:`effectiveLevel = min(4, baseLevel + min(preemptCount, 2))`;被抢占 +1 | -| 无界下潜 | 中断栈深度上限 4,超限转 `pendingInterrupts` | -| 中断请求堆积 | `pendingInterrupts` 有界 64,满则丢最老并计数 | +| 饥饿(高优先级流反复抢占) | `preemptCount` 提升有效级:`effectiveLevel = min(4, baseLevel + min(preemptCount, 2))`;被抢占 +1。**只对中断生效**——排队任务无级别,按定义可被任何中断打断 | +| 无界下潜 | 中断栈帧数上界 4(结构推论 = 中断级数);超限只做防御性计数,**不降级不丢帧** | +| 中断请求堆积 | 中断队列合计有界,满则丢最低级别里最老的一条并计数 | | 就绪队列满 | 阻塞发送方 + 计数(不静默丢) | | 同一任务反复被打断 | `preemptCount` 达阈值后有效级提升;另设**抢占冷却**:刚被抢占的任务在 `cooldown` 内不再被同级/更低级抢占 | | 任务永不结束 | 每任务 `maxTurns`(主循环目前缺失,见审查 P0)+ 每步超时 | @@ -372,18 +423,22 @@ 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, suspendStack, counters}`,测试对纯数据断言。 +- **快照断言**:`scheduler.Dump()` 返回 `{running, queue, interruptQueues[1..4], immediate, suspendStack, counters}`,测试对纯数据断言。 ### 11.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` | 中断栈深度 3;running 为最新注入者;每层 step 均为 S_LLM | -| P5 | 抢占后在安全点才生效 | running=L1 在 `S_TOOL_EXEC`;注入 L4 | 抢占**不立即生效**;工具返回后才保存/切换;`deferredPreemptions==1` | -| P6 | 临界区不可抢占 | running=L1 声明临界区;注入 L4 | 同上;L4 请求留在 `pendingInterrupts`,临界区结束立即被选中 | +| P1 | 更高中断抢占中断 | running=L2 在 `S_LLM`;注入 L3 中断 | L2 压入中断栈(step=S_LLM);L3 进 `immediate` 并变 running | +| P2 | 相等级别不抢占 | running=L2 中断在 `S_LLM`;注入 L2 | 不抢占;请求入 L2 中断队列;running 不变 | +| P3 | 更低级别不抢占 | running=L3 中断;注入 L2 | 同上,不抢占 | +| P4 | 逐级抢占嵌套 | 排队任务 → L1 → L2 → L3 → L4,均在 `S_LLM` | 中断栈深度依次 1/2/3/4;每层 step 均为 S_LLM | +| P5 | 抢占后在安全点才生效 | running=排队任务在 `S_TOOL_EXEC`;注入 L4 | 抢占**不立即生效**;工具返回后才保存/切换;`deferredPreemptions==1` | +| P6 | 临界区不可抢占 | running 声明临界区;注入 L4 | 同上;L4 请求留在中断队列,临界区结束立即被选中 | +| **P7** | **排队任务被任何中断打断** | running=排队任务;注入 **L1** 中断 | L1 也抢占成功(排队任务有效级 0) | +| **P8** | **排队输入永不抢占** | running=任意任务;注入排队输入 | 不抢占,入排队队列 | +| **P9** | **插件不能声明 L4** | `InjectOptions.Priority="L4"` | 级别被夹到 L3;`payload["priority"]` 走同一条路 | +| **P10** | **panic 产生 L4 中断** | 任务 panic | 产生一条带 `kernel=true` 的 L4 中断;L4 自身 panic 不再递归 | ### 11.2 保存现场与恢复 @@ -408,19 +463,21 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st | 编号 | 测试点 | 方式 | 预期结果 | |---|---|---|---| -| Q1 | 选择函数排序 | 同时放入不同 level 与不同 `enqueueAt` 的三个集合成员 | 取值 = `(-effectiveLevel, kind, enqueueAt)` 最小者;同级 pending 优先,其次先到先服务 | -| Q2 | 挂起任务优先恢复(同优先级) | A 挂起(早入队)+ B 就绪(晚入队),同级 | A 先被选中 | -| Q3 | 三类集合联动 | 结束 running 时 `pendingInterrupts` 与中断栈顶同时非空 | 高有效级者先;同级时 `pendingInterrupts` 优先(与 §4.1 一致) | -| Q4 | 就绪队列背压 | readyQueue 满(256)后注入 | 发送方阻塞(或按 D4 返回错误);计数 +1;不静默丢弃 | -| Q5 | pending 队列溢出 | pendingInterrupts 满(64)后注入更多 | 丢最老的 + 计数;其余保持 | +| Q1 | 中断队列按级别扫 | 四条中断队列各放一个,入队顺序与级别相反 | 取出顺序 L4→L3→L2→L1;中断耗尽后才是排队任务(FIFO) | +| Q2 | 挂起现场优先于新排队工作 | A 被抢占挂起 + B 为新排队输入 | A(栈顶)先被选中 | +| Q3 | 栈顶 vs 中断队头 | 栈顶 L3 + 队头 L2 / 栈顶 L3 + 队头 L4 / 栈顶为排队任务 + 队头 L1 | 分别取 栈顶 / 队头 / 队头 | +| Q4 | 就绪队列背压 | readyQueue 满后注入排队输入 | 发送方阻塞 + 计数 +1;不静默丢弃 | +| Q5 | 中断队列溢出 | 中断队列合计满后注入更多 | 丢**最低级别里最老**的一条 + 计数;其余保持 | +| **Q6** | **immediate 最优先** | `immediate` 非空且中断队列里有更高级别 | 取 `immediate`(抢占必须立即生效) | ### 11.5 深度、饥饿与并发 | 编号 | 测试点 | 方式 | 预期结果 | |---|---|---|---| -| D1T | 下潜深度上限 | 连续注入 6 个逐级更高的中断 | 中断栈深度 ≤ 4;超出部分在 `pendingInterrupts`;`depthRejections` 计数正确 | -| G1 | 饥饿防护(抢占提升) | 对同一 L1 任务连续抢占 5 次(同级/高级交替) | `effectiveLevel` 提升至 `min(4, 1+2)=3`;第 3 次后不再被 L1/L2 抢占 | -| G2 | 冷却生效 | 同一任务刚被抢占后立刻再注入同级中断 | 冷却期内不抢占,请求入 pending | +| D1T | 下潜深度上界(结构推论) | 挂起 3 帧后继续注入;再挂起到 4 帧 | 3 帧时 `canSuspend()==true`;4 帧(全链:排队+L1+L2+L3,L4 运行中)时为 `false` | +| G1 | 饥饿防护(抢占提升) | 对同一 **L1 中断**连续抢占 5 次(同级/高级交替) | `effectiveLevel` 提升至 `min(4, 1+2)=3`;第 3 次后不再被 L1/L2 抢占 | +| G2 | 冷却生效 | 同一中断刚被抢占后立刻再注入同级中断 | 冷却期内不抢占,请求入中断队列 | +| **G3** | **提升也必须只在中断间生效** | 排队任务被连续抢占 | 排队任务有效级恒 0(不被提升;它按定义可被任何中断打断) | | K1 | panic 隔离 | 假工具 panic | 只有该任务变 `failed`;调度器存活;后续任务正常执行 | | K2 | 竞态检查 | 全部调度用例加 `-race` | 无数据竞争报告 | | O1 | 快照一致性 | 在任意 step 边界调 `Dump()` | 返回的 `running/ready/pending/suspend` 三集合互不重叠且总数守恒 | @@ -444,7 +501,12 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st |---|---|---| | **D1** | 中断任务的上下文 | **方案 B(已定)**:中断从上一个任务之前的完整状态开始;恢复时把被挂起任务的现场加载回中断之上 | | **D2** | 阻塞 step 处置:v1 全部声明为临界区(调度器可被阻塞)还是引入异步 step | **v1 = 临界区**;异步 step 留到 v2 | -| **D3** | `pendingInterrupts` 与 `readyQueue` 是否合一 | **保持分离**(中断请求带 interrupt 语义,取出时以中断语义启动);但**共用同一个排序键** | +| **D3** | 中断队列与排队队列是否合一 | **完全分离**:中断按级别分四条队列(L4→L1 扫描),排队队列纯 FIFO,两者不共用比较器 | +| **D7** | 任务类别怎么定 | **由注入 API 决定**(`InjectInterrupt*` = 中断;`InjectText*`/`InjectInputSync*`/自循环 = 排队),**不按通道名推断** | +| **D8** | L4 归谁 | **内核独占**。唯一入口 `(*Agent).raiseKernelInterrupt`(panic / selfip);`clampPluginLevel` 把插件声明夹到 L3 | +| **D9** | L1–L3 归谁 | **插件在 `InjectOptions.Priority` 里声明**(纯追加字段);空/非法降级到 L1 | +| **D10** | 抢占者进入队列还是立即运行 | **立即运行**(`immediate` 槽)。这消除“抢占者与被挂起者同级”的比较,删除了早期的“同级 pending 优先”补丁 | +| **D11** | 中断栈帧数上界 | **结构推论 = 4**(排队 L0 + I1 + I2 + I3 挂起,I4 运行中),不是配置项;超限只计防御性计数 | | **D4** | readyQueue 满时:阻塞发送方 or 返回错误 | **阻塞发送方 + 计数**(与现状一致,避免丢用户输入) | | **D5** | 饥饿防护:抢占计数提升 or 时间老化 | **抢占计数提升**(确定性、易测);时间老化留待需要时 | | **D6** | 主循环 `max_tool_turns` 是否在本特性一并落地 | **是**(审查 P0,且调度器需要"任务可终止"这一前提) | @@ -454,16 +516,17 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st ## 13. 与发布纪律的关系 - 本特性在 `feature/input-semantics` 上开发,完成后合回 `main`,**不碰 `release/v1.2.x`**。 -- **公开 SDK 冻结**:v1 不改 `third_party/homeagent-sdk/sdk/`。验收命令: - ```bash - git diff main -- third_party/homeagent-sdk/sdk/ | wc -l # 必须为 0 - ``` -- 若 v2 需要 `ChannelDef.Priority`(纯追加),需: +- **公开 SDK 在本特性上有意新增**(feature 分支不受 rel 分支的接口冻结约束): + `sdk.InjectOptions.Priority` 与 `sdk.PriorityL1/L2/L3`。这是为了让插件能声明 + 自己中断的级别(§3.2)。 +- **追加是唯一的形态**:不改既有字段、不改签名、不改语义;`Priority` 的零值 + 等价于旧行为(L1)。 +- 合回 `main` 前需完成的发布动作: 1. 同步更新 `docs/zh/plugin-interface-matrix.md`; 2. 与 SDK 仓协同升 SDK 中版本; - 3. 遵守"只增不减、签名不改"边界。 - ---- + 3. 遵守“只增不减、签名不改”边界。 +- 内核侧接口(`internal/agent/io`、proc 桥的 `injectParams`/`injectMediaParams`) + 同步追加 `priority`,与公开 SDK 字段一一对应。 ## 14. 实现里程碑(逐个实现,每个 = 一个可独立验收的提交) @@ -474,7 +537,7 @@ v1 采纳:**`S_TOOL_EXEC` / ONNX / CAS 属于临界区,调度器在这些 st | **M2** | 调度器骨架:单 `schedulerLoop` + `readyQueue`,取代 `eventLoop` 的输入处理;无优先级(全部 L1,纯 FIFO) | Q1/Q4 通过;integration 测试通过 | | **M3a** | **前置重构(本次拆分引入)**:把一轮对话的所有权从 `processInput` 移到调度器——帧覆盖 `prepare → step… → finish`;同时移除 `process()` 整轮持有的 `a.mu`(挂起不能持锁) | 既有全部 agent 测试 + 既有 e2e 通过(行为等价);`-race` 干净 | | **M3b** | `interruptLoop` 重写 + 四级优先级 + 严格大于抢占 + 中断栈 LIFO;只支持 `S_LLM` 抢占 | P1–P4、R1、R5、K1–K2 通过;嵌套 LIFO 判据通过 | -| **M4** | 临界区 + `S_TOOL_EXEC` 声明 + `pendingInterrupts` + 深度上限 | P5–P6、D1T、Q3、Q5 通过 | +| **M4** | 临界区 + `S_TOOL_EXEC` 声明 + 中断队列 + 深度上界(**后经模型更正重做,见下**) | P5–P6、D1T、Q3、Q5 通过 | | **M5** | 饥饿防护(抢占计数提升 + 冷却) | G1–G2 通过 | | **M6** | 任务级 `responseCh` + 断链点统一为终态事件 | X1–X4 通过;`cli`/`clawhub` 不再挂起 | | **M7** | 可观测性(`Dump()`/事件/状态页)+ 既有回归 | O1–O2、E1–E3 通过;`go test -race ./internal/agent/... ./internal/plugin/...` 全绿 | @@ -500,13 +563,26 @@ go test -race -count=1 ./internal/agent/... ./internal/plugin/... ./internal/sdk | M6 | `4e4e0ad` | ✅ 新增 `task_terminal_test.go` 3 项 | | M7 | `f11de37` | ✅ 新增 `scheduler_e2e_test.go` 3 项(压力/可观测/端到端) | +#### 模型更正后的重构(2026-09-13,同一特性分支) + +用户逐条澄清后重做调度核心(**行为有意的语义变化**,非等价重构): + +| 项 | 内容 | 验收 | +|---|---|---| +| 类别化 | `TaskClass{queued,interrupt}`;类别由注入 API 决定;`newInputTask`/`newSelfTask` 为 queued,`newInterruptTask` 为 interrupt | `scheduler_kernel_test.go` P7/P8 | +| 级别归位 | `Level` 语义改为“中断级别”;`taskLevel()`(按通道名推断)删除,改为 `interruptLevel(evt)` 读 `payload["priority"]` | P9、Q1 | +| L4 内核独占 | `raiseKernelInterrupt`(panic/selfip);`requestKernelPreempt` 不夹取;panic 报告为 L4 且带递归保护 | P10、`TestKernel_PanicRaisesL4Interrupt` | +| 选择结构 | `immediate` + 四条中断队列 + 排队 FIFO + 中断栈;删除统一比较器 `pickTaskIndex`/`taskBefore` 与“同级 pending 优先”补丁 | Q1–Q3、Q6 | +| 栈上界 | `maxSuspendDepth`(配置语义)→ `maxInterruptFrames = int(LevelCritical)`(结构推论);删除“超限转 pending”降级 | D1T | +| 公开 SDK | `InjectOptions.Priority` + `PriorityL1/L2/L3`;io/proc 桥/插件模板同步透传;`example/qq` 声明 L1 | `go test ./...` 全绿 | + 实现期与设计的差异(均已回写本文档): 1. **M3 拆为 M3a/M3b**:真正挂起要求帧跨 `prepare→run→finish`,否则 `processInput` 会在挂起返回后继续提交。 2. **`a.mu` 整体移除**:它原本只包住整轮 `process()`(同一 goroutine), 移除后所有任务状态由 schedulerLoop 独占(不变量 I2/I3 可落地)。 -3. **`interceptCh` 被删除**:M3b 起中断一律走 `pendingInterrupts`,旧的 +3. **`interceptCh` 被删除**:M3b 起中断一律走中断队列(当时叫 `pendingInterrupts`),旧的 “同行注入 + 三处 drain + 批次放弃” 已无写入者,属死代码(M4 清理)。 4. **v1 未做 M0 的伪时钟**:所有抢占测试用“单次调用阻塞到 ctx 取消”的 provider 达到确定性,无需注入时钟。时序型判据(老化式提升)留待需要时。 @@ -522,5 +598,10 @@ go test -race -count=1 ./internal/agent/... ./internal/plugin/... ./internal/sdk 3. 多 agent 并行调度。 4. 与 `plan.md` §13.7 的 `RuntimeManager + 分组 worker` 合并(本设计是其前置)。 -> 已删除:“`ChannelDef.Priority` / `InjectOptions.Priority` 进入公开 SDK”—— -> 优先级是内核内部属性(§3.2),不应由插件声明。 +> **已更正**:早期稿写“`InjectOptions.Priority` 进入公开 SDK 已被删除”, +> 前提是“优先级是内核内部属性、不应由插件声明”。用户澄清后该前提被推翻: +> **L1–L3 就是给插件声明使用的**,只有 L4 归内核独占(panic / selfip)。 +> 因此 `InjectOptions.Priority` 已落地(§3.2/§13)。 +> +> 仍**不做**的是“运维可调的策略表”(`core.agent.priority.`)—— +> 那是把调度内部属性外化成配置,与“由调用方声明自己那件事有多不能等”不同。 diff --git a/internal/agent/core/eventloop.go b/internal/agent/core/eventloop.go index fd1b5f3..e436a68 100644 --- a/internal/agent/core/eventloop.go +++ b/internal/agent/core/eventloop.go @@ -50,10 +50,13 @@ func (a *Agent) interceptLoop() { clone.Payload["interrupt_source"] = evt.Source clone.Payload["interrupt_channel"] = evt.OutputChannel - // 决策交给调度器:requestPreempt 总是登记中断(进 pendingInterrupts, + // 决策交给调度器:requestPreempt 总会登记中断(进中断队列或 immediate, // 因而不会丢),仅当它会真抢占时才告诉我“该取消可取消的步骤”。 - // 本 goroutine 不碰任何帧——只写 pendingInterrupts 与让位信号。 - level := a.taskLevel(evt.Source, evt.OutputChannel) + // 本 goroutine 不碰任何帧——只写中断队列与让位信号。 + // + // 级别由插件声明(InjectOptions.Priority → payload["priority"],L1..L3); + // 未声明一律 L1。L4 只能由内核的 raiseKernelInterrupt 产生。 + level := interruptLevel(evt) if a.sched.requestPreempt(clone, level) { a.cancelCurrentLLM() } diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index cbab179..7ec31ba 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -1,23 +1,36 @@ package core -// 输入调度器(M2:骨架)。 +// 输入调度器:四级中断优先级 · 可抢占 · 现场保存/恢复。 // // 设计依据 docs/zh/input-scheduler-design.md。 // -// M2 只建立结构,不引入抢占: -// - 显式的 readyQueue 与 Task 抽象(取代 eventLoop 里隐式的 channel 排队); -// - 统一的排序键 (-Level, EnqueuedAt, ID)(设计文档 §4.1); -// - 原子快照 DumpScheduler() 与计数(可观测性); -// - **每任务 panic 隔离**:panic 只使该任务失败,调度器本身存活(不变量 I6)。 +// # 模型(两类别 + 四级) // -// M2 全部任务都是 LevelBackground(默认级),因此排序结果等价于 FIFO—— -// 与改造前的 channel 语义逐条一致。抢占、中断栈、pendingInterrupts、 -// 任务级回执在 M3–M6 加入。 +// 类别由**用哪个注入 API**决定,与通道名无关: +// - TaskInterrupt:InjectInterrupt* 注入。带级别 L1..L4,可抢占, +// 可被更高级中断打断(被打断的现场压入**中断栈**)。 +// - TaskQueued:InjectText*/InjectInputSync* 与内核自循环。**无级别**, +// 用于“不需及时处理”的场景,可被**任何**中断打断。 // -// 并发模型(不变量 I2):readyQueue/running/stats 只由 schedulerLoop 写, -// 外部只读——读取一律经 DumpScheduler() 加锁取快照。 +// 级别只属于中断: +// - L1..L3 由插件在 InjectOptions.Priority 里声明(见 clampPluginLevel); +// - L4 由内核独占,只能经 raiseKernelInterrupt 产生(panic / selfip)。 +// +// # 选择顺序 +// +// 1. immediate —— 刚抢占成功的中断(抢占必须立即生效) +// 2. 中断队列 L4→L1(同级 FIFO) +// 3. 中断栈顶(与 2 的队头比级别,取高者;栈顶无级别时中断必胜) +// 4. 排队队列(FIFO) +// +// # 并发模型(不变量 I2) +// +// queue/running/栈/stats 只由 schedulerLoop 与调度 goroutine 写; +// interruptLoop 只写中断登记与让位信号,**从不碰帧**。外部读取一律经 +// DumpScheduler() 加锁取快照。 import ( + "fmt" "log" "runtime/debug" "sync" @@ -29,27 +42,45 @@ import ( sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" ) -// Level 是任务优先级,由内核预定义四级(设计文档 §3.1)。 +// Level 是**中断**的优先级,由内核预定义四级。 // -// 取值域刻意只有四档:不引入任意整数,避免"9 级比 4 级大但没人知道怎么排"。 +// 语义:它衡量“这项工作有多不能等”,与具体通道名无关。 +// 插件在中断注入时通过 InjectOptions.Priority 声明 L1..L3; +// **L4 由内核独占**(panic、内核事件 selfip),插件声明 L4 会被夹到 L3。 +// +// 排队输入(InjectText* / InjectInputSync*)**没有级别**:它们本就是 +// “不需及时处理”的那一类,可被任何中断打断(见 TaskClass)。 type Level int const ( - // LevelBackground 后台维护:心跳蒸馏/归档/合并/复审、子任务、consolidation。 + // LevelBackground L1:完全可等。例:QQ/微信这类异步消息、批量通知。 LevelBackground Level = 1 - // LevelMessage 异步消息:QQ/微信等入站消息、插件通知。 + // LevelMessage L2:一般提醒。例:插件希望尽快看到、但不紧急的提示。 LevelMessage Level = 2 - // LevelInteractive 人机交互:用户在 CLI/WebUI 的直接对话。 + // LevelInteractive L3:需及时处理。例:时钟/定时器到达、终端输出、交互输入。 LevelInteractive Level = 3 - // LevelCritical 紧急打断:显式打断、系统告警、安全类中断。 + // LevelCritical L4:**内核独占**。panic 中断、内核事件中断(selfip)。 + // 插件不得声明此级。 LevelCritical Level = 4 ) -// DefaultLevel 是未显式声明时的优先级。 +// DefaultLevel 是未显式声明时的中断级别。 // // 取最低级是刻意的:**显式才是特权**,新插件不会默认拿到抢占权。 const DefaultLevel = LevelBackground +// clampPluginLevel 把插件声明的级别夹到允许范围(L1..L3)。 +// L4 是内核的调度内部属性,不接受外部越权。 +func clampPluginLevel(l Level) Level { + if l < LevelBackground { + return DefaultLevel + } + if l > LevelInteractive { + return LevelInteractive + } + return l +} + func (l Level) String() string { switch l { case LevelBackground: @@ -67,10 +98,37 @@ func (l Level) String() string { // ParseLevel 已删除。 // -// 为何不保留:优先级是**内核内部属性**,不是配置项——内核预定义四级 -// (L1 后台 / L2 消息 / L3 交互 / L4 紧急),由内核按内部规则为任务与中断定级。 -// 曾一度做成 `core.agent.priority.` 这种“策略表 + 字符串解析”, -// 那等于把内核的内部属性外化成运维配置,与设计意图相反。 +// 为何不保留:优先级是**内核内部属性**,不是配置项。 +// 曾一度做成 `core.agent.priority.`(配置中心可见), +// 那等于把内核的调度内部属性外化成运维配置,与设计意图相反。 +// +// 现在级别的来源只有两个(见 Task/Level 注释): +// - 插件在中断注入时声明(InjectOptions.Priority,L1..L3); +// - 内核内部产生 L4(panic / selfip)。 + +// TaskClass 是任务的两大类别——**由“用哪个注入 API”决定,与通道名无关**。 +// +// 这是模型的核心区分: +// - InjectInterrupt* → TaskInterrupt:带级别,可抢占,可被更高级中断打断(→ 中断栈) +// - InjectText* / InjectInputSync* / 内核自循环 → TaskQueued:无级别, +// 可被**任何**中断打断(“用于不需要及时处理的场景”) +type TaskClass int + +const ( + TaskQueued TaskClass = iota + TaskInterrupt +) + +func (c TaskClass) String() string { + switch c { + case TaskQueued: + return "queued" + case TaskInterrupt: + return "interrupt" + default: + return "unknown" + } +} // TaskKind 区分任务来源。 type TaskKind int @@ -94,11 +152,10 @@ func (k TaskKind) String() string { } // Task 是调度器的最小单位。 -// -// M2 只承载"一份待处理的输入";M3 起把 TaskFrame(现场)挂上来, -// 使其成为可挂起/可恢复的执行单元。 type Task struct { - ID uint64 + ID uint64 + Class TaskClass + // Level 仅对 TaskInterrupt 有意义;TaskQueued 恒为 0(无级别)。 Level Level Kind TaskKind EnqueuedAt time.Time @@ -120,11 +177,17 @@ const preemptPromotionCap = 2 // 避免高优先级流把同一任务反复打断到永不完结。 const preemptCooldown = 2 * time.Second -// effectiveLevel 返回任务的**有效**优先级(设计文档 §9 饥饿防护)。 +// effectiveLevel 返回任务的**有效**级别。 // -// 被抢占越多的任务越“值钱”,从而逐步追上抢占它的流;封顶 L4, -// 因此它永远不会反过来抢占真正的紧急输入。 +// 排队输入恒为 0(无级别):任何中断(≥ L1)都大于它——这正好实现 +// “排队输入可被任何中断打断”。 +// +// 中断则叠加饥饿防护:被抢占越多的中断越“值钱”,逐步追上抢占它的流; +// 封顶 L4,因此它永远不会反过来抢占内核紧急中断。 func effectiveLevel(t *Task) Level { + if t.Class != TaskInterrupt { + return 0 + } p := t.PreemptCount if p > preemptPromotionCap { p = preemptPromotionCap @@ -136,6 +199,21 @@ func effectiveLevel(t *Task) Level { return l } +// canPreempt 是唯一的抢占判据。 +// +// 由于 effectiveLevel(排队)=0,这一个比较同时覆盖两条规则: +// - running 是排队任务 → 任何中断(≥L1)都能抢占; +// - running 是中断 Li → 只有 Lj > Li 的中断能抢占(严格大于)。 +func canPreempt(incoming, running *Task) bool { + if incoming == nil || running == nil { + return false + } + if incoming.Class != TaskInterrupt { + return false // 排队输入从不抢占 + } + return effectiveLevel(incoming) > effectiveLevel(running) +} + // SchedulerStats 是调度器的累计计数(可观测性,设计文档 §11 O2)。 type SchedulerStats struct { Enqueued uint64 @@ -149,13 +227,20 @@ type SchedulerStats struct { // SchedulerSnapshot 是调度器的原子快照。 type SchedulerSnapshot struct { - Running *Task - Queue []*Task + Running *Task + // Queue 是排队输入队列(无级别,FIFO)。 + Queue []*Task + // InterruptQueues[level] 是四条中断队列(下标 1..4,同级 FIFO)。 + InterruptQueues [5][]*Task + // Immediate 是刚抢占成功、将在下一个安全点立即运行的中断(最多一个)。 + Immediate *Task + // PendingInterrupts = 四条中断队列 + Immediate(对外的待处理中断总数视图)。 PendingInterrupts []*Task // SuspendStack:中断栈(含嵌套抢占的多个现场),**栈顶**优先恢复。 - SuspendStack []*suspendedTask - Stats SchedulerStats - MaxSuspendDepth int + SuspendStack []*suspendedTask + Stats SchedulerStats + // MaxInterruptFrames 是中断栈帧数的结构上界(= 中断级数,不是配置项)。 + MaxInterruptFrames int } // schedulerStatus 把快照转成对外的状态 DTO(不暴露帧内容)。 @@ -168,7 +253,7 @@ func (a *Agent) schedulerStatus() sdk.SchedulerStatus { ReadyQueueDepth: len(snap.Queue), PendingInterrupts: len(snap.PendingInterrupts), SuspendStack: len(snap.SuspendStack), - MaxSuspendDepth: snap.MaxSuspendDepth, + MaxSuspendDepth: snap.MaxInterruptFrames, Enqueued: snap.Stats.Enqueued, Executed: snap.Stats.Executed, Rejected: snap.Stats.Rejected, @@ -192,9 +277,14 @@ type scheduler struct { stats SchedulerStats maxQueue int - // pendingInterrupts:因优先级不足(或运行任务在临界区)而未立即抢占的中断请求。 - // 与 readyQueue 分离:取出时以中断语义启动(设计文档 D3)。 - pendingInterrupts []*Task + // interruptQueues[level]:四条**中断队列**(level 1..4),同级 FIFO。 + // 未能立即抢占的中断(级别不足,或运行任务在临界区)按级别入队, + // nextRef 从 L4 到 L1 依次扫描。 + interruptQueues [5][]*Task + // immediate:刚抢占成功的中断。抢占必须**立即生效**,所以它不经队列, + // 在下一个安全点直接运行。这也消除了“抢占者与被抢占者同级”的比较问题—— + // 抢占者根本不需要和栈顶比。 + immediate *Task // suspendStack:**中断栈**。被抢占后保存现场的任务压栈(LIFO), // 用于“中断被中断”的嵌套场景:只有**栈顶**参与恢复选择,栈内不做优先级重排。 suspendStack []*suspendedTask @@ -208,8 +298,12 @@ type scheduler struct { // wake 用于把空闲的调度器叫醒:pendingInterrupts 不是 channel, // 没有这个信号时“空闲时到达的中断”会一直等下一次输入(设计 §5.1 ③)。 wake chan struct{} - // maxSuspendDepth:中断栈深度上限(设计文档 §6.3,默认 4)。 - maxSuspendDepth int + // maxInterruptFrames:中断栈帧数的**结构上界**,不是配置项。 + // + // 链条 = 排队(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4 运行中), + // 被挂起 4 帧;L4 之上没有更高级别,链到此为止。超限只可能是内核 bug, + // 因此这里只做防御性计数,**不降级、不丢弃帧**。 + maxInterruptFrames int } // suspendedTask 是一个被抢占任务的现场。 @@ -224,7 +318,8 @@ type nextSelection int const ( nextNone nextSelection = iota nextReady - nextPending + nextInterrupt + nextImmediate nextSuspended ) @@ -232,7 +327,11 @@ func newScheduler(maxQueue int) *scheduler { if maxQueue <= 0 { maxQueue = 256 } - return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4, wake: make(chan struct{}, 1)} + return &scheduler{ + maxQueue: maxQueue, + maxInterruptFrames: int(LevelCritical), // 结构推论:= 中断级数 + wake: make(chan struct{}, 1), + } } // signalWake 非阻塞地唤醒调度器。 @@ -249,7 +348,7 @@ func (s *scheduler) setCritical(v bool) { s.critical.Store(v) } // inCritical 报告运行任务是否在不可抢占临界区。 func (s *scheduler) inCritical() bool { return s.critical.Load() } -// hasRoom 报告就绪队列是否还能接收任务。泵入侧据此节流: +// hasRoom 报告排队队列是否还能接收任务。泵入侧据此节流: // 队列满则停止从 channel 取,让背压落回 channel 本身。 func (s *scheduler) hasRoom() bool { s.mu.Lock() @@ -257,7 +356,16 @@ func (s *scheduler) hasRoom() bool { return len(s.queue) < s.maxQueue } -// enqueue 入队;队列满返回 false(调用方负责计数)。 +// allocateIDLocked 分配任务 ID 与入队时刻(调用方持锁)。 +func (s *scheduler) allocateIDLocked(t *Task) { + s.seq++ + t.ID = s.seq + if t.EnqueuedAt.IsZero() { + t.EnqueuedAt = time.Now() + } +} + +// enqueue 把一个**排队输入**入队;队列满返回 false(调用方负责计数)。 func (s *scheduler) enqueue(t *Task) bool { s.mu.Lock() defer s.mu.Unlock() @@ -265,11 +373,7 @@ func (s *scheduler) enqueue(t *Task) bool { s.stats.Rejected++ return false } - s.seq++ - t.ID = s.seq - if t.EnqueuedAt.IsZero() { - t.EnqueuedAt = time.Now() - } + s.allocateIDLocked(t) s.stats.Enqueued++ s.queue = append(s.queue, t) return true @@ -283,74 +387,94 @@ func (s *scheduler) next() *Task { return t } -// nextRef 从三个集合中按统一排序键取出下一个任务。 +// nextRef 选出下一个任务。优先顺序: // -// 设计文档 §4.1:高有效级先;同级先到先服务。挂起任务保留其**原始**入队时刻, -// 因此同级时天然倾向“先把旧任务做完”,抑制饥饿。 +// 1. immediate —— 刚抢占成功的中断(抢占必须立即生效) +// 2. 中断队列 L4→L1(同级 FIFO) +// 3. 中断栈顶(与 2 比级别取高者;栈顶是排队任务时视为最低) +// 4. 排队队列(FIFO) func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { s.mu.Lock() defer s.mu.Unlock() - var bestTask *Task - var bestFrame *TaskFrame - bestKind := nextNone - consider := func(t *Task, k nextSelection, fr *TaskFrame) { - 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 - } - } - for _, t := range s.queue { - consider(t, nextReady, nil) - } - for _, t := range s.pendingInterrupts { - consider(t, nextPending, nil) - } - // 中断栈:只比**栈顶**(严格 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 + if s.immediate != nil { + t := s.immediate + s.immediate = nil + s.running = t + return t, nil, nextImmediate } - switch bestKind { - case nextReady: - s.queue = removeTask(s.queue, bestTask) - case nextPending: - s.pendingInterrupts = removeTask(s.pendingInterrupts, bestTask) - case nextSuspended: - // 只有栈顶可能被选中,故弹出即截断末位。 - s.suspendStack = s.suspendStack[:len(s.suspendStack)-1] + qTask, qLevel := s.highestInterruptLocked() + + // 中断栈:只比**栈顶**(严格 LIFO)。栈内不做优先级重排—— + // 嵌套抢占天然使栈自底向上级别递增,且“后被打断的先恢复”才是栈语义。 + if n := len(s.suspendStack); n > 0 { + top := s.suspendStack[n-1] + // 栈顶 vs 最高级待处理中断:取高者(持平归栈顶,维持 LIFO 与公平)。 + if qTask == nil || effectiveLevel(top.Task) >= qLevel { + s.suspendStack = s.suspendStack[:n-1] + s.stats.Resumed++ + s.running = top.Task + return top.Task, top.Frame, nextSuspended + } } - s.running = bestTask - return bestTask, bestFrame, bestKind + + if qTask != nil { + s.popInterruptLocked(qLevel) + s.running = qTask + return qTask, nil, nextInterrupt + } + + if len(s.queue) > 0 { + t := s.queue[0] + s.queue = s.queue[1:] + s.running = t + return t, nil, nextReady + } + return nil, nil, nextNone +} + +// highestInterruptLocked 返回当前最高级非空中断队列的队头及其级别。 +func (s *scheduler) highestInterruptLocked() (*Task, Level) { + for lv := LevelCritical; lv >= LevelBackground; lv-- { + if q := s.interruptQueues[lv]; len(q) > 0 { + return q[0], lv + } + } + return nil, 0 +} + +// popInterruptLocked 弹出某级别中断队列的队头(调用方已确认非空)。 +func (s *scheduler) popInterruptLocked(lv Level) { + s.interruptQueues[lv] = s.interruptQueues[lv][1:] +} + +// interruptCountLocked 统计所有待处理中断(含 immediate 槽)。 +func (s *scheduler) interruptCountLocked() int { + n := 0 + for lv := LevelBackground; lv <= LevelCritical; lv++ { + n += len(s.interruptQueues[lv]) + } + if s.immediate != nil { + n++ + } + return n +} + +// setImmediateLocked 登记一个应“立即运行”的抢占者。 +// +// 槽只有一格:若已有抢占者且新的级别更高,旧的降级入队;否则新的入队。 +func (s *scheduler) setImmediateLocked(t *Task) { + if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) { + s.enqueueInterruptLocked(t) + return + } + if s.immediate != nil { + s.enqueueInterruptLocked(s.immediate) + } + s.allocateIDLocked(t) + s.stats.Enqueued++ + s.immediate = t } func removeTask(list []*Task, target *Task) []*Task { @@ -362,52 +486,75 @@ func removeTask(list []*Task, target *Task) []*Task { return list } -// enqueueInterrupt 把一个未立即抢占的中断请求放进 pendingInterrupts。 +// enqueueInterruptLocked 把一个未立即抢占的中断按其级别入队(调用方持锁)。 // // 有界:满了丢**最老**的一条并计数(中断是提示性输入,宁可丢旧保新)。 -func (s *scheduler) enqueueInterrupt(t *Task) { - s.mu.Lock() - defer s.mu.Unlock() - s.seq++ - t.ID = s.seq - if t.EnqueuedAt.IsZero() { - t.EnqueuedAt = time.Now() +func (s *scheduler) enqueueInterruptLocked(t *Task) { + s.allocateIDLocked(t) + if s.interruptCountLocked() >= s.maxQueue { + for lv := LevelBackground; lv <= LevelCritical; lv++ { + if len(s.interruptQueues[lv]) > 0 { + s.popInterruptLocked(lv) + s.stats.Rejected++ + break + } + } } - if len(s.pendingInterrupts) >= s.maxQueue { - s.pendingInterrupts = s.pendingInterrupts[1:] - s.stats.Rejected++ + lv := t.Level + if lv < LevelBackground || lv > LevelCritical { + lv = DefaultLevel } - s.pendingInterrupts = append(s.pendingInterrupts, t) - s.signalWake() + s.interruptQueues[lv] = append(s.interruptQueues[lv], t) + s.stats.Enqueued++ } -// requestPreempt 登记一次中断请求。 +// requestPreempt 登记一次中断请求(class=TaskInterrupt)。 // // 返回 true 表示“应该尝试取消运行任务正在进行的可取消步骤(LLM 流式)”。 // -// 无论能否抢占,中断请求都进 pendingInterrupts——这样即使运行任务在抢占生效前 -// 就正常结束,中断也不会丢(它会被 nextRef 按优先级选出)。 +// 判据是 canPreempt(由优先级级别系统一承担),并受抢占冷却约束: +// - running 是排队任务 → 任何中断都抢占; +// - running 是中断 Li → 仅 Lj > Li 的中断抢占。 // -// 判据用**有效**优先级(饥饿防护),并受抢占冷却约束。 +// 能抢占时把中断放进 immediate(立即生效);否则按其级别入队,等当前任务 +// 结束或下一个安全点再处理——无论哪种,中断都不会丢。 // // 临界区(如记忆整理)内不 arm、不取消:中断只入队,等临界区结束后的安全点处理, // 这是设计 §4.3 的硬要求——那个位置的“不抢占”不能只是不让位,还必须不取消。 func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool { + return s.registerInterrupt(newInterruptTask(evt, clampPluginLevel(level))) +} + +// requestKernelPreempt 是**内核**中断入口(panic / 内核事件 selfip)。 +// +// 级别固定 L4,且**不夹取**——这是 L4 的唯一来源,插件永远够不到。 +func (s *scheduler) requestKernelPreempt(evt *agentIO.InputEvent) bool { + return s.registerInterrupt(newKernelInterruptTask(evt)) +} + +// registerInterrupt 是登记中断的公共实现(任务已带好 Class/Level)。 +func (s *scheduler) registerInterrupt(t *Task) bool { s.mu.Lock() running := s.running critical := s.critical.Load() - canPreempt := false - if !critical && running != nil && level > effectiveLevel(running) { + arm := false + if !critical && canPreempt(t, running) { if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown { - canPreempt = true + arm = true s.preemptArmed = true - s.preemptLevel = level + s.preemptLevel = t.Level + s.setImmediateLocked(t) } } + if !arm { + s.enqueueInterruptLocked(t) + } s.mu.Unlock() - s.enqueueInterrupt(newInterruptTask(evt, level)) - return canPreempt + if !arm { + s.signalWake() + } + return arm } // preemptGrantedFor 报告运行任务是否应在当前安全点让位。 @@ -429,12 +576,12 @@ func (s *scheduler) clearPreempt() { // suspend 保存现场。 // -// 深度上限(设计文档 §6.3):安全点上的 canSuspend 已提前拦下超限情况, -// 此处仅在竞态下兜底计数——绝不丢弃帧(帧丢了会丢副作用记录)。 +// 深度上界是**结构推论**(= 中断级数),不是配置项:安全点上的 canSuspend 已提前 +// 拦下超限情况,此处仅在竞态下兜底计数——绝不丢弃帧(帧丢了会丢副作用记录)。 func (s *scheduler) suspend(t *Task, f *TaskFrame) { s.mu.Lock() defer s.mu.Unlock() - if len(s.suspendStack) >= s.maxSuspendDepth { + if len(s.suspendStack) >= s.maxInterruptFrames { s.stats.Rejected++ } s.suspendStack = append(s.suspendStack, &suspendedTask{Task: t, Frame: f}) @@ -456,7 +603,7 @@ func (s *scheduler) suspend(t *Task, f *TaskFrame) { func (s *scheduler) canSuspend() bool { s.mu.Lock() defer s.mu.Unlock() - return len(s.suspendStack) < s.maxSuspendDepth + return len(s.suspendStack) < s.maxInterruptFrames } // done 标记任务执行结束。 @@ -466,7 +613,7 @@ func (s *scheduler) done(t *Task) { if s.running == t { s.running = nil } - // 任务正常结束:让位信号不再有意义(中断已在 pendingInterrupts 里)。 + // 任务正常结束:让位信号不再有意义(中断已在中断队列/immediate 里)。 s.preemptArmed = false s.preemptLevel = 0 s.stats.Executed++ @@ -475,37 +622,42 @@ func (s *scheduler) done(t *Task) { // currentLevel 返回当前正在执行任务的级别;无 running 时为默认级。 // // 用于在 prepare 段把级别写进帧(抢占比较的基准)。 -func (s *scheduler) currentLevel() Level { - s.mu.Lock() - defer s.mu.Unlock() - if s.running != nil { - return s.running.Level +// interruptLevel 返回一次**中断注入**的级别。 +// +// 级别是“这项工作有多不能等”,由插件在 InjectOptions.Priority 里声明 +// (排队注入没有级别,它们的 TaskClass 是 TaskQueued)。 +// +// 取值域 L1..L3;空/非法一律降到 DefaultLevel(L1)。 +// **L4 不在此处产生**:它由内核独占,经 raiseKernelInterrupt 直接给出 +// (panic / 内核事件 selfip),因此 clampPluginLevel 会把越权声明夹回 L3。 +func interruptLevel(evt *agentIO.InputEvent) Level { + if evt == nil || evt.Payload == nil { + return DefaultLevel } - return DefaultLevel + raw, _ := evt.Payload["priority"].(string) + l, ok := parseInterruptLevel(raw) + if !ok { + return DefaultLevel + } + return clampPluginLevel(l) } -// taskLevel 是内核为任务定级的内部规则。 -// -// ❗优先级是**内核内部属性**,不做成配置项:内核预定义四级,并按内部规则 -// 为任务与中断定级。下方规则只是 v1 的内部缺省值——它决定“谁能让位于谁”, -// 属于内核自己的隐私,不对外暴露为运维可调项。 -// -// 缺省:cli/webui/http 为人机交互(L3),system/_consolidation_ 为后台(L1), -// 其余一律默认级(L1)。 -func (a *Agent) taskLevel(source, channel string) Level { - switch channel { - case "cli", "webui", "http": - return LevelInteractive - case channelConsolidation, "system": - return LevelBackground +// parseInterruptLevel 解析插件声明的级别字符串("L1".."L3")。 +// 只认字面量:拼写错误必须降级成默认级而不是被静默当成别的级别。 +func parseInterruptLevel(s string) (Level, bool) { + switch s { + case "L1", "l1": + return LevelBackground, true + case "L2", "l2": + return LevelMessage, true + case "L3", "l3": + return LevelInteractive, true + case "L4", "l4": + // 内核级:解析出来但会被 clamp 夹到 L3。 + return LevelCritical, true + default: + return 0, false } - switch source { - case "cli", "webui": - return LevelInteractive - case "system": - return LevelBackground - } - return DefaultLevel } // inCriticalSection 报告运行任务是否处于不可抢占区。 @@ -517,43 +669,28 @@ func (a *Agent) inCriticalSection() bool { return a.currentOutputChannel == channelConsolidation } -// pickTaskIndex 返回下一个要执行的任务下标(设计文档 §4.1 的选择函数)。 -// -// 排序键:优先级降序 → 入队时刻升序 → ID 升序。 -// 纯函数:便于对抢占/优先级矩阵做确定性单测。 -func pickTaskIndex(q []*Task) int { - best := 0 - for i := 1; i < len(q); i++ { - if taskBefore(q[i], q[best]) { - best = i - } - } - return best -} - -// taskBefore 报告 x 是否应先于 y 执行(按**有效**优先级)。 -func taskBefore(x, y *Task) bool { - lx, ly := effectiveLevel(x), effectiveLevel(y) - if lx != ly { - return lx > ly - } - if !x.EnqueuedAt.Equal(y.EnqueuedAt) { - return x.EnqueuedAt.Before(y.EnqueuedAt) - } - return x.ID < y.ID -} - +// newInputTask 把一个**排队输入**包装成任务(无级别)。 func newInputTask(evt *agentIO.InputEvent) *Task { - return &Task{Kind: TaskKindInput, Level: DefaultLevel, Event: evt, EnqueuedAt: time.Now()} + return &Task{Class: TaskQueued, Kind: TaskKindInput, Event: evt, EnqueuedAt: time.Now()} } -// newInterruptTask 把一个中断请求包装成任务。 +// newInterruptTask 把一个中断请求包装成任务(带级别)。 func newInterruptTask(evt *agentIO.InputEvent, level Level) *Task { - return &Task{Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()} + return &Task{Class: TaskInterrupt, Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()} } +// newSelfTask 包装内核自循环输入——它是**排队任务**:记忆整理/子任务通知 +// 不需要及时处理,可被任何中断打断。 func newSelfTask(msg selfInputMsg) *Task { - return &Task{Kind: TaskKindSelf, Level: DefaultLevel, Self: msg, EnqueuedAt: time.Now()} + return &Task{Class: TaskQueued, Kind: TaskKindSelf, Self: msg, EnqueuedAt: time.Now()} +} + +// newKernelInterruptTask 构造一个**内核级中断**(L4)。 +// +// 这是 L4 的唯一来源:panic 中断、内核事件中断(selfip)。 +// 插件永远拿不到这个入口——它不经 InjectOptions,也不经 proc 桥。 +func newKernelInterruptTask(evt *agentIO.InputEvent) *Task { + return &Task{Class: TaskInterrupt, Kind: TaskKindInput, Level: LevelCritical, Event: evt, EnqueuedAt: time.Now()} } // DumpScheduler 返回调度器的原子快照(供状态页/测试断言)。 @@ -565,9 +702,16 @@ func (a *Agent) DumpScheduler() SchedulerSnapshot { defer a.sched.mu.Unlock() 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.Immediate = a.sched.immediate + for lv := LevelBackground; lv <= LevelCritical; lv++ { + snap.InterruptQueues[lv] = append(snap.InterruptQueues[lv], a.sched.interruptQueues[lv]...) + snap.PendingInterrupts = append(snap.PendingInterrupts, a.sched.interruptQueues[lv]...) + } + if a.sched.immediate != nil { + snap.PendingInterrupts = append(snap.PendingInterrupts, a.sched.immediate) + } snap.SuspendStack = append(snap.SuspendStack, a.sched.suspendStack...) - snap.MaxSuspendDepth = a.sched.maxSuspendDepth + snap.MaxInterruptFrames = a.sched.maxInterruptFrames return snap } @@ -630,6 +774,46 @@ func (a *Agent) pumpInbox() { } // executeTask 执行一个任务(测试与旧调用方的入口);见 executeNewTask。 +// raiseKernelInterrupt 是 **L4 的唯一入口**:panic 中断与内核事件中断(selfip)。 +// +// 它不经 io.InputChan(那是外部/插件输入),而是直接向调度器登记一条内核中断: +// 级别固定 L4、不夹取、不受插件声明影响。这正是“L4 只有内核持有”的落点。 +// +// 能否抢占由调度器按统一判据决定;若会抢占,则顺手取消可取消的 LLM 流式步骤 +// (与 interceptLoop 对插件中断的处理完全一致)。 +func (a *Agent) raiseKernelInterrupt(source, channel, text string) { + if a.sched == nil { + return + } + evt := &agentIO.InputEvent{ + Source: source, + Type: "interrupt", + OutputChannel: channel, + Payload: map[string]interface{}{ + "content": text, + "interrupt": true, + "interrupt_source": source, + "interrupt_channel": channel, + "kernel": true, + }, + } + if a.sched.requestKernelPreempt(evt) { + a.cancelCurrentLLM() + } +} + +// reportTaskPanic 把一个任务 panic 报告成内核 L4 中断。 +// +// 递归保护是**结构性**的:若 panic 的任务本身就是 L4 内核中断,则不再产生新的 +// L4——否则同一个 panic 会自我放大成中断风暴,与“内核事件”应有的语义相反。 +func (a *Agent) reportTaskPanic(t *Task, r interface{}) { + if t.Class == TaskInterrupt && t.Level >= LevelCritical { + return + } + a.raiseKernelInterrupt("kernel", "kernel", + fmt.Sprintf("内核事件:任务 #%d 发生 panic:%v(该任务已被丢弃,调度器存活)", t.ID, r)) +} + func (a *Agent) executeTask(t *Task) { a.executeNewTask(t) } @@ -647,6 +831,7 @@ func (a *Agent) executeNewTask(t *Task) { if r := recover(); r != nil { log.Printf("[agent] task#%d (%s) panic recovered: %v\n%s", t.ID, t.Level, r, debug.Stack()) + a.reportTaskPanic(t, r) } }() switch t.Kind { @@ -686,6 +871,7 @@ func (a *Agent) resumeTask(t *Task, f *TaskFrame) { if r := recover(); r != nil { log.Printf("[agent] resume task#%d panic recovered: %v\n%s", t.ID, r, debug.Stack()) + a.reportTaskPanic(t, r) a.sched.done(t) } }() diff --git a/internal/agent/core/scheduler_critical_test.go b/internal/agent/core/scheduler_critical_test.go index 28a0e30..0ecf37c 100644 --- a/internal/agent/core/scheduler_critical_test.go +++ b/internal/agent/core/scheduler_critical_test.go @@ -42,7 +42,7 @@ func TestPreempt_DeferredDuringToolExec(t *testing.T) { StageHost: sh, }) - if _, _ = enqueueTask(t, a, LevelBackground, "qq", "低优先级任务"); true { + if _, _ = enqueueQueued(t, a, "qq", "低优先级任务"); true { } lt, _, _ := a.sched.nextRef() done := make(chan struct{}) @@ -57,7 +57,7 @@ func TestPreempt_DeferredDuringToolExec(t *testing.T) { // 工具执行中注入 L4 中断。 intrEvt, _ := textEvent("cli", "紧急打断") intrEvt.Payload["interrupt"] = true - if !a.sched.requestPreempt(intrEvt, LevelCritical) { + if !a.sched.requestKernelPreempt(intrEvt) { t.Fatal("L4 应 arm 让位信号") } // 关键断言:信号已 arm,但任务仍在工具里 —— 绝不能挂起。 @@ -133,7 +133,7 @@ func TestBatch_NotAbandonedWithoutPreemption(t *testing.T) { StageHost: sh, }) - if _, _ = enqueueTask(t, a, LevelBackground, "cli", "跑两个工具"); true { + if _, _ = enqueueQueued(t, a, "cli", "跑两个工具"); true { } tt, _, _ := a.sched.nextRef() a.executeNewTask(tt) diff --git a/internal/agent/core/scheduler_e2e_test.go b/internal/agent/core/scheduler_e2e_test.go index df0c462..efae3fd 100644 --- a/internal/agent/core/scheduler_e2e_test.go +++ b/internal/agent/core/scheduler_e2e_test.go @@ -74,8 +74,12 @@ func TestScheduler_StressMixedLoad(t *testing.T) { for i := 0; i < nInputs; i++ { a.io.InjectInput("cli", "text", map[string]interface{}{"content": fmt.Sprintf("msg-%d", i)}) } + // 中断按 L1/L2/L3 轮转:把“四条中断队列按级别高→低扫描”真正压上, + // 而不只是排空一条队列。 + levels := []string{"L1", "L2", "L3"} for i := 0; i < nInterrupts; i++ { - a.io.InjectInterruptText("qq", "cli", fmt.Sprintf("intr-%d", i)) + a.io.InjectInterruptTextOpts("qq", "cli", fmt.Sprintf("intr-%d", i), + agentIO.InjectOptions{Priority: levels[i%len(levels)]}) } snap := waitQuiescent(t, a, nInputs+nInterrupts, 30*time.Second) @@ -129,7 +133,7 @@ func TestObservability_SchedulerEventsAndStatus(t *testing.T) { intrEvt, _ := textEvent("cli", "紧急") intrEvt.Payload["interrupt"] = true - a.sched.requestPreempt(intrEvt, LevelCritical) + a.sched.requestKernelPreempt(intrEvt) a.cancelCurrentLLM() <-done @@ -185,16 +189,18 @@ func TestE2E_RealLoopPreemption(t *testing.T) { a.Start() defer a.Stop() - // L1:qq 入站消息 → 阻塞在第一次 LLM 调用 - a.io.InjectInput("qq", "text", map[string]interface{}{"content": "低优先级长任务"}) + // 排队输入:qq 入站消息 → 阻塞在第一次 LLM 调用(排队任务无级别) + a.io.InjectInput("qq", "text", map[string]interface{}{"content": "长任务"}) select { case <-sp.entered: case <-time.After(5 * time.Second): t.Fatal("低优先级任务未进入 LLM") } - // L4:cli 紧急打断 → interceptLoop 应取消 LLM、登记抢占 - a.io.InjectInterruptText("cli", "cli", "紧急打断") + // 插件声明的 L3 中断:interceptLoop 应取消 LLM 并登记抢占。 + // 注意它能打断**排队任务**不是因为级别高,而是因为排队任务无级别—— + // 任何中断都大于它。 + a.io.InjectInterruptTextOpts("cli", "cli", "紧急打断", agentIO.InjectOptions{Priority: "L3"}) a.io.InjectInput("cli", "text", map[string]interface{}{"content": "后续常规输入"}) // 排空:中断任务 + 被恢复的原任务 + 后续常规输入 diff --git a/internal/agent/core/scheduler_kernel_test.go b/internal/agent/core/scheduler_kernel_test.go new file mode 100644 index 0000000..ec30dcc --- /dev/null +++ b/internal/agent/core/scheduler_kernel_test.go @@ -0,0 +1,206 @@ +package core + +// L4 的内核独占性 + 两类别抢占规则。 +// +// 模型(用户明确): +// - 类别由**用哪个注入 API** 决定,与通道名无关; +// - L1..L3 由插件在 InjectOptions.Priority 声明; +// - L4 只有内核持有(panic / 内核事件 selfip); +// - 排队输入无级别,可被**任何**中断打断。 + +import ( + "strings" + "testing" + "time" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" +) + +// 插件声明 L4 必须被夹到 L3:L4 是内核的调度内部属性,不接受外部越权。 +func TestKernel_PluginCannotClaimL4(t *testing.T) { + if got := clampPluginLevel(LevelCritical); got != LevelInteractive { + t.Fatalf("插件声明 L4 应被夹到 L3,实际 %v", got) + } + cases := []struct { + declared string + want Level + }{ + {"L1", LevelBackground}, + {"L2", LevelMessage}, + {"L3", LevelInteractive}, + {"l2", LevelMessage}, + {"L4", LevelInteractive}, // 越权 → 夹到 L3 + {"L7", DefaultLevel}, // 未知 → 默认级 + {"", DefaultLevel}, // 未声明 → 默认级 + {"紧急", DefaultLevel}, // 拼写错误 → 默认级(不得被静默当成别的级别) + } + for _, c := range cases { + evt := &agentIO.InputEvent{Payload: map[string]interface{}{}} + if c.declared != "" { + evt.Payload["priority"] = c.declared + } + if got := interruptLevel(evt); got != c.want { + t.Fatalf("声明 %q → 级别 %v,期望 %v", c.declared, got, c.want) + } + } + if got := interruptLevel(nil); got != DefaultLevel { + t.Fatalf("无事件应为默认级,实际 %v", got) + } +} + +// 排队任务可被**任何**中断打断——包括最低的 L1。 +func TestKernel_QueuedTaskIsPreemptedByAnyInterrupt(t *testing.T) { + s := newScheduler(8) + q := newInputTask(&agentIO.InputEvent{Source: "plugin", OutputChannel: "plugin"}) + s.enqueue(q) + if task, _, kind := s.nextRef(); task != q || kind != nextReady { + t.Fatalf("应取到排队任务,kind=%v", kind) + } + + evt, _ := textEvent("qq", "最低级中断") + if !s.registerInterrupt(newInterruptTask(evt, LevelBackground)) { + t.Fatal("L1 中断也必须能打断排队任务(排队任务无级别)") + } + if s.immediate == nil || s.immediate.Level != LevelBackground { + t.Fatalf("抢占者应进 immediate 槽,实际 %+v", s.immediate) + } +} + +// 排队输入从不抢占——它没有级别,也就没有“比谁高”。 +func TestKernel_QueuedInputNeverPreempts(t *testing.T) { + s := newScheduler(8) + if !s.enqueue(newInputTask(&agentIO.InputEvent{Source: "a", OutputChannel: "a"})) { + t.Fatal("入队失败") + } + s.nextRef() // running = 第一个排队任务 + if s.registerInterrupt(newInputTask(&agentIO.InputEvent{Source: "b", OutputChannel: "b"})) { + t.Fatal("排队输入不得抢占任何任务") + } +} + +// 内核 L4 入口不受夹取影响,且能抢占中断(L3)。 +func TestKernel_RequestKernelPreemptUsesL4(t *testing.T) { + s := newScheduler(8) + evt, _ := textEvent("cli", "L3 运行中") + s.registerInterrupt(newInterruptTask(evt, LevelInteractive)) + s.nextRef() // running = L3 中断 + + kevt, _ := textEvent("kernel", "panic 中断") + if !s.requestKernelPreempt(kevt) { + t.Fatal("内核 L4 应能抢占 L3 中断") + } + if s.immediate == nil || s.immediate.Level != LevelCritical { + t.Fatalf("内核中断必须是 L4,实际 %+v", s.immediate) + } +} + +// 任务 panic → 内核 L4 中断(panic 是 L4 的来源之一)。 +func TestKernel_PanicRaisesL4Interrupt(t *testing.T) { + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "已收到内核事件"}}} + a := newPreemptAgent(t, sp) + + // Event 为 nil:handleInput 解引用即 panic。 + bad := newInputTask(nil) + if !a.sched.enqueue(bad) { + t.Fatal("入队失败") + } + task, _, _ := a.sched.nextRef() + a.executeTask(task) // panic 被隔离 + + snap := a.DumpScheduler() + if snap.Immediate == nil { + t.Fatal("panic 必须产生一条内核 L4 中断") + } + if snap.Immediate.Level != LevelCritical { + t.Fatalf("panic 中断级别=%v,期望 L4", snap.Immediate.Level) + } + text, _ := snap.Immediate.Event.Payload["content"].(string) + if !strings.Contains(text, "panic") { + t.Fatalf("panic 中断应说明发生了什么,实际 %q", text) + } + if snap.Immediate.Event.Payload["kernel"] != true { + t.Fatal("内核中断必须带 kernel 标记,便于与插件中断区分") + } +} + +// 递归保护是结构性的:L4 内核中断自己 panic 时,不再产生新的 L4。 +func TestKernel_PanicInsideL4DoesNotRecurse(t *testing.T) { + sp := &scriptProvider{} + a := newPreemptAgent(t, sp) + + evt, _ := textEvent("kernel", "内核事件") + l4 := newKernelInterruptTask(evt) + a.sched.immediate = l4 + task, _, _ := a.sched.nextRef() + if task != l4 { + t.Fatal("应取到 L4 内核中断") + } + a.executeTask(&Task{ID: task.ID, Class: TaskInterrupt, Level: LevelCritical, Kind: TaskKindInput, Event: nil}) + + snap := a.DumpScheduler() + if snap.Immediate != nil || len(snap.PendingInterrupts) != 0 { + t.Fatalf("L4 自身 panic 不得再产生中断(否则自我放大),实际 immediate=%+v pending=%d", + snap.Immediate, len(snap.PendingInterrupts)) + } +} + +// 中断栈的 4 帧上界是**结构推论**:排队(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4 运行中)。 +func TestKernel_StackBoundIsFullChain(t *testing.T) { + s := newScheduler(16) + frame := func() *TaskFrame { return &TaskFrame{} } + chain := []struct { + class TaskClass + lv Level + }{ + {TaskQueued, 0}, + {TaskInterrupt, LevelBackground}, + {TaskInterrupt, LevelMessage}, + {TaskInterrupt, LevelInteractive}, + } + for i := 0; i < len(chain)-1; i++ { + s.suspend(&Task{ID: uint64(i + 1), Class: chain[i].class, Level: chain[i].lv}, frame()) + } + if !s.canSuspend() { + t.Fatal("3 帧挂起时仍应容得下 L3(第 4 级)继续下潜") + } + s.suspend(&Task{ID: 4, Class: TaskInterrupt, Level: LevelInteractive}, frame()) + if s.canSuspend() { + t.Fatal("4 帧挂起 = 全链挂起(L4 运行中),不应再有下潜余量") + } +} + +// 端到端:插件声明 Priority → io.applyInjectOpts → payload → interruptLevel → 任务级别。 +// 这条链路断在任何一环,插件声明的级别都会静默失效(降级到 L1)。 +func TestKernel_PriorityFlowsThroughIOLayer(t *testing.T) { + ioM := agentIO.NewIOManager() + ioM.InjectInterruptTextOpts("qq", "cli", "通知", agentIO.InjectOptions{Priority: "L3"}) + + select { + case evt := <-ioM.InputInterruptChan(): + if got := interruptLevel(evt); got != LevelInteractive { + t.Fatalf("经 io 层后的级别=%v,期望 L3(payload=%v)", got, evt.Payload) + } + task := newInterruptTask(evt, interruptLevel(evt)) + if task.Class != TaskInterrupt || task.Level != LevelInteractive { + t.Fatalf("中断任务类别/级别=%v/%v,期望 interrupt/L3", task.Class, task.Level) + } + case <-time.After(2 * time.Second): + t.Fatal("中断未到达 interruptCh") + } + + // 排队路径带 priority 也必须无效:排队输入没有级别。 + ioM.InjectTextOpts("qq", "cli", "普通输入", agentIO.InjectOptions{Priority: "L3"}) + select { + case evt := <-ioM.InputChan(): + task := newInputTask(evt) + if task.Class != TaskQueued || task.Level != 0 { + t.Fatalf("排队任务类别/级别=%v/%v,期望 queued/无级别", task.Class, task.Level) + } + if effectiveLevel(task) != 0 { + t.Fatalf("排队任务有效级=%v,期望 0", effectiveLevel(task)) + } + case <-time.After(2 * time.Second): + t.Fatal("排队输入未到达 inputCh") + } +} diff --git a/internal/agent/core/scheduler_preempt_test.go b/internal/agent/core/scheduler_preempt_test.go index 9531b1f..c39f1a1 100644 --- a/internal/agent/core/scheduler_preempt_test.go +++ b/internal/agent/core/scheduler_preempt_test.go @@ -75,23 +75,44 @@ func newPreemptAgent(t *testing.T, sp agentAPI.Provider) *Agent { }) } -func enqueueTask(t *testing.T, a *Agent, level Level, source, content string) (*Task, *agentIO.InputEvent) { +// enqueueQueued 入队一个**排队任务**(无级别)——对应 InjectText*/InjectInputSync*。 +func enqueueQueued(t *testing.T, a *Agent, source, content string) (*Task, *agentIO.InputEvent) { t.Helper() evt, _ := textEvent(source, content) - task := &Task{Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()} + task := newInputTask(evt) if !a.sched.enqueue(task) { t.Fatal("入队失败") } return task, evt } +// enqueueInterrupt 登记一次**中断**(走 requestPreempt:能抢占则进 immediate 槽, +// 否则按级别进中断队列),返回被登记的任务。 +func enqueueInterrupt(t *testing.T, a *Agent, level Level, source, content string) (*Task, *agentIO.InputEvent) { + t.Helper() + evt, _ := textEvent(source, content) + evt.Payload["interrupt"] = true + a.sched.requestPreempt(evt, level) + snap := a.DumpScheduler() + if snap.Immediate != nil && snap.Immediate.Event == evt { + return snap.Immediate, evt + } + q := snap.InterruptQueues[clampPluginLevel(level)] + for i := len(q) - 1; i >= 0; i-- { + if q[i].Event == evt { + return q[i], evt + } + } + return nil, evt +} + // P1 + R1 + R5 + D1=A:高优先级抢占 → 挂起在 S_LLM → 中断任务带只读前缀 → // 恢复后从 S_LLM 重发,且原任务的 msgs 未被改动。 func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { sp := newPreemptProvider("intr-done", "low-done") a := newPreemptAgent(t, sp) - lowTask, _ := enqueueTask(t, a, LevelBackground, "qq", "低优先级任务") + lowTask, _ := enqueueQueued(t, a, "qq", "低优先级任务") lt, _, kind := a.sched.nextRef() if kind != nextReady || lt != lowTask { t.Fatalf("应取到低优先级任务,kind=%v", kind) @@ -109,7 +130,7 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { // 注入 L4 中断(cli) intrEvt, _ := textEvent("cli", "紧急打断") intrEvt.Payload["interrupt"] = true - if !a.sched.requestPreempt(intrEvt, LevelCritical) { + if !a.sched.requestKernelPreempt(intrEvt) { t.Fatal("L4 应请求抢占并返回 true(应取消 LLM)") } a.cancelCurrentLLM() @@ -146,7 +167,7 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { // Q3:三集合统一比较 → 下一轮取中断(L4 > L1)。 it, _, k := a.sched.nextRef() - if k != nextPending || it.Level != LevelCritical { + if k != nextImmediate || it.Level != LevelCritical { t.Fatalf("应取到 pending 中断,kind=%v level=%v", k, it.Level) } // D1=B:中断任务在**上一个任务之前的完整状态**上开始运行,不继承本任务的现场。 @@ -194,14 +215,17 @@ func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { } } -// P2/P3:同级与更低级都不得抢占,请求进 pendingInterrupts。 +// P2/P3:同级与更低级都不得抢占,请求进中断队列。 +// +// 注意:能比“同级/更低不得抢占”的只可能是**中断之间**——排队任务无级别, +// 任何中断都能打断它(这是模型的规定,不是漏洞)。 func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) { sp := newPreemptProvider("low-done", "intr-done") a := newPreemptAgent(t, sp) - lowTask, _ := enqueueTask(t, a, LevelInteractive, "cli", "运行中的 L3") - if _, _, kind := a.sched.nextRef(); kind != nextReady { - t.Fatal("应取到运行任务") + lowTask, _ := enqueueInterrupt(t, a, LevelInteractive, "cli", "运行中的 L3") + if _, _, kind := a.sched.nextRef(); kind != nextInterrupt { + t.Fatal("应取到运行中的 L3 中断") } done := make(chan struct{}) @@ -234,7 +258,7 @@ func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) { t.Fatal("运行任务未结束") } if len(a.DumpScheduler().PendingInterrupts) != 2 { - t.Fatalf("两条未抢占中断都应保留在 pendingInterrupts,实际 %d", + t.Fatalf("两条未抢占中断都应保留在中断队列,实际 %d", len(a.DumpScheduler().PendingInterrupts)) } } @@ -243,23 +267,23 @@ func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) { func TestPreempt_DepthCapBlocksSuspension(t *testing.T) { a := newPreemptAgent(t, newPreemptProvider()) - if a.sched.maxSuspendDepth != 4 { - t.Fatalf("默认深度上限=%d,期望 4", a.sched.maxSuspendDepth) + if a.sched.maxInterruptFrames != 4 { + t.Fatalf("默认栈深上界=%d,期望 4(= 中断级数,结构推论)", a.sched.maxInterruptFrames) } 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: LevelBackground}, frame()) + for i := 0; i < a.sched.maxInterruptFrames; i++ { + a.sched.suspend(&Task{ID: uint64(i + 1), Class: TaskInterrupt, Level: LevelBackground}, frame()) } if a.sched.canSuspend() { t.Fatal("深度已达上限,canSuspend 应为 false") } // 超限兜底:仍保留帧(不丢副作用记录),但计数 Rejected。 before := a.DumpScheduler().Stats.Rejected - a.sched.suspend(&Task{ID: 99, Level: LevelBackground}, frame()) + a.sched.suspend(&Task{ID: 99, Class: TaskInterrupt, Level: LevelBackground}, frame()) if a.DumpScheduler().Stats.Rejected != before+1 { t.Fatal("超限挂起必须计数 Rejected") } - if len(a.DumpScheduler().SuspendStack) != a.sched.maxSuspendDepth+1 { + if len(a.DumpScheduler().SuspendStack) != a.sched.maxInterruptFrames+1 { t.Fatal("兜底路径必须保留帧而不是丢弃") } } @@ -273,10 +297,10 @@ func TestPreempt_IdleInterruptIsQueuedNotLost(t *testing.T) { t.Fatal("空闲时不应请求取消 LLM(没有运行任务)") } if len(a.DumpScheduler().PendingInterrupts) != 1 { - t.Fatal("空闲时的中断必须进 pendingInterrupts") + t.Fatal("空闲时的中断必须进中断队列(不能丢)") } task, _, kind := a.sched.nextRef() - if kind != nextPending || task.Level != LevelMessage { + if kind != nextInterrupt || task.Level != LevelMessage { t.Fatalf("应取到待处理中断,kind=%v", kind) } a.executeNewTask(task) diff --git a/internal/agent/core/scheduler_stack_test.go b/internal/agent/core/scheduler_stack_test.go index 3a3197d..789d4d1 100644 --- a/internal/agent/core/scheduler_stack_test.go +++ b/internal/agent/core/scheduler_stack_test.go @@ -74,7 +74,7 @@ func TestStack_NestedPreemptionResumesLIFO(t *testing.T) { a := newPreemptAgent(t, sp) // A(L1)开始运行 - if _, _ = enqueueTask(t, a, LevelBackground, "qq", "任务A"); true { + if _, _ = enqueueQueued(t, a, "qq", "任务A"); true { } at, _, _ := a.sched.nextRef() doneA := make(chan struct{}) @@ -94,7 +94,7 @@ func TestStack_NestedPreemptionResumesLIFO(t *testing.T) { // B 开始运行 bt, _, k := a.sched.nextRef() - if k != nextPending || bt.Level != LevelMessage { + if k != nextImmediate || bt.Level != LevelMessage { t.Fatalf("应取到 B(pending),kind=%v level=%v", k, bt.Level) } doneB := make(chan struct{}) @@ -113,8 +113,8 @@ func TestStack_NestedPreemptionResumesLIFO(t *testing.T) { 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[0].Task.Class != TaskQueued { + t.Fatalf("栈底应为排队任务 A(无级别),实际 %v", snap.SuspendStack[0].Task.Class) } if snap.SuspendStack[1].Task.Level != LevelMessage { t.Fatalf("栈顶应为 B(L2),实际 %v", snap.SuspendStack[1].Task.Level) @@ -122,7 +122,7 @@ func TestStack_NestedPreemptionResumesLIFO(t *testing.T) { // C 运行完毕(第三次调用,不阻塞) ct, _, k := a.sched.nextRef() - if k != nextPending || ct.Level != LevelInteractive { + if k != nextImmediate || ct.Level != LevelInteractive { t.Fatalf("应取到 C,kind=%v level=%v", k, ct.Level) } a.executeNewTask(ct) @@ -141,8 +141,8 @@ func TestStack_NestedPreemptionResumesLIFO(t *testing.T) { if k2 != nextSuspended { t.Fatalf("应继续恢复 A,kind=%v", k2) } - if rt2.Level != LevelBackground { - t.Fatalf("最后应恢复 A(L1),实际 %v", rt2.Level) + if rt2.Class != TaskQueued { + t.Fatalf("最后应恢复排队的 A(无级别),实际 %v", rt2.Class) } a.resumeTask(rt2, rf2) @@ -159,9 +159,9 @@ func TestStack_TopOnlyWinsOverHigherPrioritySuspended(t *testing.T) { // (栈自底向上基础级递增),这里专门用来区分两种实现: // · 只比栈顶 → 取 B // · 全栈扫最优 → 取 A(L3 > L2) - a.sched.suspend(&Task{ID: 1, Level: LevelInteractive, EnqueuedAt: time.Now()}, + a.sched.suspend(&Task{ID: 1, Class: TaskInterrupt, Level: LevelInteractive, EnqueuedAt: time.Now()}, a.newTaskFrame("A", a.stageCtxFromInput("A", "", ""))) - a.sched.suspend(&Task{ID: 2, Level: LevelMessage, EnqueuedAt: time.Now()}, + a.sched.suspend(&Task{ID: 2, Class: TaskInterrupt, Level: LevelMessage, EnqueuedAt: time.Now()}, a.newTaskFrame("B", a.stageCtxFromInput("B", "", ""))) rt, _, k := a.sched.nextRef() @@ -180,13 +180,13 @@ func TestStack_TopOnlyWinsOverHigherPrioritySuspended(t *testing.T) { 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()) + for i := 0; i < a.sched.maxInterruptFrames; i++ { + a.sched.suspend(&Task{ID: uint64(i + 1), Class: TaskInterrupt, 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) + if n := len(a.DumpScheduler().SuspendStack); n != a.sched.maxInterruptFrames { + t.Fatalf("栈深=%d,期望上界 %d", n, a.sched.maxInterruptFrames) } } diff --git a/internal/agent/core/scheduler_starvation_test.go b/internal/agent/core/scheduler_starvation_test.go index e414454..d8bda53 100644 --- a/internal/agent/core/scheduler_starvation_test.go +++ b/internal/agent/core/scheduler_starvation_test.go @@ -14,7 +14,7 @@ import ( ) func TestStarvation_EffectiveLevelPromotion(t *testing.T) { - base := &Task{Level: LevelBackground} + base := &Task{Class: TaskInterrupt, Level: LevelBackground} if got := effectiveLevel(base); got != LevelBackground { t.Fatalf("未抢占时有效级=%v,期望 L1", got) } @@ -33,7 +33,7 @@ func TestStarvation_EffectiveLevelPromotion(t *testing.T) { } // 封顶 L4:L3 任务被多次抢占也不会超过紧急级。 - high := &Task{Level: LevelInteractive, PreemptCount: 99} + high := &Task{Class: TaskInterrupt, Level: LevelInteractive, PreemptCount: 99} if got := effectiveLevel(high); got != LevelCritical { t.Fatalf("L3 提升后应封顶为 L4,实际 %v", got) } @@ -42,8 +42,8 @@ func TestStarvation_EffectiveLevelPromotion(t *testing.T) { func TestStarvation_CooldownBlocksImmediateRepreempt(t *testing.T) { a := newPreemptAgent(t, newPreemptProvider()) - low := &Task{ID: 1, Level: LevelBackground, EnqueuedAt: time.Now()} - a.sched.enqueue(low) + low := &Task{ID: 1, Class: TaskInterrupt, Level: LevelBackground, EnqueuedAt: time.Now()} + a.sched.immediate = low a.sched.nextRef() // running = low e1, _ := textEvent("qq", "第一次打断") @@ -64,7 +64,7 @@ func TestStarvation_CooldownBlocksImmediateRepreempt(t *testing.T) { a.sched.mu.Unlock() e2, _ := textEvent("cli", "冷却期内的紧急打断") - if a.sched.requestPreempt(e2, LevelCritical) { + if a.sched.requestKernelPreempt(e2) { t.Fatal("抢占冷却期内不得再抢占") } if a.sched.preemptGrantedFor() { @@ -75,8 +75,8 @@ func TestStarvation_CooldownBlocksImmediateRepreempt(t *testing.T) { func TestStarvation_PromotionBlocksSameLevelPreempt(t *testing.T) { a := newPreemptAgent(t, newPreemptProvider()) - low := &Task{ID: 1, Level: LevelBackground, EnqueuedAt: time.Now()} - a.sched.enqueue(low) + low := &Task{ID: 1, Class: TaskInterrupt, Level: LevelBackground, EnqueuedAt: time.Now()} + a.sched.immediate = low a.sched.nextRef() // 模拟「已被抢占过一次」:有效级 = L2。 low.PreemptCount = 1 @@ -96,23 +96,29 @@ func TestStarvation_PromotionBlocksSameLevelPreempt(t *testing.T) { } } -// 选择函数必须用有效级:被抢占过的任务在排队时应当优先于同级/更低的任务。 -func TestStarvation_SelectionUsesEffectiveLevel(t *testing.T) { - base := time.Now() - promoted := &Task{ID: 1, Level: LevelBackground, PreemptCount: 2, EnqueuedAt: base} // 有效 L3 - normal := &Task{ID: 2, Level: LevelMessage, EnqueuedAt: base.Add(time.Second)} // L2 +// 提升必须真的进入抢占判据,而不只是一个数学性质: +// 被抢占过一次的 L1 中断(有效 L2)应当顶住同级 L2 流的再次抢占。 +func TestStarvation_PromotionIsVisibleInSelection(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) - if !taskBefore(promoted, normal) { - t.Fatal("被抢占 2 次的 L1(有效 L3)应先于 L2 执行") - } - if taskBefore(normal, promoted) { - t.Fatal("选择函数不得只看基础级") + // 栈里放一个「被抢占过一次的 L1」:有效级 L2。 + a.sched.suspend(&Task{ID: 1, Class: TaskInterrupt, Level: LevelBackground, PreemptCount: 1}, + a.newTaskFrame("A", a.stageCtxFromInput("A", "", ""))) + + // 队列里来一个 L2:有效级持平(2 vs 2)→ 不得越过栈顶。 + evt, _ := textEvent("qq", "L2 中断") + a.sched.registerInterrupt(newInterruptTask(evt, LevelMessage)) + if _, _, kind := a.sched.nextRef(); kind != nextSuspended { + t.Fatalf("有效级持平应恢复栈顶,kind=%v", kind) } - // 提升不改变调度器自身的排序稳定性:同为有效级时按入队时刻。 - 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,更早 - if !taskBefore(b, a) { - t.Fatal("同有效级时应先到先服务") + // 队列里来一个 L3:严格大于 → 队头优先。 + // PreemptCount 从 0 起:suspend 内部会 +1 → 有效级 L2(正好用来卡 L2 持平)。 + a.sched.suspend(&Task{ID: 2, Class: TaskInterrupt, Level: LevelBackground}, + a.newTaskFrame("B", a.stageCtxFromInput("B", "", ""))) + evt2, _ := textEvent("cli", "L3 中断") + a.sched.registerInterrupt(newInterruptTask(evt2, LevelInteractive)) + if _, _, kind := a.sched.nextRef(); kind != nextInterrupt { + t.Fatalf("L3 > 有效 L2 应取中断队列,kind=%v", kind) } } diff --git a/internal/agent/core/scheduler_test.go b/internal/agent/core/scheduler_test.go index 03b57af..2b6add8 100644 --- a/internal/agent/core/scheduler_test.go +++ b/internal/agent/core/scheduler_test.go @@ -5,83 +5,120 @@ package core // 设计依据 docs/zh/input-scheduler-design.md §11.4(Q1/Q4)与 §11.5(O1/K1)。 import ( + "fmt" "testing" - "time" agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" ) -func mkTask(id uint64, level Level, at time.Time) *Task { - return &Task{ID: id, Level: level, EnqueuedAt: at} -} +// 新模型的选择顺序:immediate → 中断队列 L4→L1 → 栈顶(与队头比级别) → 排队 FIFO。 +func TestScheduler_SelectionOrder(t *testing.T) { + s := newScheduler(16) -// Q1:选择函数的排序键是 (-Level, EnqueuedAt, ID)。 -func TestPickTaskIndex_Ordering(t *testing.T) { - base := time.Date(2026, 9, 12, 12, 0, 0, 0, time.UTC) + // 四条中断队列各放一个,入队顺序与级别相反 —— 验证“按级别扫”而非 FIFO。 + for _, lv := range []Level{LevelBackground, LevelMessage, LevelInteractive, LevelCritical} { + evt, _ := textEvent("qq", "中断") + s.registerInterrupt(newInterruptTask(evt, lv)) + } + // 排队任务两条(无级别,FIFO)。 + s.enqueue(newSelfTask(selfInputMsg{text: "q1"})) + s.enqueue(newSelfTask(selfInputMsg{text: "q2"})) - cases := []struct { - name string - queue []*Task - want []uint64 - }{ - { - name: "高优先级先执行,与入队先后无关", - queue: []*Task{ - mkTask(1, LevelBackground, base), - mkTask(2, LevelCritical, base.Add(time.Second)), - mkTask(3, LevelMessage, base.Add(2*time.Second)), - }, - want: []uint64{2, 3, 1}, - }, - { - name: "同优先级先到先服务", - queue: []*Task{ - mkTask(1, LevelInteractive, base.Add(3*time.Second)), - mkTask(2, LevelInteractive, base.Add(time.Second)), - mkTask(3, LevelInteractive, base.Add(2*time.Second)), - }, - want: []uint64{2, 3, 1}, - }, - { - name: "同优先级同入队时刻用 ID 兜底(保证确定性)", - queue: []*Task{ - mkTask(7, LevelMessage, base), - mkTask(3, LevelMessage, base), - mkTask(5, LevelMessage, base), - }, - want: []uint64{3, 5, 7}, - }, - { - name: "四级全覆盖", - queue: []*Task{ - mkTask(1, LevelBackground, base), - mkTask(2, LevelMessage, base), - mkTask(3, LevelInteractive, base), - mkTask(4, LevelCritical, base), - }, - want: []uint64{4, 3, 2, 1}, - }, + var order []Level + for i := 0; i < 4; i++ { + task, _, kind := s.nextRef() + if kind != nextInterrupt { + t.Fatalf("第 %d 个应来自中断队列,kind=%v", i+1, kind) + } + order = append(order, task.Level) + s.done(task) + } + want := []Level{LevelCritical, LevelInteractive, LevelMessage, LevelBackground} + for i := range want { + if order[i] != want[i] { + t.Fatalf("中断执行顺序=%v,期望 %v", order, want) + } } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - q := append([]*Task(nil), c.queue...) - var got []uint64 - for len(q) > 0 { - i := pickTaskIndex(q) - got = append(got, q[i].ID) - q = append(q[:i], q[i+1:]...) - } - if len(got) != len(c.want) { - t.Fatalf("取出的任务数=%d,期望 %d", len(got), len(c.want)) - } - for i := range got { - if got[i] != c.want[i] { - t.Fatalf("执行顺序=%v,期望 %v", got, c.want) - } - } + // 中断耗尽后才是排队任务,且保持 FIFO。 + for i := 1; i <= 2; i++ { + task, _, kind := s.nextRef() + if kind != nextReady { + t.Fatalf("中断耗尽后应取排队任务,kind=%v", kind) + } + if task.Self.text != fmt.Sprintf("q%d", i) { + t.Fatalf("排队任务应 FIFO,第 %d 个=%q", i, task.Self.text) + } + s.done(task) + } + if _, _, kind := s.nextRef(); kind != nextNone { + t.Fatal("全空后应返回 nextNone") + } +} + +// immediate(刚抢占成功的中断)必须最先运行——哪怕队列里有更高级别的待处理中断。 +// 这是“抢占立即生效”的实现方式,也是它不需要和栈顶比级别的原因。 +func TestScheduler_ImmediateWins(t *testing.T) { + s := newScheduler(16) + evt1, _ := textEvent("cli", "L4 待处理") + s.registerInterrupt(newKernelInterruptTask(evt1)) + + evt2, _ := textEvent("qq", "抢占者") + preemptor := newInterruptTask(evt2, LevelBackground) + s.mu.Lock() + s.setImmediateLocked(preemptor) + s.mu.Unlock() + + task, _, kind := s.nextRef() + if kind != nextImmediate || task != preemptor { + t.Fatalf("immediate 必须先运行,kind=%v", kind) + } +} + +// 中断队列头与中断栈顶比级别,取高者;栈顶是排队任务(无级别)时任何中断都赢。 +func TestScheduler_StackTopVsInterruptQueue(t *testing.T) { + s := newScheduler(16) + // 直接构造挂起现场:不走 suspend(),避免 PreemptCount/冷却干扰本用例 + // (本用例只测“选择顺序”这一件事)。 + pushSuspended := func(id uint64, class TaskClass, lv Level) { + s.mu.Lock() + s.suspendStack = append(s.suspendStack, &suspendedTask{ + Task: &Task{ID: id, Class: class, Level: lv}, Frame: &TaskFrame{}, }) + s.mu.Unlock() + } + // 每次选取后清掉 running,让下一次 registerInterrupt 不把它当成运行任务。 + clearRunning := func() { + s.mu.Lock() + s.running = nil + s.mu.Unlock() + } + + // 栈顶 L3,队列只有 L2 → 恢复栈顶。 + pushSuspended(1, TaskInterrupt, LevelInteractive) + evt, _ := textEvent("qq", "L2 待处理") + s.registerInterrupt(newInterruptTask(evt, LevelMessage)) + if _, _, kind := s.nextRef(); kind != nextSuspended { + t.Fatalf("栈顶 L3 > 队头 L2 → 应恢复栈顶,kind=%v", kind) + } + clearRunning() + + // 栈顶 L3,队列来了 L4 → 队头优先。 + pushSuspended(2, TaskInterrupt, LevelInteractive) + evt2, _ := textEvent("cli", "L4 待处理") + s.registerInterrupt(newKernelInterruptTask(evt2)) + if _, _, kind := s.nextRef(); kind != nextInterrupt { + t.Fatalf("队头 L4 > 栈顶 L3 → 应先取中断,kind=%v", kind) + } + clearRunning() + + // 栈顶是排队任务(无级别)→ 任何中断都赢。 + pushSuspended(3, TaskQueued, 0) + evt3, _ := textEvent("qq", "L1 待处理") + s.registerInterrupt(newInterruptTask(evt3, LevelBackground)) + if _, _, kind := s.nextRef(); kind != nextInterrupt { + t.Fatalf("排队栈顶可被任何中断打断,kind=%v", kind) } } diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index 3f27d59..1d02eeb 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -126,7 +126,6 @@ type TaskFrame struct { IsInterrupt bool StartedAt time.Time Terminal taskTerminal - Level Level // PrefixLen 是 stepPrepare 构建的**基础前缀**长度(system + timeline + 用户输入)。 // 恢复时用它把「本任务自己的现场」接回重建后的前缀之上(见 rebaseFramePrefix)。 @@ -382,7 +381,6 @@ func (a *Agent) prepareInputTask(evt *agentIO.InputEvent) (*TaskFrame, taskTermi f.CleanInput = cleanInput f.IsInterrupt = isInterrupt f.StartedAt = start - f.Level = a.sched.currentLevel() return f, terminalNone } diff --git a/internal/agent/core/task_terminal_test.go b/internal/agent/core/task_terminal_test.go index 4de34f9..549eebb 100644 --- a/internal/agent/core/task_terminal_test.go +++ b/internal/agent/core/task_terminal_test.go @@ -22,7 +22,7 @@ func TestTerminal_TaskScopedReplyNotMisrouted(t *testing.T) { a := newPreemptAgent(t, sp) lowEvt, lowCh := textEvent("qq", "低优先级任务") - lowTask := &Task{Kind: TaskKindInput, Level: LevelBackground, Event: lowEvt, EnqueuedAt: time.Now()} + lowTask := newInputTask(lowEvt) if !a.sched.enqueue(lowTask) { t.Fatal("入队失败") } @@ -38,8 +38,8 @@ func TestTerminal_TaskScopedReplyNotMisrouted(t *testing.T) { intrEvt, intrCh := textEvent("cli", "紧急打断") intrEvt.Payload["interrupt"] = true - if !a.sched.requestPreempt(intrEvt, LevelCritical) { - t.Fatal("L4 应抢占 L1") + if !a.sched.requestKernelPreempt(intrEvt) { + t.Fatal("内核 L4 应抢占排队任务") } a.cancelCurrentLLM() select { @@ -50,8 +50,8 @@ func TestTerminal_TaskScopedReplyNotMisrouted(t *testing.T) { // 执行中断任务 → 只应写它自己的回执通道。 it, _, k := a.sched.nextRef() - if k != nextPending { - t.Fatalf("应取到 pending 中断,kind=%v", k) + if k != nextImmediate { + t.Fatalf("应取到立即运行的中断,kind=%v", k) } a.executeNewTask(it) if len(intrCh) != 1 { diff --git a/internal/agent/io/channel.go b/internal/agent/io/channel.go index c071115..bb96225 100644 --- a/internal/agent/io/channel.go +++ b/internal/agent/io/channel.go @@ -266,6 +266,10 @@ func applyInjectOpts(payload map[string]interface{}, opts InjectOptions) { if opts.CleanerName != "" { payload["cleaner_name"] = opts.CleanerName } + // priority 只对中断注入有意义;排队路径会忽略它(内核侧只读不写)。 + if opts.Priority != "" { + payload["priority"] = opts.Priority + } } func (m *IOManager) InjectInputOpts(source, eventType string, payload map[string]interface{}, opts InjectOptions) { diff --git a/internal/plugin/proc/corehandler.go b/internal/plugin/proc/corehandler.go index b2409eb..ab19fb5 100644 --- a/internal/plugin/proc/corehandler.go +++ b/internal/plugin/proc/corehandler.go @@ -181,7 +181,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} if err := validateContextPolicy("io.injectText", p.ContextPolicy); err != nil { return nil, err } - h.sdk.InjectTextOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName)) + h.sdk.InjectTextOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName, p.Priority)) return nil, nil case MethodIOInjectInterrupt: var p injectParams @@ -191,7 +191,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} if err := validateContextPolicy("io.injectInterrupt", p.ContextPolicy); err != nil { return nil, err } - h.sdk.InjectInterruptTextOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName)) + h.sdk.InjectInterruptTextOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName, p.Priority)) return nil, nil case MethodIOInjectTextNoMem: var p injectParams @@ -202,7 +202,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} return nil, err } // 旧 RPC 语义就是「不进记忆」,显式标志位只可能再叠上 context_policy。 - h.sdk.InjectTextOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(true, p.ContextPolicy, p.CleanerName)) + h.sdk.InjectTextOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(true, p.ContextPolicy, p.CleanerName, p.Priority)) return nil, nil case MethodIOInjectSync: var p injectParams @@ -212,7 +212,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} if err := validateContextPolicy("io.injectInputSync", p.ContextPolicy); err != nil { return nil, err } - reply := h.sdk.InjectInputSyncOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName)) + reply := h.sdk.InjectInputSyncOpts(p.Source, p.Channel, h.resolveText(p), pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName, p.Priority)) return map[string]interface{}{"reply": reply}, nil case MethodIOInjectMedia: @@ -227,7 +227,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} if err != nil { return nil, err } - h.sdk.InjectInputMediaOpts(p.Source, p.Channel, p.Text, blocks, pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName)) + h.sdk.InjectInputMediaOpts(p.Source, p.Channel, p.Text, blocks, pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName, p.Priority)) return nil, nil case MethodIOInjectMediaSync: @@ -242,7 +242,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} if err != nil { return nil, err } - reply := h.sdk.InjectInputMediaSyncOpts(p.Source, p.Channel, p.Text, blocks, pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName)) + reply := h.sdk.InjectInputMediaSyncOpts(p.Source, p.Channel, p.Text, blocks, pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName, p.Priority)) return map[string]interface{}{"reply": reply}, nil case MethodIOInjectInterruptMedia: @@ -257,7 +257,7 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} if err != nil { return nil, err } - h.sdk.InjectInterruptMediaOpts(p.Source, p.Channel, p.Text, blocks, pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName)) + h.sdk.InjectInterruptMediaOpts(p.Source, p.Channel, p.Text, blocks, pubSdkInjectOpts(p.NoMemory, p.ContextPolicy, p.CleanerName, p.Priority)) return nil, nil // ---- 生命周期(原 case 8)---- @@ -695,6 +695,8 @@ type injectParams struct { NoMemory bool `json:"no_memory,omitempty"` ContextPolicy string `json:"context_policy,omitempty"` CleanerName string `json:"cleaner_name,omitempty"` + // Priority 声明中断注入的优先级(L1..L3);L4 内核独占,见 InjectOptions。 + Priority string `json:"priority,omitempty"` } // injectMediaParams 是带媒体注入/工具块注入的参数。 @@ -715,14 +717,17 @@ type injectMediaParams struct { NoMemory bool `json:"no_memory,omitempty"` ContextPolicy string `json:"context_policy,omitempty"` CleanerName string `json:"cleaner_name,omitempty"` + Priority string `json:"priority,omitempty"` } // pubSdkInjectOpts 把 RPC 报文里的三个字段转成公开 SDK 的 InjectOptions。 // // 单独提一个转换函数是为了让「默认值」只有一个出处:零值即记入记忆 + 不裁剪, // 与旧三参数注入等价。 -func pubSdkInjectOpts(noMemory bool, policy, cleanerName string) pubsdk.InjectOptions { - return pubsdk.InjectOptions{NoMemory: noMemory, ContextPolicy: policy, CleanerName: cleanerName} +func pubSdkInjectOpts(noMemory bool, policy, cleanerName, priority string) pubsdk.InjectOptions { + return pubsdk.InjectOptions{ + NoMemory: noMemory, ContextPolicy: policy, CleanerName: cleanerName, Priority: priority, + } } // validateContextPolicy 校验上下文策略取值,与 tool.register 同一套规则。 diff --git a/third_party/homeagent-sdk/example/qq/plugin.go b/third_party/homeagent-sdk/example/qq/plugin.go index 16634bc..1dd7144 100644 --- a/third_party/homeagent-sdk/example/qq/plugin.go +++ b/third_party/homeagent-sdk/example/qq/plugin.go @@ -1435,7 +1435,12 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { if p.sdk != nil { // NoMemory:HTTP 侧来的中断提示,不是对话内容。 - p.sdk.InjectInterruptTextOpts(p.name, p.name, interrupt, sdk.InjectOptions{NoMemory: true}) + // Priority:QQ 消息是**低级别中断**——既不是时钟那样的实时工作, + // 也不是紧急工作,所以声明 L1(完全可等)。 + p.sdk.InjectInterruptTextOpts(p.name, p.name, interrupt, sdk.InjectOptions{ + NoMemory: true, + Priority: sdk.PriorityL1, + }) } w.WriteHeader(http.StatusOK) } @@ -2531,9 +2536,10 @@ func (p *Plugin) handleDownloadFile(args map[string]interface{}) (interface{}, e log.Printf("[qq] 文件下载完成: %s", savePath) if p.sdk != nil { // NoMemory:下载完成的状态通知,不是记忆内容。 + // Priority:同上,QQ 侧一律低级别中断(L1)。 p.sdk.InjectInterruptTextOpts(p.name, p.name, fmt.Sprintf("文件下载完成: %s,保存在 %s", filepath.Base(savePath), savePath), - sdk.InjectOptions{NoMemory: true}) + sdk.InjectOptions{NoMemory: true, Priority: sdk.PriorityL1}) } } else { errMsg = "下载失败,文件可能已过期" diff --git a/third_party/homeagent-sdk/sdk/plugin.go b/third_party/homeagent-sdk/sdk/plugin.go index f8b0c12..cf38db6 100644 --- a/third_party/homeagent-sdk/sdk/plugin.go +++ b/third_party/homeagent-sdk/sdk/plugin.go @@ -78,8 +78,24 @@ type InjectOptions struct { NoMemory bool ContextPolicy string CleanerName string + + // Priority 声明**中断注入**的优先级(仅 InjectInterrupt* 有意义)。 + // + // 取值 "L1"/"L2"/"L3";空等同 L1。L4 由内核独占(panic / 内核事件 selfip), + // 插件声明 L4 会被内核夹到 L3——内核的调度内部属性不接受外部越权。 + // + // 排队注入(InjectText*/InjectInputSync)没有级别:它们本就是“不需及时处理” + // 的那一类,可被任何中断打断。 + Priority string } +// 中断优先级取值(插件可用范围)。L4 不在其中:它由内核保留。 +const ( + PriorityL1 = "L1" + PriorityL2 = "L2" + PriorityL3 = "L3" +) + // ChannelDef 描述通道在记忆计算层的行为,与 ToolDef.NoMemory/Cleaner 语义一致。 // NoMemory: 此通道输入/输出不参与记忆计算(向量化/关键词提取/蒸馏),但原文保留在上下文中 // Cleaner: 计算层过滤函数,不改原文;仅在向量化/jieba/蒸馏/存档提取关键词时调用