weichenandCursor
2424f8e3f7
Why: main advanced with agent-stack audit batch-1; resolve the Bus struct conflict by keeping both hostProjectGuard and operationQuery. Evidence: go test ./internal/cmdbus ./internal/app ./internal/runmgr ./internal/server ./internal/store pass locally (SELF). Impact: scratch host guard retained alongside the new operation query port; deps gitlinks match origin/main. Refs: Agent-System-Backend#241 Co-authored-by: Cursor <cursoragent@cursor.com>
848 lines
34 KiB
Go
848 lines
34 KiB
Go
// Package cmdbus 组合 Agent-Command 生态库：注册、四入口分发（button/card/slash/chat→api
|
||
// 归一）、审计信封落盘与危险命令的内联确认卡收口（提案 T3 / D27、D30、D33）。
|
||
// 与 intentgate 之于 agent-intent 同构：schema 权威在生态库，宿主只做装配与执行适配。
|
||
package cmdbus
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
agentcommand "agent-command"
|
||
core "agent-system-core"
|
||
)
|
||
|
||
// AuditStore 审计落盘端口（store 实现；v7 建表、v18 加 owner 列）。
|
||
// owner 是请求租户（req.Owner）：审计行按租户归属，查询按 owner 过滤，
|
||
// 跨租户审计不可见（信封 Actor 是发起主体，不作归属依据）。
|
||
type AuditStore interface {
|
||
AppendAudit(owner string, env agentcommand.AuditEnvelope) error
|
||
}
|
||
|
||
// OperationRecord 是"已记录操作"的最小对账投影（R7/KTD6）。它回答的是
|
||
// “这个 operationID 在本 owner/project 名下到底是什么结果”，而不是重放入口。
|
||
// 刻意保持最小：审计信封只存状态与标识，不存理由文本，所以这里没有 Reason——
|
||
// 不为不存在的字段留位置。
|
||
type OperationRecord struct {
|
||
OperationID string
|
||
CommandID string
|
||
State agentcommand.ResultState
|
||
Time time.Time
|
||
}
|
||
|
||
// OperationQuery 是 operationID 的对账查询端口（组合根装配；nil = 不提供）。
|
||
//
|
||
// 授权口径：owner 与 projectID 都是可信过滤条件，实现必须把它们放进查询谓词
|
||
// （SQL 的 owner/project 等值条件），不得由调用方参数扩大可见范围；查不到就
|
||
// 返回 not found，不得回退为“未执行”。
|
||
//
|
||
// 诚实边界：本端口只解析“已经落成记录”的 operation。宿主侧的命令回执记录就是
|
||
// 命令审计信封（command_audit 按 owner+project 归属）；handler 自己的领域操作表
|
||
// （如文档投递 document_render_deliveries）由各自领域查询负责，不经本端口。某些
|
||
// 命令根本不持久化 req.OperationID（内置的 builtin.help、runtime.mock、
|
||
// models.set、conversation.profile、conversation.create 都是无操作表的即时用例），
|
||
// 对它们的 operationID 查询恒为 not found——这类命令只能按领域资源（会话、画像、
|
||
// 设置项）对账，不得声称可按 operationID 查询。
|
||
type OperationQuery interface {
|
||
LookupOperation(owner, projectID, operationID string) (OperationRecord, bool, error)
|
||
}
|
||
|
||
var (
|
||
// ErrOperationScope 缺少可信 owner/project/operation 作用域：fail-closed，无作用域即无可见性。
|
||
ErrOperationScope = errors.New("cmdbus: operation lookup requires owner, project and operation scope")
|
||
// ErrOperationNotRecorded 该作用域内没有此 operation 的记录（未记录 ≠ 未执行）。
|
||
ErrOperationNotRecorded = errors.New("cmdbus: no operation record in this owner and project scope")
|
||
// ErrOperationQueryUnavailable 组合根未装配对账查询端口。
|
||
ErrOperationQueryUnavailable = errors.New("cmdbus: operation query unavailable")
|
||
)
|
||
|
||
// CardPort 确认卡所需卡片能力（cardsvc.Service 实现）。
|
||
type CardPort interface {
|
||
Create(owner string, card core.ActivityCard, ref string) (core.ActivityCard, error)
|
||
SetStatus(owner, id string, status core.CardStatus, tone, note string) (core.ActivityCard, error)
|
||
}
|
||
|
||
// Handler 宿主执行适配器：同步内置用例返回 payload；异步执行自行返回 accepted 回执。
|
||
type Handler func(ctx context.Context, req Request) (any, error)
|
||
|
||
// ErrNotExecuted 标记"失败确定发生在任何副作用之前"的 handler 错误（F10/U7b）。
|
||
// 语义口径：只有带此标记的错误才可归 rejected（确定未执行、无副作用、回执不得
|
||
// 携带 operationID）；未标记的 handler 错误一律按"已执行随后不确定"归 unknown，
|
||
// 并带上本次分发已签发的 trusted operation ID 交由对账收敛。宿主不得据
|
||
// "处理方报了错"臆断未执行——那会把已写库的操作诱使成不安全重试。
|
||
//
|
||
// 进程内插件（pkg/extension.Handler）当前还没有等价的声明位：插件命令的
|
||
// handler 错误一律按 unknown 对账，直到扩展面补上对应标记为止。
|
||
var ErrNotExecuted = errors.New("cmdbus: not executed")
|
||
|
||
// NotExecutedError 是带"确定未执行"声明的错误包装。Error() 保留原始文案
|
||
// （回执 reason 不因标记而变形），Unwrap/Is 让 errors.Is 同时匹配
|
||
// ErrNotExecuted 与被包装的原始错误。
|
||
type NotExecutedError struct{ Err error }
|
||
|
||
func (e *NotExecutedError) Error() string { return e.Err.Error() }
|
||
func (e *NotExecutedError) Unwrap() error { return e.Err }
|
||
func (e *NotExecutedError) Is(target error) bool {
|
||
return target == ErrNotExecuted
|
||
}
|
||
|
||
// NotExecuted 声明该失败确定未执行（校验/权限/确认缺失/领域事务回滚等任何
|
||
// 副作用之前的拒绝）。nil 原样返回；重复标记幂等。
|
||
func NotExecuted(err error) error {
|
||
if err == nil || IsNotExecuted(err) {
|
||
return err
|
||
}
|
||
return &NotExecutedError{Err: err}
|
||
}
|
||
|
||
// IsNotExecuted 报告该错误是否声明了"确定未执行"。
|
||
func IsNotExecuted(err error) bool { return errors.Is(err, ErrNotExecuted) }
|
||
|
||
// CommandTool EXT-14：带 exposeAsTool 声明的命令的工具投影视图（runmgr 消费）。
|
||
// Risk 取 low|medium|high（工具权限声明口径）；ID 即工具名（命令 id）。
|
||
type CommandTool struct {
|
||
ID string
|
||
Title string
|
||
Description string
|
||
Params json.RawMessage
|
||
Risk string
|
||
}
|
||
|
||
// CommandTools 全部带 ExposeAsTool 的已启用命令（EXT-14 工具投影源）。
|
||
// ToolGate 是 owner 级工具注入闸（U3：禁用即不注入能力）：返回 false 时该命令
|
||
// 不投影给该 owner 的模型。组合根按需装配（如 generative-ui 按 activation 台账）。
|
||
type ToolGate func(owner, commandID string) bool
|
||
|
||
// EffectiveAccountGuard checks the current account state before a command can
|
||
// create a confirmation or invoke a handler. The composition root supplies the
|
||
// check; nil keeps standalone and auth-off fixtures unchanged. The actor is
|
||
// derived only from trusted Request identity fields, never Params. This is an
|
||
// early check, not an atomic fence across handler databases or external calls.
|
||
type EffectiveAccountGuard func(ctx context.Context, actor string) error
|
||
|
||
// SetToolGate 装配 owner 级注入闸（组合根；启动期一次）。
|
||
func (b *Bus) SetToolGate(g ToolGate) { b.toolGate = g }
|
||
|
||
func (b *Bus) CommandTools() []CommandTool {
|
||
return b.CommandToolsFor("")
|
||
}
|
||
|
||
// CommandToolsFor 按 owner 过滤工具投影（gate 为 nil 时不过滤；owner 为空时
|
||
// gated 命令一律不投影——fail-closed）。
|
||
func (b *Bus) CommandToolsFor(owner string) []CommandTool {
|
||
out := []CommandTool{}
|
||
for _, r := range b.reg.List() {
|
||
if !r.Enabled || r.Spec.ExposeAsTool == nil {
|
||
continue
|
||
}
|
||
if b.toolGate != nil && !b.toolGate(owner, r.Spec.ID) {
|
||
continue
|
||
}
|
||
risk := r.Spec.ExposeAsTool.Risk
|
||
if risk == "" {
|
||
risk = "low"
|
||
}
|
||
desc := r.Spec.Description
|
||
if desc == "" {
|
||
desc = r.Spec.Title
|
||
}
|
||
out = append(out, CommandTool{ID: r.Spec.ID, Title: r.Spec.Title,
|
||
Description: desc, Params: r.Spec.ParamsSchema, Risk: risk})
|
||
}
|
||
return out
|
||
}
|
||
|
||
// Request 一次分发请求（入口归一后的统一形态）。
|
||
type Request struct {
|
||
ID string
|
||
Params map[string]any
|
||
Entrance string // button | card | slash | chat | api（归一）
|
||
Actor string
|
||
Owner string
|
||
ConversationID string
|
||
OperationID string // overwritten by the trusted bus before handler execution
|
||
AnchorMessageID string // 确认卡锚定的发起消息（内联渲染）
|
||
Confirmed bool // 危险命令经确认卡放行后的执行
|
||
HostProjectID string // EXT-18：分发所属基座项目（空＝无项目上下文）
|
||
// HumanGrant is set only by the authenticated session path or the human
|
||
// decision resolver. Client JSON and agent tool dispatch leave it nil.
|
||
HumanGrant *HumanGrant
|
||
}
|
||
|
||
// RequiresProjectWorkDecision identifies formal RDM writes that must be bound
|
||
// to an analyzed committed-message manifest before dispatch. Read views and
|
||
// the separate exact-card structure issue/decision path are excluded.
|
||
func RequiresProjectWorkDecision(id string) bool {
|
||
for _, prefix := range []string{
|
||
"rdm.reqs.", "rdm.review.", "rdm.draft.", "rdm.cnode.", "rdm.cedge.",
|
||
"rdm.conclusion.", "rdm.section.", "rdm.stage.", "rdm.change.",
|
||
} {
|
||
if strings.HasPrefix(id, prefix) {
|
||
for _, read := range []string{"rdm.reqs.list", "rdm.reqs.exports", "rdm.reqs.underived", "rdm.review.status", "rdm.review.doc.status", "rdm.draft.proposals", "rdm.conclusion.list", "rdm.change.list"} {
|
||
if id == read {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
}
|
||
switch id {
|
||
case "rdm.structure.propose", "rdm.structure.revise", "rdm.docs.instantiate", "rdm.docs.delivery.approve",
|
||
"rdm.doc.snapshot", "rdm.ingest", "rdm.upload.confirm", "rdm.knowledge.publish", "rdm.trace.confirm",
|
||
"rdm.conflicts.resolve":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// HumanGrant ties this dispatch to one persisted human response.
|
||
// It is not a secret token and is not read from command params.
|
||
type HumanGrant struct {
|
||
CardID string
|
||
ProofID string
|
||
Actor string
|
||
Decision string
|
||
}
|
||
|
||
// Response 分发结果：回执 + 处理负载 + 可能的确认卡。
|
||
type Response struct {
|
||
Receipt agentcommand.ExecReceipt
|
||
Payload any
|
||
Card *core.ActivityCard // 高危未确认时创建的内联确认卡
|
||
}
|
||
|
||
// pending 确认卡挂起的执行（进程内；重启后确认返回 rejected expired——v1 限制）。
|
||
type pending struct {
|
||
req Request
|
||
cardID string
|
||
}
|
||
|
||
type Bus struct {
|
||
reg *agentcommand.Registry
|
||
audit AuditStore
|
||
cards CardPort
|
||
now func() time.Time
|
||
handlers map[string]Handler
|
||
slash map[string]string // "/x" → command id
|
||
pendMu sync.Mutex
|
||
pending map[string]pending
|
||
pendingStore PendingStore
|
||
intents *IntentStore // G2-2：意图声明表（自然语言路由）
|
||
toolGate ToolGate // U3：owner 级工具注入闸（nil = 不过滤）
|
||
accountGuard EffectiveAccountGuard // A1: optional command execution precheck
|
||
strictAccountWrites bool
|
||
allowImplicitLocal bool
|
||
projectWorkPreflight func(context.Context, Request) error
|
||
hostProjectGuard func(context.Context, Request) error
|
||
// operationQuery operationID 对账查询端口（nil = 不提供，见 OperationQuery）。
|
||
operationQuery OperationQuery
|
||
}
|
||
|
||
func (b *Bus) SetProjectWorkPreflight(fn func(context.Context, Request) error) {
|
||
b.projectWorkPreflight = fn
|
||
}
|
||
|
||
// SetHostProjectGuard installs the composition-root check applied to every
|
||
// dispatch that carries a base project, before the project-work preflight and
|
||
// before any handler runs. It is the single bus-level place where host-project
|
||
// capability gates live, so every entrance (slash, exec, chat intent routing,
|
||
// system.intent re-dispatch, agent tool) is covered by one decision.
|
||
func (b *Bus) SetHostProjectGuard(fn func(context.Context, Request) error) {
|
||
b.hostProjectGuard = fn
|
||
}
|
||
|
||
// PendingStore 把危险命令的确认挂起态落盘。进程内 map 仍是热路径；重启后从存储取回。
|
||
type PendingStore interface {
|
||
SavePendingConfirm(cardID, owner, payload string) error
|
||
TakePendingConfirm(cardID string) (owner, payload string, ok bool, err error)
|
||
}
|
||
|
||
func New(audit AuditStore, cards CardPort, now func() time.Time) *Bus {
|
||
if now == nil {
|
||
now = time.Now
|
||
}
|
||
return &Bus{
|
||
reg: agentcommand.NewRegistry(), audit: audit, cards: cards, now: now,
|
||
handlers: map[string]Handler{}, slash: map[string]string{}, pending: map[string]pending{},
|
||
intents: NewIntentStore(),
|
||
}
|
||
}
|
||
|
||
// Registry 暴露注册表（列清单/插件 overlay 注册用）。
|
||
func (b *Bus) Registry() *agentcommand.Registry { return b.reg }
|
||
|
||
// SetPendingStore 注入确认挂起态的持久化实现（nil = 仅进程内）。
|
||
func (b *Bus) SetPendingStore(st PendingStore) { b.pendingStore = st }
|
||
|
||
// SetOperationQuery 装配 operationID 对账查询端口（组合根；启动期一次）。
|
||
func (b *Bus) SetOperationQuery(q OperationQuery) { b.operationQuery = q }
|
||
|
||
// LookupOperation 按 owner/project 作用域查一个已记录 operation 的当前状态
|
||
// （R7/KTD6）。这是 unknown 回执的对账入口：拿到 operationID 的调用方先查
|
||
// 记录，再决定下一步，而不是重发命令。
|
||
//
|
||
// 三重 fail-closed：作用域任一为空即拒（不猜、不按 owner=“”放行）、未装配端口
|
||
// 即拒、查不到即 ErrOperationNotRecorded（不推断“未执行”）。跨 owner 查询由
|
||
// 端口的 owner 等值谓词拒绝——本层只把 owner 当可信条件往下传。
|
||
func (b *Bus) LookupOperation(owner, projectID, operationID string) (OperationRecord, error) {
|
||
if owner == "" || projectID == "" || operationID == "" {
|
||
return OperationRecord{}, ErrOperationScope
|
||
}
|
||
if b.operationQuery == nil {
|
||
return OperationRecord{}, ErrOperationQueryUnavailable
|
||
}
|
||
record, found, err := b.operationQuery.LookupOperation(owner, projectID, operationID)
|
||
if err != nil {
|
||
return OperationRecord{}, err
|
||
}
|
||
if !found || record.OperationID == "" {
|
||
return OperationRecord{}, ErrOperationNotRecorded
|
||
}
|
||
return record, nil
|
||
}
|
||
|
||
// SetEffectiveAccountGuard installs the composition-root check at startup.
|
||
// Handler Store writes still need their own final transaction fence.
|
||
func (b *Bus) SetEffectiveAccountGuard(g EffectiveAccountGuard) { b.accountGuard = g }
|
||
|
||
// RequireAccountCheckedWrites makes every Bus-owned durable sink use a final
|
||
// account transaction. The composition root enables this after wiring Store
|
||
// and Cards; standalone in-memory fixtures retain their legacy ports.
|
||
func (b *Bus) RequireAccountCheckedWrites(allowImplicitLocal bool) {
|
||
b.strictAccountWrites = true
|
||
b.allowImplicitLocal = allowImplicitLocal
|
||
}
|
||
|
||
func (b *Bus) createConfirmCard(actor string, card core.ActivityCard, ref string) (core.ActivityCard, error) {
|
||
if !b.strictAccountWrites {
|
||
return b.cards.Create(actor, card, ref)
|
||
}
|
||
checked, ok := b.cards.(interface {
|
||
CreateAccountChecked(string, core.ActivityCard, string, bool) (core.ActivityCard, error)
|
||
})
|
||
if !ok {
|
||
return core.ActivityCard{}, fmt.Errorf("cmdbus: account-checked card port required")
|
||
}
|
||
return checked.CreateAccountChecked(actor, card, ref, b.allowImplicitLocal)
|
||
}
|
||
|
||
func (b *Bus) setConfirmCardStatus(actor, id string, status core.CardStatus, tone, note string) (core.ActivityCard, error) {
|
||
if !b.strictAccountWrites {
|
||
return b.cards.SetStatus(actor, id, status, tone, note)
|
||
}
|
||
checked, ok := b.cards.(interface {
|
||
SetStatusAccountChecked(string, string, core.CardStatus, string, string, bool) (core.ActivityCard, error)
|
||
})
|
||
if !ok {
|
||
return core.ActivityCard{}, fmt.Errorf("cmdbus: account-checked card port required")
|
||
}
|
||
return checked.SetStatusAccountChecked(actor, id, status, tone, note, b.allowImplicitLocal)
|
||
}
|
||
|
||
func (b *Bus) savePending(cardID, owner, payload string) error {
|
||
if !b.strictAccountWrites {
|
||
return b.pendingStore.SavePendingConfirm(cardID, owner, payload)
|
||
}
|
||
checked, ok := b.pendingStore.(interface {
|
||
SavePendingConfirmAccountChecked(string, string, string, bool) error
|
||
})
|
||
if !ok {
|
||
return fmt.Errorf("cmdbus: account-checked pending store required")
|
||
}
|
||
return checked.SavePendingConfirmAccountChecked(cardID, owner, payload, b.allowImplicitLocal)
|
||
}
|
||
|
||
func (b *Bus) takePending(cardID, owner string) (string, string, bool, error) {
|
||
if !b.strictAccountWrites {
|
||
return b.pendingStore.TakePendingConfirm(cardID)
|
||
}
|
||
checked, ok := b.pendingStore.(interface {
|
||
TakePendingConfirmAccountChecked(string, string, bool) (string, string, bool, error)
|
||
})
|
||
if !ok {
|
||
return "", "", false, fmt.Errorf("cmdbus: account-checked pending store required")
|
||
}
|
||
return checked.TakePendingConfirmAccountChecked(cardID, owner, b.allowImplicitLocal)
|
||
}
|
||
|
||
// RegisterBuiltin 登记内置命令并绑定宿主处理器与 `/` 别名。
|
||
func (b *Bus) RegisterBuiltin(spec agentcommand.CommandSpec, slash string, h Handler) error {
|
||
spec.Binding = agentcommand.Binding{Kind: agentcommand.BindBuiltin}
|
||
if err := b.reg.Register(spec, agentcommand.SourceBuiltin); err != nil {
|
||
return err
|
||
}
|
||
if h != nil {
|
||
b.handlers[spec.ID] = h
|
||
}
|
||
if slash != "" {
|
||
b.slash[strings.ToLower(slash)] = spec.ID
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// RegisterUser 登记用户提示词命令。重复登记视为已存在。执行不走总线，发送时展开模板。
|
||
func (b *Bus) RegisterUser(spec agentcommand.CommandSpec, slash string) error {
|
||
spec.Binding = agentcommand.Binding{Kind: agentcommand.BindMessage}
|
||
if err := b.reg.Register(spec, agentcommand.SourceUser); err != nil && !errors.Is(err, agentcommand.ErrDuplicate) {
|
||
return err
|
||
}
|
||
if slash != "" {
|
||
b.slash[strings.ToLower(slash)] = spec.ID
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UnregisterUser 卸下用户命令。不存在时忽略。
|
||
func (b *Bus) UnregisterUser(id, slash string) {
|
||
_ = b.reg.Unregister(id)
|
||
if slash != "" {
|
||
delete(b.slash, strings.ToLower(slash))
|
||
}
|
||
}
|
||
|
||
// RegisterPluginCommands 插件命令 overlay（manifest provides.commands；失败 fail-closed 返回）。
|
||
func (b *Bus) RegisterPluginCommands(overlay []json.RawMessage) error {
|
||
for _, raw := range overlay {
|
||
var spec agentcommand.CommandSpec
|
||
if err := json.Unmarshal(raw, &spec); err != nil {
|
||
return fmt.Errorf("cmdbus: plugin command spec: %w", err)
|
||
}
|
||
if err := b.reg.Register(spec, agentcommand.SourcePlugin); err != nil {
|
||
return fmt.Errorf("cmdbus: register %s: %w", spec.ID, err)
|
||
}
|
||
b.bindPluginSlash(spec)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// bindPluginSlash EXT-10②：插件命令规格自带的 slash 别名绑定（与内置命令同一张
|
||
// 别名表——四入口同源：slash 命中即同一命令 id、同一审计信封）。
|
||
func (b *Bus) bindPluginSlash(spec agentcommand.CommandSpec) {
|
||
if spec.Slash != "" {
|
||
b.slash[strings.ToLower(spec.Slash)] = spec.ID
|
||
}
|
||
}
|
||
|
||
// RegisterPluginCommand 注册进程内插件命令（EXT-1）：登记规格并绑定宿主处理器。
|
||
// 仅登记规格的声明式路径仍走 RegisterPluginCommands（无 handler，分发即 rejected）。
|
||
func (b *Bus) RegisterPluginCommand(spec agentcommand.CommandSpec, h Handler) error {
|
||
if err := b.reg.Register(spec, agentcommand.SourcePlugin); err != nil {
|
||
return fmt.Errorf("cmdbus: register %s: %w", spec.ID, err)
|
||
}
|
||
if h != nil {
|
||
b.handlers[spec.ID] = h
|
||
}
|
||
b.bindPluginSlash(spec)
|
||
return nil
|
||
}
|
||
|
||
// Intents 意图声明表（G2-2：注册面与路由查询共用）。
|
||
func (b *Bus) Intents() *IntentStore { return b.intents }
|
||
|
||
// ResolveSlash `/` 入口解析：返回命令 id 与原始字段（参数由宿主处理器按命令语义解析）。
|
||
func (b *Bus) ResolveSlash(cmd string) (string, bool) {
|
||
id, ok := b.slash[strings.ToLower(strings.TrimSpace(cmd))]
|
||
return id, ok
|
||
}
|
||
|
||
// SlashTable slash 别名表（清单组装用；只读副本）。
|
||
func (b *Bus) SlashTable() map[string]string {
|
||
out := make(map[string]string, len(b.slash))
|
||
for k, v := range b.slash {
|
||
out[k] = v
|
||
}
|
||
return out
|
||
}
|
||
|
||
// HasCommand 同源判定（intentgate 命中意图后查总线是否承接）。
|
||
func (b *Bus) HasCommand(id string) bool {
|
||
_, _, err := b.reg.Resolve(id)
|
||
return err == nil
|
||
}
|
||
|
||
// List 四入口共享的命令清单（`/` 菜单、按钮、卡片 action 提示用）。
|
||
func (b *Bus) List() []agentcommand.Registered { return b.reg.List() }
|
||
|
||
// Dispatch 统一分发：账号预检→校验→危险收口→执行→审计。账号预检拒绝只返回回执，
|
||
// 避免停用账号或检查故障时继续写入审计；其余路径维持四入口审计对称。
|
||
func (b *Bus) Dispatch(ctx context.Context, req Request) Response {
|
||
start := b.now()
|
||
req.Entrance = agentcommand.CanonicalEntrance(req.Entrance)
|
||
if err := b.checkEffectiveAccount(ctx, accountActor(req)); err != nil {
|
||
// A rejected account cannot create a card, pending row, audit row, or
|
||
// invoke a handler. A failed read is also a denial.
|
||
return b.accountRejected(req.ID, start)
|
||
}
|
||
receipt := func(state agentcommand.ResultState, op, reason string) Response {
|
||
return Response{Receipt: b.auditReceipt(req, state, op, reason, start)}
|
||
}
|
||
spec, _, err := b.reg.Resolve(req.ID)
|
||
if err != nil {
|
||
return receipt(agentcommand.StateRejected, "", "unknown or disabled command: "+req.ID)
|
||
}
|
||
if err := validateParams(spec.ParamsSchema, req.Params); err != nil {
|
||
return receipt(agentcommand.StateRejected, "", "params: "+err.Error())
|
||
}
|
||
// 基座项目门禁：所有带 HostProjectID 的分发都在这里过一道，命令种类与宿主
|
||
// 能力的匹配面由组合根决定（nil 时保持原行为）。
|
||
if req.HostProjectID != "" && b.hostProjectGuard != nil {
|
||
if err := b.hostProjectGuard(ctx, req); err != nil {
|
||
return receipt(agentcommand.StateRejected, "", "host project: "+err.Error())
|
||
}
|
||
}
|
||
if RequiresProjectWorkDecision(req.ID) {
|
||
if b.projectWorkPreflight == nil {
|
||
return receipt(agentcommand.StateRejected, "", "project source preflight unavailable")
|
||
}
|
||
if err := b.projectWorkPreflight(ctx, req); err != nil {
|
||
return receipt(agentcommand.StateRejected, "", "project source preflight: "+err.Error())
|
||
}
|
||
}
|
||
// 高危收口：非确认入口先落内联确认卡（committed 语义 = 卡已创建；确认后二次分发执行）。
|
||
if spec.Danger == agentcommand.DangerHigh && !req.Confirmed {
|
||
if b.cards == nil || req.ConversationID == "" {
|
||
return receipt(agentcommand.StateRejected, "", "high-danger command requires conversation context for inline confirm")
|
||
}
|
||
if err := b.checkEffectiveAccount(ctx, accountActor(req)); err != nil {
|
||
return b.accountRejected(req.ID, start)
|
||
}
|
||
card, err := b.createConfirmCard(accountActor(req), confirmCard(spec, req), "cmd:"+spec.ID)
|
||
if err != nil {
|
||
return receipt(agentcommand.StateUnknown, "", "confirm card: "+err.Error())
|
||
}
|
||
if b.pendingStore != nil {
|
||
raw, _ := json.Marshal(req)
|
||
if err := b.savePending(card.ID, accountActor(req), string(raw)); err != nil {
|
||
// 确认卡已创建（副作用已发生），挂起态却没落库：结果不确定，且
|
||
// 卡 ID 就是本次受理的操作标识——随 unknown 一并给出，供对账。
|
||
return receipt(agentcommand.StateUnknown, card.ID, "confirm pending: "+err.Error())
|
||
}
|
||
} else if b.strictAccountWrites {
|
||
return receipt(agentcommand.StateUnknown, card.ID, "confirm pending: account-checked store required")
|
||
}
|
||
b.pendMu.Lock()
|
||
b.pending[card.ID] = pending{req: req, cardID: card.ID}
|
||
b.pendMu.Unlock()
|
||
r := b.auditReceipt(req, agentcommand.StateAccepted, card.ID, "awaiting explicit confirm", start)
|
||
return Response{Receipt: r, Card: &card}
|
||
}
|
||
h, ok := b.handlers[req.ID]
|
||
if !ok {
|
||
return receipt(agentcommand.StateRejected, "", "no host handler bound for "+req.ID)
|
||
}
|
||
if err := b.checkEffectiveAccount(ctx, accountActor(req)); err != nil {
|
||
return b.accountRejected(req.ID, start)
|
||
}
|
||
// The domain handler can persist this trusted ID with its transaction/outbox.
|
||
// Entropy, not the clock or caller Params, distinguishes concurrent operations.
|
||
// A handler error keeps the ID only on the "executed then uncertain" path
|
||
// (unknown): rejected must never carry one, and unknown without a queryable
|
||
// handle would be a dead end for reconciliation.
|
||
var opBytes [24]byte
|
||
if _, err := rand.Read(opBytes[:]); err != nil {
|
||
return receipt(agentcommand.StateUnknown, "", "operation ID unavailable")
|
||
}
|
||
req.OperationID = "op_" + hex.EncodeToString(opBytes[:])
|
||
payload, err := h(ctx, req)
|
||
if err != nil {
|
||
if IsNotExecuted(err) {
|
||
// handler 声明失败发生在任何副作用之前：确定未执行，可安全归
|
||
// rejected（回执契约禁止 rejected 携带 operationID）。
|
||
return receipt(agentcommand.StateRejected, "", "rejected: "+err.Error())
|
||
}
|
||
// handler 已开始执行才报错：宿主无法断言"无副作用"（F10）。按"已执行
|
||
// 随后不确定"归 unknown，并带上本次分发已签发的 trusted operation ID，
|
||
// 调用方据此按记录对账，而不是盲目重试。本路径不做任何自动重放。
|
||
return b.unknownAfterExecution(req, "handler failed after execution began: "+err.Error(), start)
|
||
}
|
||
return Response{Receipt: b.auditReceipt(req, agentcommand.StateCommitted, req.OperationID, "", start), Payload: payload}
|
||
}
|
||
|
||
// unknownAfterExecution 记录并返回"已执行随后不确定"的回执（F10/U7b）：审计
|
||
// 信封带 state=unknown 与已签发的 operationID（可查凭证），回执走库内受信
|
||
// 构造器 NewUnknownAfterExecution，保证 Validate() 通过且对账句柄不丢。
|
||
func (b *Bus) unknownAfterExecution(req Request, reason string, start time.Time) Response {
|
||
return Response{Receipt: b.auditReceipt(req, agentcommand.StateUnknown, req.OperationID, reason, start)}
|
||
}
|
||
|
||
// Confirm 确认卡回道：actionID=confirm 执行挂起请求（entrance 归一 card），其余视为取消。
|
||
func (b *Bus) Confirm(owner, cardID, actionID string) Response {
|
||
return b.ConfirmContext(context.Background(), owner, cardID, actionID)
|
||
}
|
||
|
||
// ConfirmContext carries the request cancellation through the confirmed
|
||
// handler and repeats the account check before consuming durable state.
|
||
func (b *Bus) ConfirmContext(ctx context.Context, owner, cardID, actionID string) Response {
|
||
// The caller supplies the authenticated card owner. Check before even
|
||
// consuming a persisted pending request; Dispatch checks the stored
|
||
// request again immediately before executing its handler.
|
||
if err := b.checkEffectiveAccount(ctx, owner); err != nil {
|
||
return b.accountRejected("unknown", b.now())
|
||
}
|
||
b.pendMu.Lock()
|
||
p, ok := b.pending[cardID]
|
||
b.pendMu.Unlock()
|
||
if ok && p.req.Owner != "" && p.req.Owner != owner {
|
||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: p.req.ID, State: agentcommand.StateRejected, Reason: "confirm owner mismatch", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||
}
|
||
if b.pendingStore != nil {
|
||
storedOwner, payload, found, err := b.takePending(cardID, owner)
|
||
if err != nil {
|
||
return b.accountRejected("unknown", b.now())
|
||
}
|
||
if found {
|
||
var req Request
|
||
if storedOwner != owner || json.Unmarshal([]byte(payload), &req) != nil {
|
||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "invalid pending confirm", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||
}
|
||
p = pending{req: req, cardID: cardID}
|
||
ok = true
|
||
} else if b.strictAccountWrites {
|
||
ok = false
|
||
}
|
||
} else if b.strictAccountWrites {
|
||
ok = false
|
||
}
|
||
if !ok {
|
||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "no pending confirm (expired or already resolved)", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||
}
|
||
req := p.req
|
||
if req.Owner != "" && req.Owner != owner {
|
||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: req.ID, State: agentcommand.StateRejected, Reason: "confirm owner mismatch", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||
}
|
||
b.pendMu.Lock()
|
||
delete(b.pending, cardID)
|
||
b.pendMu.Unlock()
|
||
req.Confirmed = true
|
||
req.Entrance = "card"
|
||
if actionID != "confirm" {
|
||
if b.cards != nil {
|
||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardResolved, "muted", "cancelled")
|
||
}
|
||
start := b.now()
|
||
return Response{Receipt: b.auditReceipt(req, agentcommand.StateRejected, "", "cancelled at confirm card", start)}
|
||
}
|
||
resp := b.Dispatch(ctx, req)
|
||
if b.cards != nil {
|
||
switch resp.Receipt.State {
|
||
case agentcommand.StateCommitted:
|
||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardResolved, "ok", "confirmed")
|
||
case agentcommand.StateUnknown:
|
||
// 已执行随后不确定：卡片必须落到终态（悬在 action_required 会诱导
|
||
// 二次确认，而挂起项已被消费，重放只会得到"已解决"），但不得断言
|
||
// 失败——tone 用 warn，note 写明不确定原因与对账用 operationID。
|
||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardFailed, "warn", uncertainNote(resp.Receipt))
|
||
default:
|
||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardFailed, "danger", resp.Receipt.Reason)
|
||
}
|
||
}
|
||
return resp
|
||
}
|
||
|
||
// uncertainNote 组成"结果未知"卡片的说明：不确定原因 + 对账句柄。
|
||
func uncertainNote(r agentcommand.ExecReceipt) string {
|
||
if r.OperationID == "" {
|
||
return r.Reason
|
||
}
|
||
return r.Reason + " (operation " + r.OperationID + ")"
|
||
}
|
||
|
||
func accountActor(req Request) string {
|
||
// Plugin and delegated commands use Actor for an audit label such as a
|
||
// plugin ID; Owner is the effective account whose state controls effects.
|
||
if req.Owner != "" {
|
||
return req.Owner
|
||
}
|
||
return req.Actor
|
||
}
|
||
|
||
func (b *Bus) checkEffectiveAccount(ctx context.Context, actor string) error {
|
||
if err := ctx.Err(); err != nil {
|
||
return err
|
||
}
|
||
if b.accountGuard == nil {
|
||
return nil
|
||
}
|
||
return b.accountGuard(ctx, actor)
|
||
}
|
||
|
||
func (b *Bus) accountRejected(commandID string, start time.Time) Response {
|
||
return Response{Receipt: agentcommand.ExecReceipt{
|
||
CommandID: commandID, State: agentcommand.StateRejected,
|
||
Reason: "effective account unavailable", SideEffects: agentcommand.SideEffectsNone, Time: start.UTC(),
|
||
}}
|
||
}
|
||
|
||
func (b *Bus) auditReceipt(req Request, state agentcommand.ResultState, op, reason string, start time.Time) agentcommand.ExecReceipt {
|
||
env := agentcommand.AuditEnvelope{
|
||
Time: start.UTC(), Actor: req.Actor, CommandID: req.ID, Entrance: agentcommand.CanonicalEntrance(req.Entrance),
|
||
ParamsDigest: digest(req.Params), State: state, DurationMs: float64(b.now().Sub(start).Milliseconds()),
|
||
OperationID: op, ConversationID: req.ConversationID,
|
||
}
|
||
if _, src, err := b.reg.Resolve(req.ID); err == nil {
|
||
env.Source = src
|
||
}
|
||
var err error
|
||
if b.strictAccountWrites {
|
||
checked, ok := b.audit.(interface {
|
||
AppendAuditAccountChecked(string, agentcommand.AuditEnvelope, bool) error
|
||
})
|
||
if !ok {
|
||
err = fmt.Errorf("cmdbus: account-checked audit store required")
|
||
} else {
|
||
err = checked.AppendAuditAccountChecked(accountActor(req), env, b.allowImplicitLocal)
|
||
}
|
||
} else {
|
||
err = b.audit.AppendAudit(req.Owner, env)
|
||
}
|
||
if err != nil {
|
||
// 审计不可用：结果标记未知，不臆断成功（四态语义）。已经签发过操作标识的
|
||
// 路径（committed/accepted/已执行后不确定）连同该标识一并返回——信封丢失
|
||
// 不等于操作没发生，调用方仍需按 operationID 对账；确定未执行的路径不带
|
||
// 标识（rejected 契约禁止带 ID，unknown 无 ID 时也无需对账）。
|
||
return unknownReceipt(req.ID, state, op, "audit unavailable: "+err.Error(), b.now().UTC())
|
||
}
|
||
if state == agentcommand.StateUnknown && op != "" {
|
||
// 已执行随后不确定：只有库内受信构造器允许 unknown 携带本流程签发的 ID。
|
||
return agentcommand.NewUnknownAfterExecution(req.ID, op, reason, b.now().UTC())
|
||
}
|
||
return agentcommand.ExecReceipt{CommandID: req.ID, State: state, OperationID: op, Reason: reason,
|
||
SideEffects: sideEffectsFor(state), Time: b.now().UTC()}
|
||
}
|
||
|
||
// unknownReceipt 构造"结果未知"回执：已签发过操作标识时走受信构造器携带该标识
|
||
// （供对账），否则返回不带标识的 unknown。state=rejected 恒不带标识。
|
||
func unknownReceipt(commandID string, state agentcommand.ResultState, op, reason string, at time.Time) agentcommand.ExecReceipt {
|
||
if op != "" && state != agentcommand.StateRejected {
|
||
return agentcommand.NewUnknownAfterExecution(commandID, op, reason, at)
|
||
}
|
||
return agentcommand.ExecReceipt{CommandID: commandID, State: agentcommand.StateUnknown, Reason: reason, Time: at}
|
||
}
|
||
|
||
// sideEffectsFor 回执的副作用声明：总线自身发起的 rejected 均为任何副作用之前
|
||
// 的拒绝（未知命令/参数校验/项目前置/账号预检/无处理器/确认取消/handler 声明的
|
||
// not-executed），故断言 none；committed/accepted 维持今日语义（未声明）。
|
||
func sideEffectsFor(state agentcommand.ResultState) agentcommand.SideEffects {
|
||
if state == agentcommand.StateRejected {
|
||
return agentcommand.SideEffectsNone
|
||
}
|
||
return agentcommand.SideEffectsUnset
|
||
}
|
||
|
||
func digest(params map[string]any) string {
|
||
if len(params) == 0 {
|
||
return ""
|
||
}
|
||
raw, _ := json.Marshal(params)
|
||
sum := sha256.Sum256(raw)
|
||
return hex.EncodeToString(sum[:8])
|
||
}
|
||
|
||
// confirmCard 构造内联确认卡：锚定发起消息、双现于收件箱、动作收口回总线。
|
||
func confirmCard(spec agentcommand.CommandSpec, req Request) core.ActivityCard {
|
||
body := []core.UINode{{Type: core.UIMarkdown, Markdown: "**" + spec.Title + "** 需要确认后执行。"}}
|
||
if spec.Description != "" {
|
||
body = append(body, core.UINode{Type: core.UIMarkdown, Markdown: spec.Description})
|
||
}
|
||
rows := [][2]string{{"入口", req.Entrance}}
|
||
for k, v := range req.Params {
|
||
rows = append(rows, [2]string{k, fmt.Sprintf("%v", v)})
|
||
}
|
||
body = append(body, core.UINode{Type: core.UIKV, KV: rows})
|
||
return core.ActivityCard{
|
||
Kind: core.CardKindConfirm, Status: core.CardActionRequired, Tone: "warn",
|
||
Title: "确认执行：" + spec.Title,
|
||
Body: body,
|
||
Actions: []core.UIAction{
|
||
{ID: "confirm", Label: "确认执行", Style: "primary"},
|
||
{ID: "cancel", Label: "取消", Style: "danger"},
|
||
},
|
||
ConversationID: core.ID(req.ConversationID),
|
||
AnchorMessageID: core.ID(req.AnchorMessageID),
|
||
}
|
||
}
|
||
|
||
// validateParams JSON Schema draft-07 子集（type:object + required + 标量类型），宿主侧校验。
|
||
func validateParams(schema json.RawMessage, params map[string]any) error {
|
||
if len(schema) == 0 {
|
||
return nil
|
||
}
|
||
var sc struct {
|
||
Type string `json:"type"`
|
||
Required []string `json:"required"`
|
||
Properties map[string]any `json:"properties"`
|
||
}
|
||
if err := json.Unmarshal(schema, &sc); err != nil {
|
||
return fmt.Errorf("bad schema: %w", err)
|
||
}
|
||
if sc.Type != "" && sc.Type != "object" {
|
||
return fmt.Errorf("schema type must be object, got %s", sc.Type)
|
||
}
|
||
for _, k := range sc.Required {
|
||
if _, ok := params[k]; !ok {
|
||
return fmt.Errorf("missing required param %s", k)
|
||
}
|
||
}
|
||
for k, v := range params {
|
||
decl, ok := sc.Properties[k]
|
||
if !ok {
|
||
continue // 未声明属性放行（宿主最小校验）
|
||
}
|
||
m, _ := decl.(map[string]any)
|
||
want, _ := m["type"].(string)
|
||
if want != "" && !scalarTypeOk(want, v) {
|
||
return fmt.Errorf("param %s: want %s", k, want)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func scalarTypeOk(want string, v any) bool {
|
||
switch want {
|
||
case "string":
|
||
_, ok := v.(string)
|
||
return ok
|
||
case "number":
|
||
return isNum(v)
|
||
case "boolean":
|
||
_, ok := v.(bool)
|
||
return ok
|
||
case "integer":
|
||
if !isNum(v) {
|
||
return false
|
||
}
|
||
return float64(toFloat(v)) == float64(int64(toFloat(v)))
|
||
}
|
||
return true
|
||
}
|
||
|
||
func isNum(v any) bool {
|
||
switch v.(type) {
|
||
case float64, float32, int, int64:
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
func toFloat(v any) float64 {
|
||
switch x := v.(type) {
|
||
case float64:
|
||
return x
|
||
case float32:
|
||
return float64(x)
|
||
case int:
|
||
return float64(x)
|
||
case int64:
|
||
return float64(x)
|
||
}
|
||
return 0
|
||
}
|