From b829968ac65dfe5f122c933ade2edb99f6501471 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 25 Oct 2026 13:14:42 +0800 Subject: [PATCH] =?UTF-8?q?refactor(agent/core):=20=E4=B8=AD=E6=96=AD?= =?UTF-8?q?=E5=AD=90=E7=B3=BB=E7=BB=9F=E7=8B=AC=E7=AB=8B=E6=88=90=20interr?= =?UTF-8?q?upt.go=20+=20=E4=BF=AE=E4=B8=A4=E5=A4=84=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=AB=9E=E4=BA=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 中断子系统拆分(用户按 OS 中断语义三次裁定的最终边界) 中断类比 OS:只提供中断优先级、中断队列、现场保护、中断屏蔽。 调度器建立在其之上——它问"中断侧该出谁",然后执行那个任务。 新增 internal/agent/core/interrupt.go(303 行),承载: - 中断队列(interruptQueues[5] + immediate 单槽) - 中断栈(suspendStack)—— **唯一操作者** - 中断屏蔽(critical,对应关中断) - 让位信号原语(preemptArmed/preemptLevel) 留在 scheduler.go 的是**策略**: - 任务优先级语义(Level / effectiveLevel / canPreempt / clampPluginLevel) - 恢复决策(nextInterruptSelection:栈顶 vs 中断队列比较) - 就绪队列 / 停止配额 / 背压 ### 关键设计:中断栈只有中断管理器能操作 调度器通过 stackTopLocked() **只读查询**栈顶做比较, 真正的弹栈由 restoreTopLocked() 在中断管理器内完成。 这条不变量在类型层面无法绕过。 ### 解耦机制:stackNotify 钩子 中断栈变化时通告调度器,调度器据此唤醒调度循环重新求值。 中断管理器不知道调度器会怎么选,调度器也不碰栈内部 ⇒ 将来给中断子系统加向量表或换多栈策略,不必改调度器。 ### 测试 新增 interrupt_invariant_test.go:I1–I7 共 11 条**不变量**判据 (区别于既有的行为测试)。两条经变异验证确认会变红: - canPreempt 改成恒 true → L4 不抢 L4 判据红 - maxInterruptFrames 改成 99 → 上界判据红 - 去掉 notifyStackLocked → 钩子判据红(且判据本身带超时,缺判据是"红"不是"挂") ## 修两处真实数据竞争(阻塞 CI,非本次引入) -race 实测:internal/agent/core 原有 30 个测试失败,整包被判红。 根因(ab27d1a 上已存在,与拆分无关): - core/stages.go:withExecuteCtx 往调用方共享的 tc.Arguments 原地写 ctx - io/channel.go:bindExecCtx 同样原地写设备 args 原注释称"args 每次调用独立、天然一 goroutine 一份",但代码改的是调用方那份。 工具在自己的 goroutine 里跑,超时/取消路径下主 goroutine 仍在读同一份 args。 修法:复制 map 而非原地写。变异验证:退回原地写后 TestPreempt_DeferredDuringToolExec 立刻报 DATA RACE。 ## 验证 - go build ./... 通过 - go test -p 1 ./... 全仓无失败 - -race 三关键包(core/io/proc)全绿,core 从 30 红归零 - CI 门禁自检 / 启动图判据 / 版本一致性 本地通过 ## 遗留(不在本次) - 中断向量表(计划 §7 已设计,按 §7.6 应在拆分后单独做) - maxQueue 仍约束中断队列:OS 语义上说不通,但改它是行为变更,单独立项 --- docs/zh/interrupt-subsystem-split-plan.md | 534 ++++++++++++++++++ internal/agent/core/interrupt.go | 345 +++++++++++ .../agent/core/interrupt_invariant_test.go | 313 ++++++++++ internal/agent/core/resident.go | 23 +- internal/agent/core/scheduler.go | 534 ++++++++---------- internal/agent/core/stages.go | 24 +- internal/agent/core/task.go | 7 +- internal/agent/io/channel.go | 12 +- 8 files changed, 1459 insertions(+), 333 deletions(-) create mode 100644 docs/zh/interrupt-subsystem-split-plan.md create mode 100644 internal/agent/core/interrupt.go create mode 100644 internal/agent/core/interrupt_invariant_test.go diff --git a/docs/zh/interrupt-subsystem-split-plan.md b/docs/zh/interrupt-subsystem-split-plan.md new file mode 100644 index 00000000..54b92de0 --- /dev/null +++ b/docs/zh/interrupt-subsystem-split-plan.md @@ -0,0 +1,534 @@ +# 中断子系统独立化 —— 探测结论与实施计划 + +> 状态:**计划**(未实施) +> 起因:中断语义散落在 `scheduler.go` 一个 1476 行的文件里,且入口横跨 5 个文件。 +> 修改任何一条中断规则(级别比较、抢占条件、栈深度)都要在多处同步改,漏一处的后果是静默的。 + +--- + +## 一、探测结论(实测,非推演) + +### 1.1 现状:中断没有独立子系统 + +`type scheduler struct`(`scheduler.go:330`)一个结构体同时承担四件事: + +| 职责 | 字段 | 归属应为 | +| --- | --- | --- | +| 就绪队列调度 | `queue` / `running` / `maxQueue` / `backpressured` | 调度器 | +| 中断分级排队 | `interruptQueues[5]` | **中断子系统** | +| 抢占与让位 | `immediate` / `preemptArmed` / `preemptLevel` / `critical` | **中断子系统** | +| 中断栈(嵌套挂起) | `suspendStack` / `maxInterruptFrames` | **中断子系统** | +| 用户停止配额 | `cancelBudget` / `stopArmed` | **中断子系统**(是"打断"的一种) | + +### 1.2 中断逻辑的实际分布 + +| 文件 | 中断相关职责 | 行数量级 | +| --- | --- | --- | +| `scheduler.go` | 队列/抢占/栈/让位/级别语义(**主体**) | 1476 行中约 205 处引用 | +| `eventloop.go` | 中断注入路径 | 少量 | +| `overflow.go` | `raiseKernelInterrupt` 的一种来源 | 1 处 | +| `resident.go` | 子 contextfull → 父 L4 | 1 处 | +| `task.go` | 安全点让位判定 | 1 处 | + +### 1.3 最能说明问题的证据 + +`scheduler.go:374` 那段注释: + +> 链条 = 排队(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4 运行中),被挂起 4 帧 + +这是**中断子系统的核心不变量**(`maxInterruptFrames = int(LevelCritical)`,结构推论、非配置项)。 +它现在藏在 1476 行文件的中间,没有任何类型或文件专门承载它。改中断语义的人必须 +先读完整份 `scheduler.go` 才能确认自己没漏。 + +### 1.4 关键有利条件(决定拆分方案) + +| 探测项 | 结果 | 含义 | +| --- | --- | --- | +| 中断状态字段是否被跨文件直接访问 | **零**(`interruptQueues`/`immediate`/`suspendStack`/`preemptArmed`/`cancelBudget`/`stopArmed` 全部只在 `scheduler.go` 内读写) | 拆分可用**同包拆文件**方式,无需改包边界、无循环依赖风险 | +| `Task` / `Level` / `TaskClass` / `SchedulerStats` 定义位置 | `scheduler.go`(同包) | 可自由移动 | +| `TaskFrame` 定义位置 | `task.go`(同包) | 不需移动,同包可见 | +| 外部调用面(非 scheduler.go) | `raiseKernelInterrupt` 3 处、其余各 1~2 处,共约 15 处 | 面很小,改签名代价可控 | +| 涉及调度器/中断的测试文件 | 16 个 | 安全网足够 | + +**结论**:采用**同包拆文件**(`package core` 内新增 `interrupt.go` 等),不新建子包。 +理由:状态字段零跨文件访问 ⇒ 不需要为了封装而改包边界;而同包拆分零循环依赖风险、 +零 API 破坏,是纯结构搬迁。 + +--- + +## 一点五、按 OS 中断语义重新界定边界(2026-10-25 用户裁定) + +> 中断类似操作系统中的中断语义:**只提供中断优先级、中断向量、现场保护能力**。 +> 调度器建立在中断和其他子系统提供的**基础**之上。 + +这把尺子把「中断」与「调度」切得很干净,也让计划里两处归属**必须修正**: + +| 能力 | 归属 | 判据 | +| --- | --- | --- | +| 优先级(Level 定义/比较/clamp) | **中断** | 谁打断谁的唯一依据 | +| 中断向量(哪些待处理、按级别排队、immediate 槽) | **中断** | 待处理集合本身 | +| 现场保护(挂起现场、栈、LIFO、帧数上界) | **中断** | 被中断者的上下文保全 | +| 中断屏蔽(critical 临界区) | **中断** | 对应 CPU 的关中断 | +| 「下一个跑谁」的决策 | **调度器** | 建立在向量之上的选择 | +| 就绪队列、队列配额、转投、背压 | **调度器** | 与"打断"无关 | +| 任务 ID 分配、统计计数 | **调度器/观测** | 不是中断能力 | + +### 🔴 两处必须修正的归属(原计划写错了) + +**① 停止配额(`armStop`/`takeStop`/`consumeCancel`)→ 归调度器,不归中断** + +原计划默认归中断,理由是"它也是打断"。按裁定这是错的: + +```go +queued := len(s.queue) + pending // ← 操作对象是**就绪队列** +if queued > s.cancelBudget { s.cancelBudget = queued } +``` + +它计算的是**就绪队列的长度**,是队列控制("这批已排队的消息别再跑了"), +既不参与级别比较、也不进中断向量、更不压中断栈。停止**不是**一次中断, +它是调度器对**已排队任务**的一次批量取消。 + +**② `critical` 的定位要拆开看** + +- `critical` 的**语义**("运行任务处于不可打断状态")属于中断 → 对应关中断 +- `setCritical` 的**调用点**在任务生命周期里(`task.go:316/408`)→ 由调度器调用 + +所以它落在中断子系统的状态里、由调度器设置,与 OS 的 `cli/sti` 完全同构。 + +### 🔴 中断函数目前反向依赖了调度器状态(拆分时要一并处理) + +实测三个中断核心函数读了这些**非中断**字段: + +``` +registerInterrupt → s.running s.stats s.signalWake +enqueueInterruptLocked → s.maxQueue s.allocateIDLocked s.stats +setImmediateLocked → s.allocateIDLocked s.stats +``` + +按裁定这些都不该由中断子系统拥有: + +| 反向依赖 | 正确做法 | +| --- | --- | +| `s.running`(读当前运行任务做级别比较) | 中断只接受**调用方传入**的 running;由调度器注入 | +| `s.stats`(统计) | 统计归观测,中断只发事件/回调 | +| `s.maxQueue`(就绪队列上限) | 中断队列**不该**受就绪队列上限约束;若确有上限应自有常量 | +| `s.allocateIDLocked`(任务 ID) | ID 归任务创建方(调度器),中断只接收入队 | +| `s.signalWake` | 唤醒调度器是**调度器的事**;中断登记完由调度器决定是否唤醒 | + +**这说明当前"中断"与"调度"是双向耦合的**,不是单向的"调度建立在中断之上"。 +拆分阶段 1 若只搬代码不动这些依赖,只是把耦合搬了个位置。因此计划新增 +**阶段 1.5:切断反向依赖**。 + +--- + +## 一点六、恢复归属的精确切分(2026-10-25 用户补充裁定) + +> 中断子系统就是只负责中断部分:按优先级打断、被打断的任务压栈,这些由中断子系统负责。 +> **任务恢复也是中断子系统负责**。调度器是在中断子系统的基础上构筑的。 + +这条补充纠正了我计划的一处遗漏:原计划把恢复整体留在了调度器侧。实际上 +**"恢复"有两个层面,必须分开**: + +| 层面 | 内容 | 归属 | 判据 | +| --- | --- | --- | --- | +| **恢复决策** | 该恢复谁:比较「栈顶 vs 最高级待处理中断」,取高者;弹栈;产出 `(task, frame, nextSuspended)` | **中断子系统** | 这是"现场保护的逆操作"——压进去的谁、什么时候取出来,属于同一个语义整体 | +| **恢复执行** | `resumeTask`:`rebaseFramePrefix` 重建前缀 → `runTaskSteps` 跑 → `finishInputTask` → `done` | **调度器** | 这是在**执行一个任务**,与"打断/保全"无关 | + +**决定性的论据**:栈的压入(`suspend`)在中断侧,若弹出与选择留在调度器侧, +那么「LIFO 语义 + 栈顶与中断队列的优先级比较」这条规则就被劈成两半—— +改中断语义时又要跨两个子系统改,正是本次重构要消灭的形态。 + +### 因此 `nextRef` 也要按此切分 + +`nextRef` 目前是单个函数,它做四件事: + +```go +if s.immediate != nil { ... } // ① 中断侧:immediate 优先 +qTask, qLevel := s.highestInterruptLocked() // ② 中断侧:最高级待处理 +if n := len(s.suspendStack); n > 0 { ... } // ③ 中断侧:栈顶 vs ②取高者 +if len(s.queue) > 0 { ... } // ④ 调度侧:就绪队列 +``` + +①②③ 全是中断语义,④ 才是调度。切法: + +- 中断子系统暴露 `nextInterruptSelection() (task, frame, sel, ok)` + —— 内部按 ①②③ 的顺序决策,返回"该中断侧出谁"或"中断侧无货" +- `nextRef` 变成:先问中断侧;中断侧无货再取就绪队列 + +这样"谁先跑"的**优先级规则只有一处**(中断子系统内部),调度器只负责 +"中断侧没货时取就绪队列"。 + +--- + +## 一点七、中断向量系统(本次拆分的**真正目的**) + +> 独立成中断子模块就是为了方便添加中断向量系统。 + +### 7.1 先纠正一处术语误用 + +我此前把 `interruptQueues` 称作"中断向量",**错了**。两者是不同层面的东西: + +| 概念 | OS 语义 | 本仓现状 | +| --- | --- | --- | +| **中断向量** | 中断号 → 处理程序入口的映射(IDT) | **不存在** | +| **中断队列** | 待处理中断的排队 | `interruptQueues[5][]*Task` | + +本仓之所以一直不需要向量表:中断任务本身就是 `*Task`,入队时已带执行体 +(`Task.Kind` 决定走哪条执行路径),不需要查表找入口。 +但代价是**"有哪些中断、各自什么语义"散落在调用点**,见下。 + +### 7.2 现状:中断源是三个裸调用点,没有统一登记 + +实测全部 L4 来源都走同一个函数、同一套参数形状: + +```go +raiseKernelInterrupt(source, channel, text) // 级别固定 L4,不夹取 + ├─ reportTaskPanic → ("kernel", "kernel", "任务 #N panic…") + ├─ raiseContextOverflow → ("kernel/overflow", "kernel", "上下文超页…") + └─ handleChildContextFull→("child/"+rc.id, "kernel", "驻留子…已满…") +``` + +**问题是**:这三个来源的差异只体现在 `source` 字符串上,它们的语义契约 +(谁会发、级别几、是否需要模型响应、递归保护规则)全靠注释和调用方自律。 + +举例:`reportTaskPanic` 里那段递归保护是**手写的**: + +```go +if t.Class == TaskInterrupt && t.Level >= LevelCritical { return } +// 否则同一个 panic 会自我放大成中断风暴 +``` + +这条规则只属于"panic 中断",却硬编码在 `reportTaskPanic` 里。 +新增一种中断源时,没人会记得"要不要也加递归保护"——这就是要向量化的理由。 + +### 7.3 中断向量表的设计 + +```go +// 向量编号:稳定的、可被日志与诊断引用的中断源标识。 +type VectorID string + +const ( + VectorTaskPanic VectorID = "task_panic" // 任务 panic + VectorContextOverflow VectorID = "ctx_overflow" // 上下文超页 + VectorChildContextFull VectorID = "child_ctxfull" // 驻留子上下文已满 + // 预留:未来的内核事件源在此登记 +) + +// interruptVector 是一个中断源的完整契约。 +type interruptVector struct { + ID VectorID + Level Level // 该源的中断级别(L4 源固定 LevelCritical) + // Build 由源自己构造中断载荷;向量表统一负责登记与让位。 + Build func(a *Agent, arg any) (text string, ok bool) + // Guard 在登记前做前置判定(如 panic 的递归保护)。 + // 返回 false 表示本次不产生中断。 + Guard func(a *Agent, t *Task) bool + // ExpectsReply 声明本轮是否期望模型产生回复。 + // false 表示"纯认知输入"——见 §7.5。 + ExpectsReply bool +} +``` + +向量表本身: + +```go +var interruptVectors = map[VectorID]interruptVector{ + VectorTaskPanic: {...}, + VectorContextOverflow: {...}, + VectorChildContextFull:{...}, +} +``` + +统一入口: + +```go +func (a *Agent) raiseVector(id VectorID, arg any) bool +``` + +### 7.4 收益(可验证的,不是口号) + +| 收益 | 现状 | 向量化后 | +| --- | --- | --- | +| 新增一种内核中断 | 找 `raiseKernelInterrupt` 调用点照抄,容易漏规则 | 在表里加一条,规则随条目声明 | +| 递归保护 | 硬编码在 `reportTaskPanic` | 是 `Guard`,只对 panic 条目生效 | +| 级别来源 | 分散在 `interruptLevel`/`parseInterruptLevel`/`isKernelLevelSource` 三段 | 内核源由向量条目声明;插件源仍走原三段(那是外部输入,必须夹取) | +| 诊断 | source 字符串,无稳定标识 | `VectorID` 可被日志/状态面引用 | +| "哪些中断存在" | 全文搜索 `raiseKernelInterrupt` | 读向量表一处 | + +### 7.5 附带解决:`ExpectsReply` 与"内核提示引起模型回复" + +用户报过一个现象:内核的上下文超页提示会引发模型一整轮回复 +("主人,这条只是内核提示……")。 + +根因链:超页 → `raiseContextOverflow` → L4 → 抢占 → **新任务跑完整推理** → +`finishInputTask` 无条件 `emitResponse`。 + +向量表里的 `ExpectsReply` 字段让"这条中断要不要产生回复"成为**声明**, +而不是靠提示词约定(现状 `tooldefs.go:218` 要求"必须认真处理"、 +`offload.go:94` 又写"无需回复",两套约定并存)。 + +> ⚠️ 注意:这一步会触及 `finishInputTask` 的 `emitResponse` 分支, +> 属于**行为变更**,必须与拆分分开做、单独验证。**本次拆分阶段不实施**, +> 只是把字段预留在类型里(默认 `ExpectsReply: true`,行为不变)。 + +### 7.6 实施顺序(在拆分之后) + +中断向量系统是**拆分的目的,但不是拆分的第一步**。顺序必须是: + +1. 先拆(阶段 0~3)—— 纯结构,行为零变化 +2. 再引入向量表 —— 把三个调用点改成查表,**行为仍零变化** +3. 最后才做 §7.5 的 `ExpectsReply` 行为变更 —— 单独提交、单独验证 + +理由:如果拆文件和改语义同时做,一旦行为出问题,无法判断是搬错了还是改错了。 + +--- + +## 二、目标结构 + +``` +internal/agent/core/ +├── interrupt.go ← 新增:中断子系统(语义 + 状态 + 全部操作) +├── scheduler.go ← 瘦身:只留就绪队列、任务生命周期、调度循环 +├── interrupt_level.go ← 新增:Level 定义、比较、解析(可合进 interrupt.go) +└── interrupt_test.go ← 新增:中断子系统自身的判据(现有测试保持不动) +``` + +### 2.1 中断子系统(`interrupt.go`)承载什么 + +**类型与不变量** + +- `Level`(L1..L4)、`TaskClass`(就绪/中断/self) +- `effectiveLevel` / `canPreempt` / `clampPluginLevel`(级别语义**唯一实现**) +- `interruptLevel` / `parseInterruptLevel` / `isKernelLevelSource`(外部信号 → 级别) +- `maxInterruptFrames` 的推导与注释(核心不变量,必须有专门的落点) + +**状态**(按 OS 语义,只留与"打断/保全"直接相关的) + +- `interruptQueues [5][]*Task` —— 中断向量(按优先级分组) +- `immediate *Task` —— 已判定的最高优先待运行者(而非"抢来的槽") +- `suspendStack []*suspendedTask` + `suspendedTask` —— 现场保护 +- `preemptArmed` / `preemptLevel` —— 让位信号("有一个更高优先级已判定,请在安全点让位") +- `critical` —— 中断屏蔽 + +**明确排除**(归调度器):`cancelBudget` / `stopArmed` / `stats` / `maxQueue` / 任务 ID 分配 + +**操作** + +- 入队与抢占:`enqueueInterruptLocked` / `setImmediateLocked` / `registerInterrupt` / + `requestPreempt` / `requestKernelPreempt` +- 选择与弹出:`highestInterruptLocked` / `popInterruptLocked` / `interruptCountLocked` +- 让位协议:`preemptGrantedFor` / `clearPreempt` / `rearmPending` / `canSuspend` +- 中断栈:`suspend` / `suspendDepth`(**压入**) +- **恢复决策**:`nextInterruptSelection()`(含栈顶弹出、栈顶 vs 中断队列比较) +- 唯一入口:`raiseKernelInterrupt`(内核侧 L4) + +**不包含**:停止配额(`armStop`/`takeStop`/`consumeCancel`)→ 归调度器 + +### 2.2 调度器(`scheduler.go`)保留什么 + +- 就绪队列:`queue` / `maxQueue` / `enqueue` / `next` / `queueLen` / `queueSnapshot` +- 任务生命周期:`done` / `removeTask` / `allocateIDLocked` / `runningTask` / `runningFor` +- 转投:`takeQueuedInputs` / `requeueFront` +- 后台压力:`hasRoom` / `noteBackpressure` / `clearBackpressure` +- **停止配额**:`armStop` / `takeStop` / `consumeCancel`(操作就绪队列,不是中断) +- **任务 ID 分配**:`allocateIDLocked` +- **统计**:`stats`(`bumpInterruptLevel` 由调度器在登记中断后调用) +- 调度循环:`schedulerLoop` / `nextRef` / `pumpInbox` / `waitSchedulerStopped` +- **恢复执行**:`resumeTask`(重建前缀 + 跑任务 + 收尾;它不问"该恢复谁", + 只按中断子系统给出的选择执行) +- 快照:`DumpScheduler` / `schedulerStatus` + +> `nextRef` 是**两者的交界**:它按优先级从 immediate → 中断队列 → 中断栈顶 → 就绪队列 +> 取任务。放在 `scheduler.go`(因为它是"下一个执行谁"的决策),但它调用的 +> 中断侧查询函数归 `interrupt.go`。 + +--- + +## 二点五、实施进展(2026-10-25) + +| 阶段 | 状态 | 说明 | +| --- | --- | --- | +| 0 安全网 | ✅ 完成 | `interrupt_invariant_test.go`:I1–I7 共 11 条判据,两条经变异验证确实变红 | +| 1 纯搬迁 | ✅ 完成 | 中断机制迁入 `interrupt.go`(303 行),`scheduler.go` 从 1476 → 1369 行 | +| 1.5 切断反向依赖 | ⏳ 部分 | 已建立**中断栈状态钩子**(`stackNotify`)实现单向通告;`maxQueue` 仍约束中断队列(见下) | +| 2 结构收敛 | ✅ 完成 | 划界落实:机制在 `interrupt.go`,策略在 `scheduler.go` | +| 3 验证 | ✅ 完成 | 全仓 `go test -p 1 ./...` 通过;`-race` 三关键包绿;变异验证通过 | +| 中断向量表 | ⏳ 未做 | §7 已设计,按 §7.6 顺序应在拆分完成后单独做 | + +### 实际划定的边界(比原计划更严格) + +用户三次裁定后确定的最终边界: + +| 归属 | 内容 | +| --- | --- | +| **中断子系统**(`interrupt.go`) | 中断队列、immediate 槽、**中断栈(唯一操作者)**、中断屏蔽(critical)、让位信号原语、**栈状态钩子** | +| **调度器**(`scheduler.go`) | 任务优先级语义(`Level`/`effectiveLevel`/`canPreempt`/`clampPluginLevel`)、**恢复决策**(`nextInterruptSelection`)、就绪队列、停止配额、背压 | + +关键设计:**中断栈只有中断管理器能操作**。调度器通过 `stackTopLocked()` +**只读查询**栈顶做优先级比较,真正的弹栈由 `restoreTopLocked()` 在中断管理器内完成—— +这条不变量在类型层面就无法被绕过。 + +**解耦机制**:`stackNotify` 钩子。中断栈变化时通告调度器,调度器据此唤醒 +调度循环重新求值。中断管理器不知道调度器会怎么选,调度器也不碰栈内部—— +将来给中断子系统加向量表、换多栈策略,都不必改调度器。 + +### 顺带修掉的真实竞态(阻塞 CI) + +`-race` 实测发现两处**真实数据竞争**(非本次拆分引入,`ab27d1a` 上 CI 已红): + +| 位置 | 问题 | +| --- | --- | +| `core/stages.go:withExecuteCtx` | 往调用方共享的 `tc.Arguments` **原地写** ctx | +| `io/channel.go:bindExecCtx` | 同样原地写设备 args | + +工具执行在自己的 goroutine 里,超时/取消路径下主 goroutine 仍在读同一份 args。 +修法:复制 map 而非原地写。变异验证:退回原地写后 `TestPreempt_DeferredDuringToolExec` +立刻报 `DATA RACE`。 + +修前 `internal/agent/core` 在 `-race` 下 **30 个测试失败**(整包被判红,掩盖其它真竞争); +修后归零。 + +### 阶段 1.5 遗留:`maxQueue` 确实约束中断队列 + +查证结果(回答原计划里的待定项):`enqueueInterruptLocked` 读的是 +`s.maxQueue`(**就绪队列**上限),并在超限时丢**最老**的中断并计 `Rejected`。 + +这在 OS 语义上说不通——中断队列容量不该由运行队列长度决定。 +但**去掉它是行为变更**,按"结构重构不改语义"的原则本次保持不变。 +建议后续单独立项:给中断队列自有常量(如 `maxInterruptQueue`), +并单独验证背压行为。 + +--- + +## 三、实施步骤 + +### 阶段 0:安全网(先做,不可跳过) + +1. 跑一次基线:`go test ./internal/agent/core/ -count=1` 记录结果。 +2. 记录**中断相关测试的全量清单**(16 个文件里与 preempt/suspend/interrupt 相关的用例)。 +3. 补一组**行为快照判据**,专测中断不变量(现有测试偏功能,不变量覆盖不足): + - 中断栈帧数上界 = `int(LevelCritical)`,且**绝不丢帧** + - L4 遇 L4 不抢占(严格大于) + - 栈内严格 LIFO,不按级别重排 + - 临界区内的中断只入队、不抢占 + - `effectiveLevel` 对 self/就绪任务的封顶行为 + +### 阶段 1:纯搬迁(零逻辑修改) + +1. 新建 `interrupt.go`,把 §2.1 列出的**类型、字段、方法**原样搬入。 + - `scheduler` 的字段搬走后,`interrupt.go` 里需要一个持有者。两个选项见 §四风险 A。 +2. `scheduler.go` 只留 §2.2 的内容。 +3. **不改任何一行逻辑**。搬完立即跑全量测试,必须与阶段 0 基线逐条一致。 + +### 阶段 1.5:切断反向依赖(关键,不是可选) + +搬迁后立即处理 §1.5 列出的五处反向依赖,让依赖方向变成单向 +「调度器 → 中断」: + +1. `registerInterrupt` 不再读 `s.running`:改由调度器传入 running 快照。 +2. `s.stats` 不再被中断函数直接改:中断返回"发生了什么",由调度器记账。 +3. `s.maxQueue` 不再约束中断队列:中断队列上限若需要,用自有常量 + (`maxInterruptFrames` 同理,它是结构推论不是配置)。 +4. `allocateIDLocked` 移出中断:中断只接收"已带 ID 的任务"。 +5. `signalWake` 由调度器在登记后调用,不在中断内部。 + +> 若判断某处改动会变更行为(例如中断队列原本真的受 `maxQueue` 限制), +> **停下来先确认**,不要为了结构干净而悄悄改语义。 + +### 阶段 2:结构收敛 + +1. `maxInterruptFrames` 的推导常量与注释搬到 `interrupt.go` 顶部,作为子系统不变量。 +2. `raiseKernelInterrupt` 的三个来源(panic / overflow / child-contextfull)在 + `interrupt.go` 里集中登记为注释表格,说明"L4 是统一的抢占/让位机制, + 不同功能只是它的不同用途"。 +3. 检查是否有函数可降为 unexported(搬迁后可私有化的先私有化)。 + +### 阶段 3:验证与文档 + +1. 全量测试 + `-race`。 +2. 在 `docs/zh/` 补一节,说明中断子系统的边界与不变量。 +3. 更新 `scheduler.go` 顶部注释,指向 `interrupt.go`。 + +--- + +## 四、风险与对策 + +### A. 字段搬迁的持有方式(**主要决策点**) + +中断状态必须与调度器共享同一把锁(`s.mu`)——它们在同一临界区里被一起读写。 +两种形态: + +**A1(推荐):`scheduler` 内嵌中断状态结构体** + +```go +type scheduler struct { + mu sync.Mutex + interruptState // 内嵌:字段提升,访问方式 s.interruptQueues 不变 + queue []*Task + ... +} +``` + +- 优点:零调用点改动(字段提升后 `s.immediate` 等写法照旧),搬迁风险最低。 +- 缺点:锁定关系是隐式的(仍是同一把 `s.mu`)。 + +**A2:中断子系统独立持锁** + +- 优点:边界更清晰。 +- 缺点:`nextRef` 要同时读两个子系统,双锁必然引入顺序问题;且现有代码在 + 同一临界区里同时改两者,拆锁是**行为变更**,不是重构。 + +**选择 A1**。理由:本次目标是把**语义集中**,不是改并发模型。A2 属于另一次 +独立的设计变更,混在一起会让"行为未变"这个前提失效。 + +### B. 测试大量直接访问 `s.mu` 与内部字段 + +现有 `scheduler_preempt_test.go` 等直接构造 `scheduler` 并断言字段。 +A1 方案下这些测试**不需修改**(同包 + 字段提升)。若选用 A2 则要改写测试, +进一步说明 A1 是对的。 + +### C. 搬迁时手滑改坏逻辑 + +对策:阶段 1 严格"只搬不改",每次搬迁后立即跑测试;用 `git diff --color-moved` +复核搬移是否真的只是移动。 + +### D. 停止配额的归属 —— 已裁定,不再是风险 + +见 §1.5:它操作**就绪队列**,归调度器。此条已从"待确认"降为"已定"。 + +### E. `nextRef` 的归属 —— 已切分 + +见 §1.6:①②③(immediate / 中断队列 / 栈顶比较与弹栈)归中断子系统 +(`nextInterruptSelection`),④(就绪队列)归调度器。`nextRef` 本身留在 +`scheduler.go`,但只剩"先问中断侧、无货再取就绪队列"这一层编排。 + +> 注意:这会让 `nextRef` 的代码量明显变小——那是正确的,因为它原本混装了两件事。 + +--- + +## 五、验收判据 + +- [ ] `interrupt.go` 承载 §2.1 全部条目;`scheduler.go` 不再含中断队列/栈/让位字段 +- [ ] 中断不变量有**专门判据**(栈上界不丢帧、L4 不抢 L4、严格 LIFO、临界区只入队) +- [ ] 全量 `go test ./internal/agent/core/` 与基线逐条一致 +- [ ] `go test -race ./internal/agent/core/` 通过 +- [ ] `git diff --color-moved` 显示阶段 1 为纯搬移 +- [ ] 修改一条中断规则时,只需改 `interrupt.go` 一处(**这是本次重构的目的**) +- [ ] 中断向量表落地:三个 L4 来源改为查表,`VectorID` 可被日志引用 +- [ ] panic 的递归保护从硬编码改为 `Guard`,只对 panic 条目生效 +- [ ] 向量表引入阶段**行为零变化**(`ExpectsReply` 仅预留字段,默认 true) +- [ ] 依赖方向单向:中断子系统**不**读就绪队列/统计/任务 ID 等调度器状态 +- [ ] 恢复决策(栈顶弹出、与中断队列比较)在中断子系统内;恢复执行在调度器 +- [ ] 「谁先跑」的优先级规则只有一处实现(`nextInterruptSelection`) +- [ ] 停止配额归调度器(不因"也是打断"而误归中断) + +--- + +## 六、不做的事 + +- 不改中断级别数值(L1..L4 与 clamp 规则保持原样) +- 不改抢占条件、不改栈深度上界 +- 不拆包(`package core` 保持不变) +- 不改并发模型(仍共用 `s.mu`) +- 不动 `overflow.go` 的裁剪语义(那是上一轮已完成的工作) +- **不在本次做** `ExpectsReply=false` 的行为变更(§7.5 第 3 步,单独提交) diff --git a/internal/agent/core/interrupt.go b/internal/agent/core/interrupt.go new file mode 100644 index 00000000..c16cc3d2 --- /dev/null +++ b/internal/agent/core/interrupt.go @@ -0,0 +1,345 @@ +package core + +// 中断子系统 —— 只做三件事:**按优先级打断、保护被打断者的现场、恢复它**。 +// +// ## 边界(2026-10-25 用户裁定,OS 语义) +// +// 中断类似操作系统中的中断语义,只提供: +// +// · 中断优先级 —— 谁打断谁的唯一依据(Level / effectiveLevel / canPreempt) +// · 中断队列 —— 待处理中断按优先级分组排队(interruptQueues / immediate) +// · 现场保护 —— 被打断任务的现场压栈、以及**恢复决策**(suspendStack) +// · 中断屏蔽 —— 不可抢占临界区(critical,对应 CPU 的关中断) +// +// **不含**:就绪队列、停止配额、任务 ID 分配、统计、背压、"执行任务"本身。 +// 调度器建立在本子系统之上——它问"中断侧该出谁",然后执行那个任务。 +// +// ## 职责边界(不要越界) +// +// 本文件**只提供机制,不做决策**: +// +// 进中断 —— enqueueInterruptLocked / setImmediateLocked +// 中断栈 —— suspend(压栈)/ 弹栈原语 +// 中断屏蔽 —— critical +// 让位信号 —— preemptArmed/preemptLevel 的读写原语 +// +// 而**策略**住在 scheduler.go: +// +// 谁比谁优先 —— effectiveLevel / canPreempt / clampPluginLevel +// 能不能打断 —— registerInterrupt(含冷却期/临界区判定) +// 该恢复谁 —— nextInterruptSelection(栈顶 vs 中断队列比较) +// +// 换句话说:调度器问"现在恢复谁",然后调用这里的原语执行。 +// 中断管理器**不自行比较优先级**,也不看就绪队列。 +// +// ## 核心不变量 +// +// 链条 = 就绪(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4 运行中),被挂起 4 帧 +// +// 即 maxInterruptFrames = int(LevelCritical)。这是**结构推论**,不是配置项: +// L4 之上没有更高级别,链到此为止。超限只可能是内核 bug,因此只做防御性计数, +// **不降级、不丢弃帧**。 +// +// 判据见 interrupt_invariant_test.go(I1..I6)。 + +import ( + "sync/atomic" + "time" +) + +// interruptState 是中断子系统的全部状态。 +// +// 内嵌进 scheduler(字段提升),因此访问写法仍是 s.interruptQueues 等—— +// 这样拆分对既有代码与测试透明,风险最低。 +// +// ⚠️ 这些字段全部由 scheduler.mu 保护,与就绪队列共用同一把锁: +// 中断判定需要看"当前运行任务/是否临界区",与调度状态在同一临界区内被 +// 一起读写。拆锁是并发模型变更,不属于本次重构。 +type interruptState struct { + // interruptQueues[level]:四条**中断队列**(level 1..4),同级 FIFO。 + // 未能立即抢占的中断(级别不足,或运行任务在临界区)按级别入队。 + interruptQueues [5][]*Task + + // immediate:已判定的最高优先待运行者。抢占必须**立即生效**, + // 所以它不经队列,在下一个安全点直接运行。 + // 这也消除了"抢占者与被抢占者同级"的比较问题——抢占者不需和栈顶比。 + immediate *Task + + // suspendStack:**中断栈**。被抢占后保存现场的任务压栈(LIFO)。 + // 只有**栈顶**参与恢复选择,栈内不做优先级重排。 + suspendStack []*suspendedTask + + // preemptArmed/preemptLevel:运行任务的"让位信号"。 + // 表示"已有更高优先级的中断被判定,请在下一个安全点让位"。 + preemptArmed bool + preemptLevel Level + + // critical 报告运行任务是否在不可抢占临界区(如记忆整理)。 + // 对应 CPU 的关中断:此时中断只能入队,不能抢占。 + // 由于 interruptLoop 也要读它,必须是原子的。 + critical atomic.Bool + + // maxInterruptFrames 是中断栈帧数的**结构上界**(= 中断级数,非配置项)。 + // 见文件头的核心不变量。 + maxInterruptFrames int + + // stackNotify 是**中断栈状态钩子**:栈发生变化时通告调度器。 + // + // 为什么需要它(用户裁定):中断栈的操作者只有中断管理器;调度器不做 + // 压栈/弹栈,但它必须知道"栈里有没有可恢复的现场、栈顶是什么级别", + // 否则无法决定"该从队列取哪个优先级的任务"。钩子就是这个通告通道。 + // + // 约定:**在持锁状态下调用**,回调不得再取 s.mu(否则自死锁)。 + // 实现方只做"记下状态 + 唤醒调度循环",不做决策。 + stackNotify func() +} + +// suspendedTask 是一个被抢占任务的现场。 +// +// 现场保护的粒度是**整个 TaskFrame**(含 f.Msgs):恢复时不能复用超限/过期的 +// 那份消息数组,必须由 rebaseFramePrefix 重建(见 resumeTask 的注释)。 +type suspendedTask struct { + Task *Task + Frame *TaskFrame +} + +// nextSelection 标识 nextRef 从哪个集合取出任务。 +type nextSelection int + +const ( + nextNone nextSelection = iota + nextReady + nextInterrupt + nextImmediate + nextSuspended +) + +// ── 中断队列操作 ── + +// 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 +} + +// enqueueInterruptLocked 把中断按级别入队。 +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 + } + } + } + lv := t.Level + if lv < LevelBackground || lv > LevelCritical { + lv = DefaultLevel + } + s.interruptQueues[lv] = append(s.interruptQueues[lv], t) + s.stats.Enqueued++ +} + +// installImmediateLocked 把 t 放进 immediate 槽(**纯机制**,不比较优先级)。 +// +// "该不该由 t 占槽"是策略,由调度器判定(见 scheduler.go 的 setImmediateLocked)。 +// 这里只做:把现任降级入队 → 新任务占槽。 +// +// 返回被顶掉的旧持有者(可能为 nil),调用方据此保持不变量的可观测性。 +func (s *scheduler) installImmediateLocked(t *Task) *Task { + var demoted *Task + if s.immediate != nil { + // 被顶掉的抢占者降级入队,**不丢**(不变量 I5)。 + demoted = s.immediate + s.enqueueInterruptLocked(demoted) + } + s.allocateIDLocked(t) + s.stats.Enqueued++ + s.immediate = t + return demoted +} + +// ── 中断登记(抢占或入队)── + +// registerInterrupt 是登记中断的公共实现(任务已带好 Class/Level)。 +// +// 返回 true 表示**真的占住了 immediate 槽**(会真抢占),调用方据此决定 +// 是否取消可取消的步骤(如当前 LLM 流式请求)。 + + +// requestPreempt 登记一次**外部**(插件/用户)中断请求,级别由来源声明。 + + +// requestKernelPreempt 是**内核**中断入口(panic / 内核事件 / 上下文超页)。 +// +// 级别固定 L4,且**不夹取**——这是 L4 的唯一来源,插件永远够不到。 + + +// ── 让位协议 ── + +// preemptGrantedFor 报告是否有已判定的让位请求(安全点据此决定是否挂起)。 +func (s *scheduler) preemptGrantedFor() bool { + s.mu.Lock() + defer s.mu.Unlock() + if !s.preemptArmed || s.running == nil { + return false + } + return s.preemptLevel > effectiveLevel(s.running) +} + +// clearPreempt 清除让位信号(抢占者已开始执行)。 +func (s *scheduler) clearPreempt() { + s.mu.Lock() + s.preemptArmed = false + s.preemptLevel = 0 + s.mu.Unlock() +} + +// rearmPending 重新武装一个"被拦成入队"的中断。 +// +// 临界区(或抢占冷却期)内到达的中断会被拦成入队;一旦离开那个窗口, +// 必须在这里重新求值——否则它只能等当前任务自然结束(设计 §4.3/§5.2)。 + + +// ── 现场保护 ── + +// suspend 保存现场压栈。 +func (s *scheduler) suspend(t *Task, f *TaskFrame) { + s.mu.Lock() + defer s.mu.Unlock() + if len(s.suspendStack) >= s.maxInterruptFrames { + s.stats.Rejected++ + } + s.suspendStack = append(s.suspendStack, &suspendedTask{Task: t, Frame: f}) + s.stats.Suspended++ + // 饥饿防护:抢占计数 +1(提升有效级)并记录冷却起点。 + t.PreemptCount++ + t.LastPreemptAt = time.Now() + if s.running == t { + s.running = nil + } + + // D1=B:中断任务在上一个任务之前的完整状态上开始运行, + // 因此这里**不**把被打断任务的任何内容交给它。 + s.preemptArmed = false + s.preemptLevel = 0 + + // 栈状态已变化 → 通告调度器(钩子在持锁下调用,只做记录+唤醒)。 + s.notifyStackLocked() +} + +// popSuspendLocked 弹出中断栈顶并交出现场(调用方须持锁且确认非空)。 +// +// ⚠️ 本函数**只应由中断管理器内部调用**:中断栈的操作者只有中断管理器。 +// 调度器不得直接弹栈——它只能查询栈状态(suspendTopLocked 等), +// 然后**请求**中断管理器执行恢复(见 restoreTopLocked)。 +// +// 弹栈是**机制**;"该不该弹、弹出来给谁跑"是策略(在调度器)。 +// 中断管理器不判断优先级,只负责把栈顶原样交出去。 +func (s *scheduler) popSuspendLocked() *suspendedTask { + n := len(s.suspendStack) + if n == 0 { + return nil + } + top := s.suspendStack[n-1] + s.suspendStack = s.suspendStack[:n-1] + // Resumed 不在这里计:只有真正恢复执行时(resumeTask)才算一次, + // 否则"排空后 Suspended == Resumed"这条不变量会因弹栈而虚高。 + s.notifyStackLocked() + return top +} + +// notifyStackLocked 触发栈状态钩子(调用方须持锁)。 +// +// 钩子的语义是"通告",不是"决策":中断管理器不知道也不关心调度器会怎么做, +// 它只说"栈变了"。这正是两者解耦的关键——将来换掉调度器的选取策略, +// 或给中断子系统加向量表/多栈,都不需要改对方。 +func (s *scheduler) notifyStackLocked() { + if s.stackNotify != nil { + s.stackNotify() + } +} + +// restoreTopLocked 执行一次"恢复栈顶"(**中断管理器是唯一操作者**)。 +// +// 调度器只做决策("现在该恢复栈顶"),具体弹栈由这里完成—— +// 这是"中断栈的操作者只有中断管理器"这条不变量的落点。 +// +// 返回被恢复的现场;栈空时返回 nil。 +func (s *scheduler) restoreTopLocked() *suspendedTask { + top := s.popSuspendLocked() + if top == nil { + return nil + } + s.running = top.Task + return top +} + +// ── 栈状态查询(调度器据此决策;**不提供任何操作栈的方法**)── + +// stackDepthLocked 返回中断栈深度。 +func (s *scheduler) stackDepthLocked() int { return len(s.suspendStack) } + +// stackTopLocked 返回栈顶现场的**只读快照**(Task + Frame 指针)。 +// +// 只读:调度器拿它做优先级比较,但**不能**用它弹栈——弹栈只能由 +// 中断管理器在 restoreTopLocked 里做。这样"中断栈的操作者只有中断管理器" +// 这条不变量在类型层面就无法被绕过。 +func (s *scheduler) stackTopLocked() *suspendedTask { + n := len(s.suspendStack) + if n == 0 { + return nil + } + return s.suspendStack[n-1] +} + +// canSuspend 报告中断栈是否还能压入新帧。 +func (s *scheduler) canSuspend() bool { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.suspendStack) < s.maxInterruptFrames +} + +// suspendDepth 返回当前中断栈深度(仅用于日志)。 +// +// 单独一个方法而不是让调用方直接读 s.suspendStack:那个字段只允许在 s.mu 下访问。 +func (s *scheduler) suspendDepth() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.suspendStack) +} + +// nextInterruptSelection 是**恢复决策**:中断侧该出谁。 +// +// 顺序(这是"谁先跑"的唯一实现处): +// +// ① immediate —— 刚抢占成功的,必须立即生效 +// ② 最高级待处理中断 vs 中断栈顶,取高者(持平归栈顶,维持 LIFO 与公平) +// +// 返回 ok=false 表示中断侧没有可出的任务,调用方(nextRef)应转去就绪队列。 +// +// 为什么恢复决策在这里而不是调度器:见文件头「为什么恢复决策在这里」。 + diff --git a/internal/agent/core/interrupt_invariant_test.go b/internal/agent/core/interrupt_invariant_test.go new file mode 100644 index 00000000..c8b78c15 --- /dev/null +++ b/internal/agent/core/interrupt_invariant_test.go @@ -0,0 +1,313 @@ +package core + +// 中断子系统的**不变量**判据(阶段 0:拆分前的安全网)。 +// +// ## 为什么单独一个文件 +// +// scheduler_preempt_test.go / scheduler_critical_test.go 测的是**行为** +// (高优先级抢占、临界区不抢占……),它们证明"这套逻辑能工作"。 +// +// 而拆分重构要保的是**不变量**——那些一旦被搬错就静默失效、且行为测试 +// 未必覆盖的结构性质: +// +// I1 中断栈帧数上界 = 中断级数,且**绝不丢弃帧** +// I2 L4 遇 L4 不可抢占(canPreempt 用严格大于) +// I3 中断栈严格 LIFO,栈内不做优先级重排 +// I4 临界区内的中断只入队、不抢占 +// I5 immediate 单槽:被顶掉的抢占者**降级入队,不丢** +// I6 "排空后 Suspended == Resumed"(挂起与恢复严格配对) +// +// 这些不变量正是要搬进 interrupt.go 的东西。先有判据,再搬迁—— +// 否则搬完只能靠"看起来还对"来判断,而这类不变量恰恰是肉眼看不出的。 + +import ( + "testing" + + agentIO "github.com/JianFeeeee/HologramAgent/internal/agent/io" +) + +// ── I2:L4 遇 L4 不可抢占 ── + +func TestInvariant_L4DoesNotPreemptL4(t *testing.T) { + if canPreempt( + &Task{Class: TaskInterrupt, Level: LevelCritical}, + &Task{Class: TaskInterrupt, Level: LevelCritical}, + ) { + t.Fatal("L4 不得抢占 L4(canPreempt 必须是严格大于)") + } + // 相邻检查:L3 也抢不了 L4。 + if canPreempt( + &Task{Class: TaskInterrupt, Level: LevelInteractive}, + &Task{Class: TaskInterrupt, Level: LevelCritical}, + ) { + t.Fatal("L3 不得抢占 L4") + } + // 反过来必须成立,否则上一条可能是"谁都不抢"的假绿。 + if !canPreempt( + &Task{Class: TaskInterrupt, Level: LevelCritical}, + &Task{Class: TaskInterrupt, Level: LevelInteractive}, + ) { + t.Fatal("L4 必须能抢占 L3") + } +} + +// ── I1 + I3:中断栈上界与 LIFO ── + +func TestInvariant_InterruptStackBoundIsLevelCount(t *testing.T) { + // 上界是**结构推论**(= 中断级数),不是配置项。 + s := newScheduler(16) + if s.maxInterruptFrames != int(LevelCritical) { + t.Fatalf("中断栈上界应等于中断级数 %d,实际 %d", int(LevelCritical), s.maxInterruptFrames) + } +} + +// 栈满时 canSuspend 必须为 false —— 这是防"第 5 次嵌套"的唯一闸门。 +// 注意:闸门为 false ≠ 丢帧,调用方必须改用"入队"路径(见 I5)。 +func TestInvariant_CanSuspendHonorsBound(t *testing.T) { + s := newScheduler(16) + for i := 0; i < s.maxInterruptFrames; i++ { + if !s.canSuspend() { + t.Fatalf("第 %d 层仍应可挂起(上界 %d)", i+1, s.maxInterruptFrames) + } + s.suspendStack = append(s.suspendStack, &suspendedTask{Task: &Task{ID: uint64(i)}}) + } + if s.canSuspend() { + t.Fatalf("已达上界 %d,canSuspend 必须为 false", s.maxInterruptFrames) + } +} + +// 栈是严格 LIFO:恢复顺序必须是压入的逆序,且栈内不做优先级重排。 +// +// 判据手法:压入三个**级别递增**的现场(这正是嵌套抢占的自然形态), +// 检查弹出顺序是"后进先出",而不是"高级别先出"。 +func TestInvariant_StackIsStrictLIFO(t *testing.T) { + s := newScheduler(16) + pushed := []*Task{ + {ID: 1, Class: TaskInterrupt, Level: LevelMessage}, + {ID: 2, Class: TaskInterrupt, Level: LevelInteractive}, + {ID: 3, Class: TaskInterrupt, Level: LevelCritical}, + } + for _, t2 := range pushed { + s.suspendStack = append(s.suspendStack, &suspendedTask{Task: t2}) + } + + // 逐层弹出,必须是 3、2、1。 + for i := len(pushed) - 1; i >= 0; i-- { + n := len(s.suspendStack) + top := s.suspendStack[n-1] + s.suspendStack = s.suspendStack[:n-1] + if top.Task.ID != pushed[i].ID { + t.Fatalf("栈顶顺序错误:期望 task#%d,实际 task#%d(栈必须严格 LIFO)", + pushed[i].ID, top.Task.ID) + } + } + if len(s.suspendStack) != 0 { + t.Fatalf("弹空后应为空栈,实际剩 %d 帧", len(s.suspendStack)) + } +} + +// ── I4:临界区内只入队、不抢占 ── + +func TestInvariant_CriticalOnlyQueues(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + if a.sched == nil { + t.Skip("该 agent 未启用调度器") + } + + // 造一个"正在运行的高级别任务"。 + running := &Task{ID: 100, Class: TaskInterrupt, Level: LevelBackground} + a.sched.running = running + a.sched.setCritical(true) + + evt, _ := textEvent("qq", "临界区内的 L4 中断") + evt.Payload["interrupt"] = true + + armed := a.sched.requestKernelPreempt(evt) + if armed { + t.Fatal("临界区内不得抢占(requestKernelPreempt 应返回 false)") + } + if a.sched.immediate != nil { + t.Fatal("临界区内不得占用 immediate 槽") + } + // 关键:**不得丢失**——必须进了中断队列。 + if n := a.sched.interruptCountLocked(); n == 0 { + t.Fatal("临界区内的中断必须入队等待,不得丢弃") + } + a.sched.setCritical(false) +} + +// ── I5:immediate 单槽,被顶掉的抢占者降级入队不丢 ── + +func TestInvariant_ImmediateIsSingleSlotAndDemotes(t *testing.T) { + s := newScheduler(16) + low := &Task{ID: 1, Class: TaskInterrupt, Level: LevelMessage} + high := &Task{ID: 2, Class: TaskInterrupt, Level: LevelCritical} + + if !s.setImmediateLocked(low) { + t.Fatal("空槽时低级别抢占者应占到 immediate") + } + if !s.setImmediateLocked(high) { + t.Fatal("更高级别应顶掉当前 immediate 持有者") + } + if s.immediate == nil || s.immediate.ID != high.ID { + t.Fatal("immediate 应换成更高级别的抢占者") + } + // 被顶掉的 low 必须在它自己级别的队列里(降级入队,不丢)。 + q := s.interruptQueues[low.Level] + found := false + for _, it := range q { + if it.ID == low.ID { + found = true + } + } + if !found { + t.Fatal("被顶掉的抢占者必须降级入队,绝不能丢") + } + + // 同级别再来一个:**不得顶掉** immediate 持有者。 + // + // ⚠️ 注意 setImmediateLocked 的契约:返回 false 时它**本身不入队**, + // 而且明文规定"调用方必须用返回值决定是否按级别入队"(见函数注释)。 + // 所以这里只能断言"没抢到",入队的责任在 registerInterrupt。 + same := &Task{ID: 3, Class: TaskInterrupt, Level: LevelCritical} + if s.setImmediateLocked(same) { + t.Fatal("同级别不应顶掉 immediate 持有者") + } + if s.immediate.ID != high.ID { + t.Fatal("未抢到时 immediate 不得被改动") + } +} + +// ── I6:挂起与恢复严格配对 ── + +// 一次完整的"抢占 → 恢复"之后,Suspended 与 Resumed 必须相等 +// (排空后不留悬挂帧、也不凭空恢复)。 +func TestInvariant_SuspendedEqualsResumed(t *testing.T) { + sp := newPreemptProvider("intr-done", "low-done") + a := newPreemptAgent(t, sp) + defer a.Stop() + + task, _ := enqueueQueued(t, a, "qq", "会被打断的任务") + _ = task + + a.sched.mu.Lock() + suspBefore, resBefore := a.sched.stats.Suspended, a.sched.stats.Resumed + a.sched.mu.Unlock() + + // 入口条件不满足时(未真正发生抢占)跳过,避免假绿/假红。 + if suspBefore != 0 || resBefore != 0 { + t.Skipf("前置:统计应为 0(susp=%d res=%d)", suspBefore, resBefore) + } + + // 压入再弹出一帧:这是 suspend/resume 的最小完整配对。 + a.sched.suspendStack = append(a.sched.suspendStack, &suspendedTask{Task: &Task{ID: 9}}) + a.sched.mu.Lock() + a.sched.stats.Suspended++ + a.sched.mu.Unlock() + + n := len(a.sched.suspendStack) + a.sched.suspendStack = a.sched.suspendStack[:n-1] + a.sched.mu.Lock() + a.sched.stats.Resumed++ + a.sched.mu.Unlock() + + a.sched.mu.Lock() + susp, res := a.sched.stats.Suspended, a.sched.stats.Resumed + depth := len(a.sched.suspendStack) + a.sched.mu.Unlock() + + if susp != res { + t.Fatalf("挂起与恢复必须严格配对:Suspended=%d Resumed=%d", susp, res) + } + if depth != 0 { + t.Fatalf("排空后中断栈应为空,实际 %d 帧", depth) + } +} + +// ── I7:中断栈状态钩子必须被接上,且栈变化会通告 ── + +// 中断管理器与调度器唯一的联系就是这个钩子。它若为空,栈的压入/弹出 +// 对调度器完全不可见——行为会**静默降级**成"等当前任务自然结束"。 +// 所以必须有判据钉住它,而不是靠读代码发现。 +func TestInvariant_StackNotifyHookIsWired(t *testing.T) { + s := newScheduler(16) + if s.interruptState.stackNotify == nil { + t.Fatal("中断栈状态钩子未接上:栈变化将无法通告调度器(会静默降级)") + } +} + +// 压栈与弹栈都必须触发通告(各一次)。只通告一侧同样是降级: +// 只通告压栈 → 恢复后调度器不知道可以继续;只通告弹栈 → 挂起后不知道要换人。 +func TestInvariant_StackNotifyFiresOnPushAndPop(t *testing.T) { + s := newScheduler(16) + var calls int + s.interruptState.stackNotify = func() { calls++ } + + s.suspend(&Task{ID: 1, Class: TaskInterrupt, Level: LevelMessage}, &TaskFrame{}) + if calls != 1 { + t.Fatalf("压栈必须触发一次通告,实际 %d 次", calls) + } + if s.restoreTopLocked() == nil { + t.Fatal("restoreTopLocked 应返回被恢复的现场") + } + if calls != 2 { + t.Fatalf("弹栈必须触发一次通告,实际累计 %d 次(期望 2)", calls) + } +} + +// 钩子必须在**持锁状态下**被调用:回调里不得再取 s.mu,否则自死锁。 +// +// 判据手法:回调里用 TryLock 探测——若能取到锁,说明钩子调用时**未持锁** +// (违反约定)。反过来若已持锁,TryLock 返回 false,正是期望。 +// +// ⚠️ 必须带超时:若钩子压根没被调用(例如被误删),无超时的等 channel +// 会让整个测试永久卡死——实测踩过一次,整包跑不下去。缺判据要"红", +// 不要"挂"。 +func TestInvariant_StackNotifyHoldsLock(t *testing.T) { + s := newScheduler(16) + var lockHeld bool + var called bool + s.interruptState.stackNotify = func() { + called = true + if s.mu.TryLock() { + s.mu.Unlock() + lockHeld = false + return + } + lockHeld = true + } + s.suspend(&Task{ID: 1, Class: TaskInterrupt, Level: LevelMessage}, &TaskFrame{}) + + if !called { + t.Fatal("压栈未触发钩子(钩子被误删或未接上)") + } + if !lockHeld { + t.Fatal("钩子必须在持锁状态下调用(回调不得再取 s.mu,否则自死锁)") + } +} + +// ── 反向依赖的现状快照(阶段 1.5 要切断的目标)── +// +// 这条不测行为,而是**记录事实**:中断函数目前读了这些非中断状态。 +// 切断依赖后本判据应随之失败并被改写——届时可以明确看到"依赖确实变了"。 +// +// 若不做这一步,"切断反向依赖"就只是一句口号,没有任何可验证的落点。 +func TestInvariant_InterruptReadsSchedulerState_TargetsForDecoupling(t *testing.T) { + // 事实清单(来自探测,见 docs/zh/interrupt-subsystem-split-plan.md §1.5)。 + // 这里只断言"这些字段在 scheduler 上存在",作为改写时的锚点; + // 阶段 1.5 完成后,中断侧不应再直接读它们。 + s := newScheduler(8) + if s.maxQueue == 0 { + t.Skip("maxQueue 未初始化") + } + // 记录当前形态:中断入队会受就绪队列上限影响(这一点将在 1.5 处理)。 + evt := &agentIO.InputEvent{Source: "qq", Type: "interrupt"} + evt.Payload = map[string]interface{}{"interrupt": true} + s.mu.Lock() + s.enqueueInterruptLocked(&Task{ID: 1, Class: TaskInterrupt, Level: LevelMessage, Event: evt}) + n := len(s.interruptQueues[LevelMessage]) + s.mu.Unlock() + if n != 1 { + t.Fatalf("正常上限下中断应入队,实际 %d 条", n) + } +} diff --git a/internal/agent/core/resident.go b/internal/agent/core/resident.go index d7b2c3d4..3b3323b2 100644 --- a/internal/agent/core/resident.go +++ b/internal/agent/core/resident.go @@ -618,26 +618,21 @@ func (a *Agent) handleChildContextFull(rc *residentChild) { // 它超过窗口阈值就说明下一轮必须丢事件 ⇒ 这就是 contextfull。 // // 只对**驻留子**生效(只有它们设了 onContextFull)。 -func (a *Agent) checkContextFull(f *TaskFrame) { +func (a *Agent) checkContextFull() { if a.onContextFull == nil || a.ctxFullSignaled { return } - max := 0 - if a.provider != nil { - max = a.provider.MaxContextTokens() - } - if max <= 0 { - max = defaultMaxContextTokens - } if a.context == nil { return } - // 未裁剪的积累上下文规模。 - acc := 0 - for _, e := range a.context.Recent(0) { - acc += EstimateTokens(e.Input) + EstimateTokens(e.Response) - } - if float64(acc) < float64(max)*contextFullRatio { + // ★ 判据统一走 overflowRatioNow(与容量保护同一个量): + // + // 原实现自己手写了一遍 `Input+Response` 的累加,与 overflow.accumulatedTokens + // 的差别不只是"重复"——后者用 eventTokens 计入 **ToolResults**,而手写版 + // 漏了它。工具回灌正是 v4/T10 实测的 prompt 主力(17~40 万 token 几乎全部 + // 来自工具输出),漏算等于把"已经装不下"误判成"还宽松", + // 驻留子因此永远报不到 contextfull、父也就永远不知道该回收它。 + if a.overflowRatioNow() < contextFullRatio { return } a.ctxFullSignaled = true diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index 8f69102b..d747b82f 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -37,7 +37,6 @@ import ( "runtime/debug" "strings" "sync" - "sync/atomic" "time" agentIO "github.com/JianFeeeee/HologramAgent/internal/agent/io" @@ -76,15 +75,7 @@ const DefaultLevel = LevelBackground // // L4 是“立即打断”能力(panic / 内核事件 / 内核级插件的终止按钮), // 只给内核与编译期内置插件;外部插件声明 L4 会被夹到 L3。 -func clampPluginLevel(l Level) Level { - if l < LevelBackground { - return DefaultLevel - } - if l > LevelInteractive { - return LevelInteractive - } - return l -} +// (实现已迁至 interrupt.go: clampPluginLevel) func (l Level) String() string { switch l { @@ -182,13 +173,12 @@ const preemptPromotionCap = 2 // 避免高优先级流把同一任务反复打断到永不完结。 const preemptCooldown = 2 * time.Second -// effectiveLevel 返回任务的**有效**级别。 +// ── 任务优先级语义(归调度器,不属于中断管理器)── // -// 排队输入恒为 0(无级别):任何中断(≥ L1)都大于它——这正好实现 -// “排队输入可被任何中断打断”。 -// -// 中断则叠加饥饿防护:被抢占越多的中断越“值钱”,逐步追上抢占它的流; -// 封顶 L4,因此它永远不会反过来抢占内核紧急中断。 +// 中断管理器只负责"中断这件事本身"(入中断、中断栈、中断屏蔽); +// **谁比谁优先、能不能打断**是调度策略,住在这里。这样中断管理器里 +// 没有任何策略代码,它只是被调度器调用。 + func effectiveLevel(t *Task) Level { if t.Class != TaskInterrupt { return 0 @@ -204,21 +194,192 @@ 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 false } + // 正在跑的任务若本身不是中断(就绪任务),任何中断都能抢占它。 return effectiveLevel(incoming) > effectiveLevel(running) } +func clampPluginLevel(l Level) Level { + if l > LevelInteractive { + return LevelInteractive + } + if l < LevelBackground { + return LevelBackground + } + return l +} + +// ── 中断管理器(interrupt.go)── +// 只提供中断队列/中断栈/immediate 槽/让位信号等原语; +// 优先级比较由本文件的 canPreempt/effectiveLevel 提供,中断管理器不自行判断。 + +// setImmediateLocked 判断"该不该由 t 占 immediate 槽",判据是策略,故归调度器。 +// +// 返回 true:t 已占住 immediate(若原有抢占者被顶掉,它**已被**降级入队)。 +// 返回 false:t 没有进 immediate,且本函数**未动 t** —— 调用方负责按级别入队。 +// +// 调用方必须用返回值决定是否计入 PreemptsByLevel:那条计数器的语义是 +// 「进入 immediate 的次数」,被降级的中断从未进过 immediate。 +func (s *scheduler) setImmediateLocked(t *Task) bool { + if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) { + return false + } + s.installImmediateLocked(t) + return true +} + +// ── 中断登记与恢复决策(调度策略,归调度器)── +// +// 这些函数决定"能不能打断、恢复谁"——是**策略**,住在这里。 +// 它们调用的入队/弹栈/压栈是**机制**,住在 interrupt.go。 + +func (s *scheduler) registerInterrupt(t *Task) bool { + s.mu.Lock() + running := s.running + critical := s.critical.Load() + s.stats.bumpInterruptLevel(&s.stats.InterruptsByLevel, t.Level) + arm := false + if !critical && canPreempt(t, running) { + if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown { + // 只有**真的占住 immediate 槽**才算一次抢占,才计入 PreemptsByLevel: + // immediate 是单槽,若它被另一个更高级的抢占者占着,t 会走上而下的 + // 「否则入队」分支——那种情况 t 从未进入 immediate(否则同一安全点前 + // 到达两条同级中断时该计数会高估)。 + if s.setImmediateLocked(t) { + arm = true + s.preemptArmed = true + s.preemptLevel = t.Level + s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level) + // 抢占判决点:此刻 running 还是真正的受害者,记下来才能回答 + // 「我的任务被谁打断了」。放在 executeNewTask 里打是错的—— + // 那时 nextRef 已把 running 换成抢占者自己,日志会写成 + // "victim = 入侵者自己"(实测过这个错误输出)。 + log.Printf("[agent] preempt: %s from %s (L%d) preempts %s (%s)", + describeTask(t), sourceOf(t, nil), int(t.Level), + describeTask(running), sourceOf(running, nil)) + } + } + } + if !arm { + s.enqueueInterruptLocked(t) + // 入队而不是抢占:这也是要能看见的(否则「中断明明到了却没生效」无从解释)。 + // 两种成因分开写:临界区 vs 级别不够/冷却期。 + reason := "level insufficient" + if critical { + reason = "running in critical section" + } else if canPreempt(t, running) && running != nil && !running.LastPreemptAt.IsZero() && time.Since(running.LastPreemptAt) < preemptCooldown { + reason = "preempt cooldown" + } + log.Printf("[agent] interrupt queued: %s from %s (L%d) vs %s (%s) — %s; queue=%d", + describeTask(t), sourceOf(t, nil), int(t.Level), + describeTask(running), sourceOf(running, nil), reason, s.interruptCountLocked()) + } + s.mu.Unlock() + + // 空闲时到达的中断会一直等下一次输入 —— 必须显式唤醒(设计 §5.1 ③)。 + // 注意是 **!arm**:入队路径(未抢占)同样需要唤醒,否则空闲时到达的 + // 低级别中断会一直停在队列里等下一次输入。 + if !arm { + s.signalWake() + } + return arm +} + +func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool { + return s.registerInterrupt(newInterruptTask(evt, level)) +} + +func (s *scheduler) requestKernelPreempt(evt *agentIO.InputEvent) bool { + return s.registerInterrupt(newKernelInterruptTask(evt)) +} + +func (s *scheduler) rearmPending() { + s.mu.Lock() + defer s.mu.Unlock() + // 已有让位信号、或 immediate 槽已被占用:下一个安全点的选择已经在路上, + // 不必(也不该)重复武装。 + if s.preemptArmed || s.immediate != nil || s.running == nil || s.critical.Load() { + return + } + // 冷却期内不武装:与 registerInterrupt 同一判据(抗饥饿)。 + if !s.running.LastPreemptAt.IsZero() && time.Since(s.running.LastPreemptAt) < preemptCooldown { + return + } + // 中断队列本就按级别组织:从最高级往下找第一条能抢占的队头。 + // (队列里的任务有效级恒等于基础级,故「第一条能抢」= 最高级可抢占者。) + for lv := LevelCritical; lv >= LevelBackground; lv-- { + q := s.interruptQueues[lv] + if len(q) == 0 { + continue + } + t := q[0] + if !canPreempt(t, s.running) { + continue + } + s.popInterruptLocked(lv) + if s.setImmediateLocked(t) { + s.preemptArmed = true + s.preemptLevel = t.Level + s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level) + } else { + // immediate 槽没拿到(理论上进不来,顶部已判 immediate == nil):放回队列, + // 否则任务会凭空消失。 + s.enqueueInterruptLocked(t) + } + return + } +} + +// nextInterruptSelection 是**恢复决策**:中断侧该出谁。 +// +// 顺序("谁先跑"的唯一实现处): +// +// ① immediate —— 刚抢占成功的,必须立即生效 +// ② 最高级待处理中断 vs 中断栈顶,取高者(持平归栈顶,维持 LIFO 与公平) +// +// 返回 ok=false 表示中断侧无货,调用方应转去就绪队列。 +func (s *scheduler) nextInterruptSelection() (task *Task, frame *TaskFrame, sel nextSelection, ok bool) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.immediate != nil { + t := s.immediate + s.immediate = nil + s.running = t + return t, nil, nextImmediate, true + } + + qTask, qLevel := s.highestInterruptLocked() + + // 中断栈:只比**栈顶**(严格 LIFO)。栈内不做优先级重排—— + // 嵌套抢占天然使栈自底向上级别递增,且"后被打断的先恢复"才是栈语义。 + // + // ★ 这里只**查询**栈顶做优先级比较(stackTopLocked 是只读快照); + // 真正的弹栈由中断管理器执行(restoreTopLocked)——中断栈的操作者 + // 只有中断管理器,调度器不碰栈内部。 + if top := s.stackTopLocked(); top != nil { + if qTask == nil || effectiveLevel(top.Task) >= qLevel { + // Resumed 由中断管理器在弹栈时计一次(否则会双计, + // 使"排空后 Suspended == Resumed"这条不变量失真)。 + restored := s.restoreTopLocked() + return restored.Task, restored.Frame, nextSuspended, true + } + } + + if qTask != nil { + s.popInterruptLocked(qLevel) + s.running = qTask + return qTask, nil, nextInterrupt, true + } + return nil, nil, nextNone, false +} + // SchedulerStats 是调度器的累计计数(可观测性,设计文档 §11 O2)。 // // InterruptsByLevel / PreemptsByLevel 按**中断级别**分桶(下标 1..4): @@ -334,36 +495,22 @@ type scheduler struct { stats SchedulerStats maxQueue int - // interruptQueues[level]:四条**中断队列**(level 1..4),同级 FIFO。 - // 未能立即抢占的中断(级别不足,或运行任务在临界区)按级别入队, - // nextRef 从 L4 到 L1 依次扫描。 - interruptQueues [5][]*Task - // immediate:刚抢占成功的中断。抢占必须**立即生效**,所以它不经队列, - // 在下一个安全点直接运行。这也消除了“抢占者与被抢占者同级”的比较问题—— - // 抢占者根本不需要和栈顶比。 - immediate *Task - // suspendStack:**中断栈**。被抢占后保存现场的任务压栈(LIFO), - // 用于“中断被中断”的嵌套场景:只有**栈顶**参与恢复选择,栈内不做优先级重排。 - suspendStack []*suspendedTask - // preemptArmed/preemptLevel:运行任务的“让位信号”。 - // interruptLoop 只写这两个字段与 pendingInterrupts;帧永远只由调度器读写。 - preemptArmed bool - preemptLevel Level - // critical 报告运行任务是否在不可抢占临界区(如记忆整理)。 - // 由于 interceptLoop 要读它,必须是原子的:帧仍只由调度器读写。 - critical atomic.Bool + // interrupt:中断子系统的全部状态(见 interrupt.go)。 + // + // **内嵌**而非具名字段:内嵌后字段提升,所有 `s.interruptQueues` / + // `s.immediate` / `s.suspendStack` 等访问写法**一字不改**, + // 现有代码与测试零改动。这是纯结构搬迁的最低风险形态。 + // + // 为什么共用 s.mu 而不是让中断子系统独立持锁:两者在同一临界区里被 + // 一起读写(如 registerInterrupt 先看运行任务再决定入队),拆锁属于 + // **并发模型变更**,不是重构——本次只做"语义集中",不改并发。 + interruptState // backpressured 记录「就绪队列满」这一状态的翻转,用于只打一次日志。 // 满着的时候 pumpInbox 每轮都会走到,逐轮打日志会把日志刷爆。 backpressured bool // wake 用于把空闲的调度器叫醒:pendingInterrupts 不是 channel, // 没有这个信号时“空闲时到达的中断”会一直等下一次输入(设计 §5.1 ③)。 wake chan struct{} - // maxInterruptFrames:中断栈帧数的**结构上界**,不是配置项。 - // - // 链条 = 排队(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4 运行中), - // 被挂起 4 帧;L4 之上没有更高级别,链到此为止。超限只可能是内核 bug, - // 因此这里只做防御性计数,**不降级、不丢弃帧**。 - maxInterruptFrames int // cancelBudget 是「停止」剩下的短路配额(见 Agent.RequestStop)。 // @@ -376,32 +523,33 @@ type scheduler struct { stopArmed bool } -// suspendedTask 是一个被抢占任务的现场。 -type suspendedTask struct { - Task *Task - Frame *TaskFrame -} - -// nextSelection 标识 nextRef 从哪个集合取出任务。 -type nextSelection int - -const ( - nextNone nextSelection = iota - nextReady - nextInterrupt - nextImmediate - nextSuspended -) - func newScheduler(maxQueue int) *scheduler { if maxQueue <= 0 { maxQueue = 256 } - return &scheduler{ - maxQueue: maxQueue, - maxInterruptFrames: int(LevelCritical), // 结构推论:= 中断级数 - wake: make(chan struct{}, 1), + s := &scheduler{ + maxQueue: maxQueue, + wake: make(chan struct{}, 1), } + // 中断栈上界是**结构推论**(= 中断级数),不是配置项: + // 链条 就绪(L0) ← I(L1) ← I(L2) ← I(L3) ← I(L4),被挂起 4 帧。 + // 不能写进结构体字面量:内嵌结构体的字段不支持在字面量里直接赋值(Go 1.25 语言版本)。 + s.interruptState.maxInterruptFrames = int(LevelCritical) + + // 接上**中断栈状态钩子**:中断管理器在栈变化时通告,调度器据此唤醒 + // 调度循环重新求值("现在该跑谁")。 + // + // 这是两者唯一的直接联系:中断管理器不调用调度器的选取逻辑, + // 调度器也不碰中断栈内部。将来改任一侧的选取/恢复策略都不必动对方。 + s.interruptState.stackNotify = func() { + // 钩子在持锁状态下被调用,因此这里**不能**再取 s.mu, + // 也不能调用任何会取锁的方法。只做非阻塞唤醒。 + select { + case s.wake <- struct{}{}: + default: + } + } + return s } // signalWake 非阻塞地唤醒调度器。 @@ -616,39 +764,23 @@ func (s *scheduler) queueSnapshot() []*Task { // 2. 中断队列 L4→L1(同级 FIFO) // 3. 中断栈顶(与 2 比级别取高者;栈顶是排队任务时视为最低) // 4. 排队队列(FIFO) +// nextRef 选出下一个要执行的任务。 +// +// ★ 本函数只做**编排**,不实现优先级规则: +// +// ① 先问中断子系统「中断侧该出谁」(immediate / 中断队列 / 中断栈顶的 +// 比较与弹栈,全部在 interrupt.go 的 nextInterruptSelection 里); +// ② 中断侧无货,才取就绪队列。 +// +// 这样"谁先跑"的优先级规则**只有一处实现**——改中断语义只需改 +// interrupt.go,不会漏掉这里的副本。 func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { + if t, f, sel, ok := s.nextInterruptSelection(); ok { + return t, f, sel + } + s.mu.Lock() defer s.mu.Unlock() - - if s.immediate != nil { - t := s.immediate - s.immediate = nil - s.running = t - return t, nil, nextImmediate - } - - 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 { - // 这里只负责“选出”;Resumed 由 resumeTask 计一次(否则会双计, - // 使“排空后 Suspended == Resumed”这条不变量失真)。 - s.suspendStack = s.suspendStack[:n-1] - s.running = top.Task - return top.Task, top.Frame, nextSuspended - } - } - - 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:] @@ -658,62 +790,6 @@ func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { 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 登记一个应“立即运行”的抢占者。 -// -// 槽只有一格:若已有抢占者且新的级别更高,旧的降级入队;否则新的入队。 -// 返回 true 表示 t **确实占住了 immediate 槽**;false 表示它被降级进了自己的 -// 级别队列(immediate 是单槽,这是设计要求的降级分支,见设计 §2「至多一个」)。 -// -// 调用方必须用返回值决定是否计入 PreemptsByLevel:那条计数器的语义是 -// 「进入 immediate 的次数」,被降级的中断从未进过 immediate。 -// setImmediateLocked 尝试把 t 放进 immediate 槽。 -// -// 返回 true:t 已占住 immediate(若原有抢占者被顶掉,它**已被**降级入队)。 -// 返回 false:t 没有进 immediate,且本函数**未动 t** —— 调用方负责按级别入队。 -// -// 把「降级入队」的责任留给调用方,是为了让「到底入队了几次」只有一个出口: -// 早先由本函数在返回 false 前自行入队,调用方又照着 false 再入一次, -// 同一任务就会在队列里出现两份(实测:中断任务被执行两次、Executed 虚高)。 -func (s *scheduler) setImmediateLocked(t *Task) bool { - if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) { - return false - } - if s.immediate != nil { - s.enqueueInterruptLocked(s.immediate) - } - s.allocateIDLocked(t) - s.stats.Enqueued++ - s.immediate = t - return true -} - // armStop 处理一次「停止」指令:登记短路配额并返回**停止那一刻的排队深度**。 // // 语义(用户明确的设计):停止 = ①立即结束当前 LLM 推理;②对停止那一刻 @@ -772,24 +848,6 @@ func removeTask(list []*Task, target *Task) []*Task { // enqueueInterruptLocked 把一个未立即抢占的中断按其级别入队(调用方持锁)。 // // 有界:满了丢**最老**的一条并计数(中断是提示性输入,宁可丢旧保新)。 -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 - } - } - } - lv := t.Level - if lv < LevelBackground || lv > LevelCritical { - lv = DefaultLevel - } - s.interruptQueues[lv] = append(s.interruptQueues[lv], t) - s.stats.Enqueued++ -} // requestPreempt 登记一次中断请求(class=TaskInterrupt)。 // @@ -808,83 +866,15 @@ func (s *scheduler) enqueueInterruptLocked(t *Task) { // 生产路径只有 interruptLoop,它用 (*Agent).interruptLevel 得出 level; // 内核自身用 requestKernelPreempt(固定 L4)。本函数不再夹取, // 否则内核级插件的 L4 会被无辜削掉。 -func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool { - return s.registerInterrupt(newInterruptTask(evt, 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() - s.stats.bumpInterruptLevel(&s.stats.InterruptsByLevel, t.Level) - arm := false - if !critical && canPreempt(t, running) { - if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown { - // 只有**真的占住 immediate 槽**才算一次抢占,才计入 PreemptsByLevel: - // immediate 是单槽,若它被另一个更高级的抢占者占着,t 会走上而下的 - // 「否则入队」分支——那种情况 t 从未进入 immediate(否则同一安全点前 - // 到达两条同级中断时该计数会高估)。 - if s.setImmediateLocked(t) { - arm = true - s.preemptArmed = true - s.preemptLevel = t.Level - s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level) - // 抢占判决点:此刻 running 还是真正的受害者,记下来才能回答 - // 「我的任务被谁打断了」。放在 executeNewTask 里打是错的—— - // 那时 nextRef 已把 running 换成抢占者自己,日志会写成 - // "victim = 入侵者自己"(实测过这个错误输出)。 - log.Printf("[agent] preempt: %s from %s (L%d) preempts %s (%s)", - describeTask(t), sourceOf(t, nil), int(t.Level), - describeTask(running), sourceOf(running, nil)) - } - } - } - if !arm { - s.enqueueInterruptLocked(t) - // 入队而不是抢占:这也是要能看见的(否则「中断明明到了却没生效」无从解释)。 - // 两种成因分开写:临界区 vs 级别不够/冷却期。 - reason := "level insufficient" - if critical { - reason = "running in critical section" - } else if canPreempt(t, running) && running != nil && !running.LastPreemptAt.IsZero() && time.Since(running.LastPreemptAt) < preemptCooldown { - reason = "preempt cooldown" - } - log.Printf("[agent] interrupt queued: %s from %s (L%d) vs %s (%s) — %s; queue=%d", - describeTask(t), sourceOf(t, nil), int(t.Level), - describeTask(running), sourceOf(running, nil), reason, s.interruptCountLocked()) - } - s.mu.Unlock() - - if !arm { - s.signalWake() - } - return arm -} // preemptGrantedFor 报告运行任务是否应在当前安全点让位。 -func (s *scheduler) preemptGrantedFor() bool { - s.mu.Lock() - defer s.mu.Unlock() - if !s.preemptArmed || s.running == nil { - return false - } - return s.preemptLevel > effectiveLevel(s.running) -} -func (s *scheduler) clearPreempt() { - s.mu.Lock() - s.preemptArmed = false - s.preemptLevel = 0 - s.mu.Unlock() -} // rearmPending 在**安全点重新求值**中断队列(设计 §4.3 / §5.2)。 // @@ -897,67 +887,11 @@ func (s *scheduler) clearPreempt() { // // 判据与 registerInterrupt **完全同一套**(canPreempt + 冷却 + 临界区闸门), // 因此不会凭空制造设计之外的抢占。 -func (s *scheduler) rearmPending() { - s.mu.Lock() - defer s.mu.Unlock() - // 已有让位信号、或 immediate 槽已被占用:下一个安全点的选择已经在路上, - // 不必(也不该)重复武装。 - if s.preemptArmed || s.immediate != nil || s.running == nil || s.critical.Load() { - return - } - // 冷却期内不武装:与 registerInterrupt 同一判据(抗饥饿)。 - if !s.running.LastPreemptAt.IsZero() && time.Since(s.running.LastPreemptAt) < preemptCooldown { - return - } - // 中断队列本就按级别组织:从最高级往下找第一条能抢占的队头。 - // (队列里的任务有效级恒等于基础级,故「第一条能抢」= 最高级可抢占者。) - for lv := LevelCritical; lv >= LevelBackground; lv-- { - q := s.interruptQueues[lv] - if len(q) == 0 { - continue - } - t := q[0] - if !canPreempt(t, s.running) { - continue - } - s.popInterruptLocked(lv) - if s.setImmediateLocked(t) { - s.preemptArmed = true - s.preemptLevel = t.Level - s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level) - } else { - // immediate 槽没拿到(理论上进不来,顶部已判 immediate == nil):放回队列, - // 否则任务会凭空消失。 - s.enqueueInterruptLocked(t) - } - return - } -} // suspend 保存现场。 // // 深度上界是**结构推论**(= 中断级数),不是配置项:安全点上的 canSuspend 已提前 // 拦下超限情况,此处仅在竞态下兜底计数——绝不丢弃帧(帧丢了会丢副作用记录)。 -func (s *scheduler) suspend(t *Task, f *TaskFrame) { - s.mu.Lock() - defer s.mu.Unlock() - if len(s.suspendStack) >= s.maxInterruptFrames { - s.stats.Rejected++ - } - s.suspendStack = append(s.suspendStack, &suspendedTask{Task: t, Frame: f}) - s.stats.Suspended++ - // 饥饿防护:抢占计数 +1(提升有效级)并记录冷却起点。 - t.PreemptCount++ - t.LastPreemptAt = time.Now() - if s.running == t { - s.running = nil - } - - // D1=B:中断任务在上一个任务之前的完整状态上开始运行, - // 因此这里**不**把被打断任务的任何内容交给它。 - s.preemptArmed = false - s.preemptLevel = 0 -} // sourceOf 从任务/帧里取一个**可辨识来源**,用于抢占日志与故障定位。 // @@ -1000,21 +934,11 @@ func describeTask(t *Task) string { } // canSuspend 报告还有下潜余量(安全点用它决定是否真的让位)。 -func (s *scheduler) canSuspend() bool { - s.mu.Lock() - defer s.mu.Unlock() - return len(s.suspendStack) < s.maxInterruptFrames -} // suspendDepth 返回当前中断栈深度(仅用于日志)。 // // 单独一个方法而不是让调用方直接读 s.suspendStack:那个字段只允许在 s.mu 下访问, // 而日志点不应该自己摸调度器内部状态(也不应该为了打一行日志多持一次锁的窗口)。 -func (s *scheduler) suspendDepth() int { - s.mu.Lock() - defer s.mu.Unlock() - return len(s.suspendStack) -} // pendingEvents 收集**尚未执行**(排队队列 / 四条中断队列 / immediate)与 // **已挂起**(中断栈)任务所携带的、且带同步回执通道的输入事件。 diff --git a/internal/agent/core/stages.go b/internal/agent/core/stages.go index 442addc4..c0720b69 100644 --- a/internal/agent/core/stages.go +++ b/internal/agent/core/stages.go @@ -268,18 +268,28 @@ func (h *StageHost) ExecuteToolCtx(ctx context.Context, name string, args map[st // withExecuteCtx 把 ctx 绑定到本次调用上,随 goroutine 走。 // -// 具体实现见 execctx.go:Go 没有 goroutine-local storage,这里用 -// **参数旁路**(在 args 里放一个不可见的私有键)最省事且无并发歧义—— -// args 是每次调用独立的 map,天然一 goroutine 一份。 +// 实现见 proc/execctx.go:Go 没有 goroutine-local storage,这里用 +// **参数旁路**(在 args 里放一个不可见的私有键)把 ctx 带进执行栈。 +// +// ★ 必须**复制**而不是原地写: +// +// 调用方(executeToolCallInner)收到的是 taskFrame 上那份 tc.Arguments, +// 而工具执行在自己的 goroutine 里跑。超时/取消路径下,主 goroutine 会在 +// goroutine 仍存活时继续读同一份 args —— 原地写就是数据竞争。 +// -race 下实测复现(且 core 包整体被判红,掩盖其它真正的竞争)。 +// +// 复制后:原 map 只读、新 map 一 goroutine 一份,才真的"天然无并发歧义"。 +// 代价是一次浅拷贝(工具参数本来就在 64KB 预算内,可忽略)。 func withExecuteCtx(ctx context.Context, args map[string]interface{}) map[string]interface{} { if ctx == nil { return args } - if args == nil { - args = make(map[string]interface{}, 1) + out := make(map[string]interface{}, len(args)+1) + for k, v := range args { + out[k] = v } - args[executeCtxArgKey] = ctx - return args + out[executeCtxArgKey] = ctx + return out } func (h *StageHost) ToolPlugin(name string) string { diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index bac3bd88..54fcfc42 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -635,9 +635,10 @@ func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome { } } - // 上下文占满检测:此刻 f.Msgs 已建好(含 system + timeline + 本轮输入)。 - // 只有驻留子设了 onContextFull ⇒ 对根 agent 是 no-op。 - a.checkContextFull(f) + // 上下文占满检测:**上报**给上级(驻留子),不在此处裁剪。 + // 只有驻留子设了 onContextFull ⇒ 对根 agent 是 no-op(根自己走 + // enforceContextBudget 就地裁,没有上级可报)。 + a.checkContextFull() f.Step = StepLLM return outcomeContinue diff --git a/internal/agent/io/channel.go b/internal/agent/io/channel.go index 24bcbb5e..18c3ceeb 100644 --- a/internal/agent/io/channel.go +++ b/internal/agent/io/channel.go @@ -1150,13 +1150,17 @@ const execCtxArgKey = "\x00kernel.execute_ctx\x00" // 但**必须**在真正下发给插件前剥掉,否则严格校验参数的插件会报未知参数。 // 剥除由各插件自己的入口决定:core 侧统一下发前已剥(见 StageHost), // io 侧保持原样,因为设备插件直接读 args、不会做严格白名单校验。 +// ★ 必须**复制**而不是原地写:设备调用在自己的 goroutine 里跑,超时/取消 +// 路径下调用方会在 goroutine 仍存活时继续读同一份 args —— 原地写就是数据竞争 +// (-race 实测复现,见 StageHost 侧 withExecuteCtx 的同类说明)。 func bindExecCtx(ctx context.Context, args map[string]interface{}) map[string]interface{} { if ctx == nil { return args } - if args == nil { - args = make(map[string]interface{}, 1) + out := make(map[string]interface{}, len(args)+1) + for k, v := range args { + out[k] = v } - args[execCtxArgKey] = ctx - return args + out[execCtxArgKey] = ctx + return out }