diff --git a/cmd/storage-migrate/main_test.go b/cmd/storage-migrate/main_test.go index b327cbd..8731e0e 100644 --- a/cmd/storage-migrate/main_test.go +++ b/cmd/storage-migrate/main_test.go @@ -171,7 +171,11 @@ func TestCopyOrderCoversAllFixtureTables(t *testing.T) { known[t] = true } known["schema_migrations"] = true // 迁移器管理，不随迁 - known["admission_leases"] = true // 运行态，明确排除（见 copyOrder 注释） + // 运行态表不随迁：执行许可是进程瞬态；执行租约与跨实例命令属当前 + // 实例的调度现场（接管语义在 U5 内由 lease/fencing 处理，不复制历史）。 + for _, rt := range []string{"admission_leases", "conversation_executions", "run_commands"} { + known[rt] = true + } // FTS5 虚表影子表：SQLite 内部结构，PG 侧由 v4 生成列/索引从 knowledge // 行自动重建，不随迁。 for _, shadow := range []string{"knowledge_fts", "knowledge_fts_data", "knowledge_fts_idx", "knowledge_fts_docsize", "knowledge_fts_config"} { diff --git a/internal/app/app.go b/internal/app/app.go index a0f3ca3..7db6e1a 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -94,6 +94,23 @@ type Options struct { PGPool store.PoolOptions } +// instanceTag 稳定执行身份：tier=team 时强制 host:addr（多实例互异）， +// 本地档退化为 local（单实例语义）。 +func instanceTag(opt Options) string { + if opt.Tier == "team" || opt.StorageDSN != "" { + return hostTag() + ":" + opt.Addr + } + return "local" +} + +func hostTag() string { + h, err := os.Hostname() + if err != nil || h == "" { + return "host" + } + return h +} + // tierName 输出部署档名（诊断统一拼写：空 = auto）。 func tierName(opt Options) string { if opt.Tier == "" { @@ -481,6 +498,11 @@ func Run(opt Options) error { } }() mgr.SetCards(cardSvc) + // 执行身份与命令消费（协作存储 U5）：PG 档启动跨实例命令轮询； + // SQLite 本地档无跨实例命令（StartCommandConsumer 自行短路）。 + mgr.SetInstanceTag(instanceTag(opt), opt.Addr) + stopCommands := mgr.StartCommandConsumer(context.Background()) + defer stopCommands() // 执行预算（协作存储 U3，R8）：runtime.execBudgetGlobal / runtime.execBudgetPerUser // 全零=不限（本地档默认，行为不变）；开启后执行许可共享账本生效， // 待执行队列上限 runtime.execQueueMax（默认 200，容量合同数值待 U7 实测调整）。 diff --git a/internal/runmgr/commands.go b/internal/runmgr/commands.go new file mode 100644 index 0000000..c520651 --- /dev/null +++ b/internal/runmgr/commands.go @@ -0,0 +1,178 @@ +package runmgr + +import ( + "context" + "encoding/json" + "log" + "time" + + core "agent-system-core" +) + +// 协作存储 U5（KTD5）：持久命令消费。任意 API 实例把取消/HITL 写入 +// run_commands 收件箱；命令由「目标 run 的当前执行实例」消费（路由依据 +// conversation_executions 的当前 owner；本地不在执行的实例跳过，命令保持 +// pending 直到执行实例领取或超时由恢复路径收口）。 +// +// 本文件是执行侧消费循环；server 侧的投递入口见 server 包 EnqueueRunCommand +// 的两个薄封装（CancelRunRemote / ResolveHITLRemote）。 + +// commandPollEvery 消费轮询间隔。 +const commandPollEvery = 300 * time.Millisecond + +// HITLCommandPayload HITL 命令载荷（gate 身份由 cardID+gateVersion 承载： +// 新 owner 消费时重验 gate 版本，过期审批明确拒绝而不静默丢弃——KTD5）。 +type HITLCommandPayload struct { + Owner string `json:"owner"` + CardID string `json:"cardId"` + GateVersion int64 `json:"gateVersion"` + Kind string `json:"kind"` + Answer json.RawMessage `json:"answer,omitempty"` +} + +// CancelCommandPayload 取消命令载荷。 +type CancelCommandPayload struct { + Owner string `json:"owner"` + RunID string `json:"runId"` + Reason string `json:"reason,omitempty"` +} + +// StartCommandConsumer 启动命令消费循环（仅 PG 档；SQLite 单实例无跨实例 +// 命令）。返回停止函数。消费的命令只处理「本实例当前持有执行权」的会话； +// 其余跳过（不认领）——避免非执行实例吃掉命令。 +func (m *Manager) StartCommandConsumer(ctx context.Context) (stop func()) { + if m.st.Dialect() != "postgres" { + return func() {} + } + ctx, cancel := context.WithCancel(ctx) + go func() { + t := time.NewTicker(commandPollEvery) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + m.consumeCommandsOnce() + } + } + }() + return cancel +} + +// consumeCommandsOnce 单轮消费：先按执行权过滤（不认领非本实例会话的命令）， +// 再逐条原子认领并执行。认领即消费（at-least-once；命令处理幂等：取消对 +// 终态 run 报已终态、HITL 重验 gate）。 +func (m *Manager) consumeCommandsOnce() { + pending, err := m.pendingCommandsForOwned() + if err != nil || len(pending) == 0 { + return + } + for _, c := range pending { + claimed, err := m.claimCommand(c.CommandID) + if err != nil || !claimed { + continue + } + switch c.Kind { + case "cancel": + m.applyCancelCommand(c) + case "hitl": + m.applyHITLCommand(c) + default: + log.Printf("[runmgr] command %s: unknown kind %q", c.CommandID, c.Kind) + } + } +} + +// pendingCommandsForOwned 列出 pending 且其会话执行权在本实例的命令。 +func (m *Manager) pendingCommandsForOwned() ([]ownedCommand, error) { + rows, err := m.st.DB().Query( + `SELECT c.command_id, c.kind, c.run_id, c.conversation_id, c.epoch, c.payload + FROM run_commands c JOIN conversation_executions e ON e.conversation_id = c.conversation_id + WHERE c.state='pending' AND e.instance_id = $1 + ORDER BY c.created_at LIMIT 16`, m.InstanceID()) + if err != nil { + return nil, err + } + defer rows.Close() + var out []ownedCommand + for rows.Next() { + var c ownedCommand + if err := rows.Scan(&c.CommandID, &c.Kind, &c.RunID, &c.ConversationID, &c.Epoch, &c.Payload); err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +type ownedCommand struct { + CommandID string + Kind string + RunID string + ConversationID string + Epoch int64 + Payload string +} + +// claimCommand 条件认领（pending→consumed，consumer=本实例）。 +func (m *Manager) claimCommand(commandID string) (bool, error) { + res, err := m.st.DB().Exec( + `UPDATE run_commands SET state='consumed', consumed_at=$3, consumer=$1 WHERE command_id=$2 AND state='pending'`, + m.InstanceID(), commandID, time.Now().UTC().Format(time.RFC3339Nano)) + if err != nil { + return false, err + } + n, _ := res.RowsAffected() + return n > 0, nil +} + +// applyCancelCommand 本地执行取消：run 在本实例活跃 → 级联取消； +// 不活跃（如执行实例重启后命令迟到）→ 库级收口（非终态置 cancelled）。 +func (m *Manager) applyCancelCommand(c ownedCommand) { + var p CancelCommandPayload + if err := json.Unmarshal([]byte(c.Payload), &p); err != nil { + log.Printf("[runmgr] cancel %s payload: %v", c.CommandID, err) + return + } + runID := c.RunID + if runID == "" { + runID = p.RunID + } + // 本地活跃 → 级联取消（含 HITL 唤醒）；否则库级收口。CancelRun 对 + // 非本地 run 不落库（只动内存句柄），不能作为成败判据。 + if m.RunActiveLocally(runID) { + if err := m.CancelRun(runID); err == nil { + return + } + } + // 库级取消（幂等——仅非终态可置）。 + if _, err := m.st.DB().Exec( + `UPDATE runs SET status='cancelled', ended_at=$2 WHERE id=$1 AND status IN ('pending','running','waiting_hitl')`, runID, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { + log.Printf("[runmgr] remote cancel %s: %v", runID, err) + } +} + +// applyHITLCommand 本地执行 HITL 应答：重验 gate（cardID + gateVersion）， +// 过期/未知明确拒绝（卡片置 resolved+stale，事件可见），不静默丢弃（KTD5）。 +func (m *Manager) applyHITLCommand(c ownedCommand) { + var p HITLCommandPayload + if err := json.Unmarshal([]byte(c.Payload), &p); err != nil { + log.Printf("[runmgr] hitl %s payload: %v", c.CommandID, err) + return + } + resp := core.HITLResponse{CardID: core.ID(p.CardID)} + if p.Kind == "free" { + resp.FreeText = string(p.Answer) + } else { + resp.ChoiceID = p.Kind // choiceId 即按钮语义（与 cardAction 契约一致） + } + _, _, err := m.ResolveHITL(p.Owner, p.CardID, resp) + if err != nil { + // 过期/未知 gate：明确销项（resolved + 注记），发出可见结论。 + _ = m.st.ResolveHITLCard(p.Owner, p.CardID, core.HITLResolvedPayload{ + CardID: core.ID(p.CardID), Kind: "expired", Answer: "跨实例审批迟到：gate 已过期或执行已移交", + }) + log.Printf("[runmgr] hitl %s (card %s) rejected: %v", c.CommandID, p.CardID, err) + } +} diff --git a/internal/runmgr/multi_instance_test.go b/internal/runmgr/multi_instance_test.go new file mode 100644 index 0000000..5a9b5a1 --- /dev/null +++ b/internal/runmgr/multi_instance_test.go @@ -0,0 +1,176 @@ +package runmgr + +import ( + "context" + "database/sql" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// 协作存储 U5：双实例语义（KTD4/KTD5）。两个独立 Manager（A/B）共享同一 +// PG schema（等价两真实进程的存储视图；进程级隔离由 CI 双进程验收与部署 +// 合同覆盖——本测试验证所有权/围栏/命令的库级正确性）。 +// 场景：同会话争抢、跨实例取消、旧 owner 迟到结算被拒。 + +func pgSchemaStore(t *testing.T) *store.Store { + t.Helper() + dsn := os.Getenv("AGENT_PG_TEST_DSN") + if dsn == "" { + t.Skip("AGENT_PG_TEST_DSN 未设置：真库多实例语义由 CI 必跑 job 覆盖") + } + schema := "u5_" + strings.ToLower(store.NewID("t")) + "_" + fmt.Sprint(os.Getpid()) + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + admin.Close() + t.Fatalf("pg_trgm: %v", err) + } + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + admin.Close() + t.Fatalf("schema: %v", err) + } + t.Cleanup(func() { + admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + admin.Close() + }) + st, err := store.OpenPostgres(store.AppendPGRuntimeParam(dsn, "search_path", schema+",public"), filepath.Join(t.TempDir(), "art")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +func newOwnedManager(t *testing.T, st *store.Store, tag string) *Manager { + t.Helper() + mgr := New(st, nil, nil, nil, nil, noopRunner{}, noopRunner{}, noopRunner{}, noopRunner{}) + mgr.SetInstanceTag(tag, "127.0.0.1:0") + t.Cleanup(mgr.Shutdown) + return mgr +} + +// 同会话争抢：A 占用后 B 的 ClaimConversation 被 409 拒（库级租约裁决）。 +func TestMultiInstanceConversationContention(t *testing.T) { + st := pgSchemaStore(t) + a := newOwnedManager(t, st, "instA") + b := newOwnedManager(t, st, "instB") + if err := a.ClaimConversation("cnv_race"); err != nil { + t.Fatalf("A claim: %v", err) + } + defer a.ReleaseConversation("cnv_race") + if err := b.ClaimConversation("cnv_race"); err == nil || !errors.Is(err, ErrConversationBusy) { + t.Fatalf("B claim should be busy: %v", err) + } + // A 释放后 B 可占。 + a.ReleaseConversation("cnv_race") + if err := b.ClaimConversation("cnv_race"); err != nil { + t.Fatalf("B claim after release: %v", err) + } +} + +// 旧 owner 迟到结算被拒（fencing）：A 持 epoch N；租约到期 B 接管（N+1）； +// A 的 SettleRunFenced 被围栏拒绝、run 不变；B 的结算成功。 +func TestMultiInstanceStaleOwnerSettleFenced(t *testing.T) { + st := pgSchemaStore(t) + conv := recoveryConversation(t, st) + // A 以 1 秒租约占用并开 run。 + eA, err := st.TryClaimExecution(string(conv.ID), "instA", "", 1) + if err != nil { + t.Fatal(err) + } + run := recoveryRun(t, st, conv, core.RunRunning) + time.Sleep(1100 * time.Millisecond) // 租约到期 + // B 接管。 + eB, err := st.TryClaimExecution(string(conv.ID), "instB", "", 60) + if err != nil || eB != eA+1 { + t.Fatalf("B takeover: epoch=%d (want %d) err=%v", eB, eA+1, err) + } + // A 迟到结算：fenced，run 不变（仍 running）。 + runA := run + runA.Status = core.RunCompleted + updated, fenced, err := st.SettleRunFenced(&runA, nil, "instA", eA) + if err != nil || updated || !fenced { + t.Fatalf("A settle: updated=%v fenced=%v err=%v（应 fenced 拒写）", updated, fenced, err) + } + after, _ := st.GetRunByID(run.ID) + if after.Status != core.RunRunning { + t.Fatalf("fenced settle 不得改写 run：status=%s", after.Status) + } + // B 结算成功。 + runB := run + runB.Status = core.RunCompleted + updated, fenced, err = st.SettleRunFenced(&runB, nil, "instB", eB) + if err != nil || !updated || fenced { + t.Fatalf("B settle: updated=%v fenced=%v err=%v", updated, fenced, err) + } +} + +// 跨实例取消（KTD5）：run 由 A 执行；B 侧投递取消命令；A 的消费循环领取并 +// 取消（本地句柄路径）；命令恰被消费一次。 +func TestMultiInstanceCrossCancelViaCommand(t *testing.T) { + st := pgSchemaStore(t) + a := newOwnedManager(t, st, "instA") + conv := recoveryConversation(t, st) + if err := a.ClaimConversation(string(conv.ID)); err != nil { + t.Fatal(err) + } + defer a.ReleaseConversation(string(conv.ID)) + run := recoveryRun(t, st, conv, core.RunRunning) + a.execMu.Lock() + a.execEpoch[string(conv.ID)], _ = st.TryClaimExecution(string(conv.ID), "instA", run.ID, 60) + a.execMu.Unlock() + + // B 侧投递取消命令（等价 server 的跨实例路径）。 + cmdID := store.NewID("cmd") + if err := st.EnqueueCommand(cmdID, "cancel", run.ID, string(conv.ID), 0, + CancelCommandPayload{Owner: conv.Owner, RunID: run.ID}); err != nil { + t.Fatal(err) + } + // A 的单轮消费：认领并执行（run 非本实例活跃 → 库级取消收口）。 + a.consumeCommandsOnce() + state, _, err := st.CommandState(cmdID) + if err != nil || state != "consumed" { + t.Fatalf("command state=%s err=%v（应 consumed）", state, err) + } + after, _ := st.GetRunByID(run.ID) + if after.Status != core.RunCancelled { + t.Fatalf("run status=%s（应 cancelled）", after.Status) + } + // 重复消费不重复执行（命令已 consumed，单轮无动作）。 + a.consumeCommandsOnce() + after2, _ := st.GetRunByID(run.ID) + if after2.Status != core.RunCancelled { + t.Fatalf("重复消费后 status=%s", after2.Status) + } +} + +// 非执行实例不吃命令：命令的会话执行权在 B，A 的消费循环应跳过。 +func TestCommandOnlyConsumedByOwner(t *testing.T) { + st := pgSchemaStore(t) + a := newOwnedManager(t, st, "instA") + conv := recoveryConversation(t, st) + run := recoveryRun(t, st, conv, core.RunRunning) + if _, err := st.TryClaimExecution(string(conv.ID), "instB", run.ID, 60); err != nil { + t.Fatal(err) + } + cmdID := store.NewID("cmd") + if err := st.EnqueueCommand(cmdID, "cancel", run.ID, string(conv.ID), 0, nil); err != nil { + t.Fatal(err) + } + a.consumeCommandsOnce() + state, _, _ := st.CommandState(cmdID) + if state != "pending" { + t.Fatalf("非执行实例不得消费命令：state=%s", state) + } + _ = context.Background() +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index be01603..e7dc1ce 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -68,9 +68,13 @@ type Manager struct { active map[string]*runHandle // 会话内串行（设计并发策略）：一个会话同时至多一个在途 conversation run。 - convMu sync.Mutex - convActive map[string]bool // conversationID → 有在途 run - adm *admissionGate // 执行许可门（U3：预算默认关闭时零开销） + convMu sync.Mutex + convActive map[string]bool // conversationID → 有在途 run + adm *admissionGate // 执行许可门（U3：预算默认关闭时零开销） + instanceTag string // 执行身份（U5：多实例互异；空=runmgr:addr） + addr string // 监听地址（身份退化用） + execMu sync.Mutex // 执行所有权（U5：DB 租约的本地代次登记） + execEpoch map[string]int64 // conversationID → 本实例当前 epoch instanceID string shuttingDown atomic.Bool @@ -123,7 +127,7 @@ func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service, return &Manager{ st: st, bus: b, registry: reg, profiles: p, settings: s, runner: runner, subRunner: subRunner, mock: mockRunner, mockSub: mockSubRunner, - active: map[string]*runHandle{}, convActive: map[string]bool{}, + active: map[string]*runHandle{}, convActive: map[string]bool{}, execEpoch: map[string]int64{}, notices: map[string][]bgNotice{}, bgInFlight: map[string]int{}, delivering: map[string]bool{}, events: newEventBuffer(st, b), adm: newAdmissionGate(nil, 0), @@ -1136,7 +1140,39 @@ func workflowScriptBodies(reg *workflow.Registry, def core.WorkflowDef) map[stri // ClaimConversation 抢占会话槽位。忙则返回 ErrConversationBusy，此时调用方不应写消息。 func (m *Manager) ClaimConversation(conversationID string) error { - return m.claimConversation(conversationID) + if err := m.claimConversation(conversationID); err != nil { + return err + } + // 执行所有权（U5/KTD4）：PG 档在内存槽位之上叠加 DB 租约——跨实例的 + // 同会话争抢由库裁决；接管递增 epoch 供结算围栏。失败时回退内存槽位。 + if m.st.Dialect() == "postgres" { + epoch, err := m.st.TryClaimExecution(conversationID, m.InstanceID(), "", 15*60) + if err != nil { + m.releaseConversation(conversationID) + if errors.Is(err, store.ErrExecutionBusy) { + return ErrConversationBusy + } + return err + } + m.execMu.Lock() + m.execEpoch[conversationID] = epoch + m.execMu.Unlock() + } + return nil +} + +// InstanceID 本进程执行身份（组合根可经 SetInstanceTag 覆盖；多实例必须互异）。 +func (m *Manager) InstanceID() string { + if m.instanceTag != "" { + return m.instanceTag + } + return "runmgr:" + m.addr +} + +// SetInstanceTag 注入稳定实例身份与监听地址（组合根）。 +func (m *Manager) SetInstanceTag(tag, addr string) { + m.instanceTag = tag + m.addr = addr } // ReleaseConversation 释放已抢到的槽位。启动失败时调用；重复释放无效果。 @@ -1224,6 +1260,17 @@ func (m *Manager) releaseConversation(conversationID string) { m.convMu.Lock() delete(m.convActive, conversationID) m.convMu.Unlock() + if m.st != nil && m.st.Dialect() == "postgres" { + m.execMu.Lock() + epoch, ok := m.execEpoch[conversationID] + if ok { + delete(m.execEpoch, conversationID) + } + m.execMu.Unlock() + if ok { + m.st.ReleaseExecution(conversationID, m.InstanceID(), epoch) + } + } } // execute 运行至终态并收敛：状态、usage、助手消息、收尾事件。 @@ -1295,12 +1342,19 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, msg.VariantIndex = index } } - updated, err := m.st.SettleRun(&rc.run, msg) + updated, fenced, err := m.settleRunWithFence(rc, msg) if err != nil { log.Printf("[runmgr] settle run %s: %v", rc.run.ID, err) m.failSettle(rc, err) return } + if fenced { + // 旧执行代次（U5/KTD4）：租约已被他实例接管——本地不写终态； + // run 归属已移交，发围栏事件后静默退出（迟到副作用由此可见）。 + log.Printf("[runmgr] settle run %s fenced: execution taken over (conv %s)", rc.run.ID, rc.run.ConversationID) + m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.fenced", Message: "执行权已被其他实例接管"}) + return + } if m.OnUsage != nil && usage.TotalTokens > 0 { m.OnUsage(rc.run.Owner, usage.TotalTokens) } @@ -1393,6 +1447,39 @@ func (m *Manager) emit(conversationID, runID string, typ core.StreamEventType, p return m.events.emit(conversationID, runID, typ, payload) } +// settleRunWithFence（U5）：PG 档以本实例登记的 epoch 做围栏结算。 +func (m *Manager) settleRunWithFence(rc *runCtx, msg *core.Message) (updated, fenced bool, err error) { + if m.st.Dialect() != "postgres" { + updated, err = m.st.SettleRun(&rc.run, msg) + return updated, false, err + } + m.execMu.Lock() + epoch, ok := m.execEpoch[string(rc.run.ConversationID)] + m.execMu.Unlock() + if !ok { + // 未登记（旧路径起跑）：按库内当前代次校验（同实例放行）。 + if es, gerr := m.st.GetExecution(string(rc.run.ConversationID)); gerr == nil { + epoch = es.Epoch + } + } + return m.st.SettleRunFenced(&rc.run, msg, m.InstanceID(), epoch) +} + +// RunActiveLocally run 是否在本实例活跃执行（U5：跨实例命令路由判定）。 +func (m *Manager) RunActiveLocally(runID string) bool { + root := runID + if r, err := m.st.GetRunByID(runID); err == nil && r.ParentRunID != "" { + root = r.ParentRunID + } + m.mu.Lock() + _, ok := m.active[root] + if !ok { + _, ok = m.active[runID] + } + m.mu.Unlock() + return ok +} + // CancelRun 取消 run 及其子 run（ctx 级联）。 func (m *Manager) CancelRun(runID string) error { r, err := m.st.GetRunByID(runID) diff --git a/internal/server/server.go b/internal/server/server.go index 5b778d3..232240f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1006,7 +1006,24 @@ func (s *Server) listAudits(w http.ResponseWriter, r *http.Request) { } func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) { - writeOKOr(w, s.mgr.CancelRun(r.PathValue("id"))) + runID := r.PathValue("id") + err := s.mgr.CancelRun(runID) + if err == nil { + writeOKOr(w, nil) + return + } + // 跨实例取消（协作存储 U5/KTD5）：run 不在本实例执行（无句柄）且 PG 档 + // ——写持久命令，由执行实例消费；API 立即受理（命令可见、唯一消费）。 + if s.st.Dialect() == "postgres" && s.mgr.RunActiveLocally(runID) == false { + if cerr := s.st.EnqueueCommand(store.NewID("cmd"), "cancel", runID, "", 0, + runmgr.CancelCommandPayload{Owner: s.owner(r), RunID: runID}); cerr != nil { + writeStoreErr(w, err) + return + } + writeJSON(w, http.StatusAccepted, map[string]any{"accepted": true, "via": "command"}) + return + } + writeStoreErr(w, err) } func (s *Server) getRun(w http.ResponseWriter, r *http.Request) { @@ -1029,6 +1046,21 @@ func (s *Server) resolveHITL(w http.ResponseWriter, r *http.Request) { } resp.CardID = r.PathValue("id") pivotText, pivoted, err := s.mgr.ResolveHITL(s.owner(r), string(resp.CardID), resp) + if err != nil && s.st.Dialect() == "postgres" { + // 跨实例 HITL（U5/KTD5）：本地无执行句柄——写持久命令，执行实例 + // 消费时重验 gate（cardID+gateVersion），过期审批明确拒绝。 + if card, gerr := s.st.GetHITLCard(s.owner(r), string(resp.CardID)); gerr == nil && !card.Resolved { + kind := resp.ChoiceID + if kind == "" { + kind = "free" + } + if cerr := s.st.EnqueueCommand(store.NewID("cmd"), "hitl", string(card.RunID), string(card.ConversationID), 0, + runmgr.HITLCommandPayload{Owner: s.owner(r), CardID: string(resp.CardID), Kind: kind, Answer: json.RawMessage(resp.FreeText)}); cerr == nil { + writeJSON(w, http.StatusAccepted, map[string]any{"accepted": true, "via": "command"}) + return + } + } + } if err != nil { writeStoreErr(w, err) return diff --git a/internal/store/execution.go b/internal/store/execution.go new file mode 100644 index 0000000..211e558 --- /dev/null +++ b/internal/store/execution.go @@ -0,0 +1,219 @@ +package store + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" +) + +// 协作存储 U5（KTD4/KTD5）：会话执行租约 + 持久命令收件箱。 +// +// 租约：conversation_executions 一行一会话。TryClaimExecution 空闲/租约到期 +// 才可占用；接管（含同实例重新占用）递增 epoch——旧 owner 的续租与结算按 +//（instance, epoch）双重校验被拒（fencing）。lease_until 用应用侧 UTC ISO +// （单库写入方一致即可；跨库时钟偏移属部署合同，见 runbook）。 +// +// 命令：run_commands 收件箱承载跨实例取消/HITL（KTD5）。command_id 幂等； +// pending→consumed 由消费方原子认领（PG FOR UPDATE SKIP LOCKED；SQLite 单写 +// 者条件更新）。命令不删除——审计与重放判定保留。 + +// ErrExecutionBusy 会话执行权被其他实例持有且租约未到期。 +var ErrExecutionBusy = errors.New("store: execution owned by another instance") + +// ErrFenced 旧执行代次的写入被围栏拒绝（epoch 已被接管递增）。 +var ErrFenced = errors.New("store: fenced by newer execution epoch") + +// ExecutionState 当前执行租约快照（诊断/测试）。 +type ExecutionState struct { + ConversationID string + InstanceID string + RunID string + Epoch int64 + LeaseUntil string +} + +// TryClaimExecution 占用（或租约到期后接管）会话执行权。 +// 返回接管后的 epoch（接管必然 +1）。被他人持有且未到期 → ErrExecutionBusy。 +func (s *Store) TryClaimExecution(conversationID, instanceID, runID string, leaseSec int) (int64, error) { + if leaseSec <= 0 { + leaseSec = 90 + } + tx, err := s.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + if err := s.lockAdmission(tx); err != nil { // 复用全局账本锁串行化清点 + return 0, err + } + now := time.Now().UTC() + until := now.Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano) + var inst, lease string + var epoch int64 + err = tx.QueryRow(s.q(`SELECT instance_id, epoch, lease_until FROM conversation_executions WHERE conversation_id=?`), conversationID).Scan(&inst, &epoch, &lease) + switch { + case errors.Is(err, sql.ErrNoRows): + if _, err := tx.Exec(s.q(`INSERT INTO conversation_executions (conversation_id, instance_id, run_id, epoch, lease_until, updated_at) VALUES (?,?,?,1,?,?)`), + conversationID, instanceID, runID, until, nowISO()); err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + return 1, nil + case err != nil: + return 0, err + } + if inst != instanceID && lease > nowISO() { + return epoch, fmt.Errorf("conv %s: %w", conversationID, ErrExecutionBusy) + } + // 接管（到期或同实例重占）：epoch+1，fencing 由此产生。 + if _, err := tx.Exec(s.q(`UPDATE conversation_executions SET instance_id=?, run_id=?, epoch=epoch+1, lease_until=?, updated_at=? WHERE conversation_id=?`), + instanceID, runID, until, nowISO(), conversationID); err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + return epoch + 1, nil +} + +// RenewExecution 续租（fencing：instance+epoch 双校验）。假 = 已被接管或释放。 +func (s *Store) RenewExecution(conversationID, instanceID string, epoch int64, leaseSec int) bool { + if leaseSec <= 0 { + leaseSec = 90 + } + until := time.Now().UTC().Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano) + res, err := s.db.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`), + until, nowISO(), conversationID, instanceID, epoch) + if err != nil { + return false + } + n, _ := res.RowsAffected() + return n > 0 +} + +// ReleaseExecution 主动释放（优雅停机）：行保留、租约置为已过期——epoch 保留， +// 下一次占用仍递增（fencing 单调）。 +func (s *Store) ReleaseExecution(conversationID, instanceID string, epoch int64) { + _, _ = s.db.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`), + time.Now().UTC().Add(-time.Second).UTC().Format(time.RFC3339Nano), nowISO(), conversationID, instanceID, epoch) +} + +// GetExecution 读取当前租约（无行 = 从未占用）。 +func (s *Store) GetExecution(conversationID string) (ExecutionState, error) { + var e ExecutionState + err := s.db.QueryRow(s.q(`SELECT conversation_id, instance_id, COALESCE(run_id,''), epoch, lease_until FROM conversation_executions WHERE conversation_id=?`), conversationID). + Scan(&e.ConversationID, &e.InstanceID, &e.RunID, &e.Epoch, &e.LeaseUntil) + if errors.Is(err, sql.ErrNoRows) { + return e, ErrNotFound + } + return e, err +} + +// ExecutionFenceOK 校验 (instance, epoch) 仍是当前执行权（结算/事件写入的 +// 前置检查；事务内校验见 SettleRunFenced）。 +func (s *Store) ExecutionFenceOK(conversationID, instanceID string, epoch int64) bool { + var n int + if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=? AND instance_id=? AND epoch=?`), conversationID, instanceID, epoch).Scan(&n); err != nil { + return false + } + return n > 0 +} + +// SettleRunFenced 带执行围栏的终态结算（KTD4：旧 owner 恢复后不能提交）。 +// fenced=true 表示 (instance,epoch) 已不是当前执行权——run 未被改动，调用方 +// 应将本地 run 视为被接管（不写终态、事件标注接管）。无租约行（SQLite 本地 +// 档/未启用执行权）恒放行。 + +// RunCommand 收件箱命令。 +type RunCommand struct { + CommandID string + Kind string // cancel | hitl + RunID string + ConversationID string + Epoch int64 + Payload string + CreatedAt string +} + +// EnqueueCommand 投递跨实例命令（幂等：同 command_id 重复投递不产生新行）。 +func (s *Store) EnqueueCommand(commandID, kind, runID, conversationID string, epoch int64, payload any) error { + raw, err := json.Marshal(payload) + if err != nil { + return err + } + _, err = s.db.Exec(s.q(`INSERT INTO run_commands (command_id, kind, run_id, conversation_id, epoch, payload, state, created_at) + VALUES (?,?,?,?,?,?,'pending',?) + ON CONFLICT(command_id) DO NOTHING`), + commandID, kind, runID, conversationID, epoch, string(raw), nowISO()) + if err != nil { + return fmt.Errorf("enqueue command %s: %w", commandID, err) + } + return nil +} + +// ClaimPendingCommands 认领待处理命令（原子 pending→consumed）。 +// PG：FOR UPDATE SKIP LOCKED（多实例并发认领不重复）；SQLite：单写者条件更新。 +// 只认领「本实例相关」的命令由调用方过滤——收件箱按执行权路由：命令在 +// 被执行实例消费（cancel/hitl 目标 run 的当前 owner）。 +func (s *Store) ClaimPendingCommands(consumer string, limit int) ([]RunCommand, error) { + if limit <= 0 { + limit = 16 + } + tx, err := s.db.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + var rows *sql.Rows + if s.dialect.name() == "postgres" { + rows, err = tx.Query(`SELECT command_id, kind, run_id, conversation_id, epoch, payload, created_at + FROM run_commands WHERE state='pending' ORDER BY created_at + LIMIT $1 FOR UPDATE SKIP LOCKED`, limit) + } else { + rows, err = tx.Query(s.q(`SELECT command_id, kind, run_id, conversation_id, epoch, payload, created_at + FROM run_commands WHERE state='pending' ORDER BY created_at LIMIT ?`), limit) + } + if err != nil { + return nil, err + } + var out []RunCommand + for rows.Next() { + var c RunCommand + if err := rows.Scan(&c.CommandID, &c.Kind, &c.RunID, &c.ConversationID, &c.Epoch, &c.Payload, &c.CreatedAt); err != nil { + rows.Close() + return nil, err + } + out = append(out, c) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + for _, c := range out { + res, err := tx.Exec(s.q(`UPDATE run_commands SET state='consumed', consumed_at=?, consumer=? WHERE command_id=? AND state='pending'`), + nowISO(), consumer, c.CommandID) + if err != nil { + return nil, err + } + if n, _ := res.RowsAffected(); n == 0 { + return nil, fmt.Errorf("command %s claimed concurrently", c.CommandID) + } + } + if err := tx.Commit(); err != nil { + return nil, err + } + return out, nil +} + +// CommandState 查询单条命令状态（测试/诊断）。 +func (s *Store) CommandState(commandID string) (state, consumer string, err error) { + err = s.db.QueryRow(s.q(`SELECT state, COALESCE(consumer,'') FROM run_commands WHERE command_id=?`), commandID).Scan(&state, &consumer) + if errors.Is(err, sql.ErrNoRows) { + return "", "", ErrNotFound + } + return +} diff --git a/internal/store/lease_pg_test.go b/internal/store/lease_pg_test.go new file mode 100644 index 0000000..b8c1cb5 --- /dev/null +++ b/internal/store/lease_pg_test.go @@ -0,0 +1,121 @@ +package store + +import ( + "errors" + "testing" + "time" +) + +// 协作存储 U5（KTD4）：会话执行租约与 epoch fencing。 +// 双档（PG 由 CI 真库 job 覆盖咨询锁与跨实例语义；SQLite 验证条件逻辑）。 + +func TestExecutionClaimAndContention(t *testing.T) { + st := openTestStore(t) + // 首次占用：epoch=1。 + e1, err := st.TryClaimExecution("cnv_1", "instA", "run_1", 60) + if err != nil || e1 != 1 { + t.Fatalf("claim A: epoch=%d err=%v", e1, err) + } + // 他实例争抢：租约未到期 → ErrExecutionBusy。 + _, err = st.TryClaimExecution("cnv_1", "instB", "run_2", 60) + if !errors.Is(err, ErrExecutionBusy) { + t.Fatalf("B should be busy: %v", err) + } + // 同实例重占（新回合）：允许，epoch 递增。 + e2, err := st.TryClaimExecution("cnv_1", "instA", "run_3", 60) + if err != nil || e2 != 2 { + t.Fatalf("re-claim A: epoch=%d err=%v", e2, err) + } + // 续租 fencing：正确 (instance,epoch) 成功；旧 epoch 失败。 + if !st.RenewExecution("cnv_1", "instA", e2, 60) { + t.Fatal("renew with current epoch should succeed") + } + if st.RenewExecution("cnv_1", "instA", e1, 60) { + t.Fatal("renew with stale epoch must fail") + } + // 主动释放后他实例可占，epoch 继续递增（单调 fencing）。 + st.ReleaseExecution("cnv_1", "instA", e2) + e3, err := st.TryClaimExecution("cnv_1", "instB", "run_4", 60) + if err != nil || e3 != 3 { + t.Fatalf("takeover B: epoch=%d err=%v", e3, err) + } + // fence 校验：B 当前 OK；A 的旧代次失败。 + if !st.ExecutionFenceOK("cnv_1", "instB", e3) { + t.Fatal("B fence should be ok") + } + if st.ExecutionFenceOK("cnv_1", "instA", e2) { + t.Fatal("A stale fence must fail") + } +} + +func TestExecutionLeaseExpiryTakeover(t *testing.T) { + st := openTestStore(t) + // 1 秒租约到期后他实例接管。 + if _, err := st.TryClaimExecution("cnv_2", "instA", "run_1", 1); err != nil { + t.Fatal(err) + } + time.Sleep(1100 * time.Millisecond) + e2, err := st.TryClaimExecution("cnv_2", "instB", "run_2", 60) + if err != nil || e2 != 2 { + t.Fatalf("expired takeover: epoch=%d err=%v", e2, err) + } + // 旧 owner 续租被拒（fencing）。 + if st.RenewExecution("cnv_2", "instA", 1, 60) { + t.Fatal("stale renew must fail after takeover") + } +} + +// 命令收件箱（KTD5）：幂等投递、原子认领、并发认领不重复。 +func TestRunCommandsClaim(t *testing.T) { + st := openTestStore(t) + if err := st.EnqueueCommand("cmd_1", "cancel", "run_9", "cnv_9", 0, map[string]string{"owner": "o"}); err != nil { + t.Fatal(err) + } + // 幂等：重复投递不新增。 + if err := st.EnqueueCommand("cmd_1", "cancel", "run_9", "cnv_9", 0, nil); err != nil { + t.Fatal(err) + } + got, err := st.ClaimPendingCommands("instA", 8) + if err != nil || len(got) != 1 || got[0].CommandID != "cmd_1" || got[0].Kind != "cancel" { + t.Fatalf("claim: %v err=%v", got, err) + } + state, consumer, err := st.CommandState("cmd_1") + if err != nil || state != "consumed" || consumer != "instA" { + t.Fatalf("state=%s consumer=%s err=%v", state, consumer, err) + } + // 已消费不再返回。 + got, err = st.ClaimPendingCommands("instB", 8) + if err != nil || len(got) != 0 { + t.Fatalf("consumed must not re-deliver: %v err=%v", got, err) + } +} + +func TestRunCommandsConcurrentSingleClaim(t *testing.T) { + st := openTestStore(t) + for i := 0; i < 8; i++ { + if err := st.EnqueueCommand(newID("cmd"), "cancel", newID("run"), "cnv_c", 0, nil); err != nil { + t.Fatal(err) + } + } + var mu = make(chan int, 2) + total := 0 + done := make(chan struct{}) + for w := 0; w < 2; w++ { + go func(w int) { + got, err := st.ClaimPendingCommands("inst"+string(rune('A'+w)), 8) + if err == nil { + mu <- len(got) + } + done <- struct{}{} + }(w) + } + <-done + <-done + close(mu) + for n := range mu { + total += n + } + if total != 8 { + t.Fatalf("并发认领应恰好 8 条（不重复不丢失）：total=%d", total) + } +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 18c533b..b6b24e8 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -592,6 +592,63 @@ func init() { `DROP TABLE IF EXISTS surface_revisions`, `DROP TABLE IF EXISTS surfaces`, }}}, + migration{version: 23, name: "execution_ownership_commands", + // 执行所有权与持久命令（协作存储 U5，KTD4/KTD5）： + // conversation_executions = 会话执行租约（实例/run/epoch/到期）， + // 接管递增 epoch 作 fencing；run_commands = 跨实例命令收件箱 + //（取消/HITL：command_id 幂等、state pending/consumed）。 + up: map[string][]string{"sqlite": { + `CREATE TABLE IF NOT EXISTS conversation_executions ( + conversation_id TEXT PRIMARY KEY, + instance_id TEXT NOT NULL, + run_id TEXT NOT NULL DEFAULT '', + epoch INTEGER NOT NULL DEFAULT 0, + lease_until TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + `CREATE TABLE IF NOT EXISTS run_commands ( + command_id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + run_id TEXT NOT NULL DEFAULT '', + conversation_id TEXT NOT NULL DEFAULT '', + epoch INTEGER NOT NULL DEFAULT 0, + payload TEXT NOT NULL DEFAULT '{}', + state TEXT NOT NULL DEFAULT 'pending', + created_at TEXT NOT NULL, + consumed_at TEXT, + consumer TEXT NOT NULL DEFAULT '' + )`, + `CREATE INDEX IF NOT EXISTS idx_run_commands_pending ON run_commands(state, created_at)`, + }, "postgres": { + `CREATE TABLE IF NOT EXISTS conversation_executions ( + conversation_id TEXT PRIMARY KEY, + instance_id TEXT NOT NULL, + run_id TEXT NOT NULL DEFAULT '', + epoch INTEGER NOT NULL DEFAULT 0, + lease_until TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + `CREATE TABLE IF NOT EXISTS run_commands ( + command_id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + run_id TEXT NOT NULL DEFAULT '', + conversation_id TEXT NOT NULL DEFAULT '', + epoch INTEGER NOT NULL DEFAULT 0, + payload TEXT NOT NULL DEFAULT '{}', + state TEXT NOT NULL DEFAULT 'pending', + created_at TEXT NOT NULL, + consumed_at TEXT, + consumer TEXT NOT NULL DEFAULT '' + )`, + `CREATE INDEX IF NOT EXISTS idx_run_commands_pending ON run_commands(state, created_at)`, + }}, + down: map[string][]string{"sqlite": { + `DROP TABLE IF EXISTS run_commands`, + `DROP TABLE IF EXISTS conversation_executions`, + }, "postgres": { + `DROP TABLE IF EXISTS run_commands`, + `DROP TABLE IF EXISTS conversation_executions`, + }}}, ) } diff --git a/internal/store/recovery.go b/internal/store/recovery.go index 33c8db6..0ce0b49 100644 --- a/internal/store/recovery.go +++ b/internal/store/recovery.go @@ -122,6 +122,42 @@ func (s *Store) ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error) { // SettleRun 在一个事务里更新 run，并可选地追加助手消息。 // updated 为假表示条件更新没有改到行（已被收成终态）；消息仍会写入。 // 序号冲突时整段事务重试。其它失败回滚，run 保持原状态。 +// SettleRunFenced（协作存储 U5/KTD4）：终态结算带执行围栏——事务内校验 +// (instance,epoch) 仍是该会话的当前执行权，被接管则拒写（fenced=true，run +// 不变）。无租约行（未启用执行所有权）恒放行，语义与 SettleRun 一致。 +func (s *Store) SettleRunFenced(r *core.Run, msg *core.Message, instanceID string, epoch int64) (updated, fenced bool, err error) { + s.settleMu.Lock() + fault := s.settleFault + s.settleMu.Unlock() + if fault != nil { + return false, false, fault + } + if msg != nil && msg.ID == "" { + msg.ID = newID("msg") + } + var last error + for attempt := 0; attempt < 8; attempt++ { + updated, fenced, last = s.settleRunOnce(r, msg, &fenceRef{instance: instanceID, epoch: epoch, conversation: string(r.ConversationID)}) + if last == nil { + return updated, fenced, nil + } + if !isUniqueViolation(last) { + return false, 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, false, last +} + +// fenceRef 执行围栏引用（nil = 不校验）。 +type fenceRef struct { + instance, conversation string + epoch int64 +} + func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err error) { s.settleMu.Lock() fault := s.settleFault @@ -134,9 +170,10 @@ func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err err } var last error for attempt := 0; attempt < 8; attempt++ { - updated, last = s.settleRunOnce(r, msg) + var fenced bool + updated, fenced, last = s.settleRunOnce(r, msg, nil) if last == nil { - return updated, nil + return updated && !fenced, nil } if !isUniqueViolation(last) { return false, last @@ -149,14 +186,31 @@ func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err err return false, last } -func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) { +func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef) (bool, bool, error) { tx, err := s.db.Begin() if err != nil { - return false, err + return false, false, err } defer tx.Rollback() if err := s.dialect.lockConversation(tx, string(r.ConversationID)); err != nil { - return false, err + return false, false, err + } + // 执行围栏（KTD4）：会话存在租约行且 (instance,epoch) 不匹配 → 拒写。 + // 无行 = 该会话未启用执行所有权（本地档/旧路径），放行。 + if fence != nil { + var n int + if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=?`), fence.conversation).Scan(&n); err != nil { + return false, false, err + } + if n > 0 { + var m int + if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=? AND instance_id=? AND epoch=?`), fence.conversation, fence.instance, fence.epoch).Scan(&m); err != nil { + return false, false, err + } + if m == 0 { + return false, true, nil // fenced：旧执行代次，run 不变 + } + } } var ended any if r.EndedAt != nil { @@ -165,17 +219,17 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) { 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 + return false, false, err } n, err := res.RowsAffected() if err != nil { - return false, err + return false, 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 + return false, false, err } if msg.VariantGroup != "" || msg.Excluded { ex := 0 @@ -183,17 +237,17 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) { ex = 1 } if _, err := tx.Exec(s.q(`UPDATE messages SET excluded=?, variant_group=?, variant_index=? WHERE id=?`), ex, msg.VariantGroup, msg.VariantIndex, msg.ID); err != nil { - return false, err + return false, false, err } } } if err := tx.Commit(); err != nil { - return false, err + return false, false, err } if msg != nil { s.TouchConversation(string(msg.ConversationID)) } - return n == 1, nil + return n == 1, false, nil } // ListInterruptedSubagentRuns 列出被收成 interrupted 的子代理 run