diff --git a/internal/app/app.go b/internal/app/app.go index c6fb41a..6a8b8e4 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -473,6 +473,16 @@ func Run(opt Options) error { // 收件箱任务卡：事件走会话事件管道（持久化/seq/重放）；关闭即 T0 基线。 cardSvc := cardsvc.New(st, mgr.EmitCardEvent) mgr.SetCards(cardSvc) + // 执行预算（协作存储 U3，R8）：runtime.execBudgetGlobal / runtime.execBudgetPerUser + // 全零=不限（本地档默认，行为不变）；开启后执行许可共享账本生效， + // 待执行队列上限 runtime.execQueueMax（默认 200，容量合同数值待 U7 实测调整）。 + mgr.SetExecutionBudgetFn(func() store.ExecutionBudget { + eff := setSvc.Effective(store.OwnerDefault, "") + return store.ExecutionBudget{ + Global: int(eff.GetNumber("runtime.execBudgetGlobal", 0)), + PerUser: int(eff.GetNumber("runtime.execBudgetPerUser", 0)), + } + }, int(setSvc.Effective(store.OwnerDefault, "").GetNumber("runtime.execQueueMax", 200))) // 命令总线（提案 T3/D27/D30）：四入口同源分发 + 审计 + 危险内联确认卡。 bus := cmdbus.New(st, cardSvc, func() time.Time { return time.Now().UTC() }) diff --git a/internal/runmgr/admission.go b/internal/runmgr/admission.go new file mode 100644 index 0000000..a261845 --- /dev/null +++ b/internal/runmgr/admission.go @@ -0,0 +1,122 @@ +package runmgr + +import ( + "context" + "errors" + "log" + "sync" + + "agent-system-backend/internal/store" +) + +// 协作存储 U3 切片 2：执行许可门（R8）。预算经 budgetFn 动态读取（设置 +// runtime.execBudgetGlobal / runtime.execBudgetPerUser，默认全零=不限，本地档 +// 行为与历史一致）。许可的共享权威在 store.admission_leases（PG 咨询锁下 +// 原子清点）；本门只负责有界排队（接纳数）与让出/重取（父等待子代理或 +// HITL 时让出执行许可、保留会话所有权——避免额度占尽时父子互等）。 + +// ErrAdmissionQueueFull 待执行队列已满（接纳被拒；HTTP 层映射 429）。 +var ErrAdmissionQueueFull = errors.New("runmgr: execution queue full") + +// admissionGate 每进程一个；释放时广播唤醒全部等待者（无公平队列，Golang +// cond 语义；公平分配留 U7 压测后按实测加）。 +type admissionGate struct { + mu sync.Mutex + broadcast chan struct{} // 释放时 close+重建 + waiters int + budgetFn func() store.ExecutionBudget + queueMax int +} + +func newAdmissionGate(budgetFn func() store.ExecutionBudget, queueMax int) *admissionGate { + if budgetFn == nil { + budgetFn = func() store.ExecutionBudget { return store.ExecutionBudget{} } + } + if queueMax <= 0 { + queueMax = 200 // 方案容量合同的默认队列上限（待实测调整） + } + return &admissionGate{broadcast: make(chan struct{}), budgetFn: budgetFn, queueMax: queueMax} +} + +// budget 加锁读取当前预算（与 SetExecutionBudgetFn 并发安全）。 +func (g *admissionGate) budget() store.ExecutionBudget { + if g == nil { + return store.ExecutionBudget{} // 直接构造的 Manager（测试）：预算关闭 + } + g.mu.Lock() + defer g.mu.Unlock() + return g.budgetFn() +} + +// QueueDepth 返回当前排队数（诊断/验收）。 +func (g *admissionGate) QueueDepth() int { + if g == nil { + return 0 + } + g.mu.Lock() + defer g.mu.Unlock() + return g.waiters +} + +// QueueFull 非阻塞探测：排队是否已满（API 边缘在落库前拒绝，429 且不写数据）。 +func (g *admissionGate) QueueFull() bool { + budget := g.budget() + if budget.Off() { + return false + } + g.mu.Lock() + defer g.mu.Unlock() + return g.waiters >= g.queueMax +} + +// acquire 阻塞获取执行许可（ctx 取消则放弃排队）。预算关闭时零开销直通。 +func (g *admissionGate) acquire(ctx context.Context, st *store.Store, runID, owner string) error { + if g == nil { + return nil + } + budget := g.budget() + if budget.Off() { + return nil + } + for { + if err := st.TryAcquireExecutionSlot(runID, owner, budget); err == nil { + return nil + } else if !errors.Is(err, store.ErrBudgetExceeded) { + return err // 存储故障：不放行（fail-closed，宁可拒绝执行） + } + g.mu.Lock() + if g.waiters >= g.queueMax { + g.mu.Unlock() + return ErrAdmissionQueueFull + } + g.waiters++ + ch := g.broadcast + g.mu.Unlock() + select { + case <-ch: + g.mu.Lock() + g.waiters-- + g.mu.Unlock() + case <-ctx.Done(): + g.mu.Lock() + g.waiters-- + g.mu.Unlock() + return ctx.Err() + } + } +} + +// release 释放执行许可并广播（幂等；预算关闭时零开销）。 +func (g *admissionGate) release(st *store.Store, runID string) { + if g == nil || g.budget().Off() { + return + } + if err := st.ReleaseExecutionSlot(runID); err != nil { + // 释放失败只影响账本整洁（下次复活即修正），不中断运行。 + log.Printf("[runmgr] admission release %s: %v", runID, err) + } + g.mu.Lock() + close(g.broadcast) + g.broadcast = make(chan struct{}) + g.mu.Unlock() +} diff --git a/internal/runmgr/admission_test.go b/internal/runmgr/admission_test.go new file mode 100644 index 0000000..6172587 --- /dev/null +++ b/internal/runmgr/admission_test.go @@ -0,0 +1,140 @@ +package runmgr + +import ( + "context" + "errors" + "testing" + "time" + + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// 协作存储 U3 切片 2（R8）：执行预算在 runmgr 侧的行为面—— +// 子代理计入全局预算、父让出避免父子互等、队列有界（429 拒绝面）。 + +// 同步子代理：父占满全局预算仍能派生前进（父让出 → 子获取 → 子完成 → +// 父重取），不发生互等死锁。 +func TestSyncSpawnUnderSaturatedBudgetMakesProgress(t *testing.T) { + runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "子结果"}}} + c, st := newSpawnEnv(t, runner) + ws := t.TempDir() + c.mgr.SetWorkspaceRoot(ws) + c.mgr.SetExecutionBudgetFn(func() store.ExecutionBudget { return store.ExecutionBudget{Global: 1} }, 10) + + parentRun := c.run + if err := st.CreateRun(&parentRun); err != nil { + t.Fatal(err) + } + // 父先占住唯一的执行许可（等价于一个正在执行的父 run）。 + if err := st.TryAcquireExecutionSlot(string(parentRun.ID), parentRun.Owner, store.ExecutionBudget{Global: 1}); err != nil { + t.Fatal(err) + } + + done := make(chan error, 1) + go func() { + _, err := c.mgr.spawnSubagent(context.Background(), parentRun, "agent.sub.explore", "检索", "任务", "") + done <- err + }() + select { + case err := <-done: + if err != nil { + t.Fatalf("饱和预算下同步派生应前进：%v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("父满额派生死锁：子代理 10s 未完成（让出语义失效）") + } + // 父重取成功：租约复活。 + if held, _, err := st.ActiveExecutionSlots(); err != nil || held != 1 { + t.Fatalf("父应重取唯一许可：held=%d err=%v", held, err) + } +} + +// 后台子代理计入全局预算：预算 1 时第二个后台子代理排队，第一个完成后 +// 才执行（不超卖）。 +func TestBackgroundChildrenShareBudget(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + runner := &gateRunner{entered: entered, release: release, parts: []core.Part{{Type: core.PartText, Text: "结果"}}} + c, st := newSpawnEnv(t, runner) + ws := t.TempDir() + c.mgr.SetWorkspaceRoot(ws) + c.mgr.SetExecutionBudgetFn(func() store.ExecutionBudget { return store.ExecutionBudget{Global: 1} }, 10) + + parentRun := c.run + if err := st.CreateRun(&parentRun); err != nil { + t.Fatal(err) + } + first, err := c.mgr.spawnBackground(parentRun, "agent.sub.explore", "A", "任务A", "") + if err != nil { + t.Fatal(err) + } + <-entered // 第一个子代理已进入执行（持有唯一许可）。 + second, err := c.mgr.spawnBackground(parentRun, "agent.sub.explore", "B", "任务B", "") + if err != nil { + t.Fatal(err) + } + // 第二个应处于排队（无活跃租约可用）——账本里只有 first。 + if held, _, _ := st.ActiveExecutionSlots(); held != 1 { + t.Fatalf("预算 1 时不应有两个活跃许可：%d", held) + } + close(release) + // 第一个完成后第二个获得许可并执行完毕（entered 的 once 已用掉，用 + // 账本最终归零 + 等待窗口断言）。 + deadline := time.Now().Add(10 * time.Second) + for { + held, _, _ := st.ActiveExecutionSlots() + if held == 0 && c.mgr.QueueDepth() == 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("第二个后台子代理未完成：held=%d queue=%d", held, c.mgr.QueueDepth()) + } + time.Sleep(50 * time.Millisecond) + } + _ = first + _ = second +} + +// 队列有界：预算占满、队列达上限后 QueueFull() 为真（API 边缘 429 拒绝面）， +// acquire 返回 ErrAdmissionQueueFull；释放后排队者获得许可。 +func TestAdmissionQueueBounded(t *testing.T) { + st := openTestStoreQuiet(t) + g := newAdmissionGate(func() store.ExecutionBudget { return store.ExecutionBudget{Global: 1} }, 1) + ctx := context.Background() + if err := g.acquire(ctx, st, "run_hold", "o"); err != nil { + t.Fatal(err) + } + queued := make(chan error, 1) + go func() { queued <- g.acquire(ctx, st, "run_wait", "o") }() + // 等待排队者入队（深度 1）。 + deadline := time.Now().Add(5 * time.Second) + for g.QueueDepth() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if !g.QueueFull() { + t.Fatal("队列达上限应 QueueFull") + } + if err := g.acquire(ctx, st, "run_over", "o"); !errors.Is(err, ErrAdmissionQueueFull) { + t.Fatalf("超队列应 ErrAdmissionQueueFull：%v", err) + } + g.release(st, "run_hold") + if err := <-queued; err != nil { + t.Fatalf("排队者应获得许可：%v", err) + } + g.release(st, "run_wait") +} + +// openTestStoreQuiet：runmgr 测试包里建一个不迁移断言的临时库（沿用 +// recovery_test 的 openRecoveryStore 会走双档；这里只需本地档）。 +func openTestStoreQuiet(t *testing.T) *store.Store { + t.Helper() + st, err := store.OpenSQLite(t.TempDir() + "/adm.db") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +var _ = core.Run{} // 保持 import 锚（newSpawnEnv 依赖 core） diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index f86f5ef..dfcaef2 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -70,6 +70,7 @@ type Manager struct { // 会话内串行（设计并发策略）：一个会话同时至多一个在途 conversation run。 convMu sync.Mutex convActive map[string]bool // conversationID → 有在途 run + adm *admissionGate // 执行许可门（U3：预算默认关闭时零开销） instanceID string shuttingDown atomic.Bool @@ -125,6 +126,7 @@ func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service, active: map[string]*runHandle{}, convActive: map[string]bool{}, notices: map[string][]bgNotice{}, bgInFlight: map[string]int{}, delivering: map[string]bool{}, events: newEventBuffer(st, b), + adm: newAdmissionGate(nil, 0), } } @@ -171,6 +173,27 @@ func (m *Manager) SetCommandSource(src CommandSource) { m.commands = src } // SetIndexer 注入检索索引入口（分块+全文+可选向量；nil = 整文档单块）。 func (m *Manager) SetIndexer(i core.DocumentIndexer) { m.indexer = i } +// SetExecutionBudgetFn 注入执行预算读取（U3，R8）：fn 返回当前预算（全零= +// 不限，本地档默认）；queueMax 为待执行队列上限（<=0 走默认 200）。 +func (m *Manager) SetExecutionBudgetFn(fn func() store.ExecutionBudget, queueMax int) { + if m.adm == nil { + m.adm = newAdmissionGate(fn, queueMax) // 直接构造的 Manager（测试） + return + } + m.adm.mu.Lock() + m.adm.budgetFn = fn + if queueMax > 0 { + m.adm.queueMax = queueMax + } + m.adm.mu.Unlock() +} + +// QueueFull 非阻塞探测待执行队列是否已满（API 边缘 429 拒绝，不落库）。 +func (m *Manager) QueueFull() bool { return m.adm.QueueFull() } + +// QueueDepth 当前排队数（诊断）。 +func (m *Manager) QueueDepth() int { return m.adm.QueueDepth() } + // SetWorkspaceRoot 注入 fs 预留能力（Hidden 工具族）的沙箱根目录。 // 实际会话沙箱为 ///；空值=未配置（工具执行报错）。 func (m *Manager) SetWorkspaceRoot(dir string) { m.workspaceRoot = dir } @@ -553,6 +576,14 @@ func (c *runCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) { if card.TimeoutSec != nil && *card.TimeoutSec > 0 { timeout = time.Duration(*card.TimeoutSec) * time.Second } + // HITL 等待让出执行许可（U3：等待审批不占执行额度；应答后重取——可能 + // 排队，取消语境下重取失败由取消路径收口）。 + c.mgr.adm.release(c.mgr.st, string(c.run.ID)) + defer func() { + if err := c.mgr.adm.acquire(c.ctx, c.mgr.st, string(c.run.ID), c.run.Owner); err != nil { + log.Printf("[runmgr] admission reacquire after HITL %s: %v", c.run.ID, err) + } + }() return waitHITL(c.ctx, c.h.hitlCh, timeout) } @@ -1145,7 +1176,22 @@ func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Messa PersistInputTexts(m.workspaceRoot, conv.Owner, conv.ID, attachments) inputPaths = ListWorkspaceInputs(m.workspaceRoot, conv.Owner, conv.ID, inputPaths) rc := &runCtx{mgr: m, run: run, prof: prof, eff: eff, ctx: ctx, h: h, userMessage: userMessage, attachments: attachments, inputPaths: inputPaths, launch: launch, files: filestate.New(), permissionMode: conversationMode(m, run.Owner, run.ConversationID)} - go m.execute(rc, runner) + go func() { + // 执行许可（U3，R8）：排队有界；获取失败（队列满/取消/存储故障）以 + // failed 收口——不执行任何模型调用，副作用无从产生。 + if err := m.adm.acquire(rc.ctx, m.st, run.ID, run.Owner); err != nil { + log.Printf("[runmgr] admission denied run %s: %v", run.ID, err) + m.setRunStatus(&run, core.RunFailed, core.PhaseThinking) + if uErr, ok := err.(error); ok { + m.st.UpdateRun(&core.Run{ID: run.ID, Status: core.RunFailed, Error: &core.ErrorPart{Code: "run.admission_denied", Message: uErr.Error()}}) + } + m.emit(run.ConversationID, run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.admission_denied", Message: "执行预算排队已满或已取消"}) + m.releaseConversation(run.ConversationID) + return + } + defer m.adm.release(m.st, run.ID) + m.execute(rc, runner) + }() return run, nil } @@ -1455,6 +1501,22 @@ func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agen if err != nil { return core.AgentResultPayload{}, err } + // 父子互等规避（U3，R8）：同步子代理阻塞父 run——父先让出执行许可，子 + // 获取自己的许可（与父同池计入全局预算），子完成后父重取（可能排队； + // 取消语境下重取失败仅记录，run 由取消路径收口）。 + m.adm.release(m.st, string(parent.ID)) + defer func() { + if err := m.adm.acquire(parentCtx, m.st, string(parent.ID), parent.Owner); err != nil { + log.Printf("[runmgr] admission reacquire after child %s: %v", parent.ID, err) + } + }() + if err := m.adm.acquire(parentCtx, m.st, rc.run.ID, rc.run.Owner); err != nil { + m.setRunStatus(&rc.run, core.RunFailed, rc.run.Phase) + m.st.UpdateRun(&core.Run{ID: rc.run.ID, Status: core.RunFailed, Error: &core.ErrorPart{Code: "run.admission_denied", Message: err.Error()}}) + m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.admission_denied", Message: "子代理执行许可排队已满或已取消"}) + return core.AgentResultPayload{}, err + } + defer m.adm.release(m.st, rc.run.ID) return m.finishChild(rc, runner, true) } @@ -1602,6 +1664,16 @@ func (m *Manager) spawnBackground(parent core.Run, agentID, name, task, hint str bgCard := m.createBackgroundCard(parent, rc.run.ID, name, task) go func() { defer m.decBackground(parent.ConversationID) + // 后台子代理同样计入执行预算（U3：子代理预算计入全局）。 + if err := m.adm.acquire(context.Background(), m.st, rc.run.ID, parent.Owner); err != nil { + log.Printf("[runmgr] admission denied background %s: %v", rc.run.ID, err) + m.setRunStatus(&rc.run, core.RunFailed, rc.run.Phase) + m.st.UpdateRun(&core.Run{ID: rc.run.ID, Status: core.RunFailed, Error: &core.ErrorPart{Code: "run.admission_denied", Message: err.Error()}}) + m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.admission_denied", Message: "后台子代理执行许可排队已满或已取消"}) + m.finishBackgroundCard(parent.Owner, bgCard, core.AgentResultPayload{}, err) + return + } + defer m.adm.release(m.st, rc.run.ID) payload, err := m.finishChild(rc, runner, false) m.finishBackgroundCard(parent.Owner, bgCard, payload, err) m.enqueueNotice(parent.ConversationID, bgNotice{owner: parent.Owner, name: name, payload: payload}) diff --git a/internal/server/action_matrix_test.go b/internal/server/action_matrix_test.go new file mode 100644 index 0000000..dec7d0d --- /dev/null +++ b/internal/server/action_matrix_test.go @@ -0,0 +1,89 @@ +package server + +import ( + "testing" + + core "agent-system-core" +) + +// 协作存储 U3 切片 2（KTD13）：动作授权矩阵。未配置的动作默认拒绝； +// 旧只读 EXT-8 适配器只保留 read 能力；action-aware 适配器注册后权威。 + +func newMatrixServer(t *testing.T) *Server { + t.Helper() + s := &Server{} + t.Cleanup(func() {}) + return s +} + +// 无任何适配器：全部动作拒绝（默认拒绝；与 EXT-8 未注册时读取 NotFound 同基调）。 +func TestActionMatrixDefaultDeny(t *testing.T) { + s := newMatrixServer(t) + for _, a := range []ProjectAction{ProjectRead, ProjectEdit, ProjectApprove, ProjectExecute, ProjectAdmin} { + if s.authorizeProjectAction("alice", "p1", a) { + t.Fatalf("未注册适配器时 %s 应默认拒绝", a) + } + } +} + +// 旧只读适配器：read 放行、其余动作不得自动获得写入能力（KTD13 兼容语义）。 +func TestActionMatrixLegacyAdapterReadOnly(t *testing.T) { + s := newMatrixServer(t) + s.SetProjectAuthorizer(func(owner string, projectID core.ID) (bool, error) { + return owner == "alice", nil + }) + if !s.authorizeProjectAction("alice", "p1", ProjectRead) { + t.Fatal("alice read 应放行（旧适配器）") + } + for _, a := range []ProjectAction{ProjectEdit, ProjectApprove, ProjectExecute, ProjectAdmin} { + if s.authorizeProjectAction("alice", "p1", a) { + t.Fatalf("旧只读适配器不得自动获得 %s 能力", a) + } + } + // 他人（动作越权）：read 同样拒绝。 + if s.authorizeProjectAction("bob", "p1", ProjectRead) { + t.Fatal("bob 越权 read 应拒绝") + } +} + +// action-aware 适配器：注册后权威（read 与显式配置的动作放行，未配置动作 +// 默认拒绝；出错 fail-closed）。 +func TestActionMatrixActionAwareAuthoritative(t *testing.T) { + s := newMatrixServer(t) + // 同时注册旧适配器：action-aware 存在时不得回落旧适配器（权威唯一）。 + s.SetProjectAuthorizer(func(owner string, projectID core.ID) (bool, error) { + return true, nil + }) + s.SetProjectActionAuthorizer(func(owner string, projectID core.ID, action ProjectAction) (bool, error) { + switch action { + case ProjectRead, ProjectEdit: + return owner == "alice", nil + default: + return false, nil // approve/execute/admin 未授予 + } + }) + if !s.authorizeProjectAction("alice", "p1", ProjectRead) || !s.authorizeProjectAction("alice", "p1", ProjectEdit) { + t.Fatal("alice read/edit 应放行") + } + for _, a := range []ProjectAction{ProjectApprove, ProjectExecute, ProjectAdmin} { + if s.authorizeProjectAction("alice", "p1", a) { + t.Fatalf("未授予的 %s 应默认拒绝", a) + } + } + if s.authorizeProjectAction("bob", "p1", ProjectEdit) { + t.Fatal("bob 越权 edit 应拒绝") + } + // 适配器出错：fail-closed（错误不区分，避免泄露存在性）。 + s.SetProjectActionAuthorizer(func(owner string, projectID core.ID, action ProjectAction) (bool, error) { + return false, errBoom + }) + if s.authorizeProjectAction("alice", "p1", ProjectRead) { + t.Fatal("适配器出错应 fail-closed") + } +} + +var errBoom = &boomErr{} + +type boomErr struct{} + +func (*boomErr) Error() string { return "boom" } diff --git a/internal/server/server.go b/internal/server/server.go index b8d9332..8c8dc60 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -63,13 +63,51 @@ type Server struct { // ProjectAuthz EXT-8 资源授权适配器（组合根注入；nil=未注册，读取入口 // owner 自有判定未命中即 NotFound——与未扩展基座行为一致）。 ProjectAuthz func(owner string, projectID core.ID) (bool, error) + // ProjectAuthzActions 动作级授权适配器（协作存储 U3，KTD13）：注册后对 + // 全部动作权威；未注册时 read 回落 ProjectAuthz（兼容），其余动作一律 + // 默认拒绝——旧只读适配器不得自动获得写入能力。公开契约的版本化扩展 + // 计划见 docs/guides/team-storage.md（KTD13 一节）。 + ProjectAuthzActions func(owner string, projectID core.ID, action ProjectAction) (bool, error) } +// ProjectAction 项目资源的动作授权矩阵（KTD13）。共享可读不等于可写或可 +// 批准：动作语义独立判定，未配置的动作默认拒绝。 +type ProjectAction string + +const ( + ProjectRead ProjectAction = "read" // 读取项目资源（会话/消息/检索/工件下载） + ProjectEdit ProjectAction = "edit" // 写入与修改（会话内发消息、卡片裁决、设置变更） + ProjectApprove ProjectAction = "approve" // 审批（HITL 确认、命令确认） + ProjectExecute ProjectAction = "execute" // 触发执行（run、工具、工作流） + ProjectAdmin ProjectAction = "admin" // 管理（成员、共享范围、销毁性操作） +) + // SetProjectAuthorizer 组合根在装载扩展后注入 EXT-8 适配器。 func (s *Server) SetProjectAuthorizer(fn func(owner string, projectID core.ID) (bool, error)) { s.ProjectAuthz = fn } +// SetProjectActionAuthorizer 组合根注入 KTD13 动作级适配器（版本化公开契约 +// 扩展落地前，宿主侧先以本槽位提供矩阵语义）。 +func (s *Server) SetProjectActionAuthorizer(fn func(owner string, projectID core.ID, action ProjectAction) (bool, error)) { + s.ProjectAuthzActions = fn +} + +// authorizeProjectAction 动作矩阵唯一判定（KTD13）： +// - action-aware 适配器已注册 → 权威（出错/拒绝一律 false，fail-closed）； +// - 未注册 → read 回落旧只读 ProjectAuthz（兼容 EXT-8 现网），其余动作 +// 默认拒绝（旧适配器不得自动获得写入能力）。 +func (s *Server) authorizeProjectAction(owner string, projectID core.ID, action ProjectAction) bool { + if s.ProjectAuthzActions != nil { + ok, err := s.ProjectAuthzActions(owner, projectID, action) + return err == nil && ok + } + if action == ProjectRead { + return s.authorizeSharedProject(owner, projectID) + } + return false +} + // authorizeSharedProject 读取路径的共享判定：适配器未注册或拒绝/出错一律 false // （fail-closed；错误不区分，避免泄露存在性）。 func (s *Server) authorizeSharedProject(owner string, projectID core.ID) bool { @@ -767,6 +805,12 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) { } claimed := false if routeKind == route.KindTurn || routeKind == route.KindWorkflow { + // 执行许可预检（U3，R8）：待执行队列已满时 429 且不写任何数据 + //（与忙 409 同位；排队中的接纳不受影响）。 + if s.mgr.QueueFull() { + http.Error(w, "execution queue full（runtime.execQueueMax）", http.StatusTooManyRequests) + return + } if err := s.mgr.ClaimConversation(convID); err != nil { writeStoreErr(w, err) return @@ -1613,6 +1657,10 @@ func writeStoreErr(w http.ResponseWriter, err error) { http.Error(w, err.Error(), http.StatusConflict) return } + if errors.Is(err, runmgr.ErrAdmissionQueueFull) { + http.Error(w, err.Error(), http.StatusTooManyRequests) + return + } http.Error(w, err.Error(), http.StatusInternalServerError) } diff --git a/internal/store/admission.go b/internal/store/admission.go new file mode 100644 index 0000000..c9393fb --- /dev/null +++ b/internal/store/admission.go @@ -0,0 +1,97 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" +) + +// 协作存储 U3 切片 2（R8/KTD「接纳与执行许可分别计数」）：执行许可的共享 +// 权威落在数据库（admission_leases），多实例天然共享；接纳数（排队中的 run） +// 由 runmgr 的有界队列管理，不进本表。预算关闭（全零）时不产生任何租约行。 + +// ExecutionBudget 执行许可预算。零值 = 不限（本地档默认，行为与历史一致）。 +type ExecutionBudget struct { + Global int // 全局并发执行上限（含子代理与索引任务）；0=不限 + PerUser int // 每租户并发执行上限；0=不限 +} + +// Off 表示预算完全关闭（不限流、不留租约）。 +func (b ExecutionBudget) Off() bool { return b.Global <= 0 && b.PerUser <= 0 } + +// ErrBudgetExceeded 预算已满（可排队）；ErrAdmissionQueueFull 是其排队也满的 +// 上层形态（runmgr 定义，映射 429）。 +var ErrBudgetExceeded = errors.New("store: execution budget exceeded") + +// TryAcquireExecutionSlot 原子获取（或重新激活）run 的执行许可：在 admission +// 咨询锁下清点活跃租约（全局 + 租户），不超预算才登记。父 run 让出后重新 +// 获取走同一路径（upsert 复活既有行）。 +func (s *Store) TryAcquireExecutionSlot(runID, owner string, budget ExecutionBudget) error { + if budget.Off() { + return nil + } + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if err := s.lockAdmission(tx); err != nil { + return err + } + var global, perUser int + if budget.Global > 0 { + if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM admission_leases WHERE released_at IS NULL AND run_id<>?`), runID).Scan(&global); err != nil { + return err + } + if global >= budget.Global { + return fmt.Errorf("exec slot %s: %w", runID, ErrBudgetExceeded) + } + } + if budget.PerUser > 0 { + if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM admission_leases WHERE released_at IS NULL AND owner=? AND run_id<>?`), owner, runID).Scan(&perUser); err != nil { + return err + } + if perUser >= budget.PerUser { + return fmt.Errorf("exec slot %s: %w", runID, ErrBudgetExceeded) + } + } + if _, err := tx.Exec(s.q(`INSERT INTO admission_leases (run_id, owner, acquired_at, released_at) VALUES (?,?,?,NULL) + ON CONFLICT(run_id) DO UPDATE SET owner=excluded.owner, acquired_at=excluded.acquired_at, released_at=NULL`), + runID, owner, nowISO()); err != nil { + return err + } + return tx.Commit() +} + +// ReleaseExecutionSlot 释放执行许可（幂等）：父 run 等待子代理/HITL 时让出， +// 恢复执行须重新获取（方案 KTD「让出执行许可、保留会话所有权」）。 +func (s *Store) ReleaseExecutionSlot(runID string) error { + _, err := s.db.Exec(s.q(`UPDATE admission_leases SET released_at=? WHERE run_id=? AND released_at IS NULL`), nowISO(), runID) + return err +} + +// ActiveExecutionSlots 返回活跃执行许可数（诊断/验收：全局与按租户）。 +func (s *Store) ActiveExecutionSlots() (global int, byOwner map[string]int, err error) { + byOwner = map[string]int{} + rows, err := s.db.Query(s.q(`SELECT owner, COUNT(*) FROM admission_leases WHERE released_at IS NULL GROUP BY owner`)) + if err != nil { + return 0, nil, err + } + defer rows.Close() + for rows.Next() { + var o string + var n int + if err := rows.Scan(&o, &n); err != nil { + return 0, nil, err + } + byOwner[o] = n + global += n + } + return global, byOwner, rows.Err() +} + +// lockAdmission 在事务内锁定执行许可账本（PG 事务级咨询锁；SQLite 单写者 +// 天然串行）。与 lockConversation/lockScope 同模式（dialect 收敛）。 +func (s *Store) lockAdmission(tx *sql.Tx) error { + return s.dialect.lockAdmission(tx) +} diff --git a/internal/store/admission_pg_test.go b/internal/store/admission_pg_test.go new file mode 100644 index 0000000..321bce5 --- /dev/null +++ b/internal/store/admission_pg_test.go @@ -0,0 +1,110 @@ +package store + +import ( + "context" + "errors" + "sync" + "testing" +) + +// 协作存储 U3 切片 2（R8）：执行许可账本。预算判定原子性、让出/复活、 +// 并发恰 N 个赢家。双档（PG 由 CI 真库 job 覆盖咨询锁与部分索引）。 + +func TestExecutionBudgetEnforced(t *testing.T) { + st := openTestStore(t) + budget := ExecutionBudget{Global: 2} + if err := st.TryAcquireExecutionSlot("run_1", "o", budget); err != nil { + t.Fatal(err) + } + if err := st.TryAcquireExecutionSlot("run_2", "o", budget); err != nil { + t.Fatal(err) + } + err := st.TryAcquireExecutionSlot("run_3", "o", budget) + if !errors.Is(err, ErrBudgetExceeded) { + t.Fatalf("global=2 时第三个应拒绝：%v", err) + } + // 释放一个 → 第三个可进。 + if err := st.ReleaseExecutionSlot("run_1"); err != nil { + t.Fatal(err) + } + if err := st.TryAcquireExecutionSlot("run_3", "o", budget); err != nil { + t.Fatal(err) + } + // 复活（父让出后重取）：释放 run_2 再复活 run_2。 + if err := st.ReleaseExecutionSlot("run_2"); err != nil { + t.Fatal(err) + } + if err := st.TryAcquireExecutionSlot("run_2", "o", budget); err != nil { + t.Fatal(err) + } + global, byOwner, err := st.ActiveExecutionSlots() + if err != nil || global != 2 || byOwner["o"] != 2 { + t.Fatalf("活跃账本：global=%d byOwner=%v err=%v", global, byOwner, err) + } + // 预算关闭：零开销直通、不留租约。 + if err := st.TryAcquireExecutionSlot("run_9", "o", ExecutionBudget{}); err != nil { + t.Fatal(err) + } +} + +func TestExecutionBudgetPerUser(t *testing.T) { + st := openTestStore(t) + budget := ExecutionBudget{PerUser: 1} + if err := st.TryAcquireExecutionSlot("run_a", "alice", budget); err != nil { + t.Fatal(err) + } + if err := st.TryAcquireExecutionSlot("run_b", "bob", budget); err != nil { + t.Fatal(err) + } + err := st.TryAcquireExecutionSlot("run_c", "alice", budget) + if !errors.Is(err, ErrBudgetExceeded) { + t.Fatalf("perUser=1 时 alice 第二个应拒绝：%v", err) + } + // 同 run 重取不占新额度（复活语义）。 + if err := st.TryAcquireExecutionSlot("run_a", "alice", budget); err != nil { + t.Fatal(err) + } +} + +// 并发获取恰 N 个赢家：预算 3、12 个竞争者 → 3 成功 9 拒绝（原子清点， +// 不超卖；PG 路径验证咨询锁串行化）。 +func TestExecutionBudgetConcurrentNoOversell(t *testing.T) { + st := openTestStore(t) + budget := ExecutionBudget{Global: 3} + const racers = 12 + var mu sync.Mutex + ok, denied := 0, 0 + var wg sync.WaitGroup + for i := 0; i < racers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + err := st.TryAcquireExecutionSlot(newID("run"), "o"+string(rune('a'+i%3)), budget) + mu.Lock() + defer mu.Unlock() + switch { + case err == nil: + ok++ + case errors.Is(err, ErrBudgetExceeded): + denied++ + default: + t.Errorf("意外错误：%v", err) + } + }(i) + } + wg.Wait() + if ok != 3 || denied != racers-3 { + t.Fatalf("应恰 3 成功：ok=%d denied=%d", ok, denied) + } +} + +// 迁移到位验收（v21 表存在）由 TestMigrationUpgradesExistingDatabase 的 +// 全量账本断言覆盖；此处补空库→全量后账本可清点的最小回归。 +func TestAdmissionLeasesTablePresent(t *testing.T) { + st := openTestStore(t) + global, _, err := st.ActiveExecutionSlots() + if err != nil || global != 0 { + t.Fatalf("空账本：global=%d err=%v", global, err) + } + _ = context.Background() +} diff --git a/internal/store/dialect.go b/internal/store/dialect.go index babc5db..8e3abd8 100644 --- a/internal/store/dialect.go +++ b/internal/store/dialect.go @@ -15,6 +15,8 @@ type dialect interface { lockConversation(tx *sql.Tx, conversationID string) error // lockScope 锁定设置 scope 的读改写（owner+scope）。 lockScope(tx *sql.Tx, owner, scope string) error + // lockAdmission 锁定执行许可账本的清点与登记（U3 admission）。 + lockAdmission(tx *sql.Tx) error } // ---- SQLite（嵌入式档）---- @@ -25,6 +27,7 @@ func (sqliteDialect) name() string { return "sqlite" func (sqliteDialect) rebind(q string) string { return q } func (sqliteDialect) lockConversation(*sql.Tx, string) error { return nil } func (sqliteDialect) lockScope(*sql.Tx, string, string) error { return nil } +func (sqliteDialect) lockAdmission(*sql.Tx) error { return nil } // ---- PostgreSQL（多用户/多实例档）---- @@ -55,3 +58,8 @@ func (postgresDialect) lockScope(tx *sql.Tx, owner, scope string) error { _, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext($1))`, "scope:"+owner+"/"+scope) return err } + +func (postgresDialect) lockAdmission(tx *sql.Tx) error { + _, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext($1))`, "admission:exec-slots") + return err +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index ee760d0..600daa7 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -486,6 +486,32 @@ func init() { }, "postgres": { `DROP TABLE IF EXISTS recovery_holds`, }}}, + migration{version: 21, name: "admission_leases", + // 执行许可账本（协作存储 U3 切片 2，R8）：预算开启时每个在执行的 + // run（会话回合与子代理同池）持有一行活跃租约；父等待让出后 + // 复活。预算关闭（全零）不产生行，历史行为不变。 + up: map[string][]string{"sqlite": { + `CREATE TABLE IF NOT EXISTS admission_leases ( + run_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + acquired_at TEXT NOT NULL, + released_at TEXT + )`, + `CREATE INDEX IF NOT EXISTS idx_admission_leases_open ON admission_leases(owner) WHERE released_at IS NULL`, + }, "postgres": { + `CREATE TABLE IF NOT EXISTS admission_leases ( + run_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + acquired_at TEXT NOT NULL, + released_at TEXT + )`, + `CREATE INDEX IF NOT EXISTS idx_admission_leases_open ON admission_leases(owner) WHERE released_at IS NULL`, + }}, + down: map[string][]string{"sqlite": { + `DROP TABLE IF EXISTS admission_leases`, + }, "postgres": { + `DROP TABLE IF EXISTS admission_leases`, + }}}, ) }