weichen
e0b1c27f1f
ci / vet-and-test (pull_request) Successful in 42s
Why: GET /api/commands/audit leaked every tenant's audit rows (command ids, entrances, conversation ids, param digests) to any authenticated user; owner was never recorded nor filtered.
Evidence: Red run at a7cc69b FAIL (alice sees bob's row); after fix go test ./... 32/32 packages ok, including TestCommandAuditOwnerScoping and the v18 backfill test (rows recorded on the task card).
Impact: v18 migration adds command_audit.owner with actor-based backfill (empty-actor rows stay invisible via API, kept in-table); AppendAudit/ListAudits gain owner params; cmdbus port and fakes updated; handler passes authenticated owner. No public extension surface change.
Refs: Task card R1; U0 proposal section 5 C7 finding.
395 lines
14 KiB
Go
395 lines
14 KiB
Go
// Package cmdbus 组合 Agent-Command 生态库：注册、四入口分发（button/card/slash/chat→api
|
||
// 归一）、审计信封落盘与危险命令的内联确认卡收口（提案 T3 / D27、D30、D33）。
|
||
// 与 intentgate 之于 agent-intent 同构：schema 权威在生态库，宿主只做装配与执行适配。
|
||
package cmdbus
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
agentcommand "agent-command"
|
||
"agent-system-core"
|
||
)
|
||
|
||
// AuditStore 审计落盘端口（store 实现；v7 建表、v18 加 owner 列）。
|
||
// owner 是请求租户（req.Owner）：审计行按租户归属，查询按 owner 过滤，
|
||
// 跨租户审计不可见（信封 Actor 是发起主体，不作归属依据）。
|
||
type AuditStore interface {
|
||
AppendAudit(owner string, env agentcommand.AuditEnvelope) error
|
||
}
|
||
|
||
// 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)
|
||
|
||
// Request 一次分发请求（入口归一后的统一形态）。
|
||
type Request struct {
|
||
ID string
|
||
Params map[string]any
|
||
Entrance string // button | card | slash | chat | api（归一）
|
||
Actor string
|
||
Owner string
|
||
ConversationID string
|
||
AnchorMessageID string // 确认卡锚定的发起消息（内联渲染）
|
||
Confirmed bool // 危险命令经确认卡放行后的执行
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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{},
|
||
}
|
||
}
|
||
|
||
// Registry 暴露注册表（列清单/插件 overlay 注册用）。
|
||
func (b *Bus) Registry() *agentcommand.Registry { return b.reg }
|
||
|
||
// SetPendingStore 注入确认挂起态的持久化实现（nil = 仅进程内）。
|
||
func (b *Bus) SetPendingStore(st PendingStore) { b.pendingStore = st }
|
||
|
||
// 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)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 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
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 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)
|
||
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())
|
||
}
|
||
// 高危收口：非确认入口先落内联确认卡（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")
|
||
}
|
||
card, err := b.cards.Create(req.Owner, confirmCard(spec, req), "cmd:"+spec.ID)
|
||
if err != nil {
|
||
return receipt(agentcommand.StateUnknown, "", "confirm card: "+err.Error())
|
||
}
|
||
b.pendMu.Lock()
|
||
b.pending[card.ID] = pending{req: req, cardID: card.ID}
|
||
b.pendMu.Unlock()
|
||
if b.pendingStore != nil {
|
||
raw, _ := json.Marshal(req)
|
||
_ = b.pendingStore.SavePendingConfirm(card.ID, req.Owner, string(raw))
|
||
}
|
||
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)
|
||
}
|
||
payload, err := h(ctx, req)
|
||
if err != nil {
|
||
return receipt(agentcommand.StateRejected, "", "rejected: "+err.Error())
|
||
}
|
||
op := fmt.Sprintf("op_%s_%d", strings.ReplaceAll(strings.TrimPrefix(req.ID, ""), ".", "_"), b.now().UnixNano())
|
||
return Response{Receipt: b.auditReceipt(req, agentcommand.StateCommitted, op, "", start), Payload: payload}
|
||
}
|
||
|
||
// Confirm 确认卡回道：actionID=confirm 执行挂起请求（entrance 归一 card），其余视为取消。
|
||
func (b *Bus) Confirm(owner, cardID, actionID string) Response {
|
||
b.pendMu.Lock()
|
||
p, ok := b.pending[cardID]
|
||
if ok && actionID == "confirm" {
|
||
delete(b.pending, cardID)
|
||
}
|
||
b.pendMu.Unlock()
|
||
if ok && actionID == "confirm" && b.pendingStore != nil {
|
||
_, _, _, _ = b.pendingStore.TakePendingConfirm(cardID)
|
||
}
|
||
if !ok && b.pendingStore != nil {
|
||
_, payload, found, err := b.pendingStore.TakePendingConfirm(cardID)
|
||
if err == nil && found {
|
||
var req Request
|
||
if json.Unmarshal([]byte(payload), &req) == nil {
|
||
p = pending{req: req, cardID: cardID}
|
||
ok = true
|
||
}
|
||
}
|
||
}
|
||
if !ok {
|
||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "no pending confirm (expired or already resolved)", Time: b.now().UTC()}}
|
||
}
|
||
req := p.req
|
||
req.Confirmed = true
|
||
req.Entrance = "card"
|
||
if actionID != "confirm" {
|
||
if b.cards != nil {
|
||
_, _ = b.cards.SetStatus(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(context.Background(), req)
|
||
if b.cards != nil {
|
||
if resp.Receipt.State == agentcommand.StateCommitted {
|
||
_, _ = b.cards.SetStatus(owner, cardID, core.CardResolved, "ok", "confirmed")
|
||
} else {
|
||
_, _ = b.cards.SetStatus(owner, cardID, core.CardFailed, "danger", resp.Receipt.Reason)
|
||
}
|
||
}
|
||
return resp
|
||
}
|
||
|
||
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
|
||
}
|
||
if err := b.audit.AppendAudit(req.Owner, env); err != nil {
|
||
// 审计不可用：结果标记未知，不臆断成功（四态语义）。
|
||
return agentcommand.ExecReceipt{CommandID: req.ID, State: agentcommand.StateUnknown, Reason: "audit unavailable: " + err.Error(), Time: b.now().UTC()}
|
||
}
|
||
return agentcommand.ExecReceipt{CommandID: req.ID, State: state, OperationID: op, Reason: reason, Time: b.now().UTC()}
|
||
}
|
||
|
||
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
|
||
}
|