Agent-System-Backend/internal/runmgr/background_test.go

package runmgr

import (
	"strings"
	"sync"
	"testing"
	"time"

	"agent-system-backend/internal/runtime"
	core "agent-system-core"
)

type gateRunner struct {
	entered chan struct{}
	release chan struct{}
	once    sync.Once
	parts   []core.Part
	err     error
	spawn   func(runtime.RunCtx) error
}

func (g *gateRunner) Name() string { return "gate" }

func (g *gateRunner) Execute(rc runtime.RunCtx) error {
	if g.entered != nil {
		g.once.Do(func() { close(g.entered) })
	}
	if g.release != nil {
		<-g.release
	}
	if g.spawn != nil {
		return g.spawn(rc)
	}
	if g.err != nil {
		return g.err
	}
	if len(g.parts) > 0 {
		return rc.SetAssistantParts(g.parts)
	}
	return nil
}

type noopRunner struct{}

func (noopRunner) Name() string { return "noop" }
func (noopRunner) Execute(rc runtime.RunCtx) error {
	return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "已汇报"}})
}

func waitUntil(t *testing.T, cond func() bool) {
	t.Helper()
	waitUntilWithin(t, 2*time.Second, cond)
}

func waitUntilWithin(t *testing.T, bound time.Duration, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(bound)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatal("timed out")
}

func systemTexts(t *testing.T, c *runCtx) []string {
	t.Helper()
	msgs, err := c.mgr.st.ListMessages(c.run.ConversationID)
	if err != nil {
		t.Fatal(err)
	}
	var out []string
	for _, m := range msgs {
		if m.Role != core.RoleSystem {
			continue
		}
		for _, p := range m.Parts {
			if p.Type == core.PartText {
				out = append(out, p.Text)
			}
		}
	}
	return out
}

// backgroundIdleBound only bounds a hang: each child settles through several
// fsync'd SQLite commits, and CI disk stalls have exceeded a 2s window.
const backgroundIdleBound = 20 * time.Second

func waitBackgroundIdle(t *testing.T, c *runCtx) {
	t.Helper()
	waitUntilWithin(t, backgroundIdleBound, func() bool {
		c.mgr.noticeMu.Lock()
		defer c.mgr.noticeMu.Unlock()
		return c.mgr.bgInFlight[c.run.ConversationID] == 0
	})
}

func TestBackgroundDoesNotBlockParent(t *testing.T) {
	entered := make(chan struct{})
	release := make(chan struct{})
	runner := &gateRunner{
		entered: entered, release: release,
		parts: []core.Part{{Type: core.PartText, Text: "后台结果"}},
	}
	c, _ := newSpawnEnv(t, runner)
	c.mgr.runner = noopRunner{}
	id, err := c.SpawnBackground("agent.sub.explore", "检索", "查资料", "")
	if err != nil || id == "" {
		t.Fatalf("spawn: %v %s", err, id)
	}
	select {
	case <-entered:
	case <-time.After(time.Second):
		t.Fatal("child did not start")
	}
	close(release)
	waitBackgroundIdle(t, c)
	if texts := systemTexts(t, c); len(texts) != 0 {
		t.Fatalf("completion manufactured foreground notice: %v", texts)
	}
}

func TestBackgroundQueuesUntilIdleThenMerges(t *testing.T) {
	runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "完成了"}}}
	c, st := newSpawnEnv(t, runner)
	c.mgr.runner = noopRunner{}
	if err := c.mgr.claimConversation(c.run.ConversationID); err != nil {
		t.Fatal(err)
	}
	if _, err := c.SpawnBackground("agent.sub.general", "甲", "任务甲", ""); err != nil {
		t.Fatal(err)
	}
	if _, err := c.SpawnBackground("agent.sub.general", "乙", "任务乙", ""); err != nil {
		t.Fatal(err)
	}
	waitBackgroundIdle(t, c)
	if len(systemTexts(t, c)) != 0 {
		t.Fatal("notice delivered while conversation was busy")
	}
	c.mgr.releaseConversation(c.run.ConversationID)
	c.mgr.deliverNotices(c.run.ConversationID)
	if len(systemTexts(t, c)) != 0 {
		t.Fatal("idle transition inserted background reply")
	}
	runs, err := st.ListRunsByConversation(c.run.ConversationID)
	if err != nil {
		t.Fatal(err)
	}
	conv := 0
	for _, r := range runs {
		if r.Kind == core.RunConversation {
			conv++
		}
	}
	if conv != 0 {
		t.Fatalf("conversation runs = %d, want 0", conv)
	}
	hist := c.History()
	seen := false
	for _, m := range hist {
		if m.Role == core.RoleSystem {
			seen = true
		}
	}
	if seen {
		t.Fatal("background result entered foreground history")
	}
}

func TestBackgroundChildCannotDelegate(t *testing.T) {
	got := make(chan error, 1)
	runner := &gateRunner{spawn: func(rc runtime.RunCtx) error {
		_, err := rc.(interface {
			SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error)
		}).SpawnAgent("agent.sub.general", "再派", "继续", "")
		got <- err
		return nil
	}}
	c, _ := newSpawnEnv(t, runner)
	c.mgr.runner = noopRunner{}
	if _, err := c.SpawnBackground("agent.sub.general", "后台", "任务", ""); err != nil {
		t.Fatal(err)
	}
	defer waitBackgroundIdle(t, c)
	select {
	case err := <-got:
		if err == nil || !strings.Contains(err.Error(), "子代理不能再派发子代理") {
			t.Fatalf("err = %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("child did not try to delegate")
	}
}

func TestBackgroundCap(t *testing.T) {
	release := make(chan struct{})
	runner := &gateRunner{release: release}
	c, _ := newSpawnEnv(t, runner)
	c.mgr.runner = noopRunner{}
	if err := c.mgr.claimConversation(c.run.ConversationID); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 3; i++ {
		if _, err := c.SpawnBackground("agent.sub.general", "n", "t", ""); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := c.SpawnBackground("agent.sub.general", "overflow", "t", ""); err == nil || !strings.Contains(err.Error(), "3") {
		t.Fatalf("cap err = %v", err)
	}
	close(release)
	waitUntil(t, func() bool {
		c.mgr.noticeMu.Lock()
		defer c.mgr.noticeMu.Unlock()
		return c.mgr.bgInFlight[c.run.ConversationID] == 0
	})
}