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

package runmgr

import (
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"
	"time"

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

type materialParentSpy struct{ calls atomic.Int32 }

type materialFocusRunner struct {
	tasks   chan string
	release chan struct{}
}

func (*materialFocusRunner) Name() string { return "material-focus" }

func (r *materialFocusRunner) Execute(rc runtime.RunCtx) error {
	r.tasks <- rc.Subtask()
	<-r.release
	return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "整理完成"}})
}

func (*materialParentSpy) Name() string { return "material-parent-spy" }

func (s *materialParentSpy) Execute(rc runtime.RunCtx) error {
	s.calls.Add(1)
	return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "剩余 4 项（含 3 项依赖接口）是否纳入第二期？"}})
}

func startMaterialAckRun(t *testing.T, c *runCtx, docs []core.ParsedDocument, userText string) core.Run {
	t.Helper()
	conv, err := c.mgr.st.GetConversation(c.run.Owner, c.run.ConversationID)
	if err != nil {
		t.Fatal(err)
	}
	user := core.Message{ConversationID: conv.ID, Role: core.RoleUser}
	if userText != "" {
		user.Parts = []core.Part{{Type: core.PartText, Text: userText}}
	}
	if err := c.mgr.st.AppendMessage(&user); err != nil {
		t.Fatal(err)
	}
	run, err := c.mgr.StartConversationRun(conv, user, docs, nil)
	if err != nil {
		t.Fatal(err)
	}
	waitUntil(t, func() bool {
		current, err := c.mgr.st.GetRunByID(run.ID)
		c.mgr.convMu.Lock()
		busy := c.mgr.convActive[conv.ID]
		c.mgr.convMu.Unlock()
		return err == nil && current.Status == core.RunCompleted && !busy
	})
	return run
}

func materialAckTextForRun(t *testing.T, c *runCtx, runID core.ID) string {
	t.Helper()
	msgs, err := c.mgr.st.ListMessages(c.run.ConversationID)
	if err != nil {
		t.Fatal(err)
	}
	visible := ""
	for _, msg := range msgs {
		if msg.RunID != runID || msg.Role != core.RoleAssistant {
			continue
		}
		for _, part := range msg.Parts {
			if part.Type == core.PartText && !part.Process {
				visible += part.Text
			}
		}
	}
	return visible
}

func waitMaterialAckTestIdle(t *testing.T, c *runCtx) {
	t.Helper()
	waitUntil(t, func() bool {
		convID := c.run.ConversationID
		c.mgr.noticeMu.Lock()
		active := c.mgr.bgInFlight[convID]
		queued := len(c.mgr.bgQueue[convID])
		delivering := c.mgr.delivering[convID]
		pending := len(c.mgr.notices[convID])
		c.mgr.noticeMu.Unlock()
		c.mgr.convMu.Lock()
		busy := c.mgr.convActive[convID]
		c.mgr.convMu.Unlock()
		return active == 0 && queued == 0 && !delivering && pending == 0 && !busy
	})
}

func TestMaterialAckAttachmentOnlyDoesNotCallParentModel(t *testing.T) {
	hold := &holdRunner{release: make(chan struct{})}
	c, _ := newCardSpawnEnv(t, hold)
	c.mgr.SetExecutionBudgetFn(nil, 0)
	parent := &materialParentSpy{}
	c.mgr.runner = parent
	t.Cleanup(func() {
		close(hold.release)
		waitMaterialAckTestIdle(t, c)
	})
	run := startMaterialAckRun(t, c, materialDocs(), "")
	if got := parent.calls.Load(); got != 0 {
		t.Fatalf("parent provider path called %d times", got)
	}
	visible := materialAckTextForRun(t, c, run.ID)
	if !strings.Contains(visible, "待确认") || strings.Contains(visible, "剩余 4 项（含 3 项依赖接口）") {
		t.Fatalf("first visible response is not a truthful host acknowledgement: %q", visible)
	}
	childIDs, err := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
	if err != nil || len(childIDs) != 1 {
		t.Fatalf("material children = %v, err = %v", childIDs, err)
	}
	cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
	if err != nil || len(cards) != 1 || cards[0].Task == nil || cards[0].Task.CheckpointSeq != 0 {
		t.Fatalf("acknowledgement changed task checkpoints: cards=%+v err=%v", cards, err)
	}
}

