344 lines
13 KiB
Go
344 lines
13 KiB
Go
package runmgr
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
agentllm "agent-native/runtime/llm"
|
||
bruntime "agent-system-backend/internal/runtime"
|
||
"agent-system-backend/internal/store"
|
||
core "agent-system-core"
|
||
)
|
||
|
||
type meteredChildRunner struct{}
|
||
|
||
func (meteredChildRunner) Name() string { return "metered-child" }
|
||
|
||
func (meteredChildRunner) Execute(rc bruntime.RunCtx) error {
|
||
rc.(interface{ AddUsage(core.Usage) }).AddUsage(core.Usage{PromptTokens: 2, CompletionTokens: 1, ToolCalls: 2})
|
||
return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "完成"}})
|
||
}
|
||
|
||
func TestSubRunPersistsToolCallUsage(t *testing.T) {
|
||
c, st := newSpawnEnv(t, meteredChildRunner{})
|
||
result, err := c.SpawnSubagent("核对", "核对材料", "")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
child, err := st.GetRunByID(result.AgentRun)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if child.Usage == nil || child.Usage.ToolCalls != 2 || child.Usage.TotalTokens != 3 {
|
||
t.Fatalf("child usage omitted tool count: %+v", child.Usage)
|
||
}
|
||
}
|
||
|
||
func TestFailedSubRunPersistsReturnedTerminalUsage(t *testing.T) {
|
||
stream := func(context.Context, string, []agentllm.Message, []agentllm.ToolDecl, agentllm.StreamOptions) <-chan agentllm.Event {
|
||
out := make(chan agentllm.Event, 1)
|
||
out <- agentllm.Event{Kind: agentllm.KindMessageEnd,
|
||
Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopError},
|
||
Usage: agentllm.Usage{PromptTokens: 12, CompletionTokens: 5, TotalTokens: 17, CacheReadTokens: 3},
|
||
Err: errors.New("tool stream ended before [DONE]"),
|
||
}
|
||
close(out)
|
||
return out
|
||
}
|
||
c, st := newSpawnEnv(t, &bruntime.SubLLMRunner{Stream: stream})
|
||
if _, err := c.SpawnSubagent("核对", "整理材料", ""); err == nil {
|
||
t.Fatal("failed subagent was accepted")
|
||
}
|
||
runs, err := st.ListRunsByConversation(c.run.ConversationID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for _, run := range runs {
|
||
if run.Kind != core.RunSubagent {
|
||
continue
|
||
}
|
||
if run.Status != core.RunFailed || run.Usage == nil || run.Usage.TotalTokens != 17 ||
|
||
run.Usage.PromptTokens != 12 || run.Usage.CompletionTokens != 5 || run.Usage.CacheReadTokens != 3 {
|
||
t.Fatalf("failed child lost returned provider Usage: %+v", run)
|
||
}
|
||
return
|
||
}
|
||
t.Fatal("failed child run was not persisted")
|
||
}
|
||
|
||
func TestFailedSubRunPersistsUsageAfterSuccessfulToolTurn(t *testing.T) {
|
||
calls := 0
|
||
stream := func(context.Context, string, []agentllm.Message, []agentllm.ToolDecl, agentllm.StreamOptions) <-chan agentllm.Event {
|
||
calls++
|
||
out := make(chan agentllm.Event, 1)
|
||
switch calls {
|
||
case 1:
|
||
out <- agentllm.Event{Kind: agentllm.KindMessageEnd,
|
||
Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopToolCalls,
|
||
ToolCalls: []agentllm.ToolCall{{ID: "call-1", Name: "time_now", Arguments: "{}"}}},
|
||
Usage: agentllm.Usage{PromptTokens: 10, CompletionTokens: 2, TotalTokens: 12, CacheReadTokens: 4},
|
||
}
|
||
case 2:
|
||
out <- agentllm.Event{Kind: agentllm.KindMessageEnd,
|
||
Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopError},
|
||
Usage: agentllm.Usage{PromptTokens: 6, CompletionTokens: 4, CacheReadTokens: 2},
|
||
Err: errors.New("tool stream ended before [DONE]"),
|
||
}
|
||
}
|
||
close(out)
|
||
return out
|
||
}
|
||
c, st := newSpawnEnv(t, &bruntime.SubLLMRunner{Stream: stream})
|
||
if _, err := c.SpawnSubagent("核对", "整理材料", ""); err == nil {
|
||
t.Fatal("failed subagent was accepted")
|
||
}
|
||
if calls != 2 {
|
||
t.Fatalf("provider calls = %d, want 2", calls)
|
||
}
|
||
runs, err := st.ListRunsByConversation(c.run.ConversationID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for _, run := range runs {
|
||
if run.Kind != core.RunSubagent {
|
||
continue
|
||
}
|
||
if run.Status != core.RunFailed || run.Usage == nil {
|
||
t.Fatalf("mixed-turn child status/usage = %s/%+v", run.Status, run.Usage)
|
||
}
|
||
if run.Usage.PromptTokens != 16 || run.Usage.CompletionTokens != 6 ||
|
||
run.Usage.TotalTokens != 22 || run.Usage.CacheReadTokens != 6 ||
|
||
run.Usage.LastPromptTokens != 6 || run.Usage.LastCompletionTokens != 4 ||
|
||
run.Usage.ToolCalls != 1 {
|
||
t.Fatalf("mixed-turn failed child usage = %+v", *run.Usage)
|
||
}
|
||
return
|
||
}
|
||
t.Fatal("failed child run was not persisted")
|
||
}
|
||
|
||
// ---- N2 WP5：子代理独立转录、中断续跑、通知持久化 ----
|
||
|
||
// 派生即落 request.json；完成时 transcript.jsonl + done.json 落定。
|
||
func TestSubagentTranscriptFilesWritten(t *testing.T) {
|
||
c, _ := newSpawnEnv(t, &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "结果文本"}}})
|
||
ws := t.TempDir()
|
||
c.mgr.SetWorkspaceRoot(ws)
|
||
res, err := c.SpawnSubagent("检索", "查资料", "提示语")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
dir := filepath.Join(ws, c.run.Owner, c.run.ConversationID, "subagents", res.AgentRun)
|
||
reqBody, err := os.ReadFile(filepath.Join(dir, "request.json"))
|
||
if err != nil {
|
||
t.Fatalf("request.json: %v", err)
|
||
}
|
||
if !strings.Contains(string(reqBody), "查资料") {
|
||
t.Fatalf("request.json 内容：%s", reqBody)
|
||
}
|
||
trBody, err := os.ReadFile(filepath.Join(dir, "transcript.jsonl"))
|
||
if err != nil || !strings.Contains(string(trBody), "结果文本") {
|
||
t.Fatalf("transcript.jsonl：%v %s", err, trBody)
|
||
}
|
||
doneBody, err := os.ReadFile(filepath.Join(dir, "done.json"))
|
||
if err != nil || !strings.Contains(string(doneBody), "completed") {
|
||
t.Fatalf("done.json：%v %s", err, doneBody)
|
||
}
|
||
}
|
||
|
||
// 中断子 run 凭 request.json 续跑：同任务重派、hint 带前次转录路径、
|
||
// resumed.json 防重复续跑。
|
||
func TestResumeInterruptedChildrenRespawns(t *testing.T) {
|
||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "续跑结果"}}}
|
||
c, st := newSpawnEnv(t, runner)
|
||
c.mgr.runner = noopRunner{}
|
||
ws := t.TempDir()
|
||
c.mgr.SetWorkspaceRoot(ws)
|
||
|
||
// 父 run 落库（续跑要经 store 取回派生上下文）。
|
||
parentRun := c.run
|
||
if err := st.CreateRun(&parentRun); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 造一个中断的子 run 现场（等价于崩溃前的派生）。
|
||
child := core.Run{
|
||
ID: store.NewID("run"), ParentRunID: c.run.ID, ConversationID: c.run.ConversationID,
|
||
ProjectID: c.run.ProjectID, Owner: c.run.Owner, Kind: core.RunSubagent, ProfileID: c.run.ProfileID,
|
||
Status: core.RunRunning, CreatedAt: time.Now().UTC(),
|
||
}
|
||
if err := st.CreateRun(&child); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
c.mgr.writeSubagentRequest(child, "agent.sub.explore", "检索", "原始任务", "原始提示", true)
|
||
if _, err := st.SealInterrupted(child.ID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
before, err := st.SubagentRunIDs(c.run.ConversationID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
c.mgr.ResumeInterruptedChildren()
|
||
|
||
ids, err := st.SubagentRunIDs(c.run.ConversationID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(ids) != len(before)+1 {
|
||
t.Fatalf("续跑应新增一个子 run：before=%v after=%v", before, ids)
|
||
}
|
||
// resumed.json 幂等：第二次调用不再新增。
|
||
c.mgr.ResumeInterruptedChildren()
|
||
ids2, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||
if len(ids2) != len(ids) {
|
||
t.Fatalf("重复续跑：%v → %v", ids, ids2)
|
||
}
|
||
waitResumedTranscripts(t, c.mgr, child, ids)
|
||
}
|
||
|
||
func waitResumedTranscripts(t *testing.T, mgr *Manager, original core.Run, ids map[string]bool) {
|
||
t.Helper()
|
||
waitUntil(t, func() bool {
|
||
for id := range ids {
|
||
if id != original.ID && doneStatus(mgr.subagentDir(original.Owner, original.ConversationID, id)) != "completed" {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
})
|
||
}
|
||
|
||
// 通知先落库后投递；模拟重启后 LoadPendingNotices 重投并清库。
|
||
func TestNoticesPersistAndRedeliverAfterRestart(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)
|
||
}
|
||
// Restore an actual legacy pending notice. New children never create these.
|
||
c.mgr.enqueueNotice(c.run.ConversationID, bgNotice{owner: c.run.Owner, name: "甲", payload: core.AgentResultPayload{Status: core.RunCompleted, Summary: "后台结果"}})
|
||
// 模拟重启：同一 store 上的新 Manager 只接通知面（会话仍忙时不应丢）。
|
||
c.mgr.releaseConversation(c.run.ConversationID)
|
||
m2 := New(st, c.mgr.bus, nil, c.mgr.profiles, c.mgr.settings, noopRunner{}, runner, nil, nil)
|
||
m2.LoadPendingNotices()
|
||
waitUntil(t, func() bool {
|
||
items, err := st.ListNotices()
|
||
return err == nil && len(items) == 0
|
||
})
|
||
texts := systemTexts(t, c)
|
||
if len(texts) != 0 {
|
||
t.Fatalf("legacy notice must not restart foreground: %v", texts)
|
||
}
|
||
}
|
||
|
||
// 适配层 Background 分支：Request.Background=true → SpawnBackground（语义
|
||
// 与产品 delegate 的 background 参数一致）。在本包内用 runCtx 直验。
|
||
func TestSpawnBackgroundRecordsDetach(t *testing.T) {
|
||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "bg"}}}
|
||
c, _ := newSpawnEnv(t, runner)
|
||
c.mgr.runner = noopRunner{}
|
||
id, err := c.SpawnBackground("agent.sub.general", "后台", "任务", "")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
run, err := c.mgr.st.GetRunByID(id)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if run.Kind != core.RunSubagent || run.Status != core.RunCompleted && run.Status != core.RunRunning {
|
||
t.Fatalf("run = %+v", run)
|
||
}
|
||
}
|
||
|
||
// N3 WP1：done.json 语义分流——优雅停机写的 interrupted 不挡续跑（一次且仅一次）；
|
||
// completed 终态挡续跑。
|
||
func TestResumeAfterGracefulInterruptMarker(t *testing.T) {
|
||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "续跑结果"}}}
|
||
c, st := newSpawnEnv(t, runner)
|
||
c.mgr.runner = noopRunner{}
|
||
ws := t.TempDir()
|
||
c.mgr.SetWorkspaceRoot(ws)
|
||
parentRun := c.run
|
||
if err := st.CreateRun(&parentRun); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
child := core.Run{
|
||
ID: store.NewID("run"), ParentRunID: c.run.ID, ConversationID: c.run.ConversationID,
|
||
ProjectID: c.run.ProjectID, Owner: c.run.Owner, Kind: core.RunSubagent, ProfileID: c.run.ProfileID,
|
||
Status: core.RunRunning, CreatedAt: time.Now().UTC(),
|
||
}
|
||
if err := st.CreateRun(&child); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
c.mgr.writeSubagentRequest(child, "agent.sub.explore", "检索", "原始任务", "原始提示", true)
|
||
// 模拟优雅停机：partial 转录 + done.json{status:"interrupted"}。
|
||
c.mgr.writeSubagentTranscript(child, []core.Part{{Type: core.PartText, Text: "半截现场"}}, "interrupted")
|
||
if _, err := st.SealInterrupted(child.ID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
before, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||
c.mgr.ResumeInterruptedChildren()
|
||
ids, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||
if len(ids) != len(before)+1 {
|
||
t.Fatalf("优雅停机现场应被续跑一次：before=%v after=%v", before, ids)
|
||
}
|
||
c.mgr.ResumeInterruptedChildren()
|
||
ids2, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||
if len(ids2) != len(ids) {
|
||
t.Fatalf("重复续跑：%v → %v", ids, ids2)
|
||
}
|
||
waitResumedTranscripts(t, c.mgr, child, ids)
|
||
}
|
||
|
||
func TestResumeSkipsCompletedMarker(t *testing.T) {
|
||
c, st := newSpawnEnv(t, &gateRunner{})
|
||
c.mgr.runner = noopRunner{}
|
||
ws := t.TempDir()
|
||
c.mgr.SetWorkspaceRoot(ws)
|
||
parentRun := c.run
|
||
if err := st.CreateRun(&parentRun); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
child := core.Run{
|
||
ID: store.NewID("run"), ParentRunID: c.run.ID, ConversationID: c.run.ConversationID,
|
||
ProjectID: c.run.ProjectID, Owner: c.run.Owner, Kind: core.RunSubagent, ProfileID: c.run.ProfileID,
|
||
Status: core.RunRunning, CreatedAt: time.Now().UTC(),
|
||
}
|
||
if err := st.CreateRun(&child); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
c.mgr.writeSubagentRequest(child, "agent.sub.explore", "检索", "原始任务", "", true)
|
||
c.mgr.writeSubagentTranscript(child, []core.Part{{Type: core.PartText, Text: "已完成"}}, "completed")
|
||
if _, err := st.SealInterrupted(child.ID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
before, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||
c.mgr.ResumeInterruptedChildren()
|
||
ids, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||
if len(ids) != len(before) {
|
||
t.Fatalf("completed 终态不得续跑：before=%v after=%v", before, ids)
|
||
}
|
||
}
|
||
|
||
// N3 WP6：子代理记账链——AddUsage 累计拆分字段，finishChild 合并进 child.Usage。
|
||
func TestAddUsageAccumulatesSplitFields(t *testing.T) {
|
||
c, _ := newSpawnEnv(t, &gateRunner{})
|
||
c.h = &runHandle{}
|
||
c.AddUsage(core.Usage{PromptTokens: 100, CompletionTokens: 10, LastPromptTokens: 100, LastCompletionTokens: 10, CacheReadTokens: 50, CacheWriteTokens: 4, ToolCalls: 1})
|
||
c.AddUsage(core.Usage{PromptTokens: 60, CompletionTokens: 6, LastPromptTokens: 60, LastCompletionTokens: 6, CacheReadTokens: 20, CacheWriteTokens: 2, ToolCalls: 2})
|
||
c.mgr.mu.Lock()
|
||
defer c.mgr.mu.Unlock()
|
||
if c.h.promptTokens != 160 || c.h.completionTokens != 16 ||
|
||
c.h.lastPromptTokens != 60 || c.h.lastCompletionTokens != 6 ||
|
||
c.h.cacheReadTokens != 70 || c.h.cacheWriteTokens != 6 || c.h.toolCalls != 3 {
|
||
t.Fatalf("记账链：%+v", c.h)
|
||
}
|
||
}
|