349 lines
11 KiB
Go
349 lines
11 KiB
Go
package runmgr
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"agent-system-backend/internal/bus"
|
|
"agent-system-backend/internal/profiles"
|
|
"agent-system-backend/internal/settings"
|
|
"agent-system-backend/internal/store"
|
|
"agent-system-backend/internal/tools"
|
|
core "agent-system-core"
|
|
)
|
|
|
|
func openRecoveryStore(t *testing.T) *store.Store {
|
|
t.Helper()
|
|
dsn := os.Getenv("AGENT_PG_TEST_DSN")
|
|
if dsn == "" {
|
|
st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "test.db"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { st.Close() })
|
|
return st
|
|
}
|
|
schema := "t_" + strings.ToLower(store.NewID("s")) + "_" + strconv.Itoa(os.Getpid())
|
|
admin, err := sql.Open("pgx", dsn)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// pg_trgm 预装 public（幂等）：避免 v4 扩展落进用例 schema 后随 CASCADE
|
|
// 删除造成的并行竞态（见 store 包 openTestStore 同名注释）。
|
|
if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
|
|
admin.Close()
|
|
t.Fatalf("preinstall pg_trgm: %v", err)
|
|
}
|
|
if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
|
|
admin.Close()
|
|
t.Fatalf("create schema: %v", err)
|
|
}
|
|
t.Cleanup(func() {
|
|
admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
|
|
admin.Close()
|
|
})
|
|
// 共用 store 的 DSN 运行参数拼接（修复盲拼 "&search_path="：对无查询串的
|
|
// URL DSN 会把参数并进库名，连不上库——svc.sh 形态 DSN 的既有缺陷）。
|
|
schemaDSN := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public")
|
|
st, err := store.OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { st.Close() })
|
|
return st
|
|
}
|
|
|
|
func newRecoveryManager(t *testing.T, st *store.Store) *Manager {
|
|
t.Helper()
|
|
reg := tools.NewRegistry()
|
|
mgr := New(st, bus.New(), reg, profiles.New(st), settings.NewService(st), nil, nil, nil, nil)
|
|
t.Cleanup(mgr.Shutdown)
|
|
return mgr
|
|
}
|
|
|
|
func recoveryConversation(t *testing.T, st *store.Store) core.Conversation {
|
|
t.Helper()
|
|
p, err := st.CreateProject(store.OwnerDefault, "p")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c, err := st.CreateConversation(store.OwnerDefault, p.ID, "c", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return c
|
|
}
|
|
|
|
func recoveryRun(t *testing.T, st *store.Store, conv core.Conversation, status core.RunStatus) core.Run {
|
|
t.Helper()
|
|
r := core.Run{
|
|
ConversationID: conv.ID, ProjectID: conv.ProjectID, Owner: conv.Owner,
|
|
Kind: core.RunConversation, Status: status, Phase: core.PhaseThinking,
|
|
}
|
|
if err := st.CreateRun(&r); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return r
|
|
}
|
|
|
|
func TestReconcileInterruptedSealsOpenRuns(t *testing.T) {
|
|
st := openRecoveryStore(t)
|
|
st.SetInstanceID("this")
|
|
mgr := newRecoveryManager(t, st)
|
|
mgr.SetInstanceID("this")
|
|
conv := recoveryConversation(t, st)
|
|
pending := recoveryRun(t, st, conv, core.RunPending)
|
|
running := recoveryRun(t, st, conv, core.RunRunning)
|
|
waiting := recoveryRun(t, st, conv, core.RunWaitingHITL)
|
|
done := recoveryRun(t, st, conv, core.RunCompleted)
|
|
card := &core.HITLCard{ID: "card_wait", RunID: waiting.ID, ConversationID: conv.ID, Question: "继续？", Kind: core.HITLKindConfirm}
|
|
if err := st.SaveHITLCard(conv.Owner, card); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := mgr.ReconcileInterrupted(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, id := range []string{pending.ID, running.ID, waiting.ID} {
|
|
got, err := st.GetRunByID(id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got.Status != core.RunInterrupted || got.EndedAt == nil || got.Error == nil || got.Error.Code != "run.interrupted" {
|
|
t.Fatalf("run %s: status=%s err=%v ended=%v", id, got.Status, got.Error, got.EndedAt)
|
|
}
|
|
}
|
|
kept, err := st.GetRunByID(done.ID)
|
|
if err != nil || kept.Status != core.RunCompleted {
|
|
t.Fatalf("terminal run changed: %+v %v", kept, err)
|
|
}
|
|
resolved, err := st.GetHITLCard(conv.Owner, "card_wait")
|
|
if err != nil || !resolved.Resolved {
|
|
t.Fatalf("card: %+v %v", resolved, err)
|
|
}
|
|
evs, err := st.EventsAfter(conv.ID, 0, 20)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
failed := 0
|
|
for _, e := range evs {
|
|
if e.Type != core.EvRunFailed {
|
|
continue
|
|
}
|
|
var p core.RunFailedPayload
|
|
if err := json.Unmarshal(e.Payload, &p); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if p.Code != "run.interrupted" {
|
|
t.Fatalf("payload %+v", p)
|
|
}
|
|
failed++
|
|
}
|
|
if failed != 3 {
|
|
t.Fatalf("run.failed events: %d", failed)
|
|
}
|
|
msgs, err := st.ListMessages(conv.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
asstByRun := map[string]int{}
|
|
for _, m := range msgs {
|
|
if m.Role == core.RoleAssistant {
|
|
asstByRun[m.RunID]++
|
|
if len(m.Parts) != 1 || m.Parts[0].Type != core.PartError || m.Parts[0].Err == nil || m.Parts[0].Err.Code != "run.interrupted" || m.Parts[0].Err.Message != "进程重启或停机，运行已中断" {
|
|
t.Fatalf("unexpected assistant message parts for run %s: %+v", m.RunID, m.Parts)
|
|
}
|
|
}
|
|
}
|
|
for _, id := range []string{pending.ID, running.ID, waiting.ID} {
|
|
if asstByRun[id] != 1 {
|
|
t.Fatalf("run %s: expected 1 assistant message, got %d", id, asstByRun[id])
|
|
}
|
|
}
|
|
if asstByRun[done.ID] != 0 {
|
|
t.Fatalf("terminal run %s: expected 0 assistant messages, got %d", done.ID, asstByRun[done.ID])
|
|
}
|
|
|
|
if err := mgr.ReconcileInterrupted(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
again, err := st.EventsAfter(conv.ID, 0, 20)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(again) != len(evs) {
|
|
t.Fatalf("second reconcile emitted more events: %d -> %d", len(evs), len(again))
|
|
}
|
|
msgsAgain, err := st.ListMessages(conv.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(msgsAgain) != len(msgs) {
|
|
t.Fatalf("second reconcile added duplicate messages: %d -> %d", len(msgs), len(msgsAgain))
|
|
}
|
|
}
|
|
|
|
func TestReconcileInterruptedAssistantNotice(t *testing.T) {
|
|
st := openRecoveryStore(t)
|
|
st.SetInstanceID("this")
|
|
mgr := newRecoveryManager(t, st)
|
|
mgr.SetInstanceID("this")
|
|
|
|
// 1. Conversation with an open run that gets sealed
|
|
conv1 := recoveryConversation(t, st)
|
|
userMsg1 := core.Message{ConversationID: conv1.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hello"}}}
|
|
if err := st.AppendMessage(&userMsg1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sealedRun := recoveryRun(t, st, conv1, core.RunRunning)
|
|
|
|
// 2. Conversation with an already terminal run
|
|
conv2 := recoveryConversation(t, st)
|
|
userMsg2 := core.Message{ConversationID: conv2.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}}
|
|
if err := st.AppendMessage(&userMsg2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
terminalRun := recoveryRun(t, st, conv2, core.RunCompleted)
|
|
|
|
// 3. Conversation with an open run that already has an assistant message
|
|
conv3 := recoveryConversation(t, st)
|
|
userMsg3 := core.Message{ConversationID: conv3.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "test"}}}
|
|
if err := st.AppendMessage(&userMsg3); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
openRunWithMsg := recoveryRun(t, st, conv3, core.RunRunning)
|
|
asstMsg3 := core.Message{ConversationID: conv3.ID, RunID: openRunWithMsg.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "already answered"}}}
|
|
if err := st.AppendMessage(&asstMsg3); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if err := mgr.ReconcileInterrupted(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// conv1: history contains exactly one assistant message with run.interrupted
|
|
msgs1, err := st.ListMessages(conv1.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var asst1 []core.Message
|
|
for _, m := range msgs1 {
|
|
if m.Role == core.RoleAssistant {
|
|
asst1 = append(asst1, m)
|
|
}
|
|
}
|
|
if len(asst1) != 1 {
|
|
t.Fatalf("conv1: expected exactly 1 assistant message, got %d", len(asst1))
|
|
}
|
|
if asst1[0].RunID != sealedRun.ID {
|
|
t.Fatalf("conv1: expected assistant message for run %s, got %s", sealedRun.ID, asst1[0].RunID)
|
|
}
|
|
if len(asst1[0].Parts) != 1 || asst1[0].Parts[0].Type != core.PartError || asst1[0].Parts[0].Err == nil || asst1[0].Parts[0].Err.Code != "run.interrupted" || asst1[0].Parts[0].Err.Message != "进程重启或停机，运行已中断" {
|
|
t.Fatalf("conv1: unexpected parts: %+v", asst1[0].Parts)
|
|
}
|
|
|
|
// conv2: already-terminal run must have NO assistant message written
|
|
kept, err := st.GetRunByID(terminalRun.ID)
|
|
if err != nil || kept.Status != core.RunCompleted {
|
|
t.Fatalf("terminal run changed: %+v %v", kept, err)
|
|
}
|
|
msgs2, err := st.ListMessages(conv2.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, m := range msgs2 {
|
|
if m.Role == core.RoleAssistant {
|
|
t.Fatalf("conv2: unexpected assistant message for already-terminal run: %+v", m)
|
|
}
|
|
}
|
|
|
|
// conv3: run had an assistant message already, so no duplicate was created
|
|
msgs3, err := st.ListMessages(conv3.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var asst3 []core.Message
|
|
for _, m := range msgs3 {
|
|
if m.Role == core.RoleAssistant {
|
|
asst3 = append(asst3, m)
|
|
}
|
|
}
|
|
if len(asst3) != 1 || asst3[0].ID != asstMsg3.ID {
|
|
t.Fatalf("conv3: expected only original assistant message, got %+v", asst3)
|
|
}
|
|
}
|
|
|
|
func TestReconcileLeavesOtherInstanceOnPostgres(t *testing.T) {
|
|
if os.Getenv("AGENT_PG_TEST_DSN") == "" {
|
|
t.Skip("AGENT_PG_TEST_DSN not set")
|
|
}
|
|
st := openRecoveryStore(t)
|
|
if st.Dialect() != "postgres" {
|
|
t.Fatal("expected postgres")
|
|
}
|
|
mgr := newRecoveryManager(t, st)
|
|
mgr.SetInstanceID("this")
|
|
conv := recoveryConversation(t, st)
|
|
st.SetInstanceID("other")
|
|
other := recoveryRun(t, st, conv, core.RunRunning)
|
|
st.SetInstanceID("this")
|
|
own := recoveryRun(t, st, conv, core.RunRunning)
|
|
if err := mgr.ReconcileInterrupted(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
gotOther, err := st.GetRunByID(other.ID)
|
|
if err != nil || gotOther.Status != core.RunRunning {
|
|
t.Fatalf("other instance: %+v %v", gotOther, err)
|
|
}
|
|
gotOwn, err := st.GetRunByID(own.ID)
|
|
if err != nil || gotOwn.Status != core.RunInterrupted {
|
|
t.Fatalf("own instance: %+v %v", gotOwn, err)
|
|
}
|
|
}
|
|
|
|
func TestFinishRunSettleFailureMarksFailed(t *testing.T) {
|
|
st := openRecoveryStore(t)
|
|
mgr := newRecoveryManager(t, st)
|
|
conv := recoveryConversation(t, st)
|
|
run := recoveryRun(t, st, conv, core.RunRunning)
|
|
st.SetSettleFault(os.ErrClosed)
|
|
rc := &runCtx{mgr: mgr, run: run, h: &runHandle{
|
|
assistant: []core.Part{{Type: core.PartText, Text: "hello"}},
|
|
started: time.Now(),
|
|
}}
|
|
mgr.finishRun(rc, core.RunCompleted, core.Usage{}, nil)
|
|
got, err := st.GetRunByID(run.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got.Status != core.RunFailed || got.Error == nil || got.Error.Code != "run.settle_failed" {
|
|
t.Fatalf("status=%s err=%v", got.Status, got.Error)
|
|
}
|
|
msgs, err := st.ListMessages(conv.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(msgs) != 0 {
|
|
t.Fatalf("assistant message must not land when settle fails: %d", len(msgs))
|
|
}
|
|
evs, err := st.EventsAfter(conv.ID, 0, 10)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(evs) != 1 || evs[0].Type != core.EvRunFailed {
|
|
t.Fatalf("events: %+v", evs)
|
|
}
|
|
var p core.RunFailedPayload
|
|
if err := json.Unmarshal(evs[0].Payload, &p); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if p.Code != "run.settle_failed" {
|
|
t.Fatalf("payload %+v", p)
|
|
}
|
|
}
|