Merge pull request 'Add shared execution budgets and the action-authz matrix (U3 slice 2)' (#141) from feature/collab-storage-u3-budget-authz into main

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() })

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()
+}

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）

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 工具族）的沙箱根目录。
 // 实际会话沙箱为 <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})

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" }

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)
 }
 

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)
+}

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()
+}

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
+}

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`,
+			}}},
 	)
 }