func TestMaterialAckQueueAndReuse(t *testing.T) {
	waiter := &gateRunner{release: make(chan struct{})}
	c, _ := newCardSpawnEnv(t, waiter)
	c.mgr.SetExecutionBudgetFn(nil, 0)
	parent := &materialParentSpy{}
	c.mgr.runner = parent
	t.Cleanup(func() {
		close(waiter.release)
		waitMaterialAckTestIdle(t, c)
	})
	for i := 0; i < 3; i++ {
		if _, err := c.SpawnBackground("agent.sub.general", fmt.Sprintf("占位%d", i), "等待", ""); err != nil {
			t.Fatal(err)
		}
	}
	first := startMaterialAckRun(t, c, materialDocs(), "")
	second := startMaterialAckRun(t, c, materialDocs(), "")
	for _, run := range []core.Run{first, second} {
		text := materialAckTextForRun(t, c, run.ID)
		if !strings.Contains(text, "排队") || strings.Contains(text, "正在整理") {
			t.Fatalf("queued task was described as active work: %q", text)
		}
	}
	if got := parent.calls.Load(); got != 0 {
		t.Fatalf("queued material triggered %d parent provider calls", got)
	}
	cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
	if err != nil {
		t.Fatal(err)
	}
	materialCards := 0
	for _, card := range cards {
		if card.Title == materialCardTitle {
			materialCards++
			if card.Task == nil || card.Task.ExecutionState != "queued" {
				t.Fatalf("material card not queued: %+v", card.Task)
			}
		}
	}
	if materialCards != 1 {
		t.Fatalf("same material was enqueued %d times", materialCards)
	}
}

func TestMaterialAckActiveReuseDoesNotClaimNewStart(t *testing.T) {
	hold := &holdRunner{release: make(chan struct{})}
	c, _ := newCardSpawnEnv(t, hold)
	c.mgr.SetExecutionBudgetFn(nil, 0)
	parent := &materialParentSpy{}
	c.mgr.runner = parent
	t.Cleanup(func() {
		close(hold.release)
		waitMaterialAckTestIdle(t, c)
	})
	_ = startMaterialAckRun(t, c, materialDocs(), "")
	second := startMaterialAckRun(t, c, materialDocs(), "")
	text := materialAckTextForRun(t, c, second.ID)
	if !strings.Contains(text, "仍在整理") || strings.Contains(text, "已重新启动") || parent.calls.Load() != 0 {
		t.Fatalf("active material reuse was announced as new work: %q calls=%d", text, parent.calls.Load())
	}
	cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
	if err != nil || len(cards) != 1 {
		t.Fatalf("same active material created another task: cards=%d err=%v", len(cards), err)
	}
}

func TestMaterialAckCompletedReuseNoTextAndUserFollowup(t *testing.T) {
	c, _ := newCardSpawnEnv(t, &captureRunner{parts: []core.Part{{Type: core.PartText, Text: "整理完成：需要核对验收条件。"}}})
	c.mgr.SetExecutionBudgetFn(nil, 0)
	parent := &materialParentSpy{}
	c.mgr.runner = parent
	bindTrustedMaterialNoticeInput(t, c)
	docs := c.attachments
	_ = startMaterialAckRun(t, c, docs, "")
	key := c.materialRevision()
	waitUntil(t, func() bool {
		cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
		if err != nil || len(cards) != 1 || cards[0].Task == nil || cards[0].Task.ExecutionState != "completed" {
			return false
		}
		c.mgr.materialMu.Lock()
		active := c.mgr.materialActive[string(c.run.ConversationID)][key]
		c.mgr.materialMu.Unlock()
		c.mgr.convMu.Lock()
		busy := c.mgr.convActive[c.run.ConversationID]
		c.mgr.convMu.Unlock()
		return !active && !busy && cards[0].Work != nil && cards[0].Work.ResultVersion > 0
	})
	c.mgr.materialKeyRegister(string(c.run.ConversationID), key)
	decision := c.mgr.materialIntakeDecision(secondRunCtx(c, docs, ""))
	c.mgr.materialKeyRelease(string(c.run.ConversationID), key)
	if decision.State != materialCompleted || decision.Note != materialDoneNote {
		t.Fatalf("completed card was described as still running: %+v", decision)
	}
	before := parent.calls.Load()
	reused := startMaterialAckRun(t, c, docs, "")
	reply := materialAckTextForRun(t, c, reused.ID)
	if !strings.Contains(reply, "已整理完成") || strings.Contains(reply, "正在整理") ||
		parent.calls.Load() != before {
		t.Fatalf("completed no-text reuse was not a safe acknowledgement: reply=%q before=%d after=%d", reply, before, parent.calls.Load())
	}
	newQuestion := startMaterialAckRun(t, c, docs, "请回答新的具体问题")
	if parent.calls.Load() != before+1 || !strings.Contains(materialAckTextForRun(t, c, newQuestion.ID), "剩余 4 项") {
		t.Fatalf("explicit follow-up did not retain the ordinary conversation path")
	}
	cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
	if err != nil || len(cards) != 1 {
		t.Fatalf("completed material was respawned: cards=%d err=%v", len(cards), err)
	}
}

