Add shared execution budgets and the action-authz matrix (U3 slice 2) #141
@@ -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() })
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
@@ -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 工具族)的沙箱根目录。
|
||||
// 实际会话沙箱为 <root>/<owner>/<conversationID>/;空值=未配置(工具执行报错)。
|
||||
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})
|
||||
|
||||
@@ -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" }
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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`,
|
||||
}}},
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user