From 90ac98ea65db3b3e0e30a6fd3af048070b8c3f7a Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 7 Oct 2026 10:29:47 +0000 Subject: [PATCH] Converge run persistence on SQLite through the Store port Why: Q1 WP5 - platform's SQLite becomes an implementation of the squad Store port so run state and orchestration state live in one database, and platform's in-process execution goes through the Executor port instead of bypassing it (X6). Evidence: all five modules go vet clean and go test ok, including engine-on-SQLite (state in DB, no run-level JSON files, second engine recovers from the same DB), SQLiteStore round trip, and the full squad compatibility net. Impact: runState/hitlGate exported so the port is implementable; Options.Store wires into engineDeps; leader persistence and checkpoints route through the port; PrepareWorkspace no longer writes run-level JSON (st.save owns it); platform RunInProcess takes squad.Executor (callers updated); serve mounts the engine on platform's RunStore - single orchestration.db for both. Refs: Documents proposal 2026-10-07-agent-stack-n1q1i1-plans.md Q1 WP5 --- adapters/sqlite/go.mod | 2 + adapters/sqlite/platform_test.go | 12 +- adapters/sqlite/runstore.go | 230 ++++++++++++++++++++++++++ adapters/sqlite/runstore_test.go | 188 +++++++++++++++++++++ adapters/sqlite/runtime.go | 53 ++++-- adapters/sqlite/stub_executor_test.go | 30 ++++ cmd/agent-squad/cli/cli.go | 20 ++- cmd/agent-squad/cli/platform_cmd.go | 19 ++- squad/engine.go | 55 +++--- squad/hitl.go | 4 +- squad/leader.go | 26 +-- squad/runstate.go | 40 ++--- squad/runstore.go | 65 +++++--- squad/termination.go | 2 +- squad/termination_test.go | 2 +- squad/workflow.go | 54 +++--- squad/workspace.go | 15 +- 17 files changed, 656 insertions(+), 161 deletions(-) create mode 100644 adapters/sqlite/runstore.go create mode 100644 adapters/sqlite/runstore_test.go create mode 100644 adapters/sqlite/stub_executor_test.go diff --git a/adapters/sqlite/go.mod b/adapters/sqlite/go.mod index bc4affd..07e5142 100644 --- a/adapters/sqlite/go.mod +++ b/adapters/sqlite/go.mod @@ -4,10 +4,12 @@ go 1.27 require ( agent-native v0.0.0 + agent-squad v0.0.0 modernc.org/sqlite v1.37.0 ) require ( + error-code-system v0.0.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/google/uuid v1.6.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect diff --git a/adapters/sqlite/platform_test.go b/adapters/sqlite/platform_test.go index 23be60e..eb54557 100644 --- a/adapters/sqlite/platform_test.go +++ b/adapters/sqlite/platform_test.go @@ -11,7 +11,7 @@ import ( "testing" "time" - "agent-native/runtime/llm" + agentsquad "agent-squad/squad" ) func TestOrchestrationFlow(t *testing.T) { @@ -116,12 +116,10 @@ func TestOrchestrationFlow(t *testing.T) { if err != nil { t.Fatal(err) } - text, err := s.RunInProcess(context.Background(), woke, work, func(ctx context.Context, model string, messages []llm.Message, td []llm.ToolDecl, opts llm.StreamOptions) <-chan llm.Event { - ch := make(chan llm.Event, 1) - ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd}} - close(ch) - return ch - }) + text, err := s.RunInProcess(context.Background(), woke, work, stubExecutor{result: &agentsquad.ExecResult{ + Success: true, Logs: "done", + Patches: map[string][]byte{"output.md": []byte("done")}, + }}) if err != nil || text != "done" { t.Fatalf("in-process %q %v", text, err) } diff --git a/adapters/sqlite/runstore.go b/adapters/sqlite/runstore.go new file mode 100644 index 0000000..8a7de25 --- /dev/null +++ b/adapters/sqlite/runstore.go @@ -0,0 +1,230 @@ +package platform + +import ( + "database/sql" + "encoding/json" + "fmt" + "os" + "time" + + "agent-squad/squad" +) + +// SQLiteStore 把 squad 的 Store 端口落到本库（Q1 WP5）：run/gate/checkpoint/leader +// 与平台其余表同一个 SQLite 文件——运行状态与平台状态不再各记一份。 +// 表名带 squad_run_ 前缀，与平台既有 23 张表并存不冲突。 +type SQLiteStore struct { + db *sql.DB +} + +// OpenRunStore 在本 Store 的底层连接上提供 squad.Store 端口（schema 随 migrate 建）。 +func (s *Store) RunStore() *SQLiteStore { + return &SQLiteStore{db: s.db} +} + +const runSchema = ` +CREATE TABLE IF NOT EXISTS 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 +); +CREATE TABLE IF NOT EXISTS squad_run_gate ( + ws_dir TEXT PRIMARY KEY, gate TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS squad_run_checkpoint ( + ws_dir TEXT NOT NULL, superstep INTEGER NOT NULL, at TEXT NOT NULL, + graph_hash TEXT NOT NULL, steps TEXT NOT NULL, + PRIMARY KEY (ws_dir, superstep) +); +CREATE TABLE IF NOT EXISTS squad_run_leader ( + ws_dir TEXT PRIMARY KEY, data TEXT NOT NULL +); +` + +func (s *SQLiteStore) migrate() error { + _, err := s.db.Exec(runSchema) + return err +} + +// OpenSQLiteStore 独立打开一个 squad.Store（不经 platform.Store 的场景，如测试）。 +func OpenSQLiteStore(path string) (*SQLiteStore, error) { + db, err := sql.Open("sqlite", path+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)") + if err != nil { + return nil, err + } + s := &SQLiteStore{db: db} + if err := s.migrate(); err != nil { + _ = db.Close() + return nil, err + } + return s, nil +} + +func j(v any) string { + b, err := json.Marshal(v) + if err != nil { + return "" + } + return string(b) +} + +func (s *SQLiteStore) SaveRun(st *squad.RunState) error { + if err := s.migrate(); err != nil { + return err + } + _, err := s.db.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 (?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(ws_dir) DO UPDATE SET + meta=excluded.meta, workflow=excluded.workflow, squad=excluded.squad, bb=excluded.bb, + template=excluded.template, approved=excluded.approved, + repair_budget_remaining=excluded.repair_budget_remaining, + repair_budget_granted=excluded.repair_budget_granted, + clarifications=excluded.clarifications, steering=excluded.steering, + usage=excluded.usage, updated_at=excluded.updated_at`, + st.Meta.WorkDir, j(st.Meta), j(st.Plan.Workflow), j(st.Plan.Squad), j(st.BB), + st.Template, j(st.Approved), st.RepairBudgetRemaining, st.RepairBudgetGranted, + j(st.Clarifications), j(st.Steering), j(st.Usage), now()) + return err +} + +func (s *SQLiteStore) LoadRun(wsDir string) (*squad.RunState, error) { + if err := s.migrate(); err != nil { + return nil, err + } + st := &squad.RunState{} + var meta, wf, sq, bb, template, approved, clar, steering, usage string + row := s.db.QueryRow(`SELECT meta, workflow, squad, bb, template, approved, + repair_budget_remaining, repair_budget_granted, clarifications, steering, usage + FROM squad_run_state WHERE ws_dir = ?`, wsDir) + var granted int + err := row.Scan(&meta, &wf, &sq, &bb, &template, &approved, + &st.RepairBudgetRemaining, &granted, &clar, &steering, &usage) + if err != nil { + if err == sql.ErrNoRows { + return nil, fmt.Errorf("load run %s: %w", wsDir, os.ErrNotExist) + } + return nil, err + } + st.RepairBudgetGranted = granted + for pair, dst := range map[string]any{ + meta: &st.Meta, wf: &st.Plan.Workflow, sq: &st.Plan.Squad, bb: &st.BB, + approved: &st.Approved, clar: &st.Clarifications, steering: &st.Steering, usage: &st.Usage, + } { + if err := json.Unmarshal([]byte(pair), dst); err != nil { + return nil, fmt.Errorf("load run %s: parse: %w", wsDir, err) + } + } + st.Template = template + if st.Approved == nil { + st.Approved = map[string]bool{} + } + return st, nil +} + +func (s *SQLiteStore) SaveGate(wsDir string, gate *squad.HitlGate) error { + if err := s.migrate(); err != nil { + return err + } + if gate == nil { + _, err := s.db.Exec(`DELETE FROM squad_run_gate WHERE ws_dir = ?`, wsDir) + return err + } + _, err := s.db.Exec(`INSERT INTO squad_run_gate (ws_dir, gate) VALUES (?,?) + ON CONFLICT(ws_dir) DO UPDATE SET gate=excluded.gate`, wsDir, j(gate)) + return err +} + +func (s *SQLiteStore) LoadGate(wsDir string) (*squad.HitlGate, error) { + if err := s.migrate(); err != nil { + return nil, err + } + var raw string + err := s.db.QueryRow(`SELECT gate FROM squad_run_gate WHERE ws_dir = ?`, wsDir).Scan(&raw) + if err != nil { + if err == sql.ErrNoRows { + return nil, nil + } + return nil, err + } + gate := &squad.HitlGate{} + if err := json.Unmarshal([]byte(raw), gate); err != nil { + return nil, fmt.Errorf("parse gate: %w", err) + } + return gate, nil +} + +func (s *SQLiteStore) SaveCheckpoint(wsDir string, cp squad.Checkpoint) error { + if err := s.migrate(); err != nil { + return err + } + _, err := s.db.Exec(`INSERT INTO squad_run_checkpoint (ws_dir, superstep, at, graph_hash, steps) + VALUES (?,?,?,?,?) + ON CONFLICT(ws_dir, superstep) DO UPDATE SET at=excluded.at, graph_hash=excluded.graph_hash, steps=excluded.steps`, + wsDir, cp.Superstep, cp.At.UTC().Format("2006-01-02T15:04:05Z07:00"), cp.GraphHash, j(cp.StepStates)) + return err +} + +func (s *SQLiteStore) LoadLatestCheckpoint(wsDir string) (*squad.Checkpoint, error) { + if err := s.migrate(); err != nil { + return nil, err + } + var cp squad.Checkpoint + var at, steps string + err := s.db.QueryRow(`SELECT superstep, at, graph_hash, steps FROM squad_run_checkpoint + WHERE ws_dir = ? ORDER BY superstep DESC LIMIT 1`, wsDir). + Scan(&cp.Superstep, &at, &cp.GraphHash, &steps) + if err != nil { + if err == sql.ErrNoRows { + return nil, nil + } + return nil, err + } + t, _ := parseTime(at) + cp.At = t + if err := json.Unmarshal([]byte(steps), &cp.StepStates); err != nil { + return nil, fmt.Errorf("parse checkpoint steps: %w", err) + } + return &cp, nil +} + +func (s *SQLiteStore) SaveLeader(wsDir string, data []byte) error { + if err := s.migrate(); err != nil { + return err + } + _, err := s.db.Exec(`INSERT INTO squad_run_leader (ws_dir, data) VALUES (?,?) + ON CONFLICT(ws_dir) DO UPDATE SET data=excluded.data`, wsDir, string(data)) + return err +} + +func (s *SQLiteStore) LoadLeader(wsDir string) ([]byte, error) { + if err := s.migrate(); err != nil { + return nil, err + } + var raw string + err := s.db.QueryRow(`SELECT data FROM squad_run_leader WHERE ws_dir = ?`, wsDir).Scan(&raw) + if err != nil { + if err == sql.ErrNoRows { + return nil, nil + } + return nil, err + } + return []byte(raw), nil +} + +// Close 只在独立打开（OpenSQLiteStore）时调用；经 platform.Store 打开时由平台统一关闭。 +func (s *SQLiteStore) Close() error { return s.db.Close() } + +var _ squad.Store = (*SQLiteStore)(nil) + +// parseTime 解析检查点时间（RFC3339Nano 与 RFC3339 两态）。 +func parseTime(s string) (time.Time, error) { + for _, layout := range []string{time.RFC3339Nano, time.RFC3339} { + if t, err := time.Parse(layout, s); err == nil { + return t, nil + } + } + return time.Time{}, fmt.Errorf("bad time %q", s) +} diff --git a/adapters/sqlite/runstore_test.go b/adapters/sqlite/runstore_test.go new file mode 100644 index 0000000..1bb1553 --- /dev/null +++ b/adapters/sqlite/runstore_test.go @@ -0,0 +1,188 @@ +package platform + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "agent-squad/squad" +) + +func newTestRunState(wsDir string) *squad.RunState { + return &squad.RunState{ + Meta: squad.RunMetadata{ + ID: "run_test", Status: "executing", WorkDir: wsDir, + CreatedAt: time.Now(), UpdatedAt: time.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"}}}, + Template: "tpl", + Approved: map[string]bool{"step:s1": true}, + RepairBudgetRemaining: 1, + RepairBudgetGranted: 0, + Clarifications: []string{"c1"}, + Steering: []string{"st1"}, + 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) != 1 || len(got.Steering) != 1 { + t.Fatalf("scalars = %+v", got) + } + 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: false, 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) + } +} + +const goodReportLite = "# 测试交付物\n\n## 概述\n\n这是由脚本后端生成的合格交付物，长度超过五十个字符以满足最小质检要求。\n" diff --git a/adapters/sqlite/runtime.go b/adapters/sqlite/runtime.go index 2706bd4..7bda807 100644 --- a/adapters/sqlite/runtime.go +++ b/adapters/sqlite/runtime.go @@ -9,8 +9,7 @@ import ( "path/filepath" "strings" - "agent-native/runtime" - "agent-native/runtime/llm" + "agent-squad/squad" ) // CLICommands are the external tools a daemon probes, in order. @@ -196,36 +195,54 @@ func (s *Store) RunCommand(ctx context.Context, task Task, command, workdir stri return s.FinishTask(task.ID, "completed", "") } -// RunInProcess executes the task with AgentSession. stream is required; the -// product supplies a provider, tests supply a script. -func (s *Store) RunInProcess(ctx context.Context, task Task, workdir string, stream llm.StreamFunc) (string, error) { +// RunInProcess executes the task through the squad.Executor port（Q1 WP5 修 X6： +// 不再绕过执行端口直驱 agent-native runtime）。调用方经 adapters/native 等 +// 适配器构造 Executor；nil 即 fail-closed。 +func (s *Store) RunInProcess(ctx context.Context, task Task, workdir string, exec squad.Executor) (string, error) { + if exec == nil { + return "", fmt.Errorf("RunInProcess requires an executor (no bypass)") + } if err := s.MarkRunning(task.ID); err != nil { return "", err } issue, _ := s.GetIssue(task.IssueID) - agent := runtime.NewAgentSession(runtime.Config{ - Model: "agent-native", - Stream: stream, - MaxTurns: 4, - OnEvent: func(ev runtime.Event) { + res, err := exec.Execute(ctx, squad.ExecRequest{ + WorkDir: workdir, + Prompt: issue.Title + "\n" + issue.Body, + Targets: []string{"output.md"}, + OnEvent: func(ev squad.ProcessEvent) { _ = s.AddTaskMessage(task.ID, ev.Type, ev.Text) }, }) - res, err := agent.Prompt(ctx, llm.Message{Role: llm.RoleUser, Content: issue.Title + "\n" + issue.Body}) if err != nil { _ = s.FinishTask(task.ID, "failed", "config") return "", err } - if res.Err != nil { - _ = s.FinishTask(task.ID, "failed", "model") - return res.Text, res.Err + if res == nil || !res.Success { + reason := "model" + if res != nil && res.ErrorReason != "" { + reason = res.ErrorReason + } + _ = s.FinishTask(task.ID, "failed", reason) + if res != nil { + return res.Logs, fmt.Errorf("executor failed: %s", res.ErrorReason) + } + return "", fmt.Errorf("executor failed") } - _ = s.AddUsage(task.ID, "agent-native", res.Usage.PromptTokens, res.Usage.CompletionTokens, res.Usage.CacheReadTokens, res.Usage.CacheWriteTokens) + if len(res.Patches) > 0 { + _, _ = squad.MaterializeTargets(workdir, []string{"output.md"}, res.Patches) + } + _ = s.AddUsage(task.ID, exec.Name(), res.PromptTokens, res.CompletionTokens, 0, 0) if err := s.FinishTask(task.ID, "completed", ""); err != nil { - return res.Text, err + return res.Logs, err } - _ = os.WriteFile(filepath.Join(workdir, "output.md"), []byte(res.Text), 0o644) - return res.Text, nil + if data, rerr := os.ReadFile(filepath.Join(workdir, "output.md")); rerr == nil { + return string(data), nil + } + if err := os.WriteFile(filepath.Join(workdir, "output.md"), []byte(res.Logs), 0o644); err != nil { + return res.Logs, err + } + return res.Logs, nil } // AgentCLI applies a task-token command: comment, status, sub-issue, or activity. diff --git a/adapters/sqlite/stub_executor_test.go b/adapters/sqlite/stub_executor_test.go new file mode 100644 index 0000000..0f7e279 --- /dev/null +++ b/adapters/sqlite/stub_executor_test.go @@ -0,0 +1,30 @@ +package platform + +import ( + "context" + + "agent-squad/squad" +) + +// stubExecutor 是测试用的 squad.Executor（X6 后调用方经端口注入）。 +type stubExecutor struct { + result *squad.ExecResult + err error +} + +func (s stubExecutor) Name() string { return "stub" } + +func (s stubExecutor) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) { + if s.result == nil { + return nil, s.err + } + // 每次调用返回拷贝：调用方会改写 Success 字段，共享指针会跨步骤串味。 + cp := *s.result + if s.result.Patches != nil { + cp.Patches = map[string][]byte{} + for k, v := range s.result.Patches { + cp.Patches[k] = v + } + } + return &cp, s.err +} diff --git a/cmd/agent-squad/cli/cli.go b/cmd/agent-squad/cli/cli.go index f120e7d..8c49dea 100644 --- a/cmd/agent-squad/cli/cli.go +++ b/cmd/agent-squad/cli/cli.go @@ -86,6 +86,7 @@ type app struct { files *store.FileEventSink st *store.FileStore log *logsys.Logger + plat *platform.Store } func dataDir() string { @@ -159,12 +160,20 @@ func assemble(dir, plannerName, modelOverride, intentRoot string) (*app, error) } } } + // 单库收编（Q1 WP5）：引擎运行状态与平台编排状态进同一个 SQLite。 + // 库打不开即失败（与 serve 现状一致），不允许静默退回文件态。 + plat, err := platform.Open(filepath.Join(dir, "orchestration.db")) + if err != nil { + return nil, fmt.Errorf("编排库打开失败: %w", err) + } engine, err := squad.NewEngine(squad.Options{ DataDir: dir, Registry: reg, Blueprints: bps, Planner: planner, Backends: backends, Sink: fileSink, IntentGate: gate, IntentCatalog: catalog, IntentRunner: runner, + Store: plat.RunStore(), }) if err != nil { + _ = plat.Close() return nil, err } logCfg, err := config.Load("") @@ -178,7 +187,7 @@ func assemble(dir, plannerName, modelOverride, intentRoot string) (*app, error) if err != nil { return nil, err } - return &app{engine: engine, sink: fanout, files: fileSink, st: st, log: logger}, nil + return &app{engine: engine, sink: fanout, files: fileSink, st: st, log: logger, plat: plat}, nil } func (a *app) close() { @@ -223,12 +232,9 @@ func cmdServe(args []string) int { srv := server.New(a.engine, a.sink, a.st) srv.SetToken(*token) srv.SetLogger(a.log) - if plat, err := platform.Open(filepath.Join(*data, "orchestration.db")); err != nil { - fmt.Fprintln(os.Stderr, "编排库打开失败:", err) - return 1 - } else { - srv.SetPlatform(plat) - defer plat.Close() + if a.plat != nil { + srv.SetPlatform(a.plat) + defer a.plat.Close() } stopSweep := srv.StartWakeupSweeper(15 * time.Second) defer stopSweep() diff --git a/cmd/agent-squad/cli/platform_cmd.go b/cmd/agent-squad/cli/platform_cmd.go index 2d83a45..18b53f0 100644 --- a/cmd/agent-squad/cli/platform_cmd.go +++ b/cmd/agent-squad/cli/platform_cmd.go @@ -9,8 +9,8 @@ import ( "path/filepath" "time" - "agent-native/runtime/llm" platform "agent-squad/adapters/sqlite" + "agent-squad/squad" ) func cmdPlatform(args []string) int { @@ -99,7 +99,7 @@ func cmdPlatform(args []string) int { return 1 } } - text, err := st.RunInProcess(context.Background(), task, fs.Arg(1), scriptedOK) + text, err := st.RunInProcess(context.Background(), task, fs.Arg(1), scriptedExecutor{}) return printPlatform(map[string]string{"text": text}, err) default: fmt.Fprintln(os.Stderr, "未知编排命令", args[0], args[1]) @@ -111,11 +111,16 @@ func taskByID(st *platform.Store, id string) (platform.Task, error) { return st.TaskByToken(id) } -func scriptedOK(ctx context.Context, model string, messages []llm.Message, td []llm.ToolDecl, opts llm.StreamOptions) <-chan llm.Event { - ch := make(chan llm.Event, 1) - ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{Role: llm.RoleAssistant, Content: "ok", StopReason: llm.StopEnd}} - close(ch) - return ch +// scriptedExecutor 是 daemon run-native 冒烟路径的脚本执行器（X6 后经端口注入）。 +type scriptedExecutor struct{} + +func (scriptedExecutor) Name() string { return "scripted" } + +func (scriptedExecutor) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) { + return &squad.ExecResult{ + Success: true, Logs: "ok", + Patches: map[string][]byte{"output.md": []byte("ok")}, + }, nil } func printPlatform(v any, err error) int { diff --git a/squad/engine.go b/squad/engine.go index 7fca98d..9e886e8 100644 --- a/squad/engine.go +++ b/squad/engine.go @@ -35,6 +35,9 @@ type Options struct { HitlPolicy HitlPolicy // RepairBudget 质检失败自动修复次数（默认 1）。 RepairBudget int + // Store 运行状态持久化端口（Q1 WP4/WP5）；nil 回落 FileStore。 + // 注入 SQLiteStore 后运行状态进 SQLite（装配期一次，跨 run 一致）。 + Store Store // IntentGate 可选意图闸门：Run 入口一次确定性裁决（nil = 不启用， // 行为与原实现一致）。见 intent.go；实现方为 squadintent 适配器。 IntentGate IntentGate @@ -73,7 +76,7 @@ func (d engineDeps) storeOr() Store { } // loadRun 经端口加载运行状态，并让它携带同一端口（后续 save 自然走同一实现）。 -func (e *SquadEngine) loadRun(wsDir string) (*runState, error) { +func (e *SquadEngine) loadRun(wsDir string) (*RunState, error) { st, err := e.deps.storeOr().LoadRun(wsDir) if err != nil { return nil, err @@ -153,7 +156,7 @@ func NewEngine(opts Options) (*SquadEngine, error) { Registry: opts.Registry, Backends: opts.Backends, Evaluator: opts.Evaluator, Checker: opts.Checker, Sink: opts.Sink, Workspaces: ws, StepTimeout: opts.StepTimeout, HitlPolicy: opts.HitlPolicy, - IntentRunner: opts.IntentRunner, + IntentRunner: opts.IntentRunner, Store: opts.Store, } return &SquadEngine{opts: opts, deps: deps, ws: ws, runLocks: map[string]*sync.Mutex{}}, nil } @@ -264,12 +267,12 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (* } } - st := &runState{Meta: meta, Approved: map[string]bool{}, RepairBudgetRemaining: e.opts.RepairBudget, store: e.deps.storeOr()} + st := &RunState{Meta: meta, Approved: map[string]bool{}, RepairBudgetRemaining: e.opts.RepairBudget, store: e.deps.storeOr()} return e.planAndExecute(ctx, st, runOpts) } // planAndExecute 规划 → 挂起或执行；st 可为已有运行状态（澄清续跑复用）。 -func (e *SquadEngine) planAndExecute(ctx context.Context, st *runState, runOpts RunOptions) (*RunResult, error) { +func (e *SquadEngine) planAndExecute(ctx context.Context, st *RunState, runOpts RunOptions) (*RunResult, error) { drv := &flowDriver{opts: e.deps} meta := &st.Meta @@ -324,7 +327,7 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *runState, runOpts Context: fmt.Sprintf("原因：%s\n问题：\n%s", planOut.Clarification.Reason, strings.Join(qs, "\n")), CreatedAt: time.Now(), Status: "pending", } - gate := &hitlGate{Request: *req, PrevStatus: RunPlanning} + gate := &HitlGate{Request: *req, PrevStatus: RunPlanning} // 澄清挂起也需要工作区承载 pending.json。 if err := e.materializeWorkspace(meta, st, &SquadPlan{Workflow: WorkflowPlan{WorkflowID: NewPlanID(), Status: "pending", Steps: nil}}); err != nil { return nil, err @@ -364,7 +367,7 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *runState, runOpts CreatedAt: time.Now(), Status: "pending", Reason: "plan", } - gate := &hitlGate{Request: *req, Plan: planOut.Plan, PrevStatus: RunPlanning} + gate := &HitlGate{Request: *req, Plan: planOut.Plan, PrevStatus: RunPlanning} if err := st.saveGate(gate); err != nil { return nil, err } @@ -398,7 +401,7 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *runState, runOpts } // initLeaderState 按运行选项初始化 leader 循环状态。 -func (e *SquadEngine) initLeaderState(st *runState) { +func (e *SquadEngine) initLeaderState(st *RunState) { leaderRole := st.Meta.Metadata["leader_role"] if leaderRole == "" { if _, ok := findAgent(st.Plan.Squad.Agents, "leader"); ok { @@ -427,12 +430,12 @@ func parseIntDefault(s string, def int) int { } // isLeaderRun 该运行是否处于 leader-delegation 模式。 -func (e *SquadEngine) isLeaderRun(st *runState) bool { +func (e *SquadEngine) isLeaderRun(st *RunState) bool { return st.Meta.Metadata["mode"] == "leader" } // resumeAfterHold 挂起解除后的统一续跑入口（按模式分发）。 -func (e *SquadEngine) resumeAfterHold(ctx context.Context, st *runState) (*RunResult, error) { +func (e *SquadEngine) resumeAfterHold(ctx context.Context, st *RunState) (*RunResult, error) { if e.isLeaderRun(st) { out, gate, err := e.runLeaderLoop(ctx, st) if err != nil { @@ -453,16 +456,16 @@ func (e *SquadEngine) resumeAfterHold(ctx context.Context, st *runState) (*RunRe } // materializeWorkspace 落盘工作区与子工作区（已存在则只更新计划并重建待执行子目录）。 -func (e *SquadEngine) materializeWorkspace(meta *RunMetadata, st *runState, plan *SquadPlan) error { +func (e *SquadEngine) materializeWorkspace(meta *RunMetadata, st *RunState, plan *SquadPlan) error { if st.Meta.WorkDir == "" { st.Meta = *meta - wsDir, err := e.ws.PrepareWorkspace(&st.Meta, plan) + _, err := e.ws.PrepareWorkspace(&st.Meta, plan) if err != nil { return err } st.Template = plan.Template - bb, _ := loadBlackboard(wsDir) - st.BB = *bb + // 新运行的黑板在内存初始化；持久化由后续 st.save() 经 Store 端口完成。 + st.BB = BlackboardState{Fields: map[string]string{}} } else { // 复用既有工作区：磁盘黑板为权威（澄清续跑/计划更新场景）。 if bb, err := loadBlackboard(st.Meta.WorkDir); err == nil { @@ -568,7 +571,7 @@ func choiceHoldKind(kind HumanRequestKind) bool { } } -func (e *SquadEngine) noteAndKeepHold(st *runState, gate *hitlGate, decision RespondDecision) (*RunResult, error) { +func (e *SquadEngine) noteAndKeepHold(st *RunState, gate *HitlGate, decision RespondDecision) (*RunResult, error) { if err := st.addSteering(decision.Answer); err != nil { return nil, err } @@ -580,7 +583,7 @@ func (e *SquadEngine) noteAndKeepHold(st *runState, gate *hitlGate, decision Res return heldResult(st, gate), nil } -func clearSameGate(st *runState, requestID string) error { +func clearSameGate(st *RunState, requestID string) error { current, err := loadGate(st.Meta.WorkDir) if err != nil || current == nil || current.Request.ID != requestID { return err @@ -588,7 +591,7 @@ func clearSameGate(st *runState, requestID string) error { return st.saveGate(nil) } -func (e *SquadEngine) applyDecision(ctx context.Context, st *runState, gate *hitlGate, decision RespondDecision) (*RunResult, error) { +func (e *SquadEngine) applyDecision(ctx context.Context, st *RunState, gate *HitlGate, decision RespondDecision) (*RunResult, error) { req := &gate.Request now := time.Now() drv := &flowDriver{opts: e.deps} @@ -756,7 +759,7 @@ func (e *SquadEngine) applyDecision(ctx context.Context, st *runState, gate *hit } // respondClarification 澄清回应：答案并入目标后重新规划。 -func (e *SquadEngine) respondClarification(ctx context.Context, st *runState, gate *hitlGate, decision RespondDecision) (*RunResult, error) { +func (e *SquadEngine) respondClarification(ctx context.Context, st *RunState, gate *HitlGate, decision RespondDecision) (*RunResult, error) { choice := strings.ToLower(decision.Choice) if choice == "abort" { return e.cancelRun(st, "人工中止（澄清阶段）"), nil @@ -791,7 +794,7 @@ func (e *SquadEngine) respondClarification(ctx context.Context, st *runState, ga } // bootstrapMinimalSquad 注册通用最小编制（organizer/specialist/writer/leader）。 -func (e *SquadEngine) bootstrapMinimalSquad(st *runState) { +func (e *SquadEngine) bootstrapMinimalSquad(st *RunState) { backend := st.Meta.Backend if backend == "" { backend = e.opts.DefaultBackend @@ -811,7 +814,7 @@ func (e *SquadEngine) bootstrapMinimalSquad(st *runState) { } // manualPass 人工放行质检：标记 ManualPass 并收口。 -func (e *SquadEngine) manualPass(st *runState, decision RespondDecision) (*RunResult, error) { +func (e *SquadEngine) manualPass(st *RunState, decision RespondDecision) (*RunResult, error) { report := e.deps.Checker.Check(CheckRequest{ RunID: st.Meta.ID, Workspace: st.Meta.WorkDir, Deliverable: "output.md", Plan: planPtr(st), Squad: &st.Plan.Squad, Template: st.Template, @@ -840,7 +843,7 @@ func (e *SquadEngine) manualPass(st *runState, decision RespondDecision) (*RunRe Summary: summarize(st, *report, status) + "（质检人工放行）"}, nil } -func (e *SquadEngine) cancelRun(st *runState, reason string) *RunResult { +func (e *SquadEngine) cancelRun(st *RunState, reason string) *RunResult { now := time.Now() st.Meta.Status = string(RunCancelled) st.Meta.CompletedAt = &now @@ -854,11 +857,11 @@ func (e *SquadEngine) cancelRun(st *runState, reason string) *RunResult { } // failStepResume 权限拒绝后的失败步骤处理（关键失败/非关键降级）。 -func (d *flowDriver) failStepResume(ctx context.Context, st *runState, idx int) (*RunResult, *hitlGate, error) { +func (d *flowDriver) failStepResume(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) { return d.failStep(ctx, st, idx) } -func (e *SquadEngine) respondStepApproval(ctx context.Context, st *runState, req *HumanRequest, decision RespondDecision) (*RunResult, error) { +func (e *SquadEngine) respondStepApproval(ctx context.Context, st *RunState, req *HumanRequest, decision RespondDecision) (*RunResult, error) { switch strings.ToLower(decision.Choice) { case "approve": st.Approved["step:"+req.StepID] = true @@ -876,7 +879,7 @@ func (e *SquadEngine) respondStepApproval(ctx context.Context, st *runState, req return e.resumeAfterHold(ctx, st) } -func (e *SquadEngine) respondInterrupted(ctx context.Context, st *runState, req *HumanRequest, decision RespondDecision) (*RunResult, error) { +func (e *SquadEngine) respondInterrupted(ctx context.Context, st *RunState, req *HumanRequest, decision RespondDecision) (*RunResult, error) { switch strings.ToLower(decision.Choice) { case "retry": if !markStep(st, req.StepID, "pending", "") { @@ -896,7 +899,7 @@ func (e *SquadEngine) respondInterrupted(ctx context.Context, st *runState, req return e.resumeAfterHold(ctx, st) } -func markStep(st *runState, id, status, errText string) bool { +func markStep(st *RunState, id, status, errText string) bool { for i := range st.Plan.Workflow.Steps { if st.Plan.Workflow.Steps[i].ID != id { continue @@ -910,7 +913,7 @@ func markStep(st *runState, id, status, errText string) bool { // recoverInterrupted 处理崩溃时停在 running 的步骤。 // interactive 挂起确认，不自动再执行；auto 允许一次重试；disabled 直接失败。 -func (e *SquadEngine) recoverInterrupted(ctx context.Context, st *runState) (*RunResult, error) { +func (e *SquadEngine) recoverInterrupted(ctx context.Context, st *RunState) (*RunResult, error) { for i := range st.Plan.Workflow.Steps { step := st.Plan.Workflow.Steps[i] if step.Status != "running" { @@ -1310,7 +1313,7 @@ func appendAnswer(d RespondDecision) string { return "（" + d.Answer + "）" } -func lastCompletedStepID(st *runState) string { +func lastCompletedStepID(st *RunState) string { id := "" for _, s := range st.Plan.Workflow.Steps { if s.Status == "completed" { diff --git a/squad/hitl.go b/squad/hitl.go index e94d602..8ededde 100644 --- a/squad/hitl.go +++ b/squad/hitl.go @@ -136,8 +136,8 @@ func expiredDecision(req HumanRequest, policy HitlPolicy, now time.Time) (Respon return RespondDecision{RequestID: req.ID, Choice: choice, Answer: "expired"}, true } -// hitlGate 引擎内部：一次挂起的完整上下文（落盘 pending.json）。 -type hitlGate struct { +// HitlGate 引擎内部：一次挂起的完整上下文（落盘 pending.json）。 +type HitlGate struct { Request HumanRequest `json:"request"` Plan *SquadPlan `json:"plan,omitempty"` // confirmation 时携带候选计划 Steps []StepSpec `json:"steps,omitempty"` // adjustment：建议追加的步骤 diff --git a/squad/leader.go b/squad/leader.go index f2372a8..276de06 100644 --- a/squad/leader.go +++ b/squad/leader.go @@ -13,7 +13,7 @@ import ( // leaderExecRequest 构造 leader 轮次的执行请求：与标准步骤同等装配 // （过程事件桥、ToolProfile/Skills/工具白名单、执行标识与预算）—— // 修平此前 leader 轮丢弃 OnEvent 与 ToolProfile 的缺口。 -func leaderExecRequest(drv *flowDriver, st *runState, leader RoleSpec, workDir, prompt, instructions string, targets []string, timeout time.Duration) ExecRequest { +func leaderExecRequest(drv *flowDriver, st *RunState, leader RoleSpec, workDir, prompt, instructions string, targets []string, timeout time.Duration) ExecRequest { stepID := fmt.Sprintf("leader_turn_%d", st.Leader.Turns) var budget *ExecBudget if timeout > 0 { @@ -80,22 +80,24 @@ type delegation struct { const leaderSubWorkspace = "tasks/00_leader" -func (st *runState) loadLeader() { +func (st *RunState) loadLeader() { if st.Leader != nil { return } st.Leader = &LeaderState{MaxTurns: DefaultMaxLeaderTurns} - data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "leader.json")) - if err == nil { + data, err := st.storeOr().LoadLeader(st.Meta.WorkDir) + if err == nil && data != nil { _ = json.Unmarshal(data, st.Leader) } } -func (st *runState) saveLeader() { +func (st *RunState) saveLeader() { if st.Leader == nil { return } - _ = writeJSON(filepath.Join(st.Meta.WorkDir, "leader.json"), st.Leader) + if data, err := json.Marshal(st.Leader); err == nil { + _ = st.storeOr().SaveLeader(st.Meta.WorkDir, data) + } } // leaderPrompt 组装 leader 的系统指令（协调契约 + 通用守则）。 @@ -132,7 +134,7 @@ func leaderPrompt(members []RoleSpec, catalogMd string) string { } // leaderTaskBrief 组装 leader 本轮的 TASK.md（目标 + 黑板 + 委派历史）。 -func leaderTaskBrief(st *runState) string { +func leaderTaskBrief(st *RunState) string { var b strings.Builder b.WriteString("# Leader 轮次：评估并委派\n\n") fmt.Fprintf(&b, "## 运行目标\n%s\n\n", st.Meta.Goal.Text) @@ -175,7 +177,7 @@ func leaderTaskBrief(st *runState) string { // runLeaderLoop leader-delegation 主循环：委派→执行→唤醒，直到 // close/failed/上限/挂起（escalation/gap/权限等由 HITL 处理后重入本函数）。 -func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResult, *hitlGate, error) { +func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResult, *HitlGate, error) { st.loadLeader() drv := &flowDriver{opts: e.deps} leader, ok := findAgent(st.Plan.Squad.Agents, st.Leader.LeaderRole) @@ -221,7 +223,7 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResu }, CreatedAt: time.Now(), Status: "pending", } - gate := &hitlGate{Request: *req, PrevStatus: RunExecuting} + gate := &HitlGate{Request: *req, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunWaitingHuman) _ = st.save() @@ -293,7 +295,7 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResu }, CreatedAt: time.Now(), Status: "pending", } - gate := &hitlGate{Request: *req, PrevStatus: RunExecuting} + gate := &HitlGate{Request: *req, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunWaitingHuman) _ = st.save() @@ -383,7 +385,7 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *runState) (*RunResu } // leaderFail 统一的 leader 模式失败收口。 -func (e *SquadEngine) leaderFail(st *runState, reason string) *RunResult { +func (e *SquadEngine) leaderFail(st *RunState, reason string) *RunResult { now := time.Now() st.Meta.Status = string(RunFailed) st.Meta.CompletedAt = &now @@ -397,7 +399,7 @@ func (e *SquadEngine) leaderFail(st *runState, reason string) *RunResult { } // snapshotDeliverablesToLeader 把成员交付文件复制进 leader 工作区 inbox/<步骤id>/。 -func snapshotDeliverablesToLeader(st *runState, step StepSpec) error { +func snapshotDeliverablesToLeader(st *RunState, step StepSpec) error { srcDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace)) dstDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(leaderSubWorkspace), "inbox", step.ID) for _, t := range step.Targets { diff --git a/squad/runstate.go b/squad/runstate.go index 20fd1b0..d66e06f 100644 --- a/squad/runstate.go +++ b/squad/runstate.go @@ -10,9 +10,9 @@ import ( "time" ) -// runState 引擎内部运行状态（磁盘即权威：workspace.json / workflow.json / +// RunState 引擎内部运行状态（磁盘即权威：workspace.json / workflow.json / // squad.json / blackboard.json / pending.json 全部落盘，进程重启可恢复）。 -type runState struct { +type RunState struct { Meta RunMetadata `json:"meta"` Plan SquadPlan `json:"plan"` BB BlackboardState `json:"bb"` @@ -39,14 +39,14 @@ type runState struct { } // storeOr 返回注入的 Store，缺省 FileStore。 -func (st *runState) storeOr() Store { +func (st *RunState) storeOr() Store { if st.store != nil { return st.store } return FileStore{} } -func (st *runState) policy() HitlPolicy { +func (st *RunState) policy() HitlPolicy { return mergeHitlPolicy(HitlPolicy{}, st.Meta.HitlPolicyOf()) } @@ -62,14 +62,14 @@ func (m RunMetadata) HitlPolicyOf() *HitlPolicy { return p } -func (st *runState) save() error { +func (st *RunState) save() error { st.Meta.UpdatedAt = time.Now() return st.storeOr().SaveRun(st) } // persist 是 FileStore 语义的低层落盘（全部写文件）。Store.SaveRun 调它； -// runState.save 走 Store 端口，替换实现（SQLiteStore）在装配期注入后自然接管。 -func (st *runState) persist() error { +// RunState.save 走 Store 端口，替换实现（SQLiteStore）在装配期注入后自然接管。 +func (st *RunState) persist() error { if err := writeJSON(filepath.Join(st.Meta.WorkDir, "workspace.json"), &st.Meta); err != nil { return err } @@ -103,13 +103,13 @@ type diskRunState struct { } // Workflow 返回计划的编排清单（简洁访问器）。 -func (st *runState) Workflow() WorkflowPlan { return st.Plan.Workflow } +func (st *RunState) Workflow() WorkflowPlan { return st.Plan.Workflow } // setWorkflow 写回编排清单。 -func (st *runState) setWorkflow(wf WorkflowPlan) { st.Plan.Workflow = wf } +func (st *RunState) setWorkflow(wf WorkflowPlan) { st.Plan.Workflow = wf } // saveGate 落盘 HITL 挂起上下文（pending.json）。 -func (st *runState) saveGate(gate *hitlGate) error { +func (st *RunState) saveGate(gate *HitlGate) error { if gate == nil { err := os.Remove(filepath.Join(st.Meta.WorkDir, "pending.json")) if os.IsNotExist(err) { @@ -121,8 +121,8 @@ func (st *runState) saveGate(gate *hitlGate) error { } // loadRunState 从工作区恢复运行状态。 -func loadRunState(wsDir string) (*runState, error) { - st := &runState{} +func loadRunState(wsDir string) (*RunState, error) { + st := &RunState{} if err := readJSON(filepath.Join(wsDir, "workspace.json"), &st.Meta); err != nil { return nil, fmt.Errorf("load workspace.json: %w", err) } @@ -165,7 +165,7 @@ func loadRunState(wsDir string) (*runState, error) { } // loadGate 恢复挂起上下文；无挂起返回 nil。 -func loadGate(wsDir string) (*hitlGate, error) { +func loadGate(wsDir string) (*HitlGate, error) { data, err := os.ReadFile(filepath.Join(wsDir, "pending.json")) if err != nil { if os.IsNotExist(err) { @@ -173,7 +173,7 @@ func loadGate(wsDir string) (*hitlGate, error) { } return nil, err } - gate := &hitlGate{} + gate := &HitlGate{} if err := json.Unmarshal(data, gate); err != nil { return nil, fmt.Errorf("parse pending.json: %w", err) } @@ -189,7 +189,7 @@ func readJSON(path string, v any) error { } // setStep 并行安全地回填步骤状态（与 save 的全量 marshal 互斥）。 -func (st *runState) setStep(idx int, s StepSpec) { +func (st *RunState) setStep(idx int, s StepSpec) { st.mu.Lock() st.Plan.Workflow.Steps[idx] = s st.mu.Unlock() @@ -197,7 +197,7 @@ func (st *runState) setStep(idx int, s StepSpec) { // injectSteering 把挂起的运行中人工补充指令追加到 prompt（下一步骤边界注入）， // 注入后清空并落盘。注入发生在步骤开始前，不会污染正在执行的步骤。 -func (st *runState) injectSteering(prompt string) string { +func (st *RunState) injectSteering(prompt string) string { st.mu.Lock() defer st.mu.Unlock() if len(st.Steering) == 0 { @@ -216,7 +216,7 @@ func (st *runState) injectSteering(prompt string) string { } // addSteering 追加一条尚未注入的人工补充，并立刻落盘。 -func (st *runState) addSteering(text string) error { +func (st *RunState) addSteering(text string) error { text = strings.TrimSpace(text) if text == "" { return nil @@ -228,7 +228,7 @@ func (st *runState) addSteering(text string) error { } // loadSteering 恢复未消费的补充指令。 -func (st *runState) loadSteering() { +func (st *RunState) loadSteering() { data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "steering.json")) if err != nil { return @@ -240,7 +240,7 @@ func (st *runState) loadSteering() { } // addUsage 累计角色用量。 -func (st *runState) addUsage(role string, promptTokens, completionTokens int) { +func (st *RunState) addUsage(role string, promptTokens, completionTokens int) { if promptTokens == 0 && completionTokens == 0 { return } @@ -260,7 +260,7 @@ func (st *runState) addUsage(role string, promptTokens, completionTokens int) { } // usageSnapshot 输出用量快照（值拷贝）。 -func (st *runState) usageSnapshot() map[string]Usage { +func (st *RunState) usageSnapshot() map[string]Usage { if len(st.Usage) == 0 { return nil } diff --git a/squad/runstore.go b/squad/runstore.go index 0b95522..acc23f2 100644 --- a/squad/runstore.go +++ b/squad/runstore.go @@ -13,35 +13,54 @@ import ( // SQLite 实现进 adapters/sqlite（WP5）。事件流仍归 store/eventsink（子包）， // 不在本端口内重复。 type Store interface { - SaveRun(st *runState) error - LoadRun(wsDir string) (*runState, error) - SaveGate(wsDir string, gate *hitlGate) error - LoadGate(wsDir string) (*hitlGate, error) + SaveRun(st *RunState) error + LoadRun(wsDir string) (*RunState, error) + SaveGate(wsDir string, gate *HitlGate) error + LoadGate(wsDir string) (*HitlGate, error) SaveCheckpoint(wsDir string, cp Checkpoint) error LoadLatestCheckpoint(wsDir string) (*Checkpoint, error) + // SaveLeader/LoadLeader 持久化 leader 循环状态（raw JSON bytes； + // 文件态即 leader.json）。LoadLeader 不存在时返回 (nil, nil)。 + SaveLeader(wsDir string, data []byte) error + LoadLeader(wsDir string) ([]byte, error) } // FileStore 把 runs// 目录布局呈现为 Store（现状文件态）。 type FileStore struct{} -func (FileStore) SaveRun(st *runState) error { +func (FileStore) SaveRun(st *RunState) error { return st.persist() } -func (FileStore) LoadRun(wsDir string) (*runState, error) { +func (FileStore) LoadRun(wsDir string) (*RunState, error) { return loadRunState(wsDir) } -func (FileStore) SaveGate(wsDir string, gate *hitlGate) error { - st := &runState{} +func (FileStore) SaveGate(wsDir string, gate *HitlGate) error { + st := &RunState{} st.Meta.WorkDir = wsDir return st.saveGate(gate) } -func (FileStore) LoadGate(wsDir string) (*hitlGate, error) { +func (FileStore) LoadGate(wsDir string) (*HitlGate, error) { return loadGate(wsDir) } +func (FileStore) SaveLeader(wsDir string, data []byte) error { + return os.WriteFile(filepath.Join(wsDir, "leader.json"), data, 0o644) +} + +func (FileStore) LoadLeader(wsDir string) ([]byte, error) { + data, err := os.ReadFile(filepath.Join(wsDir, "leader.json")) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + return data, nil +} + func (FileStore) SaveCheckpoint(wsDir string, cp Checkpoint) error { dir := filepath.Join(wsDir, "checkpoints") if err := os.MkdirAll(dir, 0o755); err != nil { @@ -80,9 +99,9 @@ type Checkpoint struct { wsDir string `json:"-"` } -// write 立即同步落盘（FileStore 语义）。 -func (cp Checkpoint) write() error { - return FileStore{}.SaveCheckpoint(cp.wsDir, cp) +// write 立即同步落盘（经当前 Store 端口——SQLite 后端时进数据库）。 +func (cp Checkpoint) write(store Store) error { + return store.SaveCheckpoint(cp.wsDir, cp) } // StepSnapshot 是检查点里的单步快照。 @@ -106,16 +125,20 @@ const ( // checkpointWriter 按档位执行边界快照；async 档单 goroutine 串行落盘。 type checkpointWriter struct { - mode CheckpointMode - ch chan Checkpoint - done chan error + mode CheckpointMode + store Store + ch chan Checkpoint + done chan error } -func newCheckpointWriter(mode CheckpointMode) *checkpointWriter { +func newCheckpointWriter(mode CheckpointMode, store Store) *checkpointWriter { if mode == "" { mode = CheckpointSync } - w := &checkpointWriter{mode: mode} + if store == nil { + store = FileStore{} + } + w := &checkpointWriter{mode: mode, store: store} if mode == CheckpointAsync { w.ch = make(chan Checkpoint, 16) w.done = make(chan error, 1) @@ -125,7 +148,7 @@ func newCheckpointWriter(mode CheckpointMode) *checkpointWriter { if first != nil { continue } - if err := cp.write(); err != nil { + if err := cp.write(w.store); err != nil { first = err } } @@ -150,7 +173,7 @@ func (w *checkpointWriter) boundary(wsDir string, cp Checkpoint) error { case CheckpointOnExit: return nil default: - return cp.write() + return cp.write(w.store) } } @@ -162,7 +185,7 @@ func (w *checkpointWriter) flush(wsDir string, cp Checkpoint) error { close(w.ch) return <-w.done case CheckpointOnExit: - return cp.write() + return cp.write(w.store) default: return nil } @@ -172,7 +195,7 @@ func (w *checkpointWriter) flush(wsDir string, cp Checkpoint) error { // 最新检查点存在且其图指纹与当前 Steps 重编译不一致时拒绝恢复—— // 崩溃期间图形态被改过，继续推进会把已完成的结果接到错误的图上。 // 检查点不存在（旧运行）或指纹一致则放行。 -func validateGraphContinuity(st *runState) error { +func validateGraphContinuity(st *RunState) error { latest, err := st.storeOr().LoadLatestCheckpoint(st.Meta.WorkDir) if err != nil || latest == nil { return err diff --git a/squad/termination.go b/squad/termination.go index 1074004..6455086 100644 --- a/squad/termination.go +++ b/squad/termination.go @@ -28,7 +28,7 @@ func DefaultTermination() Termination { } // terminationOf 取本运行的停止条件：引擎选项与运行状态覆盖默认值。 -func (e *SquadEngine) terminationOf(st *runState) Termination { +func (e *SquadEngine) terminationOf(st *RunState) Termination { t := DefaultTermination() if e.opts.RepairBudget > 0 { t.RepairBudget = e.opts.RepairBudget diff --git a/squad/termination_test.go b/squad/termination_test.go index 0c4221a..e5633ba 100644 --- a/squad/termination_test.go +++ b/squad/termination_test.go @@ -58,7 +58,7 @@ func TestRepairGrantCap(t *testing.T) { }, CreatedAt: time.Now(), Status: "pending", } - if err := st.saveGate(&hitlGate{Request: *req, PrevStatus: RunExecuting}); err != nil { + if err := st.saveGate(&HitlGate{Request: *req, PrevStatus: RunExecuting}); err != nil { t.Fatal(err) } if err := st.save(); err != nil { diff --git a/squad/workflow.go b/squad/workflow.go index 3d37bc2..ebc53fc 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -13,7 +13,7 @@ import ( "error-code-system/errcode" ) -// flowDriver 工作流驱动器：推进 runState 直到终态或 HITL 挂起边界。 +// flowDriver 工作流驱动器：推进 RunState 直到终态或 HITL 挂起边界。 type flowDriver struct { opts engineDeps } @@ -22,14 +22,14 @@ type flowDriver struct { // 推进模型（Q1 WP1/WP2）：每轮从权威 Steps 现场编译调度图，按 superstep // 分层推进——层内无写集冲突的步骤并行（runParallel 复用），有冲突的按声明序 // 串行（含动态调整）；动态调整改 Steps 后下一轮自然生效。 -func (d *flowDriver) advance(ctx context.Context, st *runState) (res *RunResult, gate *hitlGate, err error) { +func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult, gate *HitlGate, err error) { wf := st.Workflow() wf.Status = "in_progress" st.setWorkflow(wf) d.emit(st, Event{Type: EvPhase, Text: "开始执行工作流"}) superNo := 0 - cpw := newCheckpointWriter(CheckpointMode(st.Meta.Metadata["checkpoint_mode"])) + cpw := newCheckpointWriter(CheckpointMode(st.Meta.Metadata["checkpoint_mode"]), st.storeOr()) // 推进退出前 flush：async 落完、on_exit 写最终快照（Q1 WP4 三档的出口语义）。 var lastGraph *graph defer func() { @@ -150,7 +150,7 @@ func snapshotOf(superNo int, g *graph, steps []StepSpec) Checkpoint { // gateBeforeRun 在步骤真正执行前检查权限升级和显式审批。 // 返回非空结果或挂起时，调用方不得执行该步骤，也不得执行它所在的并行组。 -func (d *flowDriver) gateBeforeRun(ctx context.Context, st *runState, i int) (*RunResult, *hitlGate, error) { +func (d *flowDriver) gateBeforeRun(ctx context.Context, st *RunState, i int) (*RunResult, *HitlGate, error) { step := st.Plan.Workflow.Steps[i] role, _ := findAgent(st.Plan.Squad.Agents, step.Role) if role.BashPermission == BashPermissionAsk && !st.Approved["role:"+role.Role] { @@ -191,13 +191,13 @@ func (d *flowDriver) gateBeforeRun(ctx context.Context, st *runState, i int) (*R } // runParallel 并发执行一个并行组；关键步骤失败即终止整个运行。 -func (d *flowDriver) runParallel(ctx context.Context, st *runState, idxs []int) (*RunResult, *hitlGate, error) { +func (d *flowDriver) runParallel(ctx context.Context, st *RunState, idxs []int) (*RunResult, *HitlGate, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() var wg sync.WaitGroup var mu sync.Mutex failedCritical := -1 - var gate *hitlGate + var gate *HitlGate for _, idx := range idxs { wg.Add(1) go func(idx int) { @@ -228,7 +228,7 @@ func (d *flowDriver) runParallel(ctx context.Context, st *runState, idxs []int) } // runOneStep 执行单个步骤（含瞬态重试与交付物校验）。 -func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*RunResult, *hitlGate, error) { +func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) { if err := ctx.Err(); err != nil { return &RunResult{RunID: st.Meta.ID, Status: RunCancelled, Error: err.Error()}, nil, err } @@ -427,7 +427,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *runState, idx int) (*Ru } // failStep 按关键性处理失败步骤：关键 → 运行失败；非关键 → 标记后继续（返回 nil 让循环推进）。 -func (d *flowDriver) failStep(ctx context.Context, st *runState, idx int) (*RunResult, *hitlGate, error) { +func (d *flowDriver) failStep(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) { st.mu.Lock() step := st.Plan.Workflow.Steps[idx] critical := step.IsCritical() @@ -451,7 +451,7 @@ func (d *flowDriver) failStep(ctx context.Context, st *runState, idx int) (*RunR } // checkBlockingGaps 检查未解决阻塞缺口；interactive 挂起，auto 记录兜底假设并继续。 -func (d *flowDriver) checkBlockingGaps(st *runState) *hitlGate { +func (d *flowDriver) checkBlockingGaps(st *RunState) *HitlGate { gaps := st.BB.UnresolvedBlockingGaps() if len(gaps) == 0 { return nil @@ -493,7 +493,7 @@ func (d *flowDriver) checkBlockingGaps(st *runState) *hitlGate { Context: strings.Join(lines, "\n"), CreatedAt: time.Now(), Status: "pending", } - gate := &hitlGate{Request: *req, PrevStatus: RunExecuting} + gate := &HitlGate{Request: *req, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunWaitingHuman) _ = st.save() @@ -503,7 +503,7 @@ func (d *flowDriver) checkBlockingGaps(st *runState) *hitlGate { } // applyDynamicAdjustment 评估步骤产出并应用/审批动态调整。 -func (d *flowDriver) applyDynamicAdjustment(ctx context.Context, st *runState, idx int) (*RunResult, *hitlGate, error) { +func (d *flowDriver) applyDynamicAdjustment(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) { step := st.Plan.Workflow.Steps[idx] subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace)) output := readFileOr(filepath.Join(subDir, firstTarget(step.Targets)), "") @@ -543,7 +543,7 @@ func (d *flowDriver) applyDynamicAdjustment(ctx context.Context, st *runState, i evalRes.Reason, strings.Join(addLines, "\n"), strings.Join(skipLines, "\n")), CreatedAt: time.Now(), Status: "pending", } - gate := &hitlGate{Request: *req, Steps: evalRes.DynamicStepsToAdd, SkipIDs: evalRes.StepsToSkip, PrevStatus: RunExecuting} + gate := &HitlGate{Request: *req, Steps: evalRes.DynamicStepsToAdd, SkipIDs: evalRes.StepsToSkip, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunWaitingHuman) _ = st.save() @@ -556,7 +556,7 @@ func (d *flowDriver) applyDynamicAdjustment(ctx context.Context, st *runState, i } // applyAdjustment 应用动态调整：新步骤插到最后一步之前，跳过步骤标记 skipped。 -func applyAdjustment(st *runState, add []StepSpec, skipIDs []string) { +func applyAdjustment(st *RunState, add []StepSpec, skipIDs []string) { steps := st.Plan.Workflow.Steps for _, id := range skipIDs { for i := range steps { @@ -583,7 +583,7 @@ func applyAdjustment(st *runState, add []StepSpec, skipIDs []string) { } // finalize 收口：汇总交付物 → 质检 → 自动修复 → （必要时）人工裁决 → 清单。 -func (d *flowDriver) finalize(ctx context.Context, st *runState) (*RunResult, *hitlGate, error) { +func (d *flowDriver) finalize(ctx context.Context, st *RunState) (*RunResult, *HitlGate, error) { wf := st.Workflow() degraded := false for _, s := range wf.Steps { @@ -665,7 +665,7 @@ func (d *flowDriver) finalize(ctx context.Context, st *runState) (*RunResult, *h Context: string(marshalJSON(report)), CreatedAt: time.Now(), Status: "pending", } - gate := &hitlGate{Request: *req, PrevStatus: RunExecuting} + gate := &HitlGate{Request: *req, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunQualityReview) _ = st.save() @@ -699,7 +699,7 @@ func (d *flowDriver) finalize(ctx context.Context, st *runState) (*RunResult, *h } // rerunForRepair 重置交付物源步骤并重新执行到收口。 -func (d *flowDriver) rerunForRepair(ctx context.Context, st *runState, stepID string) (*RunResult, *hitlGate, error) { +func (d *flowDriver) rerunForRepair(ctx context.Context, st *RunState, stepID string) (*RunResult, *HitlGate, error) { for i := range st.Plan.Workflow.Steps { if st.Plan.Workflow.Steps[i].ID == stepID { st.Plan.Workflow.Steps[i].Status = "pending" @@ -717,7 +717,7 @@ func (d *flowDriver) rerunForRepair(ctx context.Context, st *runState, stepID st } // heldResult 由挂起上下文构造对外结果（挂起路径统一返回非 nil 结果）。 -func heldResult(st *runState, gate *hitlGate) *RunResult { +func heldResult(st *RunState, gate *HitlGate) *RunResult { status := RunWaitingHuman switch gate.Request.Kind { case HitlClarification: @@ -747,7 +747,7 @@ func waitConditionFrom(gaps []GapItem) *WaitCondition { } // hold 生成挂起请求并返回结果。 -func (d *flowDriver) hold(st *runState, kind HumanRequestKind, stepID, role, question string, options []HumanOption, context, reason string, extra *hitlGate) (*RunResult, *hitlGate, error) { +func (d *flowDriver) hold(st *RunState, kind HumanRequestKind, stepID, role, question string, options []HumanOption, context, reason string, extra *HitlGate) (*RunResult, *HitlGate, error) { req := &HumanRequest{ ID: newRequestID(), RunID: st.Meta.ID, StepID: stepID, Kind: kind, Severity: SeverityActionRequired, @@ -763,7 +763,7 @@ func (d *flowDriver) hold(st *runState, kind HumanRequestKind, stepID, role, que } gate := extra if gate == nil { - gate = &hitlGate{} + gate = &HitlGate{} } gate.Request = *req gate.PrevStatus = RunExecuting @@ -780,7 +780,7 @@ func (d *flowDriver) hold(st *runState, kind HumanRequestKind, stepID, role, que // policy 合并引擎全局策略与 Goal 覆盖（窄者胜）。 // policyRole 在引擎全局/Goal 链之上叠加角色级 HitlPolicy // （窄者胜；Goal 仍最高）。 -func (d *flowDriver) policyRole(st *runState, role RoleSpec) HitlPolicy { +func (d *flowDriver) policyRole(st *RunState, role RoleSpec) HitlPolicy { p := d.policy(st) if role.HitlPolicy != nil { p = mergeHitlPolicy(p, role.HitlPolicy) @@ -788,11 +788,11 @@ func (d *flowDriver) policyRole(st *runState, role RoleSpec) HitlPolicy { return p } -func (d *flowDriver) policy(st *runState) HitlPolicy { +func (d *flowDriver) policy(st *RunState) HitlPolicy { return mergeHitlPolicy(d.opts.HitlPolicy, st.Meta.HitlPolicyOf()) } -func (d *flowDriver) emit(st *runState, ev Event) { +func (d *flowDriver) emit(st *RunState, ev Event) { if d.opts.Sink == nil { return } @@ -803,11 +803,11 @@ func (d *flowDriver) emit(st *runState, ev Event) { d.opts.Sink.Emit(ev) } -func (d *flowDriver) emitHuman(st *runState, req *HumanRequest) { +func (d *flowDriver) emitHuman(st *RunState, req *HumanRequest) { d.emit(st, Event{Type: EvHumanInputRequired, Text: req.Question, Data: marshalJSON(req)}) } -func summarize(st *runState, report QualityReport, status RunStatus) string { +func summarize(st *RunState, report QualityReport, status RunStatus) string { var b strings.Builder fmt.Fprintf(&b, "运行 %s：%s。步骤 %d 个（完成 %d/跳过 %d/失败 %d）。质检得分 %.0f。", st.Meta.ID, status, len(st.Plan.Workflow.Steps), @@ -818,7 +818,7 @@ func summarize(st *runState, report QualityReport, status RunStatus) string { return b.String() } -func countStatus(st *runState, status string) int { +func countStatus(st *RunState, status string) int { n := 0 for _, s := range st.Plan.Workflow.Steps { if s.Status == status { @@ -828,7 +828,7 @@ func countStatus(st *runState, status string) int { return n } -func buildManifest(st *runState, deliverable string, report QualityReport) Manifest { +func buildManifest(st *RunState, deliverable string, report QualityReport) Manifest { files := map[string]string{} _ = filepath.Walk(st.Meta.WorkDir, func(path string, info os.FileInfo, err error) error { if err != nil || info.IsDir() { @@ -887,7 +887,7 @@ func appendLog(dir, line string) { _, _ = f.WriteString(line) } -func planPtr(st *runState) *WorkflowPlan { +func planPtr(st *RunState) *WorkflowPlan { wf := st.Workflow() return &wf } diff --git a/squad/workspace.go b/squad/workspace.go index 686ddf0..26fd246 100644 --- a/squad/workspace.go +++ b/squad/workspace.go @@ -34,18 +34,9 @@ func (e *WorkspaceEngine) PrepareWorkspace(meta *RunMetadata, plan *SquadPlan) ( } meta.WorkDir = wsDir meta.UpdatedAt = time.Now() - if err := writeJSON(filepath.Join(wsDir, "workspace.json"), meta); err != nil { - return "", err - } - if err := writeJSON(filepath.Join(wsDir, "squad.json"), &plan.Squad); err != nil { - return "", err - } - if err := writeJSON(filepath.Join(wsDir, "workflow.json"), &plan.Workflow); err != nil { - return "", err - } - if err := saveBlackboard(wsDir, &BlackboardState{Fields: map[string]string{}}); err != nil { - return "", err - } + // 注意：run 级状态 JSON 不在此处写——由 st.save() 经 Store 端口统一落盘 + // （Q1 WP5：SQLite 后端时这些状态进数据库而非文件）。这里只建目录与 + // 上下文/模板等运行无关状态的工作区资产。 ctxIdx, err := e.writeContextAssets(wsDir, &meta.Goal, plan) if err != nil { return "", err