feat(scheduler): M4 临界区显式化 + 清掉被 pendingInterrupts 取代的 interceptCh

设计依据 docs/zh/input-scheduler-design.md §4.3、§11.1(P5/P6)。

- 临界区语义显式化:让位检查**只在 step 之间**做,执行中的 step
  (工具 RPC / ONNX / CAS 落盘)天然不可抢占;_consolidation_ 整任务
  经 inCriticalSection() 判为不可抢占(它直接改图库)
- 删除 interceptCh 与 drainInterrupts:M3b 起中断一律走 pendingInterrupts,
  旧的「同行注入 + 三处 drain + 批次放弃」已无写入者,属死代码
- 新增 scheduler_critical_test.go 3 项:
  P5/P6 工具执行中 arm 了让位信号也不得挂起、必须等工具返回后的安全点;
  _consolidation_ 判为临界区;无抢占时同批工具必须全部执行(新语义回归)
- 验收:agent 全量 + -race;全仓 build/vet 通过
This commit is contained in:
JianFeeeee
2026-09-13 00:34:39 +08:00
parent a77e1aacae
commit 0f78d7094d
5 changed files with 151 additions and 109 deletions

View File

@ -455,13 +455,6 @@ func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome {
// 取消(context.Canceled 且 agent 未退出)时**留在本 step 并 Turn++**——等价于
// 原实现的 `continue`:重新排空中断、补占位、重新请求。抢占挂起将在 M3 从这里接管。
func (a *Agent) stepLLM(f *TaskFrame) stepOutcome {
for _, interrupt := range a.drainInterrupts() {
f.Msgs = append(f.Msgs, agentAPI.Message{
Role: "system",
Content: "[中断消息] " + interrupt,
})
}
// zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验),
// 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。
f.Msgs = dropContinuationPlaceholders(f.Msgs)
@ -569,22 +562,6 @@ func (a *Agent) stepToolBegin(f *TaskFrame) stepOutcome {
}
tc := f.PendingTools[f.ToolIdx]
if len(a.interceptCh) > 0 {
for _, interrupt := range a.drainInterrupts() {
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
}
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"plugin": a.resolveToolPlugin(tc.Name),
"args": tc.Arguments,
"status": "interrupted",
"reason": "user interrupt before execution",
"channel": a.currentOutputChannel,
})
f.Step = StepTurnEnd
return outcomeContinue
}
f.ToolsUsed = append(f.ToolsUsed, tc.Name)
pluginName := a.resolveToolPlugin(tc.Name)
log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID)
@ -744,13 +721,6 @@ func (a *Agent) stepToolAfter(f *TaskFrame) stepOutcome {
})
f.ToolIdx++
if len(a.interceptCh) > 0 {
for _, interrupt := range a.drainInterrupts() {
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
}
f.Step = StepTurnEnd
return outcomeContinue
}
f.Step = StepToolBegin
return outcomeContinue
}