HomeAgent/internal/agent/core/scheduler_critical_test.go

package core

// M4 验收测试：临界区语义显式化 + 抢占延迟到安全点 + 批次不再被中断放弃。
//
// 设计依据 docs/zh/input-scheduler-design.md §4.3（临界区）、§11.1（P5/P6）。
//
// 关键结构事实：让位检查**只在 step 之间**进行，因此任何正在执行的 step
// （工具 RPC、ONNX、CAS 落盘）天然不可抢占——中断只能等它返回。

import (
	"sync"
	"testing"
	"time"

	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
)

// P5/P6：工具执行期间到达的高优先级中断不得立即抢占；必须等工具返回后的安全点。
func TestPreempt_DeferredDuringToolExec(t *testing.T) {
	sh := NewStageHost()
	entered := make(chan struct{})
	release := make(chan struct{})
	var once sync.Once
	sh.RegisterTool("t_slow", sdk.ToolDef{Name: "t_slow", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
		once.Do(func() { close(entered) })
		<-release
		return "slow-done", nil
	})

	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
		{Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_slow")}}, // 低优先级任务调用慢工具
		{Content: "low-done"},  // 恢复后收尾
		{Content: "intr-done"}, // 中断任务
	}}
	a := New(AgentConfig{
		ID:              "crit",
		Provider:        sp,
		ProviderManager: agentAPI.NewProviderManager(),
		IO:              agentIO.NewIOManager(),
		StageHost:       sh,
	})

	if _, _ = enqueueQueued(t, a, "qq", "低优先级任务"); true {
	}
	lt, _, _ := a.sched.nextRef()
	done := make(chan struct{})
	go func() { a.executeNewTask(lt); close(done) }()

	select {
	case <-entered:
	case <-time.After(3 * time.Second):
		t.Fatal("慢工具未被调用")
	}

	// 工具执行中注入 L4 中断。
	intrEvt, _ := textEvent("cli", "紧急打断")
	intrEvt.Payload["interrupt"] = true
	if !a.sched.requestKernelPreempt(intrEvt) {
		t.Fatal("L4 应 arm 让位信号")
	}
	// 关键断言：信号已 arm，但任务仍在工具里 —— 绝不能挂起。
	if !a.sched.preemptGrantedFor() {
		t.Fatal("让位信号应已 arm")
	}
	if a.DumpScheduler().Running == nil {
		t.Fatal("工具执行中不得挂起（StepToolExec 是临界区）")
	}
	if len(a.DumpScheduler().SuspendStack) != 0 {
		t.Fatal("工具执行中 suspendStack 应为空")
	}

	// 放行工具 → 工具返回后的安全点才挂起。
	close(release)
	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("工具返回后未挂起")
	}

	snap := a.DumpScheduler()
	if len(snap.SuspendStack) != 1 {
		t.Fatalf("工具返回后 suspendStack=%d，期望 1", len(snap.SuspendStack))
	}
	if snap.SuspendStack[0].Frame.Step != StepToolAfter {
		t.Fatalf("应在工具执行后的安全点挂起（StepToolAfter），实际 %v", snap.SuspendStack[0].Frame.Step)
	}
	if len(snap.PendingInterrupts) != 1 {
		t.Fatalf("中断请求不得丢失，pendingInterrupts=%d", len(snap.PendingInterrupts))
	}
}

// _consolidation_ 整任务视为不可抢占（它直接改图库）。
func TestCriticalSection_ConsolidationMarked(t *testing.T) {
	a := newPreemptAgent(t, newPreemptProvider())

	// 判定：只有记忆整理通道是整任务临界区。
	if isCriticalChannel("cli") {
		t.Fatal("普通通道不应被判为临界区")
	}
	if !isCriticalChannel(channelConsolidation) {
		t.Fatal("记忆整理必须是不可抢占临界区")
	}

	// 集成：标志的推导链「输入事件 → 通道 → isCriticalChannel → scheduler.critical」
	// 必须成立（N0 之后通道只从事件推导，不再有内核可变字段）。
	for _, c := range []struct {
		channel string
		want    bool
	}{
		{"cli", false},
		{channelConsolidation, true},
	} {
		evt, _ := textEvent("tc", "x")
		evt.OutputChannel = c.channel
		a.sched.setCritical(isCriticalChannel(outputChannelOf(evt)))
		if got := a.sched.inCritical(); got != c.want {
			t.Fatalf("通道 %q → critical=%v，期望 %v", c.channel, got, c.want)
		}
	}
}

// 新语义：没有抢占时，同批的多个工具必须全部执行——不再有「中断放弃剩余批」。
func TestBatch_NotAbandonedWithoutPreemption(t *testing.T) {
	sh := NewStageHost()
	var mu sync.Mutex
	var ran []string
	reg := func(name string) {
		sh.RegisterTool(name, sdk.ToolDef{Name: name, Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
			mu.Lock()
			ran = append(ran, name)
			mu.Unlock()
			return name + "-out", nil
		})
	}
	reg("t_a")
	reg("t_b")

	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
		{Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_a"), tc("c2", "t_b")}},
		{Content: "全部完成"},
	}}
	a := New(AgentConfig{
		ID:              "batch",
		Provider:        sp,
		ProviderManager: agentAPI.NewProviderManager(),
		IO:              agentIO.NewIOManager(),
		StageHost:       sh,
	})

	if _, _ = enqueueQueued(t, a, "cli", "跑两个工具"); true {
	}
	tt, _, _ := a.sched.nextRef()
	a.executeNewTask(tt)

	if len(ran) != 2 || ran[0] != "t_a" || ran[1] != "t_b" {
		t.Fatalf("同批工具应全部按序执行，实际 %v", ran)
	}
	snap := a.DumpScheduler()
	if len(snap.SuspendStack) != 0 || len(snap.PendingInterrupts) != 0 {
		t.Fatalf("无抢占时不应有挂起或待处理中断：%+v", snap)
	}
	if snap.Stats.Executed != 1 {
		t.Fatalf("Executed=%d，期望 1", snap.Stats.Executed)
	}
}