From 44aa764b5625a8d8ac7486dd1e782cdb9930da55 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 3 Oct 2026 14:37:49 +0000 Subject: [PATCH] Recover interrupted runs and commit terminal state once Why: A hard kill left runs in pending, running, or waiting_hitl, a busy send stored a message before the 409, finishRun could mark a run completed without an assistant message, and a full SSE buffer dropped events without closing the stream. Evidence: GOWORK=off go vet ./... and GOWORK=off go test -count=1 ./... passed, including store, runmgr, bus, and server recovery tests. Impact: Migration v14 adds runs.instance_id and messages.request_id. Startup reconciliation and shutdown share one seal. Turn and workflow sends claim a slot before writing. Duplicate request ids return the existing message. SettleRun updates the run and appends the assistant message in one transaction. A full subscriber buffer closes the channel so SSE can replay from the cursor. Refs: Dev feature/pi-durable-borrow Co-authored-by: Cursor --- cmd/agent-system/main.go | 7 +- internal/app/app.go | 14 ++ internal/bus/bus.go | 39 ++++- internal/bus/bus_test.go | 37 +++++ internal/config/config.go | 10 +- internal/runmgr/recovery_test.go | 220 +++++++++++++++++++++++++ internal/runmgr/runmgr.go | 168 ++++++++++++++++--- internal/server/admission_test.go | 190 +++++++++++++++++++++ internal/server/server.go | 143 ++++++++++++---- internal/server/workflow_route_test.go | 42 +++++ internal/store/messages.go | 23 ++- internal/store/migrate.go | 21 +++ internal/store/recovery.go | 188 +++++++++++++++++++++ internal/store/recovery_test.go | 134 +++++++++++++++ internal/store/store.go | 28 +++- internal/store/store_test.go | 4 +- 16 files changed, 1188 insertions(+), 80 deletions(-) create mode 100644 internal/bus/bus_test.go create mode 100644 internal/runmgr/recovery_test.go create mode 100644 internal/server/admission_test.go create mode 100644 internal/store/recovery.go create mode 100644 internal/store/recovery_test.go diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 14d9b48..13d8aa7 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -41,13 +41,14 @@ func main() { dsn := fs.String("dsn", "", "PostgreSQL 档 DSN（空 = 嵌入式 SQLite 档）") artifacts := fs.String("artifacts", "", "工件对象目录（默认 /artifacts；PG 档多实例时指向共享卷）") auth := fs.String("auth", "auto", "API key 认证：auto（PG 档开、嵌入式关）| on | off") + instance := fs.String("instance", "", "本进程标识（空 = 主机名:监听地址；启动对账只收口这个实例的 run）") fs.Parse(os.Args[1:]) cliSet := map[string]bool{} fs.Visit(func(f *flag.Flag) { cliSet[f.Name] = true }) resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{ Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock, - StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, + StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance, }, cliSet) if err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) @@ -56,6 +57,7 @@ func main() { if err := app.Run(app.Options{ Addr: resolved.Addr, DataDir: resolved.DataDir, PluginsDir: resolved.PluginsDir, ForceMock: resolved.Mock, StorageDSN: resolved.StorageDSN, ArtifactsDir: resolved.ArtifactsDir, AuthMode: resolved.AuthMode, + InstanceID: resolved.Instance, }); err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) os.Exit(1) @@ -77,12 +79,13 @@ func runConfig(args []string) int { dsn := fs.String("dsn", "", "") artifacts := fs.String("artifacts", "", "") auth := fs.String("auth", "auto", "") + instance := fs.String("instance", "", "") fs.Parse(args[1:]) cliSet := map[string]bool{} fs.Visit(func(f *flag.Flag) { cliSet[f.Name] = true }) resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{ Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock, - StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, + StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance, }, cliSet) if err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) diff --git a/internal/app/app.go b/internal/app/app.go index 2e770cd..e689370 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -58,6 +58,7 @@ type Options struct { StorageDSN string // 空 = 嵌入式 SQLite 档；非空 = PostgreSQL 档（多用户/多实例） ArtifactsDir string // PG 档工件对象目录（空 = /artifacts） AuthMode string // auto（默认：PG 档开、嵌入式关）| on | off + InstanceID string // 空 = 主机名:监听地址；写入 runs.instance_id，启动对账只收口本实例 } // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。 @@ -187,6 +188,19 @@ func Run(opt Options) error { subRunner := &runtime.SubLLMRunner{Prompts: promptSvc} mockRunner := &runtime.MockRunner{} mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{}) + instanceID := opt.InstanceID + if instanceID == "" { + host, err := os.Hostname() + if err != nil || host == "" { + host = "local" + } + instanceID = host + ":" + opt.Addr + } + st.SetInstanceID(instanceID) + mgr.SetInstanceID(instanceID) + if err := mgr.ReconcileInterrupted(); err != nil { + return fmt.Errorf("reconcile interrupted runs: %w", err) + } mgr.SetSquadRunner(&runtime.SquadRunner{}) // 意图闸门：插件意图 overlay → agent-intent 注册表（未装载意图 = 行为回退）。 diff --git a/internal/bus/bus.go b/internal/bus/bus.go index e6e7011..d731103 100644 --- a/internal/bus/bus.go +++ b/internal/bus/bus.go @@ -1,5 +1,6 @@ // Package bus 是 streamhub：按会话分发已持久化的事件（SSE 通道）。 -// 事件权威在 store（seq 游标）；bus 只做在线转发，溢出丢弃由客户端从游标重放补偿。 +// 事件权威在 store（seq 游标）；bus 只做在线转发。 +// 订阅缓冲满时关闭该订阅，客户端用 Last-Event-ID 从游标重放，不再静默丢事件。 package bus import ( @@ -11,42 +12,64 @@ import ( type subscriber struct { ch chan core.StreamEvent filter string // conversationID；空 = 全部 + closed bool } type Bus struct { mu sync.Mutex subs map[int64]*subscriber next int64 + buf int } -func New() *Bus { return &Bus{subs: map[int64]*subscriber{}} } +func New() *Bus { return NewWithBuffer(512) } -// Subscribe 返回新订阅；close(unsubscribe) 注销。 +// NewWithBuffer 指定订阅缓冲长度。测试用小缓冲观察溢出关闭。 +func NewWithBuffer(n int) *Bus { + if n < 1 { + n = 1 + } + return &Bus{subs: map[int64]*subscriber{}, buf: n} +} + +// Subscribe 返回新订阅；close(unsubscribe) 注销。注销是幂等的。 func (b *Bus) Subscribe(conversationID string) (<-chan core.StreamEvent, func()) { b.mu.Lock() id := b.next b.next++ - ch := make(chan core.StreamEvent, 512) + ch := make(chan core.StreamEvent, b.buf) b.subs[id] = &subscriber{ch: ch, filter: conversationID} b.mu.Unlock() return ch, func() { b.mu.Lock() - delete(b.subs, id) + b.dropLocked(id) b.mu.Unlock() } } -// Publish 非阻塞广播；满则丢（客户端 Last-Event-ID 重放补齐）。 +// Publish 非阻塞广播。缓冲满的订阅被移除并关闭，避免静默丢事件。 func (b *Bus) Publish(e core.StreamEvent) { b.mu.Lock() defer b.mu.Unlock() - for _, s := range b.subs { - if s.filter != "" && s.filter != e.ConversationID { + for id, s := range b.subs { + if s.closed || (s.filter != "" && s.filter != e.ConversationID) { continue } select { case s.ch <- e: default: + b.dropLocked(id) } } } + +// dropLocked 在 b.mu 下关闭并移除订阅。重复调用无效果。 +func (b *Bus) dropLocked(id int64) { + s, ok := b.subs[id] + if !ok || s.closed { + return + } + s.closed = true + delete(b.subs, id) + close(s.ch) +} diff --git a/internal/bus/bus_test.go b/internal/bus/bus_test.go new file mode 100644 index 0000000..038e4a5 --- /dev/null +++ b/internal/bus/bus_test.go @@ -0,0 +1,37 @@ +package bus + +import ( + "testing" + + core "agent-system-core" +) + +func TestPublishClosesOverflowedSubscriber(t *testing.T) { + b := NewWithBuffer(1) + ch, unsub := b.Subscribe("c1") + defer unsub() + b.Publish(core.StreamEvent{ConversationID: "c1", Seq: 1, Type: "a"}) + b.Publish(core.StreamEvent{ConversationID: "c1", Seq: 2, Type: "b"}) + n := 0 + for { + e, ok := <-ch + if !ok { + break + } + n++ + if e.Seq != 1 { + t.Fatalf("buffered event seq %d", e.Seq) + } + } + if n != 1 { + t.Fatalf("delivered %d events before close", n) + } + unsub() +} + +func TestUnsubscribeIsIdempotent(t *testing.T) { + b := NewWithBuffer(1) + _, unsub := b.Subscribe("c1") + unsub() + unsub() +} diff --git a/internal/config/config.go b/internal/config/config.go index 951e9f1..d1f2160 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,7 @@ type Config struct { ArtifactsDir string `json:"artifacts"` AuthMode string `json:"auth"` Mock bool `json:"mock"` + Instance string `json:"instance"` } var defaults = Config{ @@ -40,6 +41,7 @@ var envKey = map[string]string{ "artifacts": "AGENT_SYSTEM_ARTIFACTS", "auth": "AGENT_SYSTEM_AUTH", "mock": "AGENT_SYSTEM_MOCK", + "instance": "AGENT_SYSTEM_INSTANCE", } // Source 配置来源层级（数值越大优先级越高）。 @@ -152,6 +154,8 @@ func loadFile(path string) (Config, map[string]bool, error) { dst = &c.AuthMode case "mock": dst = &c.Mock + case "instance": + dst = &c.Instance default: return Config{}, nil, fmt.Errorf("config file %s: unknown key %q", path, k) } @@ -177,6 +181,8 @@ func stringField(c Config, f string) string { return c.ArtifactsDir case "auth": return c.AuthMode + case "instance": + return c.Instance } return "" } @@ -195,6 +201,8 @@ func setString(c *Config, f, v string) { c.ArtifactsDir = v case "auth": c.AuthMode = v + case "instance": + c.Instance = v } } @@ -216,7 +224,7 @@ func (r Resolved) Explain(w io.Writer, file string) { fmt.Fprintln(w, "config file: (none)") } fmt.Fprintln(w, "field value source") - for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock"} { + for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock", "instance"} { v := stringField(r.Config, f) if f == "mock" { v = fmt.Sprintf("%t", r.Mock) diff --git a/internal/runmgr/recovery_test.go b/internal/runmgr/recovery_test.go new file mode 100644 index 0000000..d976933 --- /dev/null +++ b/internal/runmgr/recovery_test.go @@ -0,0 +1,220 @@ +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) + } + 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() + }) + st, err := store.OpenPostgres(dsn+"&search_path="+schema, 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) + } + 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)) + } +} + +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) + } +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 34e35f5..d943b65 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -7,9 +7,11 @@ import ( "context" "errors" "fmt" + "log" "path/filepath" "strings" "sync" + "sync/atomic" "time" "agent-system-backend/internal/bus" @@ -54,6 +56,9 @@ type Manager struct { convMu sync.Mutex convActive map[string]bool // conversationID → 有在途 run + instanceID string + shuttingDown atomic.Bool + // 后台子代理的完成通知只留在内存里。重启后未投递的通知丢失。 noticeMu sync.Mutex notices map[string][]bgNotice @@ -103,6 +108,9 @@ func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service, // Shutdown 停机清空事件缓冲（在 store.Close 之前调用）。 func (m *Manager) Shutdown() { m.events.stop() } +// SetInstanceID 记录本进程标识。启动对账在 PG 档只收口这个实例的 run。 +func (m *Manager) SetInstanceID(id string) { m.instanceID = id } + // SetEventNotifier 注入落盘后回调（PG 档 pgbus.Notifier；SQLite 档不注入）。 func (m *Manager) SetEventNotifier(fn func(conversationID string, evs []core.StreamEvent)) { m.events.onPersisted = fn @@ -799,6 +807,37 @@ func workflowScriptBodies(reg *workflow.Registry, def core.WorkflowDef) map[stri return out } +// ClaimConversation 抢占会话槽位。忙则返回 ErrConversationBusy，此时调用方不应写消息。 +func (m *Manager) ClaimConversation(conversationID string) error { + return m.claimConversation(conversationID) +} + +// ReleaseConversation 释放已抢到的槽位。启动失败时调用；重复释放无效果。 +func (m *Manager) ReleaseConversation(conversationID string) { + m.releaseConversation(conversationID) +} + +// StartClaimedConversationRun 在调用方已经抢到槽位之后启动回合。 +// CreateRun 失败时释放槽位；成功后由 execute 在终态释放。 +func (m *Manager) StartClaimedConversationRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) { + return m.startClaimedRun(conv, userMessage, attachments, inputs, nil) +} + +// StartClaimedWorkflowRun 在调用方已经抢到槽位之后启动工作流回合。 +func (m *Manager) StartClaimedWorkflowRun(conv core.Conversation, userMessage core.Message, def core.WorkflowDef, slots map[string]string, missing []workflow.SlotQuestion) (core.Run, error) { + copiedSlots := map[string]string{} + for k, v := range slots { + copiedSlots[k] = v + } + launch := &workflow.Launch{ + Def: workflow.Clone(def), + Slots: copiedSlots, + Missing: append([]workflow.SlotQuestion(nil), missing...), + Scripts: workflowScriptBodies(m.workflows, def), + } + return m.startClaimedRun(conv, userMessage, nil, nil, launch) +} + func (m *Manager) claimConversation(conversationID string) error { m.convMu.Lock() defer m.convMu.Unlock() @@ -882,38 +921,48 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, usage.CompletionTokens = rc.h.completionTokens usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens m.mu.Unlock() + if m.shuttingDown.Load() && status == core.RunCancelled { + status = core.RunInterrupted + errPart = &core.ErrorPart{Code: "run.interrupted", Message: "进程停机，运行已中断"} + } now := time.Now().UTC() rc.run.EndedAt = &now rc.run.Status = status rc.run.Phase = core.PhaseDone rc.run.Usage = &usage rc.run.Error = errPart - m.st.UpdateRun(&rc.run) - if m.OnUsage != nil && usage.TotalTokens > 0 { - m.OnUsage(rc.run.Owner, usage.TotalTokens) - } - // 终态权威消息树：落库助手消息（含过程部件标记）。 + // 终态权威消息树与 run 状态同一次提交。 m.mu.Lock() parts := append([]core.Part{}, rc.h.assistant...) m.mu.Unlock() if errPart != nil { parts = append(parts, core.Part{Type: core.PartError, Err: errPart}) } + var msg *core.Message if len(parts) > 0 { - msg := core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: parts} - if err := m.st.AppendMessage(&msg); err == nil { - // 消息树权威已落库；hitl 摘要部件的卡片状态由前端对账收敛。 - // submit_card 产物卡回填锚点：卡→消息互定位（D32 方向补全）。 - m.mu.Lock() - cardIDs := append([]core.ID{}, rc.h.submittedCards...) - m.mu.Unlock() - if m.cards != nil { - for _, id := range cardIDs { - if card, err := m.cards.Get(rc.run.Owner, string(id)); err == nil { - card.AnchorMessageID = msg.ID - _ = m.cards.Update(rc.run.Owner, card) - } + msg = &core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: parts} + } + updated, err := m.st.SettleRun(&rc.run, msg) + if err != nil { + log.Printf("[runmgr] settle run %s: %v", rc.run.ID, err) + m.failSettle(rc, err) + return + } + if m.OnUsage != nil && usage.TotalTokens > 0 { + m.OnUsage(rc.run.Owner, usage.TotalTokens) + } + if msg != nil { + // 消息树权威已落库；hitl 摘要部件的卡片状态由前端对账收敛。 + // submit_card 产物卡回填锚点：卡→消息互定位（D32 方向补全）。 + m.mu.Lock() + cardIDs := append([]core.ID{}, rc.h.submittedCards...) + m.mu.Unlock() + if m.cards != nil { + for _, id := range cardIDs { + if card, err := m.cards.Get(rc.run.Owner, string(id)); err == nil { + card.AnchorMessageID = msg.ID + _ = m.cards.Update(rc.run.Owner, card) } } } @@ -937,6 +986,10 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, } } + // 停机对账已经发出 run.failed 时，这里只补消息，不再发第二次。 + if !updated { + return + } switch status { case core.RunCompleted: usagePayload := core.RunCompletedPayload{Usage: usage} @@ -948,10 +1001,34 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, m.mu.Unlock() m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunCompleted, usagePayload) case core.RunFailed: - m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: errPart.Code, Message: errPart.Message}) + code, message := "run.failed", "" + if errPart != nil { + code, message = errPart.Code, errPart.Message + } + m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: code, Message: message}) + case core.RunInterrupted: + code, message := "run.interrupted", "进程停机，运行已中断" + if errPart != nil && errPart.Code != "" { + code, message = errPart.Code, errPart.Message + } + m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: code, Message: message}) } } +// failSettle 在终态提交失败后另起一次写入，把 run 标成 failed 并发出 run.failed。 +func (m *Manager) failSettle(rc *runCtx, cause error) { + now := time.Now().UTC() + rc.run.EndedAt = &now + rc.run.Status = core.RunFailed + rc.run.Phase = core.PhaseDone + rc.run.Error = &core.ErrorPart{Code: "run.settle_failed", Message: cause.Error()} + if err := m.st.UpdateRun(&rc.run); err != nil { + log.Printf("[runmgr] mark settle failure %s: %v", rc.run.ID, err) + return + } + m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.settle_failed", Message: cause.Error()}) +} + // setRunStatus 更新并广播 phase（状态不单独出事件，phase 事件携带）。 func (m *Manager) setRunStatus(r *core.Run, status core.RunStatus, phase core.RunPhase) { r.Status = status @@ -1328,8 +1405,10 @@ func (m *Manager) startNoticeRun(convID string, items []bgNotice) error { return nil } -// InterruptRunning 服务停机前把在途 run 标记为 interrupted（4.2 恢复策略）。 +// InterruptRunning 服务停机前把在途 run 收成 interrupted（4.2 恢复策略）。 +// 与启动对账共用 sealRunInterrupted，终态形状一致。 func (m *Manager) InterruptRunning() { + m.shuttingDown.Store(true) m.mu.Lock() ids := make([]string, 0, len(m.active)) for id := range m.active { @@ -1346,12 +1425,53 @@ func (m *Manager) InterruptRunning() { h.cancel() } for _, id := range ids { - r, err := m.st.GetRunByID(id) - if err != nil || r.Status == core.RunCompleted || r.Status == core.RunFailed || r.Status == core.RunCancelled { + m.sealRunInterrupted(id) + } +} + +// ReconcileInterrupted 在开始监听之前调用。 +// SQLite 档扫全部非终态 run。PG 档只扫本实例，以及升级前 instance_id 为空的行。 +func (m *Manager) ReconcileInterrupted() error { + all := m.st.Dialect() != "postgres" + runs, err := m.st.ListRecoverableRuns(m.instanceID, all) + if err != nil { + return err + } + for _, r := range runs { + m.sealRunInterrupted(r.ID) + } + return nil +} + +// sealRunInterrupted 条件更新成功才关 HITL 并发事件。并发调用只有一个赢家。 +func (m *Manager) sealRunInterrupted(runID string) { + ok, err := m.st.SealInterrupted(runID) + if err != nil || !ok { + return + } + r, err := m.st.GetRunByID(runID) + if err != nil { + return + } + m.closePendingHITL(r) + code, message := "run.interrupted", "进程重启或停机，运行已中断" + if r.Error != nil && r.Error.Code != "" { + code, message = r.Error.Code, r.Error.Message + } + m.emit(r.ConversationID, r.ID, core.EvRunFailed, core.RunFailedPayload{Code: code, Message: message}) +} + +func (m *Manager) closePendingHITL(r core.Run) { + cards, err := m.st.ActiveHITLCardsByRun(r.ID) + if err != nil { + return + } + for _, c := range cards { + res := core.HITLResolvedPayload{CardID: c.ID, Kind: "cancelled"} + if err := m.st.ResolveHITLCard(r.Owner, string(c.ID), res); err != nil { continue } - r.Status = core.RunInterrupted - m.st.UpdateRun(&r) + m.emit(r.ConversationID, r.ID, core.EvHITLResolved, res) } } diff --git a/internal/server/admission_test.go b/internal/server/admission_test.go new file mode 100644 index 0000000..e16e5b8 --- /dev/null +++ b/internal/server/admission_test.go @@ -0,0 +1,190 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "agent-system-backend/internal/bus" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/plugins" + "agent-system-backend/internal/profiles" + "agent-system-backend/internal/runmgr" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/internal/tools" + core "agent-system-core" +) + +func TestBusySendWritesNothing(t *testing.T) { + dir := t.TempDir() + st, err := store.OpenSQLite(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + reg := tools.NewRegistry() + reg.RegisterBuiltin() + setSvc := settings.NewService(st) + profSvc := profiles.New(st) + profSvc.EnsureDefault() + release := make(chan struct{}) + blocking := blockingRunner{block: make(chan struct{}, 1), release: release} + mgr := runmgr.New(st, bus.New(), reg, profSvc, setSvc, blocking, blocking, &runtime.MockRunner{}, &runtime.MockSubagent{}) + t.Cleanup(mgr.Shutdown) + ph := plugins.New(st, reg, setSvc, filepath.Join(dir, "plugins")) + srv := New(st, mgr, reg, profSvc, setSvc, ph) + srv.Bus = cmdbus.New(st, nil, nil) + srv.WireBuiltinCommands() + h := srv.Handler() + _, convID := setupConversation(t, h) + + code, _ := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "先占住"}) + if code != 200 { + t.Fatalf("first send: %d", code) + } + before, err := st.ListMessages(convID) + if err != nil { + t.Fatal(err) + } + code, _ = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "忙的时候这条不该落库"}) + if code != http.StatusConflict { + t.Fatalf("busy status %d", code) + } + after, err := st.ListMessages(convID) + if err != nil { + t.Fatal(err) + } + if len(after) != len(before) { + t.Fatalf("busy send wrote a message: %d -> %d", len(before), len(after)) + } + helpCode, helpOut := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "/help"}) + if helpCode != 200 || helpOut["command"] == nil { + t.Fatalf("slash command while busy: %d %v", helpCode, helpOut) + } + close(release) +} + +func TestRequestIDReturnsExistingMessage(t *testing.T) { + srv, st := newTestServer(t) + h := srv.Handler() + _, convID := setupConversation(t, h) + body := map[string]any{"text": "同一请求", "requestId": "req-1"} + code, first := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", body) + if code != 200 { + t.Fatalf("first %d %v", code, first) + } + code, second := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", body) + if code != 200 { + t.Fatalf("second %d %v", code, second) + } + m1 := first["message"].(map[string]any)["id"] + m2 := second["message"].(map[string]any)["id"] + if m1 != m2 { + t.Fatalf("message ids %v %v", m1, m2) + } + msgs, err := st.ListMessages(convID) + if err != nil { + t.Fatal(err) + } + user := 0 + for _, m := range msgs { + if m.Role == core.RoleUser { + user++ + } + } + if user != 1 { + t.Fatalf("user messages: %d", user) + } + runs, err := st.ListRunsByConversation(convID) + if err != nil { + t.Fatal(err) + } + if len(runs) != 1 { + t.Fatalf("runs: %d", len(runs)) + } +} + +func TestSSEReplaysBacklogBeyondOnePage(t *testing.T) { + srv, st := newTestServer(t) + h := srv.Handler() + _, convID := setupConversation(t, h) + const total = 2500 + batch := make([]core.StreamEvent, 0, 500) + for i := 0; i < total; i++ { + batch = append(batch, core.StreamEvent{ + ConversationID: convID, RunID: "r", TS: time.Now().UTC().Format(time.RFC3339Nano), + V: core.ProtocolVersion, Type: "note", + }) + if len(batch) == 500 { + if _, err := st.AppendEvents(convID, batch); err != nil { + t.Fatal(err) + } + batch = batch[:0] + } + } + srv.Subscribe = func(string) (<-chan core.StreamEvent, func()) { + ch := make(chan core.StreamEvent) + close(ch) + return ch, func() {} + } + req := httptest.NewRequest(http.MethodGet, "/api/events?conversationId="+convID, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + ids := sseIDs(rec.Body.String()) + if len(ids) != total { + t.Fatalf("replayed %d, want %d", len(ids), total) + } + for i, id := range ids { + if id != i+1 { + t.Fatalf("seq[%d]=%d", i, id) + } + } +} + +func TestSSEReconnectFillsMissingSeq(t *testing.T) { + srv, st := newTestServer(t) + h := srv.Handler() + _, convID := setupConversation(t, h) + batch := make([]core.StreamEvent, 5) + for i := range batch { + batch[i] = core.StreamEvent{ + ConversationID: convID, RunID: "r", TS: time.Now().UTC().Format(time.RFC3339Nano), + V: core.ProtocolVersion, Type: "note", + } + } + if _, err := st.AppendEvents(convID, batch); err != nil { + t.Fatal(err) + } + srv.Subscribe = func(string) (<-chan core.StreamEvent, func()) { + ch := make(chan core.StreamEvent) + close(ch) + return ch, func() {} + } + req := httptest.NewRequest(http.MethodGet, "/api/events?conversationId="+convID+"&since=2", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + ids := sseIDs(rec.Body.String()) + if len(ids) != 3 || ids[0] != 3 || ids[2] != 5 { + t.Fatalf("reconnect seqs %v", ids) + } +} + +func sseIDs(body string) []int { + var ids []int + for _, line := range strings.Split(body, "\n") { + if !strings.HasPrefix(line, "id: ") { + continue + } + n := 0 + for _, c := range line[4:] { + n = n*10 + int(c-'0') + } + ids = append(ids, n) + } + return ids +} diff --git a/internal/server/server.go b/internal/server/server.go index 927f17a..1164c01 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -310,8 +310,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { } lastSent := since - replay, err := s.st.EventsAfter(conversationID, since, 1000) - if err == nil { + for { + replay, err := s.st.EventsAfter(conversationID, lastSent, 1000) + if err != nil || len(replay) == 0 { + break + } for _, e := range replay { if e.Seq <= lastSent { continue @@ -322,8 +325,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { } lastSent = e.Seq } + if len(replay) < 1000 { + break + } } - // 心跳 + 在线转发。 + // 心跳 + 在线转发。通道关闭表示缓冲溢出，结束响应让 EventSource 带 Last-Event-ID 重连。 heartbeat := time.NewTicker(15 * time.Second) defer heartbeat.Stop() for { @@ -334,7 +340,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { case <-heartbeat.C: fmt.Fprint(w, ": ping\n\n") flusher.Flush() - case e := <-live: + case e, ok := <-live: + if !ok { + stop() + return + } if e.ConversationID != conversationID || e.Seq <= lastSent { continue } @@ -523,6 +533,7 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) { var req struct { Text string `json:"text"` AttachmentIDs []string `json:"attachmentIds"` + RequestID string `json:"requestId"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) @@ -533,6 +544,17 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) { http.Error(w, "text or attachments required", http.StatusBadRequest) return } + if req.RequestID != "" { + existing, err := s.st.MessageByRequest(convID, req.RequestID) + if err == nil { + s.writeExistingMessage(w, convID, existing) + return + } + if !errors.Is(err, store.ErrNotFound) { + writeStoreErr(w, err) + return + } + } // 技能命令：提示词库命令触发命中时保持原文、跳过命令路由，走正常 run—— // system 组装时命令触发照常命中（命令是显式意图，不受附件门槛约束）； @@ -570,45 +592,98 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) { parsedDocs = append(parsedDocs, doc) parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}}) } + + // 先路由再落库。回合和工作流要先抢槽位，忙则 409 且不写任何数据。 + // 意图命中的命令不占槽位，会话忙时也能用。 + route := "turn" + var recognized intentgate.MatchResult + if s.Gate != nil && !strings.HasPrefix(text, "/") { + recognized = s.recognize(r.Context(), text) + if recognized.Branch == "matched" && s.Bus != nil && s.Bus.HasCommand(recognized.ID) { + route = "command" + } else if recognized.Branch == "matched" && s.Workflows != nil { + if _, ok := s.Workflows.ByIntent(recognized.ID); ok { + route = "workflow" + } + } + } + claimed := false + if route == "turn" || route == "workflow" { + if err := s.mgr.ClaimConversation(convID); err != nil { + writeStoreErr(w, err) + return + } + claimed = true + } msg := core.Message{ConversationID: convID, Role: core.RoleUser, Parts: parts} - if err := s.st.AppendMessage(&msg); err != nil { + if err := s.st.AppendMessageWithRequest(&msg, req.RequestID); err != nil { + var dup *store.DuplicateRequestError + if errors.As(err, &dup) { + if claimed { + s.mgr.ReleaseConversation(convID) + } + s.writeExistingMessage(w, convID, dup.Existing) + return + } + if claimed { + s.mgr.ReleaseConversation(convID) + } writeStoreErr(w, err) return } - // 先命令总线，再绑定工作流，最后才是普通回合。 - if s.Gate != nil && !strings.HasPrefix(text, "/") { - recognized := s.recognize(r.Context(), text) - if recognized.Branch == "matched" && s.Bus != nil && s.Bus.HasCommand(recognized.ID) { - resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{ - ID: recognized.ID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner, - Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID, - }) - out := map[string]any{"message": msg, "receipt": resp.Receipt, "command": recognized.ID} - if resp.Card != nil { - out["card"] = resp.Card - } - if m, ok := resp.Payload.(map[string]any); ok { - for k, v := range m { - out[k] = v - } + if route == "command" { + resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{ + ID: recognized.ID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner, + Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID, + }) + out := map[string]any{"message": msg, "receipt": resp.Receipt, "command": recognized.ID} + if resp.Card != nil { + out["card"] = resp.Card + } + if m, ok := resp.Payload.(map[string]any); ok { + for k, v := range m { + out[k] = v } + } + writeJSON(w, http.StatusOK, out) + return + } + var run core.Run + var startErr error + if route == "workflow" { + def, _ := s.Workflows.ByIntent(recognized.ID) + missing := make([]workflow.SlotQuestion, len(recognized.Missing)) + for i, q := range recognized.Missing { + missing[i] = workflow.SlotQuestion{Key: q.Key, Question: q.Question} + } + run, startErr = s.mgr.StartClaimedWorkflowRun(conv, msg, def, recognized.Slots, missing) + } else { + run, startErr = s.mgr.StartClaimedConversationRun(conv, msg, parsedDocs, inputs) + } + if startErr != nil { + s.mgr.ReleaseConversation(convID) + writeStoreErr(w, startErr) + return + } + _ = s.st.SetMessageRunID(msg.ID, run.ID) + msg.RunID = run.ID + writeJSON(w, http.StatusOK, map[string]any{"message": msg, "run": run}) +} + +// writeExistingMessage 把同一 requestId 的已有消息和仍在进行的 run 原样返回。 +func (s *Server) writeExistingMessage(w http.ResponseWriter, convID string, msg core.Message) { + out := map[string]any{"message": msg} + if msg.RunID != "" { + if run, err := s.st.GetRunByID(msg.RunID); err == nil { + out["run"] = run writeJSON(w, http.StatusOK, out) return } - if recognized.Branch == "matched" && s.Workflows != nil { - if def, ok := s.Workflows.ByIntent(recognized.ID); ok { - missing := make([]workflow.SlotQuestion, len(recognized.Missing)) - for i, q := range recognized.Missing { - missing[i] = workflow.SlotQuestion{Key: q.Key, Question: q.Question} - } - run, err := s.mgr.StartWorkflowRun(conv, msg, def, recognized.Slots, missing) - writeJSONOr(w, map[string]any{"message": msg, "run": run}, err) - return - } - } } - run, err := s.mgr.StartConversationRun(conv, msg, parsedDocs, inputs) - writeJSONOr(w, map[string]any{"message": msg, "run": run}, err) + if run, err := s.st.LatestOpenRun(convID); err == nil { + out["run"] = run + } + writeJSON(w, http.StatusOK, out) } // handleCommand `/` 入口 → 命令总线（D27 四入口同源；slash 语义解析后统一分发+审计）。 diff --git a/internal/server/workflow_route_test.go b/internal/server/workflow_route_test.go index 2a7be82..dc6ec67 100644 --- a/internal/server/workflow_route_test.go +++ b/internal/server/workflow_route_test.go @@ -106,6 +106,48 @@ func TestIntentStartsWorkflowRun(t *testing.T) { } } +func TestWorkflowRouteClaimsBeforePersist(t *testing.T) { + srv, st := newTestServer(t) + if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": false}}); err != nil { + t.Fatal(err) + } + gate := intentgate.New() + if err := gate.Rebuild(context.Background(), []json.RawMessage{briefIntent()}); err != nil { + t.Fatal(err) + } + srv.Gate = gate + reg := workflow.NewRegistry() + reg.Set([]core.WorkflowDef{{ + ID: "demo.brief", Title: "简报", IntentID: "demo.brief", + Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}}, + }}, nil) + srv.Workflows = reg + _, convID := setupConversation(t, srv.Handler()) + conv, err := st.GetConversation(store.OwnerDefault, convID) + if err != nil { + t.Fatal(err) + } + if err := srv.mgr.ClaimConversation(convID); err != nil { + t.Fatal(err) + } + before, err := st.ListMessages(convID) + if err != nil { + t.Fatal(err) + } + code, out := doJSON(t, srv.Handler(), "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "写一份简报"}) + if code != 409 { + t.Fatalf("status %d body %v", code, out) + } + after, err := st.ListMessages(convID) + if err != nil { + t.Fatal(err) + } + if len(after) != len(before) { + t.Fatalf("workflow route wrote a message while busy") + } + srv.mgr.ReleaseConversation(conv.ID) +} + func briefIntent() json.RawMessage { return json.RawMessage(`{"id":"demo.brief","domain":"demo","version":1,"matcher":{"includes_any":["简报"]},"slots":[{"key":"topic","required":true,"pattern":"主题[:：]\\s*(\\S+)","question":"主题是什么？"}]}`) } diff --git a/internal/store/messages.go b/internal/store/messages.go index ceb0637..9349ed8 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -28,14 +28,20 @@ func isUniqueViolation(err error) bool { // AppendMessage 以单语句原子分配会话内序号（INSERT ... SELECT MAX+1）； // 并发同会话追加由 uq_messages_conv_seq 唯一约束兜底，冲突有界重试。 func (s *Store) AppendMessage(m *core.Message) error { + return s.AppendMessageWithRequest(m, "") +} + +// AppendMessageWithRequest 同 AppendMessage，并记录可选的 request_id。 +// 同一会话里非空 request_id 已存在时，把 *m 换成已有行并返回 DuplicateRequestError。 +func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) error { if m.ID == "" { m.ID = newID("msg") } m.CreatedAt = m.CreatedAt.UTC() - insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at) - SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,? FROM messages WHERE conversation_id=?` + insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) + SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` for attempt := 0; attempt < 8; attempt++ { - _, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), m.ConversationID) + _, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID) if err == nil { s.TouchConversation(m.ConversationID) return nil @@ -43,6 +49,13 @@ func (s *Store) AppendMessage(m *core.Message) error { if !isUniqueViolation(err) { return err } + if requestID != "" { + existing, lookErr := s.MessageByRequest(string(m.ConversationID), requestID) + if lookErr == nil { + *m = existing + return &DuplicateRequestError{Existing: existing} + } + } // 并发同会话追加冲突（PG 多连接竞态）：短抖动退避后重算序号。 time.Sleep(time.Duration(attempt) * time.Millisecond) } @@ -106,8 +119,8 @@ func (s *Store) CreateRun(r *core.Run) error { r.ID = newID("run") } r.CreatedAt = r.CreatedAt.UTC() - _, err := s.db.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)`), - r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO()) + _, err := s.db.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,instance_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`), + r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO(), s.instanceID) return err } diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 66c8cfe..850f9c3 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -369,6 +369,27 @@ func init() { `DROP TABLE IF EXISTS plugin_grants`, }, }}, + migration{version: 14, name: "run_instance_and_request_id", + // 启动对账按实例收口在途 run；发消息按 request_id 去重。 + // 空 request_id 不参与唯一约束，旧消息与命令路径仍可重复追加。 + up: map[string][]string{"sqlite": { + `ALTER TABLE runs ADD COLUMN instance_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE messages ADD COLUMN request_id TEXT NOT NULL DEFAULT ''`, + `CREATE UNIQUE INDEX IF NOT EXISTS uq_messages_conv_request ON messages(conversation_id, request_id) WHERE request_id <> ''`, + }, "postgres": { + `ALTER TABLE runs ADD COLUMN instance_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE messages ADD COLUMN request_id TEXT NOT NULL DEFAULT ''`, + `CREATE UNIQUE INDEX IF NOT EXISTS uq_messages_conv_request ON messages(conversation_id, request_id) WHERE request_id <> ''`, + }}, + down: map[string][]string{"sqlite": { + `DROP INDEX IF EXISTS uq_messages_conv_request`, + `ALTER TABLE messages DROP COLUMN request_id`, + `ALTER TABLE runs DROP COLUMN instance_id`, + }, "postgres": { + `DROP INDEX IF EXISTS uq_messages_conv_request`, + `ALTER TABLE messages DROP COLUMN request_id`, + `ALTER TABLE runs DROP COLUMN instance_id`, + }}}, ) } diff --git a/internal/store/recovery.go b/internal/store/recovery.go new file mode 100644 index 0000000..d892219 --- /dev/null +++ b/internal/store/recovery.go @@ -0,0 +1,188 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + "time" + + core "agent-system-core" +) + +// DuplicateRequestError 带上已落库的那条消息。 +type DuplicateRequestError struct { + Existing core.Message +} + +func (e *DuplicateRequestError) Error() string { return ErrDuplicateRequest.Error() } + +func (e *DuplicateRequestError) Unwrap() error { return ErrDuplicateRequest } + +// MessageByRequest 按会话和 request_id 查找已落库消息。 +func (s *Store) MessageByRequest(conversationID, requestID string) (core.Message, error) { + if requestID == "" { + return core.Message{}, ErrNotFound + } + var m core.Message + var parts, ca string + err := s.db.QueryRow(s.q(`SELECT id,conversation_id,run_id,role,parts,created_at FROM messages WHERE conversation_id=? AND request_id=?`), + conversationID, requestID).Scan(&m.ID, &m.ConversationID, &m.RunID, &m.Role, &parts, &ca) + if errors.Is(err, sql.ErrNoRows) { + return core.Message{}, ErrNotFound + } + if err != nil { + return core.Message{}, err + } + m.Role = core.Role(m.Role) + m.Parts, _ = unmarshalJSON[[]core.Part](parts) + m.CreatedAt, _ = parseTime(ca) + return m, nil +} + +// SetMessageRunID 把用户消息挂到随后启动的 run 上，供同一 request_id 重试时带回。 +func (s *Store) SetMessageRunID(messageID, runID string) error { + _, err := s.db.Exec(s.q(`UPDATE messages SET run_id=? WHERE id=?`), runID, messageID) + return err +} + +// LatestOpenRun 返回会话里最新的非终态、非子代理 run。 +// 幂等重试在用户消息还没写上 run_id 时，用它避免绑到更早的已完成 run。 +func (s *Store) LatestOpenRun(conversationID string) (core.Run, error) { + r, err := scanRun(func(dst ...any) error { + return s.db.QueryRow(s.q(`SELECT `+runColumns+` FROM runs WHERE conversation_id=? AND kind<>? AND status IN ('pending','running','waiting_hitl') ORDER BY created_at DESC LIMIT 1`), + conversationID, string(core.RunSubagent)).Scan(dst...) + }) + if errors.Is(err, sql.ErrNoRows) { + return r, ErrNotFound + } + return r, err +} + +// ListRecoverableRuns 列出仍停在非终态的 run。 +// all 为真时扫全部（SQLite 单进程）。否则只扫本实例，以及升级前 instance_id 为空的行。 +func (s *Store) ListRecoverableRuns(instanceID string, all bool) ([]core.Run, error) { + q := `SELECT ` + runColumns + ` FROM runs WHERE status IN ('pending','running','waiting_hitl')` + args := []any{} + if !all { + q += ` AND (instance_id=? OR instance_id='')` + args = append(args, instanceID) + } + q += ` ORDER BY created_at` + rows, err := s.db.Query(s.q(q), args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []core.Run{} + for rows.Next() { + r, err := scanRun(rows.Scan) + if err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// SealInterrupted 把仍在进行的 run 收成 interrupted。 +// 条件更新保证并发对账时只有一个赢家；已经是终态的行不受影响。 +func (s *Store) SealInterrupted(id string) (bool, error) { + errPart := marshalJSON(&core.ErrorPart{Code: "run.interrupted", Message: "进程重启或停机，运行已中断"}) + res, err := s.db.Exec(s.q(`UPDATE runs SET status=?, phase=?, error=?, ended_at=? WHERE id=? AND status IN ('pending','running','waiting_hitl')`), + string(core.RunInterrupted), string(core.PhaseDone), errPart, nowISO(), id) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + return n == 1, err +} + +// ActiveHITLCardsByRun 返回某个 run 上尚未关闭的 HITL 卡片。 +func (s *Store) ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error) { + rows, err := s.db.Query(s.q(`SELECT payload FROM hitl_cards WHERE run_id=? AND resolved=0 ORDER BY created_at`), runID) + if err != nil { + return nil, err + } + defer rows.Close() + out := []core.HITLCard{} + for rows.Next() { + var payload string + if err := rows.Scan(&payload); err != nil { + return nil, err + } + c, err := unmarshalJSON[core.HITLCard](payload) + if err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +// SettleRun 在一个事务里更新 run，并可选地追加助手消息。 +// updated 为假表示条件更新没有改到行（已被收成终态）；消息仍会写入。 +// 序号冲突时整段事务重试。其它失败回滚，run 保持原状态。 +func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err error) { + s.settleMu.Lock() + fault := s.settleFault + s.settleMu.Unlock() + if fault != nil { + return false, fault + } + if msg != nil && msg.ID == "" { + msg.ID = newID("msg") + } + var last error + for attempt := 0; attempt < 8; attempt++ { + updated, last = s.settleRunOnce(r, msg) + if last == nil { + return updated, nil + } + if !isUniqueViolation(last) { + return false, last + } + time.Sleep(time.Duration(attempt) * time.Millisecond) + } + if last == nil { + last = fmt.Errorf("store: settle retries exhausted (run %s)", r.ID) + } + return false, last +} + +func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) { + tx, err := s.db.Begin() + if err != nil { + return false, err + } + defer tx.Rollback() + if err := s.dialect.lockConversation(tx, string(r.ConversationID)); err != nil { + return false, err + } + var ended any + if r.EndedAt != nil { + ended = nowISO() + } + res, err := tx.Exec(s.q(`UPDATE runs SET status=?, phase=?, usage=?, error=?, ended_at=? WHERE id=? AND status IN ('pending','running','waiting_hitl')`), + string(r.Status), string(r.Phase), marshalJSON(r.Usage), marshalJSON(r.Error), ended, r.ID) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + if err != nil { + return false, err + } + if msg != nil { + insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) + SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` + if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), nowISO(), "", msg.ConversationID); err != nil { + return false, err + } + } + if err := tx.Commit(); err != nil { + return false, err + } + if msg != nil { + s.TouchConversation(string(msg.ConversationID)) + } + return n == 1, nil +} diff --git a/internal/store/recovery_test.go b/internal/store/recovery_test.go new file mode 100644 index 0000000..351c4b6 --- /dev/null +++ b/internal/store/recovery_test.go @@ -0,0 +1,134 @@ +package store + +import ( + "errors" + "testing" + "time" + + core "agent-system-core" +) + +func seedConversation(t *testing.T, st *Store) core.Conversation { + t.Helper() + p, err := st.CreateProject(OwnerDefault, "p") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation(OwnerDefault, p.ID, "c", "") + if err != nil { + t.Fatal(err) + } + return c +} + +func seedRun(t *testing.T, st *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 TestMessageRequestIDDedup(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + first := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}} + if err := st.AppendMessageWithRequest(&first, "req-1"); err != nil { + t.Fatal(err) + } + second := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi again"}}} + err := st.AppendMessageWithRequest(&second, "req-1") + var dup *DuplicateRequestError + if !errors.As(err, &dup) || dup.Existing.ID != first.ID { + t.Fatalf("duplicate request: %v %#v", err, dup) + } + blankA := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "a"}}} + blankB := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "b"}}} + if err := st.AppendMessageWithRequest(&blankA, ""); err != nil { + t.Fatal(err) + } + if err := st.AppendMessageWithRequest(&blankB, ""); err != nil { + t.Fatal(err) + } + msgs, err := st.ListMessages(conv.ID) + if err != nil { + t.Fatal(err) + } + if len(msgs) != 3 { + t.Fatalf("messages: %d", len(msgs)) + } +} + +func TestCreateRunRecordsInstance(t *testing.T) { + st := openTestStore(t) + st.SetInstanceID("host:8787") + conv := seedConversation(t, st) + run := seedRun(t, st, conv, core.RunPending) + var got string + if err := st.db.QueryRow(st.q(`SELECT instance_id FROM runs WHERE id=?`), run.ID).Scan(&got); err != nil { + t.Fatal(err) + } + if got != "host:8787" { + t.Fatalf("instance_id %q", got) + } +} + +func TestRecoverableRunsRespectInstance(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + st.SetInstanceID("inst-a") + own := seedRun(t, st, conv, core.RunPending) + st.SetInstanceID("inst-b") + other := seedRun(t, st, conv, core.RunRunning) + st.SetInstanceID("") + legacy := seedRun(t, st, conv, core.RunWaitingHITL) + done := seedRun(t, st, conv, core.RunCompleted) + + all := st.Dialect() != "postgres" + got, err := st.ListRecoverableRuns("inst-a", all) + if err != nil { + t.Fatal(err) + } + ids := map[string]bool{} + for _, r := range got { + ids[r.ID] = true + } + if !ids[own.ID] || !ids[legacy.ID] || ids[done.ID] { + t.Fatalf("recoverable ids: %v", ids) + } + if st.Dialect() == "postgres" && ids[other.ID] { + t.Fatal("postgres reconcile must leave another instance's run alone") + } + if st.Dialect() != "postgres" && !ids[other.ID] { + t.Fatal("sqlite reconcile scans every open run") + } +} + +func TestSettleRunRollsBackWhenMessageInsertFails(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + run := seedRun(t, st, conv, core.RunRunning) + existing := core.Message{ID: "msg_fixed", ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "user"}}} + if err := st.AppendMessage(&existing); err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + run.Status = core.RunCompleted + run.Phase = core.PhaseDone + run.EndedAt = &now + dup := core.Message{ID: "msg_fixed", ConversationID: conv.ID, RunID: run.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "assistant"}}} + if _, err := st.SettleRun(&run, &dup); err == nil { + t.Fatal("duplicate message id must fail the settle") + } + got, err := st.GetRunByID(run.ID) + if err != nil { + t.Fatal(err) + } + if got.Status != core.RunRunning { + t.Fatalf("rolled-back status: %s", got.Status) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index b805841..ae92c08 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -16,6 +16,7 @@ import ( "fmt" "os" "path/filepath" + "sync" "time" "agent-system-backend/internal/blobstore" @@ -27,14 +28,22 @@ import ( var ErrNotFound = errors.New("store: not found") var ErrConflict = errors.New("store: conflict") +// ErrDuplicateRequest 表示同一会话的 request_id 已经落库。 +// 调用方应返回已有消息，不要映射成 409：客户端把 409 当成会话忙并重新入队。 +var ErrDuplicateRequest = errors.New("store: duplicate request id") + // OwnerDefault 是单租户默认 owner（嵌入式档 / 未启用认证时的隐式身份）。 const OwnerDefault = "local" type Store struct { - db *sql.DB - dialect dialect - dataDir string // 工件对象目录（备份路径等诊断用） - blobs blobstore.Store // 工件对象存储（本地档；多实例切共享卷/对象存储） + db *sql.DB + dialect dialect + dataDir string // 工件对象目录（备份路径等诊断用） + blobs blobstore.Store // 工件对象存储（本地档；多实例切共享卷/对象存储） + instanceID string // 本进程标识；CreateRun 写入 runs.instance_id + + settleMu sync.Mutex + settleFault error // 非空时 SettleRun 在写入前失败，用于终态提交失败测试 } // OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。 @@ -101,6 +110,17 @@ func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Stor func (s *Store) Close() error { return s.db.Close() } +// SetInstanceID 记录本进程标识。之后 CreateRun 把它写入 runs.instance_id。 +// 只在开始接请求前调用一次。 +func (s *Store) SetInstanceID(id string) { s.instanceID = id } + +// SetSettleFault 让下一次 SettleRun 在写入前失败。nil 表示恢复正常。 +func (s *Store) SetSettleFault(err error) { + s.settleMu.Lock() + s.settleFault = err + s.settleMu.Unlock() +} + // Dialect 返回当前存储档名（sqlite | postgres；诊断/健康检查用）。 func (s *Store) Dialect() string { return s.dialect.name() } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 18fb493..6b37a60 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -56,8 +56,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) { if err != nil { t.Fatal(err) } - if len(got) != 13 { - t.Fatalf("expected versions [1..13], got %v", got) + if len(got) != 14 { + t.Fatalf("expected versions [1..14], got %v", got) } // 老库形态（v2 前）：events 全局自增、settings 单列主键。 if err := st.MigrateDown(1); err != nil {