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

package runmgr

import (
	"agent-system-backend/internal/cardsvc"
	core "agent-system-core"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

func TestFailedTaskDoesNotPromiseAutomaticContinuation(t *testing.T) {
	c := checkpointEnvironment(t)
	if err := c.mgr.mutateTask(c.task, func(s *core.TaskSnapshot) error {
		s.NextStep = "输出最终结构化整理结果"
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if err := c.mgr.finishTaskProjection(c.task, core.AgentResultPayload{}, errors.New("budget_exceeded: run token budget exhausted")); err != nil {
		t.Fatal(err)
	}
	card, err := c.mgr.st.GetCard(c.run.Owner, c.task.cardID)
	if err != nil {
		t.Fatal(err)
	}
	if card.Task.Phase.ID != "failed" || card.Task.Phase.Label != "执行已停止" || strings.Contains(card.Task.NextStep, "输出最终") || strings.Contains(card.Task.Conclusion, "budget_exceeded") {
		t.Fatalf("stopped task retained a process promise: %+v", card.Task)
	}
	if !strings.Contains(card.Task.Conclusion, "尚无可用") || strings.Contains(card.Task.NextStep, "已有发现") || card.Task.Remaining == "待任务进展后更新" {
		t.Fatalf("zero accepted findings were presented as retained results: %+v", card.Task)
	}
}

func TestTaskConclusionSkipsProcessText(t *testing.T) {
	c, _ := newCardSpawnEnv(t, &gateRunner{parts: []core.Part{{Type: core.PartText, Process: true, Text: "我先检查知识库。"}, {Type: core.PartText, Text: "日期仍待裁定，预算审批尚未完成。"}}})
	c.mgr.runner = noopRunner{}
	if _, err := c.mgr.spawnBackground(c.run, "agent.sub.general", "整理", "核对资料", ""); err != nil {
		t.Fatal(err)
	}
	waitUntil(t, func() bool {
		records, err := c.mgr.st.TaskCards()
		return err == nil && len(records) == 1 && records[0].Card.Task.Terminal()
	})
	records, err := c.mgr.st.TaskCards()
	if err != nil {
		t.Fatal(err)
	}
	conclusion := records[0].Card.Task.Conclusion
	if strings.Contains(conclusion, "检查知识库") || conclusion != "日期仍待裁定，预算审批尚未完成。" {
		t.Fatalf("process text became the terminal conclusion: %q", conclusion)
	}
}

func TestTaskBoundInterruptedChildDoesNotRespawn(t *testing.T) {
	c, _ := newCardSpawnEnv(t, noopRunner{})
	st := c.mgr.st
	c.mgr.runner = noopRunner{}
	c.mgr.SetWorkspaceRoot(t.TempDir())
	parent := c.run
	if err := st.CreateRun(&parent); err != nil {
		t.Fatal(err)
	}
	child := core.Run{ID: "run_bound_interrupted", ParentRunID: parent.ID, Owner: parent.Owner, ProjectID: parent.ProjectID, ConversationID: parent.ConversationID, Kind: core.RunSubagent, ProfileID: parent.ProfileID, Status: core.RunRunning}
	if err := st.CreateRun(&child); err != nil {
		t.Fatal(err)
	}
	b, _, err := c.mgr.registerBackgroundTask(parent, bgSpawnOpts{name: "整理"})
	if err != nil {
		t.Fatal(err)
	}
	if err = c.mgr.startTaskProjection(b, child.ID); err != nil {
		t.Fatal(err)
	}
	c.mgr.writeSubagentRequest(child, "agent.sub.explore", "整理", "旧请求", "旧提示", true)
	c.mgr.writeSubagentTranscript(child, nil, "interrupted")
	if _, err = st.SealInterrupted(child.ID); err != nil {
		t.Fatal(err)
	}
	c.mgr.ResumeInterruptedChildren()
	ids, err := st.SubagentRunIDs(parent.ConversationID)
	if err != nil {
		t.Fatal(err)
	}
	if len(ids) != 1 {
		t.Fatalf("canonical task was automatically dispatched as a new attempt: %v", ids)
	}
	dir := c.mgr.subagentDir(child.Owner, child.ConversationID, child.ID)
	if _, err = os.Stat(filepath.Join(dir, "resumed.json")); !os.IsNotExist(err) {
		t.Fatalf("bound task acquired legacy resume marker: %v", err)
	}
}

func TestTaskQueueKeepsIdentityWhenPromoted(t *testing.T) {
	runner := &holdRunner{release: make(chan struct{})}
	c, _ := newCardSpawnEnv(t, runner)
	c.mgr.runner = noopRunner{}
	for i := 0; i < 4; i++ {
		if _, e := c.mgr.spawnBackgroundOpts(c.run, bgSpawnOpts{name: "整理", task: "读材料", queue: true}); e != nil {
			t.Fatal(e)
		}
	}
	cards, e := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
	if e != nil || len(cards) != 4 {
		t.Fatal(cards, e)
	}
	var queued core.ActivityCard
	for _, card := range cards {
		if card.Task == nil {
			t.Fatal("background work has no canonical task")
		}
		if card.Task.ExecutionState == "queued" {
			queued = card
		}
	}
	if queued.ID == "" || queued.Task.RunID != "" {
		t.Fatal("queue did not precede execution")
	}
	close(runner.release)
	waitUntil(t, func() bool {
		got, e := c.mgr.st.GetCard(c.run.Owner, queued.ID)
		return e == nil && got.Task != nil && got.Task.ExecutionState == "completed"
	})
	got, _ := c.mgr.st.GetCard(c.run.Owner, queued.ID)
	if got.Task.TaskID != queued.Task.TaskID || got.Task.AttemptID != queued.Task.AttemptID || got.Task.RunID == "" {
		t.Fatal("promotion replaced identity")
	}
}
func TestTaskReconcileOwnedQueueAndForeignLease(t *testing.T) {
	st := openRecoveryStore(t)
	m := newRecoveryManager(t, st)
	m.SetInstanceID("A")
	m.SetCards(cardsvc.New(st, nil))
	conv := recoveryConversation(t, st)
	parent := recoveryRun(t, st, conv, core.RunRunning)
	_, own, e := m.registerBackgroundTask(parent, bgSpawnOpts{name: "自己的队列"})
	if e != nil {
		t.Fatal(e)
	}
	other := newRecoveryManager(t, st)
	other.SetInstanceID("B")
	other.SetCards(cardsvc.New(st, nil))
	_, foreign, e := other.registerBackgroundTask(parent, bgSpawnOpts{name: "其他实例"})
	if e != nil {
		t.Fatal(e)
	}
	if e = m.ReconcileInterrupted(); e != nil {
		t.Fatal(e)
	}
	got, _ := st.GetCard(conv.Owner, own.ID)
	if got.Task.ExecutionState != "interrupted" {
		t.Fatal("orphan queue retained promise", got.Task)
	}
	if !strings.Contains(got.Task.NextStep, "重新提出请求") {
		t.Fatal("interrupted queue promised automatic continuation", got.Task.NextStep)
	}
	got, _ = st.GetCard(conv.Owner, foreign.ID)
	if st.Dialect() == "postgres" && got.Task.ExecutionState != "queued" {
		t.Fatal("foreign lease was sealed")
	}
}

func TestTaskProjectionStaleExecutionCannotOverwrite(t *testing.T) {
	st := openRecoveryStore(t)
	m := newRecoveryManager(t, st)
	m.SetInstanceID("A")
	m.SetCards(cardsvc.New(st, nil))
	conv := recoveryConversation(t, st)
	parent := recoveryRun(t, st, conv, core.RunRunning)
	binding, card, e := m.registerBackgroundTask(parent, bgSpawnOpts{name: "整理"})
	if e != nil {
		t.Fatal(e)
	}
	st.ReleaseExecution(taskLeaseKey(binding.taskID), binding.instanceID, binding.epoch)
	if _, e = st.TryClaimExecution(taskLeaseKey(binding.taskID), "B", "", 60); e != nil {
		t.Fatal(e)
	}
	if e = m.startTaskProjection(binding, "old-run"); e == nil {
		t.Fatal("old executor overwrote new lease")
	}
	got, _ := st.GetCard(conv.Owner, card.ID)
	if got.Task.RunID != "" || got.Task.Revision != 1 {
		t.Fatal("fenced write leaked", got.Task)
	}
}