func TestMaterialAckCompletedReuseRequiresTrustedSources(t *testing.T) {
	c, _ := newCardSpawnEnv(t, &captureRunner{parts: []core.Part{{Type: core.PartText, Text: "整理完成"}}})
	c.mgr.SetExecutionBudgetFn(nil, 0)
	c.mgr.runner = &materialParentSpy{}
	docs := materialDocs() // parsed-only legacy material has no artifact/hash binding
	c.attachments = docs
	_ = startMaterialAckRun(t, c, docs, "")
	waitUntil(t, func() bool {
		cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
		return err == nil && len(cards) == 1 && cards[0].Task != nil && cards[0].Task.ExecutionState == "completed"
	})
	waitMaterialAckTestIdle(t, c)
	key := c.materialRevision()
	c.mgr.materialKeyRegister(string(c.run.ConversationID), key)
	defer c.mgr.materialKeyRelease(string(c.run.ConversationID), key)
	decision := c.mgr.materialIntakeDecision(secondRunCtx(c, docs, ""))
	if decision.State == materialCompleted || strings.Contains(materialAckRunner{decision: decision, attachmentCount: len(docs)}.reply(), "已整理完成") {
		t.Fatalf("unbound completed card was reused as verified: %+v", decision)
	}
}

func TestMaterialAckPrefersCurrentAttemptOverRecentlyUpdatedHistory(t *testing.T) {
	c, _ := newCardSpawnEnv(t, &captureRunner{})
	c.attachments = materialDocs()
	key := c.materialRevision()
	opts := bgSpawnOpts{name: materialCardTitle, material: true, taskKey: key, documents: c.attachments}
	oldBinding, oldCard, err := c.mgr.registerBackgroundTask(c.run, opts)
	if err != nil {
		t.Fatal(err)
	}
	if err := c.mgr.finishTaskProjection(oldBinding, core.AgentResultPayload{}, errors.New("old attempt failed")); err != nil {
		t.Fatal(err)
	}
	newBinding, newCard, err := c.mgr.registerBackgroundTask(c.run, opts)
	if err != nil {
		t.Fatal(err)
	}
	if err := c.mgr.startTaskProjection(newBinding, "run-new"); err != nil {
		t.Fatal(err)
	}
	if _, err := c.mgr.cards.SetPinned(c.run.Owner, oldCard.ID, true); err != nil {
		t.Fatal(err)
	}
	cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
	if err != nil || len(cards) != 2 || cards[0].ID != oldCard.ID {
		t.Fatalf("fixture did not put old attempt first: %+v err=%v", cards, err)
	}
	decision := c.mgr.materialDecisionForNote(c, key, materialPendingNote)
	if decision.CardID != newCard.ID || decision.State != materialRunning {
		t.Fatalf("older updated attempt masked current work: %+v", decision)
	}
}

