371 lines
14 KiB
Go
371 lines
14 KiB
Go
package runmgr
|
|
|
|
import (
|
|
"agent-system-backend/internal/store"
|
|
core "agent-system-core"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
)
|
|
|
|
func TestWorkActionRechecksProjectBeforeCommit(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
calls := 0
|
|
c.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { calls++; return calls == 1, nil })
|
|
_, err := c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.submit", core.WorkCardAction{ResultVersion: 1, SelectedIDs: []string{"summary"}})
|
|
if !errors.Is(err, store.ErrNotFound) {
|
|
t.Fatalf("revoked project allowed commit: %v", err)
|
|
}
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
if got.Work.Decision.State != "open" {
|
|
t.Fatal("revocation persisted choice")
|
|
}
|
|
ids, _ := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
|
if len(ids) != 1 {
|
|
t.Fatal("revocation launched delivery")
|
|
}
|
|
}
|
|
|
|
func TestLegacyWorkProjectionCanBeReferencedAndMarkedRead(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
card.Work = nil
|
|
if err := c.mgr.cards.Update(c.run.Owner, card); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
card, _ = c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
view := c.mgr.PresentWorkCards(c.run.Owner, []core.ActivityCard{card})[0]
|
|
if view.Work == nil || view.Work.Result == "" {
|
|
t.Fatal("historical fullresult missing")
|
|
}
|
|
stored, version, err := c.mgr.st.GetCardVersion(c.run.Owner, card.ID)
|
|
if err != nil || stored.Work == nil || !stored.UpdatedAt.Equal(card.UpdatedAt) || stored.Work.ReadVersion != 1 {
|
|
t.Fatal("legacy projection was not cached without rewriting historical time")
|
|
}
|
|
c.mgr.PresentWorkCards(c.run.Owner, []core.ActivityCard{stored})
|
|
_, nextVersion, _ := c.mgr.st.GetCardVersion(c.run.Owner, card.ID)
|
|
if version != nextVersion {
|
|
t.Fatal("cached projection rewritten on every read")
|
|
}
|
|
if err := c.mgr.ValidateWorkReference(c.run.Owner, c.run.ConversationID, WorkReference{card.ID, 1}); err != nil {
|
|
t.Fatalf("visible reference rejected: %v", err)
|
|
}
|
|
if _, err := c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.read", core.WorkCardAction{ResultVersion: 1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
if got.Work == nil || got.Work.ReadVersion != 1 {
|
|
t.Fatal("historical read did not persist snapshot")
|
|
}
|
|
}
|
|
|
|
func TestBackgroundCompletionPreservesConcurrentDismissAndPin(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
original := card
|
|
card.Work = nil
|
|
card.Status = core.CardDismissed
|
|
card.Pinned = true
|
|
if err := c.mgr.cards.Update(c.run.Owner, card); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c.mgr.finishBackgroundCard(c.run.Owner, original, core.AgentResultPayload{Summary: "整理完毕"}, nil, backgroundCardKV(card))
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
if got.Work == nil || got.Work.Result == "" || got.Status != core.CardDismissed || !got.Pinned {
|
|
t.Fatalf("completion overwrote user controls: %+v", got)
|
|
}
|
|
}
|
|
|
|
func TestSavedChildRepairsMissingCompletionProjectionWithoutExecution(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
full := card.Work.Result
|
|
// Task cards keep the status owned by their projection; only the work
|
|
// projection is missing here.
|
|
card.Work = &core.WorkCard{}
|
|
card.Pinned = true
|
|
if err := c.mgr.cards.Update(c.run.Owner, card); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
before, _ := c.mgr.st.SubagentRunIDs(card.ConversationID)
|
|
view := c.mgr.PresentWorkCards(c.run.Owner, []core.ActivityCard{card})[0]
|
|
if view.Status != core.CardDone || view.Work == nil || view.Work.Result != full || !view.Pinned {
|
|
t.Fatalf("saved result not restored: %+v", view)
|
|
}
|
|
after, _ := c.mgr.st.SubagentRunIDs(card.ConversationID)
|
|
if len(before) != len(after) {
|
|
t.Fatal("repair executed a model task")
|
|
}
|
|
}
|
|
|
|
func TestMissingSavedRunDoesNotSealAnEmptyCompletion(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
card.Work = &core.WorkCard{}
|
|
card.SourceRunID = "missing-run"
|
|
if err := c.mgr.cards.Update(c.run.Owner, card); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c.mgr.finishBackgroundCard(c.run.Owner, card, core.AgentResultPayload{Summary: "完成"}, nil, nil)
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
if got.Status != card.Status || got.Work.ResultVersion != 0 {
|
|
t.Fatal("missing durable result sealed as completed")
|
|
}
|
|
}
|
|
|
|
func TestDispatchFailureWithoutChildStillFinishesTheWorkCard(t *testing.T) {
|
|
c, _ := completedMaterialWork(t)
|
|
card := c.mgr.createBackgroundCard(c.run, "", "尚未启动的任务", "", nil, "排队中")
|
|
c.mgr.finishBackgroundCard(c.run.Owner, card, core.AgentResultPayload{}, errors.New("执行未获准启动"), nil)
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
if got.Status != core.CardFailed || got.Work == nil || !strings.Contains(got.Work.Result, "执行未获准启动") {
|
|
t.Fatalf("failed dispatch stranded: %+v", got)
|
|
}
|
|
}
|
|
|
|
func completedMaterialWork(t *testing.T) (*runCtx, core.ActivityCard) {
|
|
t.Helper()
|
|
c, _ := newCardSpawnEnv(t, &captureRunner{parts: []core.Part{{Type: core.PartText, Text: "有依据的材料整理结果及末尾来源。"}}})
|
|
parent := c.run
|
|
if err := c.mgr.st.CreateRun(&parent); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c.attachments = materialDocs()
|
|
c.mgr.materialIntake(c)
|
|
waitBackgroundIdle(t, c)
|
|
cards, err := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
|
if err != nil || len(cards) != 1 {
|
|
t.Fatalf("cards %v %v", cards, err)
|
|
}
|
|
return c, cards[0]
|
|
}
|
|
|
|
func TestWorkChoiceDeferReadReopenAndFreeInput(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
a := core.WorkCardAction{ResultVersion: 1}
|
|
got, err := c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.defer", a)
|
|
if err != nil || got.Work.Decision.State != "deferred" {
|
|
t.Fatalf("defer %v %v", got.Work, err)
|
|
}
|
|
got, err = c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.read", a)
|
|
if err != nil || got.Work.ReadVersion != 1 || got.Work.Decision.State != "deferred" {
|
|
t.Fatalf("read %v %v", got.Work, err)
|
|
}
|
|
got, err = c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.reopen", a)
|
|
if err != nil || got.Work.Decision.State != "open" {
|
|
t.Fatalf("reopen %v %v", got.Work, err)
|
|
}
|
|
if _, err = c.mgr.WorkCardAction("other-owner", card.ID, "work.submit", a); !errors.Is(err, store.ErrNotFound) {
|
|
t.Fatalf("cross-owner action: %v", err)
|
|
}
|
|
if _, err = c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.read", core.WorkCardAction{ResultVersion: 2}); !errors.Is(err, store.ErrConflict) {
|
|
t.Fatalf("stale version: %v", err)
|
|
}
|
|
a.FreeText = "只整理培训重点,不另写评审报告"
|
|
got, err = c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.submit", a)
|
|
if err != nil || got.Work.Decision.State != "submitted" || got.Work.Decision.FreeText != a.FreeText || got.Work.Decision.RespondedAt == nil {
|
|
t.Fatalf("free submission %v %v", got.Work, err)
|
|
}
|
|
waitBackgroundIdle(t, c)
|
|
if len(systemTexts(t, c)) != 0 {
|
|
t.Fatal("choice acceptance manufactured conversation reply")
|
|
}
|
|
}
|
|
|
|
func TestConcurrentWorkSubmissionsLaunchOnce(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
a := core.WorkCardAction{ResultVersion: 1, SelectedIDs: []string{"summary"}}
|
|
var wg sync.WaitGroup
|
|
for range 8 {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
_, err := c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.submit", a)
|
|
if err != nil && !errors.Is(err, store.ErrConflict) {
|
|
t.Errorf("submit: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
waitBackgroundIdle(t, c)
|
|
ids, _ := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
|
if len(ids) != 2 {
|
|
t.Fatalf("original + one delivery expected, got %v", ids)
|
|
}
|
|
if _, err := c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.submit", a); err != nil {
|
|
t.Fatalf("idempotent retry: %v", err)
|
|
}
|
|
ids, _ = c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
|
if len(ids) != 2 {
|
|
t.Fatal("retry spawned another delivery")
|
|
}
|
|
}
|
|
|
|
func TestDirectMaterialSummaryUsesExistingForegroundOnly(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
conv, err := c.mgr.st.GetConversation(c.run.Owner, c.run.ConversationID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
msg := core.Message{Parts: []core.Part{{Type: core.PartText, Text: "总结一下这份材料的摘要吧,我先看看里面有什么重点的信息"}}}
|
|
context, err := c.mgr.bindWorkContext(conv, msg, nil)
|
|
if err != nil || !strings.Contains(context, card.Work.Result) {
|
|
t.Fatalf("binding %q %v", context, err)
|
|
}
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
if got.Work.Decision.State != "submitted" || got.Work.Decision.FreeText != messageText(msg) {
|
|
t.Fatalf("decision %v", got.Work.Decision)
|
|
}
|
|
ids, _ := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
|
if len(ids) != 1 {
|
|
t.Fatal("direct request launched duplicate background")
|
|
}
|
|
if ref := c.mgr.directWorkReference(conv, "帮我总结一下昨天的会议"); ref != nil {
|
|
t.Fatal("unrelated topic bound material choice")
|
|
}
|
|
}
|
|
|
|
func TestExplicitWorkReferenceAllowsNewSummaryAfterPreviousChoice(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
_, err := c.mgr.WorkCardAction(c.run.Owner, card.ID, "work.submit", core.WorkCardAction{ResultVersion: 1, SelectedIDs: []string{"summary"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
waitBackgroundIdle(t, c)
|
|
conv, err := c.mgr.st.GetConversation(c.run.Owner, c.run.ConversationID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
msg := core.Message{Parts: []core.Part{{Type: core.PartText, Text: "请把这份材料的重点总结为英文"}}}
|
|
context, err := c.mgr.bindWorkContext(conv, msg, &WorkReference{card.ID, 1})
|
|
if err != nil || strings.HasPrefix(context, workReceiptPrefix) || !strings.Contains(context, card.Work.Result) {
|
|
t.Fatalf("explicit follow-up request discarded: %q %v", context, err)
|
|
}
|
|
}
|
|
|
|
func TestImplicitWorkBindingDoesNotCaptureQuestionsAboutMaterial(t *testing.T) {
|
|
for _, question := range []string{"这份材料的重点适合儿童患者吗?", "这份材料的摘要是否覆盖维护要求?", "这份文档为什么把这个列为重点?"} {
|
|
if materialPurposeRequest(question) {
|
|
t.Fatalf("ordinary question captured: %s", question)
|
|
}
|
|
}
|
|
if !materialPurposeRequest("总结一下这份材料的摘要吧,我先看看里面有什么重点的信息") {
|
|
t.Fatal("explicit summary request no longer recognized")
|
|
}
|
|
}
|
|
|
|
func TestButtonAndDirectSummaryRaceKeepsOneExecution(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
hold := &holdRunner{release: make(chan struct{})}
|
|
c.mgr.subRunner = hold
|
|
conv, _ := c.mgr.st.GetConversation(c.run.Owner, c.run.ConversationID)
|
|
foreground := core.Run{ID: store.NewID("run"), ConversationID: conv.ID, ProjectID: conv.ProjectID, Owner: conv.Owner, Kind: core.RunConversation, Status: core.RunRunning}
|
|
if err := c.mgr.st.CreateRun(&foreground); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
msg := core.Message{Parts: []core.Part{{Type: core.PartText, Text: "总结一下这份材料的摘要吧,我先看看里面有什么重点的信息"}}}
|
|
start := make(chan struct{})
|
|
var wg sync.WaitGroup
|
|
wg.Add(2)
|
|
var context string
|
|
var bindErr, errorSubmit error
|
|
go func() {
|
|
defer wg.Done()
|
|
<-start
|
|
_, errorSubmit = c.mgr.WorkCardAction(conv.Owner, card.ID, "work.submit", core.WorkCardAction{ResultVersion: 1, SelectedIDs: []string{"summary"}})
|
|
}()
|
|
go func() {
|
|
defer wg.Done()
|
|
<-start
|
|
context, bindErr = c.mgr.bindWorkContext(conv, msg, nil, foreground.ID)
|
|
}()
|
|
close(start)
|
|
wg.Wait()
|
|
if bindErr != nil {
|
|
t.Fatalf("accepted foreground intent lost: %v", bindErr)
|
|
}
|
|
if errorSubmit != nil && !errors.Is(errorSubmit, store.ErrConflict) {
|
|
t.Fatalf("button %v", errorSubmit)
|
|
}
|
|
ids, _ := c.mgr.st.SubagentRunIDs(conv.ID)
|
|
if strings.HasPrefix(context, workReceiptPrefix) {
|
|
if len(ids) != 2 {
|
|
t.Fatalf("button winner missing child: %v", ids)
|
|
}
|
|
} else if len(ids) != 1 || !strings.Contains(context, card.Work.Result) {
|
|
t.Fatalf("foreground winner duplicated execution: ids %v context %q", ids, context)
|
|
}
|
|
close(hold.release)
|
|
waitBackgroundIdle(t, c)
|
|
}
|
|
|
|
func TestWorkDispatchRecoveryDoesNotRetryModels(t *testing.T) {
|
|
for _, accepted := range []bool{false, true} {
|
|
t.Run(fmt.Sprint(accepted), func(t *testing.T) {
|
|
c, card := completedMaterialWork(t)
|
|
card.Work.Decision.State = "dispatching"
|
|
card.Work.Decision.RunID = store.NewID("run")
|
|
if accepted {
|
|
run := core.Run{ID: card.Work.Decision.RunID, Owner: c.run.Owner, ProjectID: c.run.ProjectID, ConversationID: c.run.ConversationID, ParentRunID: c.run.ID, Kind: core.RunSubagent, Status: core.RunFailed}
|
|
if err := c.mgr.st.CreateRun(&run); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if err := c.mgr.cards.Update(c.run.Owner, card); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
before, _ := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
|
c.mgr.recoverWorkDispatches()
|
|
after, _ := c.mgr.st.SubagentRunIDs(c.run.ConversationID)
|
|
if len(after) != len(before) {
|
|
t.Fatal("recovery retried model")
|
|
}
|
|
got, _ := c.mgr.st.GetCard(c.run.Owner, card.ID)
|
|
want := "deferred"
|
|
if accepted {
|
|
want = "submitted"
|
|
}
|
|
if got.Work.Decision.State != want {
|
|
t.Fatalf("state %s want %s", got.Work.Decision.State, want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBackgroundWorkKeepsCompleteFinalResultWithoutForegroundNotice(t *testing.T) {
|
|
final := strings.Repeat("材料事实。", 100) + "末尾依据仍然保留"
|
|
c, _ := newCardSpawnEnv(t, &captureRunner{parts: []core.Part{
|
|
{Type: core.PartThinking, Text: "已公开的推理过程", Process: true},
|
|
{Type: core.PartText, Text: "我来检索", Process: true},
|
|
{Type: core.PartText, Text: final},
|
|
}})
|
|
c.mgr.runner = noopRunner{}
|
|
if _, err := c.SpawnBackground("agent.sub.general", "完整结果", "整理", ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
waitUntil(t, func() bool {
|
|
cards, _ := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
|
return len(cards) == 1 && cards[0].Status == core.CardDone
|
|
})
|
|
cards, _ := c.mgr.cards.List(c.run.Owner, c.run.ConversationID)
|
|
if cards[0].Work == nil || cards[0].Work.Result != final {
|
|
t.Fatalf("complete final result missing: %+v", cards[0].Work)
|
|
}
|
|
if got := systemTexts(t, c); len(got) != 0 {
|
|
t.Fatalf("background completion inserted chat notice: %v", got)
|
|
}
|
|
}
|
|
|
|
func TestMaterialKeyDistinguishesEqualLengthContentAndParser(t *testing.T) {
|
|
a := materialDocs()[:1]
|
|
b := materialDocs()[:1]
|
|
b[0].Blocks = []core.DocBlock{{Type: core.BlockPara, Text: strings.Repeat("甲", len([]rune(a[0].PlainText())))}}
|
|
if materialKey(a) == materialKey(b) {
|
|
t.Fatal("equal length different content shares material key")
|
|
}
|
|
b = materialDocs()[:1]
|
|
b[0].Meta.Parser = "another-parser"
|
|
if materialKey(a) == materialKey(b) {
|
|
t.Fatal("different parser shares material key")
|
|
}
|
|
}
|