mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-03 15:53:56 +00:00
fix(scheduler): 补齐 M3 与设计稿的两处语义偏离(发现即修)
两处都不是风格差异,而是真的偏离设计语义(其一为回归),
已按“先写判据确认失败、再修”的方式处理,判据保留为回归测试。
1. 空闲时到达的中断永远不会被处理(设计 §5.1 ③ 未落地)
调度器空闲时只阻塞在 select{InputChan, selfInputCh, ctx.Done},
而 pendingInterrupts 不是 channel——interceptLoop 把中断入队后
没有任何东西唤醒调度器,中断要等“下一条输入”才被看到。
修复:调度器加 wake channel,enqueueInterrupt 非阻塞 signalWake,
空闲分支增加 wake 分支。
2. 临界区内只“不让位”却仍被“取消”(设计 §4.3/§5.2)
requestPreempt 不判临界区,interceptLoop 照常 cancelLLM,
于是正在流式的记忆整理被中断,stepLLM 以 error 提前结束——
整理任务被砍掉一半,而设计要求的是“请求排队等它结束”。
修复:scheduler 增加原子 critical 标志(帧仍只由调度器读写),
runInputTask 在 prepare 后设置、结束(含挂起)时清除;
requestPreempt 在临界区内不 arm、不取消,中断只入队。
- 新增 scheduler_regression_test.go 2 项(先失败后通过)
- 验收:agent 全量 + -race;全仓 vet 通过
This commit is contained in:
@ -21,6 +21,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"runtime/debug"
|
"runtime/debug"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||||||
@ -197,6 +198,12 @@ type scheduler struct {
|
|||||||
// interruptLoop 只写这两个字段与 pendingInterrupts;帧永远只由调度器读写。
|
// interruptLoop 只写这两个字段与 pendingInterrupts;帧永远只由调度器读写。
|
||||||
preemptArmed bool
|
preemptArmed bool
|
||||||
preemptLevel Level
|
preemptLevel Level
|
||||||
|
// critical 报告运行任务是否在不可抢占临界区(如记忆整理)。
|
||||||
|
// 由于 interceptLoop 要读它,必须是原子的:帧仍只由调度器读写。
|
||||||
|
critical atomic.Bool
|
||||||
|
// wake 用于把空闲的调度器叫醒:pendingInterrupts 不是 channel,
|
||||||
|
// 没有这个信号时“空闲时到达的中断”会一直等下一次输入(设计 §5.1 ③)。
|
||||||
|
wake chan struct{}
|
||||||
// maxSuspendDepth:suspendPool 深度上限(设计文档 §6.3,默认 4)。
|
// maxSuspendDepth:suspendPool 深度上限(设计文档 §6.3,默认 4)。
|
||||||
maxSuspendDepth int
|
maxSuspendDepth int
|
||||||
}
|
}
|
||||||
@ -221,9 +228,23 @@ func newScheduler(maxQueue int) *scheduler {
|
|||||||
if maxQueue <= 0 {
|
if maxQueue <= 0 {
|
||||||
maxQueue = 256
|
maxQueue = 256
|
||||||
}
|
}
|
||||||
return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4}
|
return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4, wake: make(chan struct{}, 1)}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// signalWake 非阻塞地唤醒调度器。
|
||||||
|
func (s *scheduler) signalWake() {
|
||||||
|
select {
|
||||||
|
case s.wake <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// setCritical 由调度器 goroutine 在任务进入/离开临界区时设置。
|
||||||
|
func (s *scheduler) setCritical(v bool) { s.critical.Store(v) }
|
||||||
|
|
||||||
|
// inCritical 报告运行任务是否在不可抢占临界区。
|
||||||
|
func (s *scheduler) inCritical() bool { return s.critical.Load() }
|
||||||
|
|
||||||
// hasRoom 报告就绪队列是否还能接收任务。泵入侧据此节流:
|
// hasRoom 报告就绪队列是否还能接收任务。泵入侧据此节流:
|
||||||
// 队列满则停止从 channel 取,让背压落回 channel 本身。
|
// 队列满则停止从 channel 取,让背压落回 channel 本身。
|
||||||
func (s *scheduler) hasRoom() bool {
|
func (s *scheduler) hasRoom() bool {
|
||||||
@ -329,6 +350,7 @@ func (s *scheduler) enqueueInterrupt(t *Task) {
|
|||||||
s.stats.Rejected++
|
s.stats.Rejected++
|
||||||
}
|
}
|
||||||
s.pendingInterrupts = append(s.pendingInterrupts, t)
|
s.pendingInterrupts = append(s.pendingInterrupts, t)
|
||||||
|
s.signalWake()
|
||||||
}
|
}
|
||||||
|
|
||||||
// requestPreempt 登记一次中断请求。
|
// requestPreempt 登记一次中断请求。
|
||||||
@ -339,11 +361,15 @@ func (s *scheduler) enqueueInterrupt(t *Task) {
|
|||||||
// 就正常结束,中断也不会丢(它会被 nextRef 按优先级选出)。
|
// 就正常结束,中断也不会丢(它会被 nextRef 按优先级选出)。
|
||||||
//
|
//
|
||||||
// 判据用**有效**优先级(饥饿防护),并受抢占冷却约束。
|
// 判据用**有效**优先级(饥饿防护),并受抢占冷却约束。
|
||||||
|
//
|
||||||
|
// 临界区(如记忆整理)内不 arm、不取消:中断只入队,等临界区结束后的安全点处理,
|
||||||
|
// 这是设计 §4.3 的硬要求——那个位置的“不抢占”不能只是不让位,还必须不取消。
|
||||||
func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool {
|
func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
running := s.running
|
running := s.running
|
||||||
|
critical := s.critical.Load()
|
||||||
canPreempt := false
|
canPreempt := false
|
||||||
if running != nil && level > effectiveLevel(running) {
|
if !critical && running != nil && level > effectiveLevel(running) {
|
||||||
if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown {
|
if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown {
|
||||||
canPreempt = true
|
canPreempt = true
|
||||||
s.preemptArmed = true
|
s.preemptArmed = true
|
||||||
@ -549,12 +575,14 @@ func (a *Agent) schedulerLoop() {
|
|||||||
|
|
||||||
t, f, kind := a.sched.nextRef()
|
t, f, kind := a.sched.nextRef()
|
||||||
if kind == nextNone {
|
if kind == nextNone {
|
||||||
// 无待办:阻塞等新输入或退出。
|
// 无待办:阻塞等新输入、新中断(wake)或退出。
|
||||||
select {
|
select {
|
||||||
case evt := <-a.io.InputChan():
|
case evt := <-a.io.InputChan():
|
||||||
a.sched.enqueue(newInputTask(evt))
|
a.sched.enqueue(newInputTask(evt))
|
||||||
case msg := <-a.selfInputCh:
|
case msg := <-a.selfInputCh:
|
||||||
a.sched.enqueue(newSelfTask(msg))
|
a.sched.enqueue(newSelfTask(msg))
|
||||||
|
case <-a.sched.wake:
|
||||||
|
// 中断已入 pendingInterrupts,回到循环顶部重新挑选。
|
||||||
case <-a.ctx.Done():
|
case <-a.ctx.Done():
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
101
internal/agent/core/scheduler_regression_test.go
Normal file
101
internal/agent/core/scheduler_regression_test.go
Normal file
@ -0,0 +1,101 @@
|
|||||||
|
package core
|
||||||
|
|
||||||
|
// 回归判据:M3 实现与设计稿 §5.1/§4.3 的两处偏离。
|
||||||
|
//
|
||||||
|
// 这两条是**先写判据、确认失败、再修**的(修完保留为回归测试)。
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||||||
|
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
|
||||||
|
)
|
||||||
|
|
||||||
|
// cancelAwareProvider 第一次调用进入后阻塞,直到 ctx 取消;记录是否被取消。
|
||||||
|
type cancelAwareProvider struct {
|
||||||
|
once sync.Once
|
||||||
|
entered chan struct{}
|
||||||
|
canceled atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newCancelAwareProvider() *cancelAwareProvider {
|
||||||
|
return &cancelAwareProvider{entered: make(chan struct{})}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *cancelAwareProvider) Name() string { return "cancel-aware" }
|
||||||
|
func (p *cancelAwareProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) {
|
||||||
|
p.once.Do(func() { close(p.entered) })
|
||||||
|
<-ctx.Done()
|
||||||
|
p.canceled.Store(true)
|
||||||
|
return nil, ctx.Err()
|
||||||
|
}
|
||||||
|
func (p *cancelAwareProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) {
|
||||||
|
return nil, context.Canceled
|
||||||
|
}
|
||||||
|
func (p *cancelAwareProvider) MaxContextTokens() int { return 8192 }
|
||||||
|
|
||||||
|
// 设计 §5.1 ③:interruptLoop 定级/决策后必须**唤醒调度器**。
|
||||||
|
// 现状:中断只被放进 pendingInterrupts,而调度器空闲时阻塞在 select(只看
|
||||||
|
// InputChan/selfInputCh/ctx.Done)——没有任何东西会把它叫醒。
|
||||||
|
func TestGap_IdleInterruptIsProcessed(t *testing.T) {
|
||||||
|
sp := &countingProvider{}
|
||||||
|
a := New(AgentConfig{
|
||||||
|
ID: "idle-intr",
|
||||||
|
Provider: sp,
|
||||||
|
ProviderManager: agentAPI.NewProviderManager(),
|
||||||
|
IO: agentIO.NewIOManager(),
|
||||||
|
StageHost: NewStageHost(),
|
||||||
|
})
|
||||||
|
a.Start()
|
||||||
|
defer a.Stop()
|
||||||
|
|
||||||
|
// 完全空闲时投递一条中断(模拟定时器通知/插件提醒)。
|
||||||
|
a.io.InjectInterruptText("qq", "cli", "空闲时的通知")
|
||||||
|
|
||||||
|
deadline := time.Now().Add(3 * time.Second)
|
||||||
|
for sp.n.Load() == 0 && time.Now().Before(deadline) {
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
}
|
||||||
|
if sp.n.Load() == 0 {
|
||||||
|
snap := a.DumpScheduler()
|
||||||
|
t.Fatalf("空闲时到达的中断未被处理:pendingInterrupts=%d(调度器未被唤醒)",
|
||||||
|
len(snap.PendingInterrupts))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 设计 §4.3/§5.2:`_consolidation_` 是整任务临界区——抢占请求必须排队等它结束,
|
||||||
|
// 而不是取消它。现状:requestPreempt 不判临界区,interceptLoop 照常 cancelLLM,
|
||||||
|
// 于是正在流式的记忆整理被中断 → stepLLM 直接以 error 结束(整理丢一半)。
|
||||||
|
func TestGap_ConsolidationMustNotBeCancelled(t *testing.T) {
|
||||||
|
sp := newCancelAwareProvider()
|
||||||
|
a := New(AgentConfig{
|
||||||
|
ID: "consol",
|
||||||
|
Provider: sp,
|
||||||
|
ProviderManager: agentAPI.NewProviderManager(),
|
||||||
|
IO: agentIO.NewIOManager(),
|
||||||
|
StageHost: NewStageHost(),
|
||||||
|
})
|
||||||
|
a.Start()
|
||||||
|
defer a.Stop()
|
||||||
|
|
||||||
|
// 走自循环通道发起一次记忆整理。
|
||||||
|
a.selfInputCh <- selfInputMsg{text: "合并实体", channel: channelConsolidation}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-sp.entered:
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
t.Fatal("记忆整理未进入 LLM 调用")
|
||||||
|
}
|
||||||
|
|
||||||
|
// L4 中断到达。
|
||||||
|
a.io.InjectInterruptText("cli", "cli", "L4 打断")
|
||||||
|
|
||||||
|
time.Sleep(500 * time.Millisecond)
|
||||||
|
if sp.canceled.Load() {
|
||||||
|
t.Fatal("记忆整理是临界区,其 LLM 不该被取消(设计 §4.3/§5.2)")
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -191,6 +191,10 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
|
|||||||
//
|
//
|
||||||
// M3a 还没有抢占,因此 outcomeSuspended 只会由 M3b 的抢占检查产生。
|
// M3a 还没有抢占,因此 outcomeSuspended 只会由 M3b 的抢占检查产生。
|
||||||
func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) (*TaskFrame, stepOutcome) {
|
func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) (*TaskFrame, stepOutcome) {
|
||||||
|
// 临界区标记由调度器 goroutine 维护,任务结束(含挂起)即清。
|
||||||
|
// interceptLoop 读它来决定“能不能取消”,因此必须是原子的。
|
||||||
|
defer a.sched.setCritical(false)
|
||||||
|
|
||||||
f, term := a.prepareInputTask(evt)
|
f, term := a.prepareInputTask(evt)
|
||||||
switch term {
|
switch term {
|
||||||
case terminalSkipped, terminalStageShortCircuit, terminalConsolidation:
|
case terminalSkipped, terminalStageShortCircuit, terminalConsolidation:
|
||||||
@ -234,6 +238,9 @@ func (a *Agent) prepareInputTask(evt *agentIO.InputEvent) (*TaskFrame, taskTermi
|
|||||||
if a.currentOutputChannel == "" {
|
if a.currentOutputChannel == "" {
|
||||||
a.currentOutputChannel = evt.Source
|
a.currentOutputChannel = evt.Source
|
||||||
}
|
}
|
||||||
|
// 进入本任务的临界区属性(记忆整理整任务不可抢占)。
|
||||||
|
// 必须在 processConsolidation 之前设置——它就在下面同步执行。
|
||||||
|
a.sched.setCritical(a.inCriticalSection())
|
||||||
|
|
||||||
if evt.OutputChannel == channelConsolidation {
|
if evt.OutputChannel == channelConsolidation {
|
||||||
a.processConsolidation(evt, in.text)
|
a.processConsolidation(evt, in.text)
|
||||||
|
|||||||
Reference in New Issue
Block a user