func TestMaterialAckFailedCardDoesNotClaimStartFailure(t *testing.T) {
	c, _ := newCardSpawnEnv(t, &captureRunner{})
	c.attachments = materialDocs()
	key := c.materialRevision()
	binding, card, err := c.mgr.registerBackgroundTask(c.run, bgSpawnOpts{
		name: materialCardTitle, material: true, taskKey: key, documents: c.attachments,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := c.mgr.finishTaskProjection(binding, core.AgentResultPayload{}, errors.New("worker stopped after starting")); err != nil {
		t.Fatal(err)
	}
	c.mgr.materialKeyRegister(string(c.run.ConversationID), key)
	defer c.mgr.materialKeyRelease(string(c.run.ConversationID), key)
	decision := c.mgr.materialIntakeDecision(c)
	text := materialAckRunner{decision: decision, attachmentCount: len(c.attachments)}.reply()
	if decision.CardID != card.ID || decision.State != materialFailed ||
		!strings.Contains(text, "未能完成") || strings.Contains(text, "未能启动") {
		t.Fatalf("failed work was mislabeled as never started: state=%+v reply=%q", decision, text)
	}
}

func TestMaterialAckPreservesUserFocusAndSmallPath(t *testing.T) {
	child := &materialFocusRunner{tasks: make(chan string, 1), release: make(chan struct{})}
	c, _ := newCardSpawnEnv(t, child)
	c.mgr.SetExecutionBudgetFn(nil, 0)
	parent := &materialParentSpy{}
	c.mgr.runner = parent
	t.Cleanup(func() {
		close(child.release)
		waitMaterialAckTestIdle(t, c)
	})
	long := []core.ParsedDocument{{Meta: core.DocMeta{Filename: "长材料.txt", Parser: "txt"},
		Blocks: []core.DocBlock{{Type: core.BlockPara, Text: strings.Repeat("材料", 6100)}}}}
	focus := "重点核对预算审批"
	run := startMaterialAckRun(t, c, long, focus)
	select {
	case task := <-child.tasks:
		if !strings.Contains(task, "用户本次工作重点：\n"+focus) {
			t.Fatalf("original user focus missing from slow task: %q", task)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("material child did not receive the original user focus")
	}
	if parent.calls.Load() != 0 || !strings.Contains(materialAckTextForRun(t, c, run.ID), "工作重点") {
		t.Fatalf("long material did not use a focus-aware host acknowledgement")
	}
	small := []core.ParsedDocument{{Meta: core.DocMeta{Filename: "短说明.txt", Parser: "txt"},
		Blocks: []core.DocBlock{{Type: core.BlockPara, Text: "这是一条短说明。"}}}}
	followup := startMaterialAckRun(t, c, small, "请直接回答这份短说明")
	if parent.calls.Load() != 1 || !strings.Contains(materialAckTextForRun(t, c, followup.ID), "剩余 4 项") {
		t.Fatalf("text with small attachment did not retain the ordinary runner")
	}
}

func TestMaterialAckSkipsSquadEntrypoint(t *testing.T) {
	hold := &holdRunner{release: make(chan struct{})}
	c, _ := newCardSpawnEnv(t, hold)
	c.mgr.SetExecutionBudgetFn(nil, 0)
	direct := &materialParentSpy{}
	squad := &materialParentSpy{}
	c.mgr.runner, c.mgr.squad = direct, squad
	if err := c.mgr.settings.MergeUser(c.run.Owner, map[string]any{"runtime.entry": "squad"}); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		close(hold.release)
		waitMaterialAckTestIdle(t, c)
	})
	run := startMaterialAckRun(t, c, materialDocs(), "")
	if direct.calls.Load() != 0 || squad.calls.Load() != 0 ||
		!strings.Contains(materialAckTextForRun(t, c, run.ID), "待确认") {
		t.Fatalf("automatic material intake reached a model runner: direct=%d squad=%d", direct.calls.Load(), squad.calls.Load())
	}
}

func TestMaterialAckFailureAndUnknown(t *testing.T) {
	t.Run("spawn-failed", func(t *testing.T) {
		c, _ := newCardSpawnEnv(t, &captureRunner{})
		c.mgr.SetExecutionBudgetFn(nil, 0)
		parent := &materialParentSpy{}
		c.mgr.runner = parent
		c.mgr.shuttingDown.Store(true)
		run := startMaterialAckRun(t, c, materialDocs(), "")
		reply := materialAckTextForRun(t, c, run.ID)
		if parent.calls.Load() != 0 || !strings.Contains(reply, "未能启动") || strings.Contains(reply, "正在整理") {
			t.Fatalf("failed spawn produced a false acknowledgement: calls=%d reply=%q", parent.calls.Load(), reply)
		}
	})
	t.Run("card-unavailable", func(t *testing.T) {
		child := &gateRunner{entered: make(chan struct{}), release: make(chan struct{})}
		c, _ := newCardSpawnEnv(t, child)
		c.mgr.SetExecutionBudgetFn(nil, 0)
		parent := &materialParentSpy{}
		c.mgr.runner = parent
		cards := c.mgr.cards
		c.mgr.cards = nil
		t.Cleanup(func() {
			c.mgr.cards = cards
			close(child.release)
			waitMaterialAckTestIdle(t, c)
		})
		run := startMaterialAckRun(t, c, materialDocs(), "")
		select {
		case <-child.entered:
		case <-time.After(2 * time.Second):
			t.Fatal("material child never started")
		}
		reply := materialAckTextForRun(t, c, run.ID)
		if parent.calls.Load() != 0 || !strings.Contains(reply, "无法确认整理状态") ||
			strings.Contains(reply, "正在整理") || strings.Contains(reply, "查看工作卡") {
			t.Fatalf("unavailable card was described as running: calls=%d reply=%q", parent.calls.Load(), reply)
		}
	})
}