weichen
cf83d30743
Why: F11 - SQLite runs lost mailbox/handoffs/board across restarts, identical JSON entries collapsed during decode, and checkpoint or terminal-save failures were swallowed behind success events. Evidence: store and checkpoint tests pass incl. injected-failure and round-trip tables (verify-u8.json, exit=0); schema migration preserves rows. Impact: runs survive restarts with collaboration state; save failures are visible before any completion event; async/on_exit semantics kept. Refs: feature/agent-stack-audit-remediation-b2 U8
760 lines
28 KiB
Go
760 lines
28 KiB
Go
package platform
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"reflect"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"agent-squad/squad"
|
||
)
|
||
|
||
func newTestRunState(wsDir string) *squad.RunState {
|
||
// 固定时刻：多次构造出的状态除工作区路径外完全一致，可直接比对两个后端的读回结果。
|
||
now := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||
return &squad.RunState{
|
||
Meta: squad.RunMetadata{
|
||
ID: "run_test", Status: "executing", WorkDir: wsDir,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
Metadata: map[string]string{"mode": ""},
|
||
},
|
||
Plan: squad.SquadPlan{
|
||
Workflow: squad.WorkflowPlan{
|
||
WorkflowID: "wf1", Status: "in_progress",
|
||
Steps: []squad.StepSpec{
|
||
{ID: "s1", Role: "researcher", Status: "completed", Attempts: 1, AttemptID: "run_test/s1/1"},
|
||
{ID: "s2", Role: "writer", Status: "pending"},
|
||
},
|
||
},
|
||
Squad: squad.SquadConfig{SquadName: "demo"},
|
||
},
|
||
BB: squad.BlackboardState{Facts: []squad.FactItem{{ID: "f1", Claim: "x"}}, Fields: map[string]string{"topic": "demo"}},
|
||
Template: "tpl",
|
||
Approved: map[string]bool{"step:s1": true},
|
||
RepairBudgetRemaining: 1,
|
||
RepairBudgetGranted: 0,
|
||
// 内容完全相同的重复条目（且 clarifications 与 steering 的 JSON 文本一样）：
|
||
// 内容键寻址的解码会把它们互相覆盖掉——这里逐条都要活下来。
|
||
Clarifications: []string{"同一句澄清", "同一句澄清"},
|
||
Steering: []string{"同一句澄清", "同一句澄清"},
|
||
Mailbox: []squad.MailMessage{
|
||
{ID: "m1", From: "researcher", To: "writer", Text: "同一句澄清", At: now},
|
||
{ID: "m2", From: "researcher", To: "writer", Text: "同一句澄清", At: now},
|
||
},
|
||
Handoffs: []squad.HandoffRecord{
|
||
{FromRole: "researcher", FromStep: "s1", ToRole: "writer", Output: "同一句澄清", At: now},
|
||
{FromRole: "writer", FromStep: "s2", ToRole: "reviewer", Output: "同一句澄清", At: now},
|
||
},
|
||
Board: []squad.BoardTask{
|
||
{ID: "s1", Status: "done", Attempts: 1},
|
||
{ID: "s2", Status: "claimed", ClaimedBy: "writer", LeaseExpires: now.Add(time.Minute), Attempts: 2},
|
||
},
|
||
Usage: map[string]*squad.Usage{"writer": {PromptTokens: 10, CompletionTokens: 5, Steps: 1}},
|
||
}
|
||
}
|
||
|
||
// TestSQLiteStoreRoundTrip：run/gate/checkpoint/leader 全链路（Q1 WP5 端口实现）。
|
||
func TestSQLiteStoreRoundTrip(t *testing.T) {
|
||
st, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer st.Close()
|
||
ws := filepath.Join(t.TempDir(), "runs", "run_test")
|
||
orig := newTestRunState(ws)
|
||
if err := st.SaveRun(orig); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
got, err := st.LoadRun(ws)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got.Meta.ID != "run_test" || got.Plan.Workflow.WorkflowID != "wf1" || got.Template != "tpl" {
|
||
t.Fatalf("loaded = %+v", got.Meta)
|
||
}
|
||
if !got.Approved["step:s1"] || got.RepairBudgetRemaining != 1 ||
|
||
len(got.Clarifications) != 2 || len(got.Steering) != 2 {
|
||
t.Fatalf("scalars = %+v", got)
|
||
}
|
||
// 协作态必须真的落库（F11a）：重启后邮箱/移交/任务板不能静默丢。
|
||
if len(got.Mailbox) != 2 || got.Mailbox[0].ID != "m1" || got.Mailbox[1].ID != "m2" {
|
||
t.Fatalf("mailbox = %+v", got.Mailbox)
|
||
}
|
||
if len(got.Handoffs) != 2 || got.Handoffs[1].ToRole != "reviewer" {
|
||
t.Fatalf("handoffs = %+v", got.Handoffs)
|
||
}
|
||
if len(got.Board) != 2 || got.Board[1].Status != "claimed" || got.Board[1].ClaimedBy != "writer" {
|
||
t.Fatalf("board = %+v", got.Board)
|
||
}
|
||
if got.Usage["writer"].PromptTokens != 10 {
|
||
t.Fatalf("usage = %+v", got.Usage)
|
||
}
|
||
if _, err := st.LoadRun(filepath.Join(t.TempDir(), "nope")); err == nil {
|
||
t.Fatal("missing run should error")
|
||
}
|
||
|
||
// gate：写→读→删。
|
||
gate := &squad.HitlGate{PrevStatus: squad.RunExecuting}
|
||
if err := st.SaveGate(ws, gate); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if g2, err := st.LoadGate(ws); err != nil || g2 == nil {
|
||
t.Fatalf("load gate: %v %+v", err, g2)
|
||
}
|
||
if err := st.SaveGate(ws, nil); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if g2, err := st.LoadGate(ws); err != nil || g2 != nil {
|
||
t.Fatalf("gate after delete: %v %+v", err, g2)
|
||
}
|
||
|
||
// checkpoint：两层，latest 取最大 superstep。
|
||
if err := st.SaveCheckpoint(ws, squad.Checkpoint{Superstep: 1, At: time.Now(), GraphHash: "h1", StepStates: []squad.StepSnapshot{{ID: "s1", Status: "completed", AttemptID: "run_test/s1/1"}}}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := st.SaveCheckpoint(ws, squad.Checkpoint{Superstep: 2, At: time.Now(), GraphHash: "h2"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
latest, err := st.LoadLatestCheckpoint(ws)
|
||
if err != nil || latest == nil {
|
||
t.Fatalf("latest: %v", err)
|
||
}
|
||
if latest.Superstep != 2 || latest.GraphHash != "h2" {
|
||
t.Fatalf("latest = %+v", latest)
|
||
}
|
||
if l2, err := st.LoadLatestCheckpoint(filepath.Join(t.TempDir(), "nope")); err != nil || l2 != nil {
|
||
t.Fatalf("latest for unknown: %v %+v", err, l2)
|
||
}
|
||
|
||
// leader bytes。
|
||
if err := st.SaveLeader(ws, []byte(`{"turns":2}`)); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
data, err := st.LoadLeader(ws)
|
||
if err != nil || string(data) != `{"turns":2}` {
|
||
t.Fatalf("leader = %q %v", data, err)
|
||
}
|
||
}
|
||
|
||
// TestEngineRunsOnSQLiteStore：引擎注入 SQLiteStore 后运行状态进 SQLite、
|
||
// 不再写 run 级 JSON 文件；第二引擎可从中恢复（验收第三条的实证形态）。
|
||
func TestEngineRunsOnSQLiteStore(t *testing.T) {
|
||
dbPath := filepath.Join(t.TempDir(), "runs.db")
|
||
runStore, err := OpenSQLiteStore(dbPath)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer runStore.Close()
|
||
dir := t.TempDir()
|
||
reg := squad.NewRoleRegistry()
|
||
for _, spec := range []squad.RoleSpec{{Role: "researcher", Backend: "script"}, {Role: "writer", Backend: "script"}, {Role: "reviewer", Backend: "script"}} {
|
||
if err := reg.Register(spec); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
backends := squad.NewBackendResolver()
|
||
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{Success: true, Logs: "ok", Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")}}})
|
||
blueprints := squad.NewBlueprintRegistry()
|
||
if err := blueprints.Register(squad.Blueprint{
|
||
Key: "demo", Title: "入库演示", Match: []string{"入库演示"},
|
||
Steps: []squad.StepSpec{
|
||
{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"notes.md", "fields.json"}},
|
||
{ID: "s2", Role: "writer", Title: "成文", Targets: []string{"output.md"}},
|
||
{ID: "s3", Role: "reviewer", Title: "终审", Targets: []string{"output.md"}},
|
||
},
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
eng, err := squad.NewEngine(squad.Options{
|
||
DataDir: dir, Registry: reg, Blueprints: blueprints,
|
||
Planner: squad.NewRulePlanner(),
|
||
Backends: backends, Sink: squad.NopSink{},
|
||
Store: runStore,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
res, err := eng.Run(context.Background(), squad.Goal{Text: "做一个三步入库演示", HitlMode: squad.HitlAuto}, squad.RunOptions{})
|
||
if err != nil || res.Status != squad.RunCompleted {
|
||
t.Fatalf("run: %v %+v", err, res)
|
||
}
|
||
// run 级 JSON 不在盘上（状态全进了 SQLite）。
|
||
for _, f := range []string{"workspace.json", "workflow.json", "squad.json", "blackboard.json", "runstate.json"} {
|
||
if _, err := os.Stat(filepath.Join(res.WorkDir, f)); !os.IsNotExist(err) {
|
||
t.Fatalf("%s should not be on disk with SQLite store (err=%v)", f, err)
|
||
}
|
||
}
|
||
// 状态可经端口读回。
|
||
st, err := runStore.LoadRun(res.WorkDir)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if st.Plan.Workflow.Status != "completed" {
|
||
t.Fatalf("workflow status = %s", st.Plan.Workflow.Status)
|
||
}
|
||
latest, err := runStore.LoadLatestCheckpoint(res.WorkDir)
|
||
if err != nil || latest == nil {
|
||
t.Fatalf("latest checkpoint: %v", err)
|
||
}
|
||
// 第二引擎（同库）恢复可用。
|
||
eng2, err := squad.NewEngine(squad.Options{
|
||
DataDir: dir, Registry: reg, Blueprints: blueprints,
|
||
Planner: squad.NewRulePlanner(),
|
||
Backends: backends, Sink: squad.NopSink{},
|
||
Store: runStore,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := eng2.Inspect(res.RunID); err != nil {
|
||
t.Fatalf("inspect from second engine: %v", err)
|
||
}
|
||
}
|
||
|
||
// TestEngineFailedResultWithCompletePatchesStaysFailedOnSQLiteStore：F05/U3B 的
|
||
// 不变式在 SQLite-store 集成层同样成立——executor 报失败就是失败，patches
|
||
// （交付物）齐全也不得把运行翻成 completed：文件只是诊断信息，不是权威结果。
|
||
func TestEngineFailedResultWithCompletePatchesStaysFailedOnSQLiteStore(t *testing.T) {
|
||
runStore, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer runStore.Close()
|
||
dir := t.TempDir()
|
||
reg := squad.NewRoleRegistry()
|
||
if err := reg.Register(squad.RoleSpec{Role: "researcher", Backend: "script"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
backends := squad.NewBackendResolver()
|
||
// 失败结果但携带全套交付物：旧语义会把它提升为成功，U3B 起必须保持失败。
|
||
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{
|
||
Success: false, Logs: "stub-refused",
|
||
Patches: map[string][]byte{
|
||
"output.md": []byte(goodReportLite),
|
||
"notes.md": []byte(goodReportLite),
|
||
"fields.json": []byte("{}"),
|
||
},
|
||
}})
|
||
blueprints := squad.NewBlueprintRegistry()
|
||
if err := blueprints.Register(squad.Blueprint{
|
||
Key: "faildemo", Title: "失败演示", Match: []string{"失败演示"},
|
||
Steps: []squad.StepSpec{{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"output.md"}}},
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
eng, err := squad.NewEngine(squad.Options{
|
||
DataDir: dir, Registry: reg, Blueprints: blueprints,
|
||
Planner: squad.NewRulePlanner(),
|
||
Backends: backends, Sink: squad.NopSink{},
|
||
Store: runStore,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
res, err := eng.Run(context.Background(), squad.Goal{Text: "做一个失败演示", HitlMode: squad.HitlAuto}, squad.RunOptions{})
|
||
if err != nil {
|
||
t.Fatalf("run: %v", err)
|
||
}
|
||
if res.Status == squad.RunCompleted {
|
||
t.Fatalf("failed executor result must not complete (patches are diagnostic only)")
|
||
}
|
||
if res.Status != squad.RunFailed {
|
||
t.Fatalf("status = %s, want failed", res.Status)
|
||
}
|
||
if res.Plan == nil || res.Plan.Steps[0].Status != "failed" {
|
||
t.Fatalf("s1 should be failed, plan = %+v", res.Plan)
|
||
}
|
||
if !strings.Contains(res.Error, "stub-refused") {
|
||
t.Fatalf("failure reason not preserved: %q", res.Error)
|
||
}
|
||
// 同一断言穿过 Store 端口：SQLite 里落盘的也是失败，而非 completed。
|
||
st, err := runStore.LoadRun(res.WorkDir)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if st.Plan.Workflow.Status != "failed" {
|
||
t.Fatalf("persisted workflow status = %s, want failed", st.Plan.Workflow.Status)
|
||
}
|
||
if got := st.Plan.Workflow.Steps[0]; got.Status != "failed" || !strings.Contains(got.Error, "stub-refused") {
|
||
t.Fatalf("persisted step = %+v", got)
|
||
}
|
||
// patches 确已物化（文件齐全），依然不翻转状态：权威是执行结果，不是文件。
|
||
sub := filepath.FromSlash(st.Plan.Workflow.Steps[0].SubWorkspace)
|
||
if _, err := os.Stat(filepath.Join(res.WorkDir, sub, "output.md")); err != nil {
|
||
t.Fatalf("diagnostic patch should still be materialized: %v", err)
|
||
}
|
||
}
|
||
|
||
const goodReportLite = "# 测试交付物\n\n## 概述\n\n这是由脚本后端生成的合格交付物，长度超过五十个字符以满足最小质检要求。\n"
|
||
|
||
// ---- F11a/F11b：SQLite 与 FileStore 同一契约、重复条目不丢、旧库升级 ----
|
||
|
||
// assertSameRunState 逐字段比对两个后端读回的运行状态（同一契约）。
|
||
// 字段按规范 JSON 比对：时间等值字段（创建/更新时间、消息时刻、租约到期）在两侧
|
||
// 都经历过 JSON 往返，按墙钟等价判定，不受内部表示（时区指针/单调时钟）影响。
|
||
func assertSameRunState(t *testing.T, got, want *squad.RunState) {
|
||
t.Helper()
|
||
sameJSON := func(label string, a, b any) {
|
||
t.Helper()
|
||
ab, err := json.Marshal(a)
|
||
if err != nil {
|
||
t.Fatalf("%s: marshal got: %v", label, err)
|
||
}
|
||
bb, err := json.Marshal(b)
|
||
if err != nil {
|
||
t.Fatalf("%s: marshal want: %v", label, err)
|
||
}
|
||
if string(ab) != string(bb) {
|
||
t.Fatalf("%s differs:\n got %s\nwant %s", label, ab, bb)
|
||
}
|
||
}
|
||
sameJSON("meta", got.Meta, want.Meta)
|
||
sameJSON("plan", got.Plan, want.Plan)
|
||
sameJSON("blackboard", got.BB, want.BB)
|
||
sameJSON("template", got.Template, want.Template)
|
||
sameJSON("approved", got.Approved, want.Approved)
|
||
sameJSON("repair_budget_remaining", got.RepairBudgetRemaining, want.RepairBudgetRemaining)
|
||
sameJSON("repair_budget_granted", got.RepairBudgetGranted, want.RepairBudgetGranted)
|
||
sameJSON("clarifications", got.Clarifications, want.Clarifications)
|
||
sameJSON("steering", got.Steering, want.Steering)
|
||
sameJSON("mailbox", got.Mailbox, want.Mailbox)
|
||
sameJSON("handoffs", got.Handoffs, want.Handoffs)
|
||
sameJSON("board", got.Board, want.Board)
|
||
sameJSON("usage", got.Usage, want.Usage)
|
||
}
|
||
|
||
// TestRunStoreContractParity：FileStore 与 SQLiteStore 满足同一份 Store 契约——
|
||
// 同一份运行状态（含 mailbox/handoffs/board）写进去，读回来必须一致。
|
||
func TestRunStoreContractParity(t *testing.T) {
|
||
sq, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer sq.Close()
|
||
|
||
fileWS := filepath.Join(t.TempDir(), "runs", "run_file")
|
||
sqlWS := filepath.Join(t.TempDir(), "runs", "run_sql")
|
||
// FileStore 侧的模板文件由工作区装配期写出（PrepareWorkspace），先补齐，
|
||
// 使两侧的 SaveRun/LoadRun 覆盖同一批字段。
|
||
if err := os.MkdirAll(fileWS, 0o755); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := os.WriteFile(filepath.Join(fileWS, "template.md"), []byte("tpl"), 0o644); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
want := newTestRunState(fileWS)
|
||
if err := (squad.FileStore{}).SaveRun(want); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
gotFile, err := (squad.FileStore{}).LoadRun(fileWS)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sqlState := newTestRunState(sqlWS)
|
||
if err := sq.SaveRun(sqlState); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
gotSQL, err := sq.LoadRun(sqlWS)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
// SQLite 读回的内容与写进去的一致。
|
||
assertSameRunState(t, gotSQL, sqlState)
|
||
// 两个后端的读回结果逐字段一致（除工作区路径这一运行自身的键）。
|
||
assertSameRunState(t, gotFile, want)
|
||
fileMeta, sqlMeta := gotFile.Meta, gotSQL.Meta
|
||
fileMeta.WorkDir, sqlMeta.WorkDir = "", ""
|
||
fileJSON, _ := json.Marshal(fileMeta)
|
||
sqlJSON, _ := json.Marshal(sqlMeta)
|
||
if string(fileJSON) != string(sqlJSON) {
|
||
t.Fatalf("meta parity broken:\n file %s\n sql %s", fileJSON, sqlJSON)
|
||
}
|
||
}
|
||
|
||
// TestSQLiteRunStoreKeepsIdenticalEntries：内容相同的两条澄清/补充指令必须都活着
|
||
// （旧实现拿 JSON 文本当键，两条同文条目只剩一条）。
|
||
func TestSQLiteRunStoreKeepsIdenticalEntries(t *testing.T) {
|
||
sq, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer sq.Close()
|
||
ws := filepath.Join(t.TempDir(), "runs", "run_dup")
|
||
st := newTestRunState(ws)
|
||
if err := sq.SaveRun(st); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
got, err := sq.LoadRun(ws)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !reflect.DeepEqual(got.Clarifications, []string{"同一句澄清", "同一句澄清"}) {
|
||
t.Fatalf("clarifications = %#v, want both identical entries", got.Clarifications)
|
||
}
|
||
if !reflect.DeepEqual(got.Steering, []string{"同一句澄清", "同一句澄清"}) {
|
||
t.Fatalf("steering = %#v, want both identical entries", got.Steering)
|
||
}
|
||
// 邮箱正文相同、ID 不同：两条都要在（不靠正文寻址）。
|
||
if len(got.Mailbox) != 2 || got.Mailbox[0].ID != "m1" || got.Mailbox[1].ID != "m2" {
|
||
t.Fatalf("mailbox = %+v", got.Mailbox)
|
||
}
|
||
}
|
||
|
||
// legacyRunStateDDL 是 F11 修复前的 squad_run_state（无协作态列），用于验证升级路径。
|
||
const legacyRunStateDDL = `
|
||
CREATE TABLE squad_run_state (
|
||
ws_dir TEXT PRIMARY KEY,
|
||
meta TEXT NOT NULL, workflow TEXT NOT NULL, squad TEXT NOT NULL, bb TEXT NOT NULL,
|
||
template TEXT, approved TEXT, repair_budget_remaining INTEGER NOT NULL DEFAULT 0,
|
||
repair_budget_granted INTEGER NOT NULL DEFAULT 0,
|
||
clarifications TEXT, steering TEXT, usage TEXT, updated_at TEXT NOT NULL
|
||
);
|
||
`
|
||
|
||
// TestSQLiteRunStoreUpgradesLegacySchema：旧库（缺 mailbox/handoffs/board 列）打开后
|
||
// 补列升级，既有一行原样读出，协作态为空默认不报错、不丢其他字段。
|
||
func TestSQLiteRunStoreUpgradesLegacySchema(t *testing.T) {
|
||
path := filepath.Join(t.TempDir(), "runs.db")
|
||
legacy, err := sql.Open("sqlite", path)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := legacy.Exec(legacyRunStateDDL); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ws := filepath.Join(t.TempDir(), "runs", "run_legacy")
|
||
meta, _ := json.Marshal(squad.RunMetadata{ID: "run_legacy", WorkDir: ws, Status: "executing"})
|
||
wf, _ := json.Marshal(squad.WorkflowPlan{WorkflowID: "wf_old", Status: "in_progress"})
|
||
sqCfg, _ := json.Marshal(squad.SquadConfig{SquadName: "legacy"})
|
||
bb, _ := json.Marshal(squad.BlackboardState{Fields: map[string]string{"k": "v"}})
|
||
approved, _ := json.Marshal(map[string]bool{"step:s1": true})
|
||
clar, _ := json.Marshal([]string{"旧澄清", "旧澄清"})
|
||
usage, _ := json.Marshal(map[string]*squad.Usage{"writer": {PromptTokens: 7, CompletionTokens: 3, Steps: 2}})
|
||
if _, err := legacy.Exec(`INSERT INTO squad_run_state
|
||
(ws_dir, meta, workflow, squad, bb, template, approved, repair_budget_remaining,
|
||
repair_budget_granted, clarifications, steering, usage, updated_at)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||
ws, meta, wf, sqCfg, bb, "旧模板", approved, 2, 1, clar, `["旧补充"]`, usage, "2026-01-01T00:00:00Z"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := legacy.Close(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
sq, err := OpenSQLiteStore(path)
|
||
if err != nil {
|
||
t.Fatalf("open legacy db: %v", err)
|
||
}
|
||
defer sq.Close()
|
||
|
||
got, err := sq.LoadRun(ws)
|
||
if err != nil {
|
||
t.Fatalf("load upgraded row: %v", err)
|
||
}
|
||
// 旧数据无损。
|
||
if got.Meta.ID != "run_legacy" || got.Plan.Workflow.WorkflowID != "wf_old" ||
|
||
got.Plan.Squad.SquadName != "legacy" || got.Template != "旧模板" {
|
||
t.Fatalf("legacy row corrupted: %+v", got)
|
||
}
|
||
if !got.Approved["step:s1"] || got.RepairBudgetRemaining != 2 || got.RepairBudgetGranted != 1 {
|
||
t.Fatalf("legacy scalars lost: %+v", got)
|
||
}
|
||
if !reflect.DeepEqual(got.Clarifications, []string{"旧澄清", "旧澄清"}) {
|
||
t.Fatalf("legacy clarifications lost: %#v", got.Clarifications)
|
||
}
|
||
if !reflect.DeepEqual(got.Steering, []string{"旧补充"}) {
|
||
t.Fatalf("legacy steering lost: %#v", got.Steering)
|
||
}
|
||
if got.Usage["writer"].PromptTokens != 7 || got.BB.Fields["k"] != "v" {
|
||
t.Fatalf("legacy usage/blackboard lost: %+v %+v", got.Usage, got.BB)
|
||
}
|
||
// 缺列读出为空默认。
|
||
if len(got.Mailbox) != 0 || len(got.Handoffs) != 0 || len(got.Board) != 0 {
|
||
t.Fatalf("new columns should default to empty: %+v %+v %+v", got.Mailbox, got.Handoffs, got.Board)
|
||
}
|
||
// 版本已记档，补列只做一次。
|
||
var version int
|
||
if err := sq.db.QueryRow(`SELECT version FROM squad_run_schema_version WHERE id = 1`).Scan(&version); err != nil {
|
||
t.Fatalf("schema version not recorded: %v", err)
|
||
}
|
||
if version != runSchemaVersion {
|
||
t.Fatalf("schema version = %d, want %d", version, runSchemaVersion)
|
||
}
|
||
// 升级后写入协作态并再次读回。
|
||
got.Mailbox = []squad.MailMessage{{ID: "m1", From: "a", To: "b", Text: "升级后写入"}}
|
||
got.Board = []squad.BoardTask{{ID: "s1", Status: "done"}}
|
||
got.Meta.Status = "completed"
|
||
if err := sq.SaveRun(got); err != nil {
|
||
t.Fatalf("save after upgrade: %v", err)
|
||
}
|
||
again, err := sq.LoadRun(ws)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(again.Mailbox) != 1 || again.Mailbox[0].Text != "升级后写入" ||
|
||
len(again.Board) != 1 || again.Board[0].ID != "s1" {
|
||
t.Fatalf("post-upgrade collaboration state lost: %+v %+v", again.Mailbox, again.Board)
|
||
}
|
||
if !reflect.DeepEqual(again.Clarifications, []string{"旧澄清", "旧澄清"}) {
|
||
t.Fatalf("upgrade write clobbered legacy clarifications: %#v", again.Clarifications)
|
||
}
|
||
}
|
||
|
||
// TestSQLiteStoreKeepsCollaborationStateAcrossRestart：进程重启（新连接）后，
|
||
// 邮箱消息、所有权移交记录与任务板状态都还在——恢复不必从头重建协作上下文。
|
||
func TestSQLiteStoreKeepsCollaborationStateAcrossRestart(t *testing.T) {
|
||
dbPath := filepath.Join(t.TempDir(), "runs.db")
|
||
runStore, err := OpenSQLiteStore(dbPath)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer runStore.Close()
|
||
|
||
dir := t.TempDir()
|
||
reg := squad.NewRoleRegistry()
|
||
for _, role := range []string{"researcher", "writer", "reviewer"} {
|
||
if err := reg.Register(squad.RoleSpec{Role: role, Backend: "script"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
backends := squad.NewBackendResolver()
|
||
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{
|
||
Success: true, Logs: "ok",
|
||
Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")},
|
||
}})
|
||
blueprints := squad.NewBlueprintRegistry()
|
||
if err := blueprints.Register(squad.Blueprint{
|
||
Key: "handoff", Title: "移交演示", Match: []string{"移交演示"},
|
||
Steps: []squad.StepSpec{
|
||
{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"notes.md", "fields.json"}},
|
||
{ID: "s2", Role: "writer", Title: "成文", Targets: []string{"output.md"}, Handoff: "reviewer"},
|
||
{ID: "s3", Role: "reviewer", Title: "终审", Targets: []string{"output.md"}},
|
||
},
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := blueprints.Register(squad.Blueprint{
|
||
Key: "board", Title: "任务板演示", Match: []string{"任务板演示"},
|
||
Steps: []squad.StepSpec{
|
||
{ID: "a", Role: "researcher", Title: "任务a", Targets: []string{"notes.md"}},
|
||
{ID: "b", Role: "writer", Title: "任务b", Needs: []string{"a"}, Targets: []string{"output.md"}},
|
||
},
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
eng, err := squad.NewEngine(squad.Options{
|
||
DataDir: dir, Registry: reg, Blueprints: blueprints,
|
||
Planner: squad.NewRulePlanner(),
|
||
Backends: backends, Sink: squad.NopSink{},
|
||
Store: runStore,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
ctx := context.Background()
|
||
handed, err := eng.Run(ctx, squad.Goal{Text: "做一个移交演示", HitlMode: squad.HitlAuto}, squad.RunOptions{})
|
||
if err != nil || handed.Status != squad.RunCompleted {
|
||
t.Fatalf("handoff run: %v %+v", err, handed)
|
||
}
|
||
if err := eng.PostMessage(ctx, handed.RunID, squad.MailMessage{From: "human", To: "writer", Text: "补充口径"}); err != nil {
|
||
t.Fatalf("post message: %v", err)
|
||
}
|
||
boarded, err := eng.Run(ctx, squad.Goal{Text: "做一个任务板演示", HitlMode: squad.HitlAuto},
|
||
squad.RunOptions{Topology: squad.TopologyTaskBoard})
|
||
if err != nil || boarded.Status != squad.RunCompleted {
|
||
t.Fatalf("board run: %v %+v", err, boarded)
|
||
}
|
||
|
||
// 重启 = 新连接（新进程会重新 migrate）。
|
||
restarted, err := OpenSQLiteStore(dbPath)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer restarted.Close()
|
||
|
||
gotHanded, err := restarted.LoadRun(handed.WorkDir)
|
||
if err != nil {
|
||
t.Fatalf("reload handoff run: %v", err)
|
||
}
|
||
if len(gotHanded.Handoffs) != 1 || gotHanded.Handoffs[0].FromRole != "writer" || gotHanded.Handoffs[0].ToRole != "reviewer" {
|
||
t.Fatalf("handoffs lost across restart: %+v", gotHanded.Handoffs)
|
||
}
|
||
if len(gotHanded.Mailbox) != 1 || gotHanded.Mailbox[0].Text != "补充口径" || gotHanded.Mailbox[0].To != "writer" {
|
||
t.Fatalf("mailbox lost across restart: %+v", gotHanded.Mailbox)
|
||
}
|
||
gotBoarded, err := restarted.LoadRun(boarded.WorkDir)
|
||
if err != nil {
|
||
t.Fatalf("reload board run: %v", err)
|
||
}
|
||
if len(gotBoarded.Board) != 2 {
|
||
t.Fatalf("board lost across restart: %+v", gotBoarded.Board)
|
||
}
|
||
for _, task := range gotBoarded.Board {
|
||
if task.Status != "done" {
|
||
t.Fatalf("board task %s = %s, want done", task.ID, task.Status)
|
||
}
|
||
}
|
||
}
|
||
|
||
// faultingStore 在真实 SQLiteStore 之上注入落盘故障（其余方法透传）。
|
||
type faultingStore struct {
|
||
inner squad.Store
|
||
failCheckpoint bool
|
||
failTerminal bool
|
||
}
|
||
|
||
func (f *faultingStore) SaveRun(st *squad.RunState) error {
|
||
if f.failTerminal && squad.RunStatus(st.Meta.Status).Terminal() {
|
||
return fmt.Errorf("injected sqlite run state save failure (%s)", st.Meta.Status)
|
||
}
|
||
return f.inner.SaveRun(st)
|
||
}
|
||
|
||
func (f *faultingStore) LoadRun(wsDir string) (*squad.RunState, error) { return f.inner.LoadRun(wsDir) }
|
||
|
||
func (f *faultingStore) SaveCheckpoint(wsDir string, cp squad.Checkpoint) error {
|
||
if f.failCheckpoint {
|
||
return fmt.Errorf("injected sqlite checkpoint save failure (superstep %d)", cp.Superstep)
|
||
}
|
||
return f.inner.SaveCheckpoint(wsDir, cp)
|
||
}
|
||
|
||
func (f *faultingStore) LoadLatestCheckpoint(wsDir string) (*squad.Checkpoint, error) {
|
||
return f.inner.LoadLatestCheckpoint(wsDir)
|
||
}
|
||
|
||
func (f *faultingStore) SaveGate(wsDir string, gate *squad.HitlGate) error {
|
||
return f.inner.SaveGate(wsDir, gate)
|
||
}
|
||
|
||
func (f *faultingStore) LoadGate(wsDir string) (*squad.HitlGate, error) {
|
||
return f.inner.LoadGate(wsDir)
|
||
}
|
||
|
||
func (f *faultingStore) SaveLeader(wsDir string, data []byte) error {
|
||
return f.inner.SaveLeader(wsDir, data)
|
||
}
|
||
|
||
func (f *faultingStore) LoadLeader(wsDir string) ([]byte, error) { return f.inner.LoadLeader(wsDir) }
|
||
|
||
// eventTally 收集事件类型计数（断言没有成功事件）。
|
||
type eventTally struct {
|
||
mu sync.Mutex
|
||
seen []string
|
||
sink squad.ProgressSink
|
||
}
|
||
|
||
func newEventTally() *eventTally {
|
||
t := &eventTally{}
|
||
t.sink = squad.ProgressFunc(func(ev squad.Event) {
|
||
t.mu.Lock()
|
||
defer t.mu.Unlock()
|
||
t.seen = append(t.seen, ev.Type)
|
||
})
|
||
return t
|
||
}
|
||
|
||
func (t *eventTally) has(typ string) bool {
|
||
t.mu.Lock()
|
||
defer t.mu.Unlock()
|
||
for _, got := range t.seen {
|
||
if got == typ {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func (t *eventTally) types() []string {
|
||
t.mu.Lock()
|
||
defer t.mu.Unlock()
|
||
return append([]string(nil), t.seen...)
|
||
}
|
||
|
||
// TestSQLiteStoreSurfacesSaveFailures：SQLite 后端上的落盘故障必须可见——
|
||
// 三档 checkpoint 模式下都不发 run_completed，运行要么报错、要么以失败收场。
|
||
func TestSQLiteStoreSurfacesSaveFailures(t *testing.T) {
|
||
for _, mode := range []string{"sync", "async", "on_exit"} {
|
||
for _, fault := range []string{"checkpoint", "terminal"} {
|
||
t.Run(mode+"/"+fault, func(t *testing.T) {
|
||
runStore, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer runStore.Close()
|
||
dir := t.TempDir()
|
||
reg := squad.NewRoleRegistry()
|
||
for _, role := range []string{"researcher", "writer", "reviewer"} {
|
||
if err := reg.Register(squad.RoleSpec{Role: role, Backend: "script"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
backends := squad.NewBackendResolver()
|
||
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{
|
||
Success: true, Logs: "ok",
|
||
Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")},
|
||
}})
|
||
blueprints := squad.NewBlueprintRegistry()
|
||
if err := blueprints.Register(squad.Blueprint{
|
||
Key: "demo", Title: "故障演示", Match: []string{"故障演示"},
|
||
Steps: []squad.StepSpec{
|
||
{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"notes.md", "fields.json"}},
|
||
{ID: "s2", Role: "writer", Title: "成文", Targets: []string{"output.md"}},
|
||
{ID: "s3", Role: "reviewer", Title: "终审", Targets: []string{"output.md"}},
|
||
},
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
store := &faultingStore{inner: runStore,
|
||
failCheckpoint: fault == "checkpoint", failTerminal: fault == "terminal"}
|
||
tally := newEventTally()
|
||
eng, err := squad.NewEngine(squad.Options{
|
||
DataDir: dir, Registry: reg, Blueprints: blueprints,
|
||
Planner: squad.NewRulePlanner(), Backends: backends,
|
||
Sink: tally.sink, Store: store,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
res, err := eng.Run(context.Background(),
|
||
squad.Goal{Text: "做一个故障演示", HitlMode: squad.HitlAuto},
|
||
squad.RunOptions{CheckpointMode: mode})
|
||
if tally.has(squad.EvRunCompleted) {
|
||
t.Fatalf("run_completed emitted despite %s save failure: %v", fault, tally.types())
|
||
}
|
||
if fault == "checkpoint" && mode == "sync" {
|
||
if err == nil {
|
||
t.Fatalf("sync boundary failure must surface as error: %+v", res)
|
||
}
|
||
if !strings.Contains(err.Error(), "persistence failure") {
|
||
t.Fatalf("error not marked as persistence failure: %v", err)
|
||
}
|
||
return
|
||
}
|
||
if err != nil {
|
||
t.Fatalf("%s/%s: unexpected infra error: %v", mode, fault, err)
|
||
}
|
||
if res.Status != squad.RunFailed {
|
||
t.Fatalf("%s/%s: want failed, got %s", mode, fault, res.Status)
|
||
}
|
||
if !strings.Contains(res.Error, "persistence failure") {
|
||
t.Fatalf("%s/%s: failure not explained: %q", mode, fault, res.Error)
|
||
}
|
||
if !tally.has(squad.EvRunFailed) {
|
||
t.Fatalf("%s/%s: no run_failed event: %v", mode, fault, tally.types())
|
||
}
|
||
})
|
||
}
|
||
}
|
||
}
|