Add cmdbus: registry-driven command dispatch with audit and inline confirm (T3) #10

Merged
weichen merged 1 commits from feature/command-bus-r1 into main 2026-09-30 07:54:53 +00:00
14 changed files with 1002 additions and 50 deletions
+4 -3
View File
@@ -3,15 +3,18 @@ module agent-system-backend
go 1.27
require (
agent-command v0.0.0
agent-intent v0.0.0
agent-native v0.0.0
agent-squad v0.0.0-00010101000000-000000000000
agent-system-core v0.0.0
github.com/jackc/pgx/v5 v5.7.6
github.com/ledongthuc/pdf v0.0.0-20260907135840-6c8c28e0e8a0
github.com/redis/go-redis/v9 v9.14.0
modernc.org/sqlite v1.60.1
)
require (
agent-squad v0.0.0-00010101000000-000000000000 // indirect
error-code-system v0.0.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
@@ -19,11 +22,9 @@ require (
github.com/google/uuid v1.6.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/pgx/v5 v5.7.6 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
github.com/mattn/go-isatty v0.0.24 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/redis/go-redis/v9 v9.14.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
golang.org/x/crypto v0.37.0 // indirect
golang.org/x/sync v0.23.0 // indirect
+11
View File
@@ -1,6 +1,12 @@
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
@@ -25,6 +31,7 @@ github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsRe
github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/redis/go-redis/v9 v9.14.0 h1:u4tNCjXOyzfgeLN+vAZaW1xUooqWDqVEsZN0U01jfAE=
github.com/redis/go-redis/v9 v9.14.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
@@ -33,6 +40,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qq
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c=
@@ -47,6 +56,8 @@ golang.org/x/tools v0.50.0 h1:c2ifzfcuY7L90lZ2aKd8S4K2NpASF08SZx9ZuJkHmSU=
golang.org/x/tools v0.50.0/go.mod h1:7ulVMw3831Mwi5EZD6RomGyffr4VFjuNYXf2BbCEAV0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
modernc.org/cc/v4 v4.29.7 h1:q+NXGJ0bK3b4TXFYQQVr9pYETGnmwFWkrUzJnMya/Tg=
modernc.org/cc/v4 v4.29.7/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI=
modernc.org/ccgo/v4 v4.36.1 h1:ZNIUZAryN0UgnJwtyxrdEzcFc3yD4Cu4AzjfPXsLsIE=
+8
View File
@@ -17,6 +17,7 @@ import (
"agent-system-backend/internal/bus"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/pgbus"
"agent-system-backend/internal/plugins"
@@ -106,8 +107,15 @@ func Run(opt Options) error {
cardSvc := cardsvc.New(st, mgr.EmitCardEvent)
mgr.SetCards(cardSvc)
// 命令总线(提案 T3/D27/D30):四入口同源分发 + 审计 + 危险内联确认卡。
bus := cmdbus.New(st, cardSvc, func() time.Time { return time.Now().UTC() })
pluginHost.OnCommands = bus.RegisterPluginCommands
srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost)
srv.Cards = cardSvc
srv.Bus = bus
srv.Gate = gate
srv.WireBuiltinCommands()
srv.Retriever = pipe
srv.Indexer = pipe
+332
View File
@@ -0,0 +1,332 @@
// 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"
"fmt"
"strings"
"sync"
"time"
agentcommand "agent-command"
"agent-system-core"
)
// AuditStore 审计落盘端口(store 实现;v7 command_audit)。
type AuditStore interface {
AppendAudit(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
}
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 }
// 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
}
// 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
}
// 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()
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 {
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(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
}
+213
View File
@@ -0,0 +1,213 @@
package cmdbus
import (
"context"
"encoding/json"
"errors"
"strings"
"sync"
"testing"
"time"
agentcommand "agent-command"
"agent-system-core"
)
// memAudit / memCards 测试替身。
type memAudit struct {
mu sync.Mutex
logs []agentcommand.AuditEnvelope
fail bool
}
func (m *memAudit) AppendAudit(env agentcommand.AuditEnvelope) error {
if m.fail {
return errors.New("audit down")
}
m.mu.Lock()
defer m.mu.Unlock()
m.logs = append(m.logs, env)
return nil
}
type memCards struct{ created []core.ActivityCard }
func (m *memCards) Create(owner string, c core.ActivityCard, ref string) (core.ActivityCard, error) {
c.ID = core.ID("card_x1")
m.created = append(m.created, c)
return c, nil
}
func (m *memCards) SetStatus(owner, id string, st core.CardStatus, tone, note string) (core.ActivityCard, error) {
return core.ActivityCard{ID: core.ID(id), Status: st}, nil
}
func newTestBus(a AuditStore, c CardPort) *Bus {
b := New(a, c, func() time.Time { return time.Unix(1700000000, 0).UTC() })
_ = b.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.ping", Title: "Ping"}, "/ping",
func(ctx context.Context, req Request) (any, error) { return map[string]any{"pong": true}, nil })
_ = b.RegisterBuiltin(agentcommand.CommandSpec{
ID: "demo.risky", Title: "危险演示", Danger: agentcommand.DangerHigh,
Description: "高危命令演示",
ParamsSchema: json.RawMessage(`{"type":"object","required":["target"],"properties":{"target":{"type":"string"}}}`),
}, "/risky", func(ctx context.Context, req Request) (any, error) {
t, _ := req.Params["target"].(string)
if t == "boom" {
return nil, errors.New("handler rejected")
}
return map[string]any{"did": t}, nil
})
return b
}
func TestDispatchFourEntrancesSameAudit(t *testing.T) {
a := &memAudit{}
b := newTestBus(a, &memCards{})
for _, ent := range []string{"button", "slash", "chat", "card", "api", "weird-unknown"} {
resp := b.Dispatch(context.Background(), Request{ID: "demo.ping", Entrance: ent})
if resp.Receipt.State != agentcommand.StateCommitted {
t.Fatalf("entrance %s: state=%s reason=%s", ent, resp.Receipt.State, resp.Receipt.Reason)
}
}
if len(a.logs) != 6 {
t.Fatalf("audit logs = %d, want 6", len(a.logs))
}
want := map[string]string{"button": "button", "slash": "slash", "chat": "chat", "card": "card", "api": "api", "weird-unknown": "api"}
for i, env := range a.logs {
if env.Source != agentcommand.SourceBuiltin {
t.Fatalf("log %d source = %s", i, env.Source)
}
}
_ = want
// 入口归一:未知名归 api
if got := a.logs[5].Entrance; got != "api" {
t.Fatalf("unknown entrance canonicalized to %s, want api", got)
}
if a.logs[0].OperationID == "" {
t.Fatal("committed receipt must carry operationID")
}
}
func TestDispatchUnknownAndBadParams(t *testing.T) {
a := &memAudit{}
b := newTestBus(a, &memCards{})
resp := b.Dispatch(context.Background(), Request{ID: "nope.x"})
if resp.Receipt.State != agentcommand.StateRejected || !strings.Contains(resp.Receipt.Reason, "unknown") {
t.Fatalf("unknown: %+v", resp.Receipt)
}
resp = b.Dispatch(context.Background(), Request{ID: "demo.risky", Confirmed: true, Params: map[string]any{}})
if resp.Receipt.State != agentcommand.StateRejected || !strings.Contains(resp.Receipt.Reason, "required param target") {
t.Fatalf("missing required: %+v", resp.Receipt)
}
}
func TestHighDangerConfirmFlow(t *testing.T) {
a := &memAudit{}
cards := &memCards{}
b := newTestBus(a, cards)
// 未确认:创建内联确认卡,回执 accepted + operationID=卡 ID。
resp := b.Dispatch(context.Background(), Request{ID: "demo.risky", Params: map[string]any{"target": "prod"}, Entrance: "button", ConversationID: "cnv_1", AnchorMessageID: "msg_1"})
if resp.Receipt.State != agentcommand.StateAccepted || resp.Receipt.OperationID != "card_x1" {
t.Fatalf("pre-confirm: %+v", resp.Receipt)
}
if resp.Card == nil || resp.Card.Kind != core.CardKindConfirm || resp.Card.Status != core.CardActionRequired {
t.Fatalf("confirm card not created properly: %+v", resp.Card)
}
if resp.Card.ConversationID != "cnv_1" || resp.Card.AnchorMessageID == "" {
t.Fatalf("card anchoring missing: %+v", resp.Card)
}
// 确认:执行挂起请求,committed。
resp2 := b.Confirm("owner", "card_x1", "confirm")
if resp2.Receipt.State != agentcommand.StateCommitted {
t.Fatalf("confirm: %+v", resp2.Receipt)
}
if m, _ := resp2.Payload.(map[string]any); m == nil || m["did"] != "prod" {
t.Fatalf("confirm payload: %+v", resp2.Payload)
}
// 二次确认(已消费):rejected expired。
resp3 := b.Confirm("owner", "card_x1", "confirm")
if resp3.Receipt.State != agentcommand.StateRejected || !strings.Contains(resp3.Receipt.Reason, "expired") {
t.Fatalf("re-confirm: %+v", resp3.Receipt)
}
// 审计链:accepted(button) → committed(card);过期确认未触碰命令,不产生审计。
if len(a.logs) != 2 {
t.Fatalf("audit logs = %d, want 2", len(a.logs))
}
if a.logs[0].State != agentcommand.StateAccepted || a.logs[0].Entrance != "button" ||
a.logs[1].State != agentcommand.StateCommitted || a.logs[1].Entrance != "card" {
t.Fatalf("audit chain wrong: %+v", a.logs)
}
}
func TestHighDangerCancelAndHandlerReject(t *testing.T) {
a := &memAudit{}
b := newTestBus(a, &memCards{})
b.Dispatch(context.Background(), Request{ID: "demo.risky", Params: map[string]any{"target": "x"}, ConversationID: "cnv_1"})
if r := b.Confirm("o", "card_x1", "cancel"); r.Receipt.State != agentcommand.StateRejected || !strings.Contains(r.Receipt.Reason, "cancelled") {
t.Fatalf("cancel: %+v", r.Receipt)
}
// handler 拒绝路径(确认后执行失败)→ rejected + 理由。
b.Dispatch(context.Background(), Request{ID: "demo.risky", Params: map[string]any{"target": "boom"}, ConversationID: "cnv_1"})
r := b.Confirm("o", "card_x1", "confirm")
if r.Receipt.State != agentcommand.StateRejected || !strings.Contains(r.Receipt.Reason, "handler rejected") {
t.Fatalf("handler reject: %+v", r.Receipt)
}
}
func TestHighDangerWithoutCardPortRejected(t *testing.T) {
b := newTestBus(&memAudit{}, nil)
resp := b.Dispatch(context.Background(), Request{ID: "demo.risky", Params: map[string]any{"target": "x"}, ConversationID: "cnv_1"})
if resp.Receipt.State != agentcommand.StateRejected || !strings.Contains(resp.Receipt.Reason, "conversation context") {
t.Fatalf("no card port: %+v", resp.Receipt)
}
}
func TestAuditUnavailableYieldsUnknown(t *testing.T) {
a := &memAudit{fail: true}
b := newTestBus(a, &memCards{})
resp := b.Dispatch(context.Background(), Request{ID: "demo.ping"})
if resp.Receipt.State != agentcommand.StateUnknown || !strings.Contains(resp.Receipt.Reason, "audit unavailable") {
t.Fatalf("audit down: %+v", resp.Receipt)
}
}
func TestPluginCommandsAndSlash(t *testing.T) {
b := newTestBus(&memAudit{}, &memCards{})
if err := b.RegisterPluginCommands([]json.RawMessage{json.RawMessage(`{"id":"plug.do","title":"插件命令"}`)}); err != nil {
t.Fatal(err)
}
if err := b.RegisterPluginCommands([]json.RawMessage{json.RawMessage(`{"id":"bad id","title":"x"}`)}); err == nil {
t.Fatal("invalid plugin spec must be rejected")
}
if id, ok := b.ResolveSlash("/risky"); !ok || id != "demo.risky" {
t.Fatalf("slash resolve: %s %v", id, ok)
}
if _, ok := b.ResolveSlash("/nope"); ok {
t.Fatal("unknown slash must not resolve")
}
if !b.HasCommand("plug.do") || b.HasCommand("plug.missing") {
t.Fatal("HasCommand wrong")
}
}
func TestValidateParamsScalars(t *testing.T) {
schema := json.RawMessage(`{"type":"object","required":["n"],"properties":{"n":{"type":"number"},"s":{"type":"string"},"b":{"type":"boolean"},"i":{"type":"integer"}}}`)
if err := validateParams(schema, map[string]any{"n": 1.5, "s": "x", "b": true, "i": 3}); err != nil {
t.Fatalf("valid params rejected: %v", err)
}
if err := validateParams(schema, map[string]any{}); err == nil || !strings.Contains(err.Error(), "required") {
t.Fatalf("missing required not caught: %v", err)
}
if err := validateParams(schema, map[string]any{"n": "not-a-number"}); err == nil {
t.Fatal("type mismatch not caught")
}
if err := validateParams(schema, map[string]any{"n": 1, "i": 2.5}); err == nil {
t.Fatal("integer fraction not caught")
}
if err := validateParams(nil, map[string]any{"any": 1}); err != nil {
t.Fatalf("nil schema must pass: %v", err)
}
}
+35 -2
View File
@@ -26,6 +26,7 @@ type Loaded struct {
Cards map[string][]core.UINode // kind → 模板节点
Workflows []core.WorkflowDef
Intents []json.RawMessage // agent-intent overlay 形态(目录侧消费)
Commands []json.RawMessage // Agent-Command spec overlay(cmdbus 注册)
Prompts map[string]string // 文件名 → 内容(画像模板)
PluginSettings []core.SettingDesc
Defaults map[string]any // 插件设置默认值(并入有效设置)
@@ -41,6 +42,8 @@ type Host struct {
loaded map[string]*Loaded // id → loaded
// OnIntents 在装载集合变化后回调全部意图 overlay(宿主接识别闸门;nil = 不接)。
OnIntents func(overlay []json.RawMessage) error
// OnCommands 在装载集合变化后回调全部命令 spec overlay(宿主接命令总线;nil = 不接)。
OnCommands func(overlay []json.RawMessage) error
}
func New(st *store.Store, reg *tools.Registry, s *settings.Service, dir string) *Host {
@@ -75,8 +78,14 @@ func (h *Host) LoadDir() map[string]error {
}
}
overlay := h.collectIntentsLocked()
onIntents := h.OnIntents
cmdOverlay := h.collectCommandsLocked()
onIntents, onCommands := h.OnIntents, h.OnCommands
h.mu.Unlock()
if onCommands != nil {
if err := onCommands(cmdOverlay); err != nil {
results["<commands>"] = err
}
}
if onIntents != nil {
if err := onIntents(overlay); err != nil {
results["<intents>"] = err
@@ -170,6 +179,15 @@ func (h *Host) loadOne(dir string) error {
ld.Intents = append(ld.Intents, specs...)
}
// commands/*.json:Agent-Command spec overlay 形态透出(cmdbuf 注册;不在此校验语义)。
for _, name := range m.Provides.Commands {
var specs []json.RawMessage
if err := readAsset(dir, "commands", name, &specs); err != nil {
return fmt.Errorf("plugins: %s: commands %s: %w", m.ID, name, err)
}
ld.Commands = append(ld.Commands, specs...)
}
// prompts/*.md:画像模板(source=plugin,可被用户复制微调)。
for _, name := range m.Provides.Prompts {
content, err := os.ReadFile(filepath.Join(dir, "prompts", name+".md"))
@@ -194,9 +212,15 @@ func (h *Host) loadOne(dir string) error {
h.mu.Lock()
h.loaded[m.ID] = ld
overlay := h.collectIntentsLocked()
onIntents := h.OnIntents
cmdOverlay := h.collectCommandsLocked()
onIntents, onCommands := h.OnIntents, h.OnCommands
h.mu.Unlock()
_ = h.st.UpsertPlugin(m.ID, dir, m)
if onCommands != nil {
if err := onCommands(cmdOverlay); err != nil {
return fmt.Errorf("plugins: %s: commands rejected by command bus (kept previous): %w", m.ID, err)
}
}
if onIntents != nil {
if err := onIntents(overlay); err != nil {
return fmt.Errorf("plugins: %s: intents rejected by host gate (kept previous): %w", m.ID, err)
@@ -214,6 +238,15 @@ func (h *Host) collectIntentsLocked() []json.RawMessage {
return out
}
// collectCommandsLocked 汇总全部已装载插件的命令 overlay(调用方持锁)。
func (h *Host) collectCommandsLocked() []json.RawMessage {
var out []json.RawMessage
for _, ld := range h.loaded {
out = append(out, ld.Commands...)
}
return out
}
// bindToolExecutor:声明式 HTTP 工具绑定执行器;纯描述型(无 http)工具暂不支持(代码插件为远期 sidecar 档)。
func (h *Host) bindToolExecutor(desc core.ToolDesc) (tools.Executor, error) {
if desc.HTTP == nil {
+89 -5
View File
@@ -2,9 +2,13 @@ package server
import (
"bufio"
"context"
"bytes"
"context"
"encoding/json"
agentcommand "agent-command"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/cmdbus"
"io"
"mime/multipart"
"net/http"
@@ -15,7 +19,6 @@ import (
"testing"
"time"
core "agent-system-core"
"agent-system-backend/internal/bus"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/plugins"
@@ -25,6 +28,7 @@ import (
"agent-system-backend/internal/settings"
"agent-system-backend/internal/store"
"agent-system-backend/internal/tools"
core "agent-system-core"
)
// newTestServer 组合根(测试版):mock runner 驱动全链路,不依赖模型。
@@ -49,6 +53,9 @@ func newTestServer(t *testing.T) (*Server, *store.Store) {
os.MkdirAll(pluginDir, 0o755)
ph := plugins.New(st, reg, setSvc, pluginDir)
srv := New(st, mgr, reg, profSvc, setSvc, ph)
srv.Cards = cardsvc.New(st, func(cid, runID string, ev core.StreamEventType, payload any) error { return nil })
srv.Bus = cmdbus.New(st, srv.Cards, nil)
srv.WireBuiltinCommands()
srv.Subscribe = func(cid string) (<-chan core.StreamEvent, func()) { return b.Subscribe(cid) }
return srv, st
}
@@ -467,6 +474,83 @@ func TestCommandsRoute(t *testing.T) {
}
}
func TestDangerCommandConfirmFlowE2E(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
// 注册一条高危命令(模拟插件/扩展注册),API 入口 button 直发。
_ = srv.Bus.RegisterBuiltin(agentcommand.CommandSpec{
ID: "e2e.destroy", Title: "销毁演示", Danger: agentcommand.DangerHigh,
ParamsSchema: json.RawMessage(`{"type":"object","required":["what"],"properties":{"what":{"type":"string"}}}`),
}, "", func(ctx context.Context, req cmdbus.Request) (any, error) {
w, _ := req.Params["what"].(string)
return map[string]any{"destroyed": w}, nil
})
// 清单可见(数组契约,legacy {name,description} 兼容):原始请求解码。
req := httptest.NewRequest("GET", "/api/commands", nil)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != 200 {
t.Fatalf("list commands: %d", rec.Code)
}
var cl []map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &cl); err != nil {
t.Fatalf("list decode: %v", err)
}
found, legacySlash := false, false
for _, m := range cl {
if m["id"] == "e2e.destroy" && m["danger"] == "high" {
found = true
}
if m["name"] == "/help" {
legacySlash = true
}
}
if !found || !legacySlash {
t.Fatalf("listing wrong (found=%v legacy=%v): %v", found, legacySlash, cl)
}
// button 入口未确认 → 确认卡 + accepted 回执。
code, out := doJSON(t, h, "POST", "/api/commands/e2e.destroy", map[string]any{
"params": map[string]any{"what": "世界"}, "entrance": "button", "conversationId": convID})
if code != 200 {
t.Fatalf("dispatch: %d %v", code, out)
}
card, _ := out["card"].(map[string]any)
if card == nil {
t.Fatalf("confirm card missing: %v", out)
}
cardID, _ := card["id"].(string)
// 卡上确认 → 执行 committed + 审计可查。
code, out2 := doJSON(t, h, "POST", "/api/cards/"+cardID+"/actions/confirm", map[string]any{})
if code != 200 {
t.Fatalf("confirm action: %d %v", code, out2)
}
if rec, _ := out2["receipt"].(map[string]any); rec == nil || rec["state"] != "committed" {
t.Fatalf("confirm receipt: %v", out2)
}
code, audits := doJSON(t, h, "GET", "/api/commands/audit?conversationId="+convID, nil)
if code != 200 {
t.Fatalf("audit query: %d", code)
}
rows, _ := audits["audits"].([]any)
if len(rows) < 2 {
t.Fatalf("audit rows = %d, want >=2 (accepted+committed)", len(rows))
}
states := []string{}
for _, r := range rows {
m, _ := r.(map[string]any)
st, _ := m["state"].(string)
states = append(states, st)
}
if states[1] != "accepted" || states[0] != "committed" {
t.Fatalf("audit order wrong: %v (latest first)", states)
}
}
func TestStaticServed(t *testing.T) {
srv, _ := newTestServer(t)
h := srv.Handler()
@@ -641,10 +725,10 @@ func TestModelRoutingOverrideChain(t *testing.T) {
set(map[string]any{"runtime": map[string]any{"mock": false}}, "/api/settings/user")
prof := mustCode(t, h, "POST", "/api/profiles", map[string]any{
"id": "prof_route_c", "name": "路由C", "systemPrompt": "x",
"model": map[string]any{"chat": "model-C"},
"tools": map[string]any{"allow": []string{"*"}},
"model": map[string]any{"chat": "model-C"},
"tools": map[string]any{"allow": []string{"*"}},
"context": map[string]any{"maxHistoryMessages": 10},
"hitl": map[string]any{"askOnAmbiguity": true}, "budget": map[string]any{},
"hitl": map[string]any{"askOnAmbiguity": true}, "budget": map[string]any{},
})
if prof["id"] != "prof_route_c" {
t.Fatalf("profile save: %v", prof)
+213 -38
View File
@@ -3,6 +3,7 @@
package server
import (
"context"
"encoding/json"
"errors"
"fmt"
@@ -15,7 +16,9 @@ import (
"strings"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/ingest"
"agent-system-backend/internal/plugins"
"agent-system-backend/internal/profiles"
@@ -37,6 +40,8 @@ type Server struct {
LimitLookup func() int // 当前限流值(次/分;<=0 不限;组合根缓存注入)
mgr *runmgr.Manager
Cards *cardsvc.Service // 收件箱任务卡(nil = 不启用;组合根注入)
Bus *cmdbus.Bus // 命令总线(nil = 不启用;组合根注入;四入口同源 D27)
Gate NLIntentGate // 意图闸门(chat 入口同源判定;nil = 不接)
reg *tools.Registry
prof *profiles.Service
set *settings.Service
@@ -54,6 +59,69 @@ func New(st *store.Store, mgr *runmgr.Manager, reg *tools.Registry, p *profiles.
return &Server{st: st, mgr: mgr, reg: reg, prof: p, set: s, plugins: ph, static: sub}
}
// NLIntentGate 意图闸门端口(intentgate.Gate 实现;避免包依赖方向反转)。
type NLIntentGate interface {
Match(ctx context.Context, input string) (string, string)
}
// WireBuiltinCommands 注册内置命令(`/` 菜单全清单的数据源;处理器为宿主用例适配)。
// 响应保持既有 {command, ...} 契约(frontend store.sendMessage 按此消费)。
func (s *Server) WireBuiltinCommands() {
b := s.Bus
if b == nil {
return
}
must := func(err error) {
if err != nil {
panic("server: builtin command: " + err.Error()) // 装配期错误属编程错误
}
}
must(b.RegisterBuiltin(agentcommand.CommandSpec{ID: "builtin.help", Title: "帮助", Description: "查看可用命令与入口说明"},
"/help", func(ctx context.Context, req cmdbus.Request) (any, error) {
return map[string]any{"reply": fmt.Sprintf("可用命令(注册表 %d 条)——/new 新会话 · /mock on|off 演示模式 · /model <名称> 设置主模型 · /profile <id> 切换画像 · /help 帮助", len(b.List()))}, nil
}))
must(b.RegisterBuiltin(agentcommand.CommandSpec{ID: "conversation.create", Title: "新建会话", Description: "在当前项目下新建会话",
ParamsSchema: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string"}}}`)},
"/new", func(ctx context.Context, req cmdbus.Request) (any, error) {
conv, err := s.st.GetConversation(req.Owner, req.ConversationID)
if err != nil {
return nil, err
}
c, err := s.st.CreateConversation(conv.Owner, conv.ProjectID, "新会话", conv.ProfileID)
if err != nil {
return nil, err
}
return map[string]any{"conversation": c}, nil
}))
must(b.RegisterBuiltin(agentcommand.CommandSpec{ID: "runtime.mock", Title: "演示模式", Description: "切换 mock 演示模式(on|off)",
ParamsSchema: json.RawMessage(`{"type":"object","required":["on"],"properties":{"on":{"type":"boolean"}}}`)},
"/mock", func(ctx context.Context, req cmdbus.Request) (any, error) {
on, _ := req.Params["on"].(bool)
if err := s.set.MergeUser(req.Owner, map[string]any{"runtime": map[string]any{"mock": on}}); err != nil {
return nil, err
}
return map[string]any{"runtime.mock": on}, nil
}))
must(b.RegisterBuiltin(agentcommand.CommandSpec{ID: "models.set", Title: "设置主模型", Description: "设置主模型名称",
ParamsSchema: json.RawMessage(`{"type":"object","required":["model"],"properties":{"model":{"type":"string"}}}`)},
"/model", func(ctx context.Context, req cmdbus.Request) (any, error) {
m, _ := req.Params["model"].(string)
if err := s.set.MergeUser(req.Owner, map[string]any{"models": map[string]any{"chat": map[string]any{"model": m}}}); err != nil {
return nil, err
}
return map[string]any{"model": m}, nil
}))
must(b.RegisterBuiltin(agentcommand.CommandSpec{ID: "conversation.profile", Title: "切换画像", Description: "切换会话画像",
ParamsSchema: json.RawMessage(`{"type":"object","required":["profileId"],"properties":{"profileId":{"type":"string"}}}`)},
"/profile", func(ctx context.Context, req cmdbus.Request) (any, error) {
pid, _ := req.Params["profileId"].(string)
if err := s.st.UpdateConversation(req.Owner, req.ConversationID, "", pid); err != nil {
return nil, err
}
return map[string]any{"profile": pid}, nil
}))
}
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
@@ -110,6 +178,8 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("GET /api/conversations/{id}/plan", s.getPlan)
mux.HandleFunc("GET /api/conversations/{id}/todos", s.getTodos)
mux.HandleFunc("GET /api/commands", s.listCommands)
mux.HandleFunc("POST /api/commands/{id}", s.execCommand)
mux.HandleFunc("GET /api/commands/audit", s.listAudits)
mux.HandleFunc("GET /api/parsers", s.listParsers)
mux.HandleFunc("POST /api/projects/{id}/knowledge", s.addKnowledge)
mux.HandleFunc("GET /api/knowledge/search", s.searchKnowledge)
@@ -397,42 +467,131 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
writeStoreErr(w, err)
return
}
// chat 入口同源(D27):意图闸门命中且命令总线承接 → 统一分发(同审计),不起 run。
if s.Bus != nil && s.Gate != nil && !strings.HasPrefix(text, "/") {
if branch, topID := s.Gate.Match(r.Context(), text); branch == "matched" && s.Bus.HasCommand(topID) {
resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{
ID: topID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner,
Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID,
})
out := map[string]any{"message": msg, "receipt": resp.Receipt, "command": topID}
if resp.Card != nil {
out["card"] = resp.Card
}
if m, ok := resp.Payload.(map[string]any); ok {
for k, v := range m {
out[k] = v
}
}
writeJSON(w, http.StatusOK, out)
return
}
}
run, err := s.mgr.StartConversationRun(conv, msg, parsedDocs)
writeJSONOr(w, map[string]any{"message": msg, "run": run}, err)
}
// handleCommand 内置命令(注册表数据驱动,插件可扩展 commands 后补)。
// handleCommand `/` 入口 → 命令总线(D27 四入口同源;slash 语义解析后统一分发+审计)。
func (s *Server) handleCommand(w http.ResponseWriter, r *http.Request, conv core.Conversation, text string) {
fields := strings.Fields(text)
cmd := strings.ToLower(fields[0])
switch cmd {
case "/help":
writeJSON(w, http.StatusOK, map[string]any{"command": cmd, "reply": "可用命令:/new 新会话 · /mock on|off 演示模式 · /model <名称> 设置主模型 · /profile <id> 切换画像 · /help 帮助"})
case "/new":
c, err := s.st.CreateConversation(conv.Owner, conv.ProjectID, "新会话", conv.ProfileID)
writeJSONOr(w, map[string]any{"command": cmd, "conversation": c}, err)
case "/mock":
if len(fields) < 2 || (fields[1] != "on" && fields[1] != "off") {
http.Error(w, "usage: /mock on|off", http.StatusBadRequest)
return
}
val := fields[1] == "on"
writeJSONOr(w, map[string]any{"command": cmd, "runtime.mock": val}, s.set.MergeUser(conv.Owner, map[string]any{"runtime": map[string]any{"mock": val}}))
case "/model":
if len(fields) < 2 {
http.Error(w, "usage: /model <model-name>", http.StatusBadRequest)
return
}
writeJSONOr(w, map[string]any{"command": cmd, "model": fields[1]}, s.set.MergeUser(conv.Owner, map[string]any{"models": map[string]any{"chat": map[string]any{"model": fields[1]}}}))
case "/profile":
if len(fields) < 2 {
http.Error(w, "usage: /profile <id>", http.StatusBadRequest)
return
}
writeJSONOr(w, map[string]any{"command": cmd, "profile": fields[1]}, s.st.UpdateConversation(conv.Owner, conv.ID, "", fields[1]))
default:
http.Error(w, "未知命令 "+cmd+"(/help 查看)", http.StatusBadRequest)
slash := strings.ToLower(fields[0])
if s.Bus == nil {
http.Error(w, "commands disabled", http.StatusServiceUnavailable)
return
}
id, ok := s.Bus.ResolveSlash(slash)
if !ok {
http.Error(w, "未知命令 "+slash+"(/help 查看)", http.StatusBadRequest)
return
}
params, err := parseSlashParams(id, fields[1:])
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{
ID: id, Params: params, Entrance: "slash", Actor: s.owner(r),
Owner: s.owner(r), ConversationID: conv.ID,
})
out := map[string]any{"command": slash, "receipt": resp.Receipt}
if m, ok := resp.Payload.(map[string]any); ok {
for k, v := range m {
out[k] = v
}
}
if resp.Receipt.State == agentcommand.StateRejected {
out["error"] = resp.Receipt.Reason
}
writeJSON(w, http.StatusOK, out)
}
// parseSlashParams slash 参数 → 命令参数(按命令语义;未知参数原样透传由 schema 校验拒绝)。
func parseSlashParams(id string, args []string) (map[string]any, error) {
switch id {
case "runtime.mock":
if len(args) != 1 || (args[0] != "on" && args[0] != "off") {
return nil, errors.New("usage: /mock on|off")
}
return map[string]any{"on": args[0] == "on"}, nil
case "models.set":
if len(args) < 1 {
return nil, errors.New("usage: /model <model-name>")
}
return map[string]any{"model": strings.Join(args, " ")}, nil
case "conversation.profile":
if len(args) < 1 {
return nil, errors.New("usage: /profile <id>")
}
return map[string]any{"profileId": args[0]}, nil
case "conversation.create":
if len(args) > 0 {
return map[string]any{"title": strings.Join(args, " ")}, nil
}
return map[string]any{}, nil
default:
return map[string]any{}, nil
}
}
// execCommand 通用命令入口(button/card/chat 等 API 客户端统一走此路由;D27)。
func (s *Server) execCommand(w http.ResponseWriter, r *http.Request) {
if s.Bus == nil {
http.Error(w, "commands disabled", http.StatusServiceUnavailable)
return
}
var body struct {
Params map[string]any `json:"params"`
Entrance string `json:"entrance"`
ConversationID string `json:"conversationId"`
AnchorMessageID string `json:"anchorMessageId"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
if body.Params == nil {
body.Params = map[string]any{}
}
resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{
ID: r.PathValue("id"), Params: body.Params, Entrance: body.Entrance, Actor: s.owner(r),
Owner: s.owner(r), ConversationID: body.ConversationID, AnchorMessageID: body.AnchorMessageID,
})
code := http.StatusOK
if resp.Receipt.State == agentcommand.StateRejected {
code = http.StatusBadRequest
}
out := map[string]any{"receipt": resp.Receipt}
if resp.Payload != nil {
out["payload"] = resp.Payload
}
if resp.Card != nil {
out["card"] = resp.Card
}
writeJSON(w, code, out)
}
// listAudits 审计查询(会话维度或全局;入口对称的可核查凭证)。
func (s *Server) listAudits(w http.ResponseWriter, r *http.Request) {
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
audits, err := s.st.ListAudits(r.URL.Query().Get("conversationId"), limit)
writeJSONOr(w, map[string]any{"audits": audits}, err)
}
func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) {
@@ -485,6 +644,14 @@ func (s *Server) cardAction(w http.ResponseWriter, r *http.Request) {
FreeText string `json:"freeText,omitempty"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
// 命令确认卡(ref=cmd:<id>)走总线回道:confirm=执行挂起命令,其余=取消。
if ref, err := s.st.CardRef(owner, r.PathValue("id")); err == nil && strings.HasPrefix(ref, "cmd:") {
if card, err := s.Cards.Get(owner, r.PathValue("id")); err == nil && card.Status == core.CardActionRequired {
resp := s.Bus.Confirm(owner, card.ID, r.PathValue("actionId"))
writeJSON(w, http.StatusOK, map[string]any{"card": card, "receipt": resp.Receipt, "payload": resp.Payload})
return
}
}
card, label, err := s.Cards.Action(owner, r.PathValue("id"), r.PathValue("actionId"), body.FreeText)
if err != nil {
writeStoreErr(w, err)
@@ -812,17 +979,25 @@ func (s *Server) getTodos(w http.ResponseWriter, r *http.Request) {
writeJSONOr(w, l, err)
}
var builtinCommands = []map[string]string{
{"name": "/help", "description": "查看可用命令"},
{"name": "/new", "description": "新建会话"},
{"name": "/mock", "description": "演示模式 on|off(不调用真实模型)"},
{"name": "/model", "description": "设置主模型名称"},
{"name": "/profile", "description": "切换会话画像"},
func (s *Server) listCommands(w http.ResponseWriter, r *http.Request) {
out := []map[string]any{}
if s.Bus != nil {
slashOf := map[string]string{}
for slash, id := range s.Bus.SlashTable() {
slashOf[id] = slash
}
for _, c := range s.Bus.List() {
item := map[string]any{"id": c.Spec.ID, "title": c.Spec.Title, "description": c.Spec.Description, "danger": string(c.Spec.Danger), "source": string(c.Source), "enabled": c.Enabled}
if slash, ok := slashOf[c.Spec.ID]; ok {
item["name"] = slash // legacy 兼容:`/` 菜单按 name 消费
}
out = append(out, item)
}
}
writeJSON(w, http.StatusOK, out)
}
func (s *Server) listCommands(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, builtinCommands)
}
func (s *Server) listParsers(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, ingest.ParsersDescription())
+55
View File
@@ -0,0 +1,55 @@
package store
import (
"database/sql"
"encoding/json"
"time"
agentcommand "agent-command"
)
// AppendAudit 追加命令审计信封(cmdbus 端口;v7 command_audit)。
func (s *Store) AppendAudit(env agentcommand.AuditEnvelope) error {
raw, err := json.Marshal(env)
if err != nil {
return err
}
_, err = s.db.Exec(s.q(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, payload)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`),
env.Time.UTC().Format(time.RFC3339Nano), env.CommandID, string(env.Source), env.Entrance, string(env.State),
env.OperationID, env.ConversationID, string(raw))
return err
}
// ListAudits 审计查询(时间倒序;conversationID 空 = 全局)。
func (s *Store) ListAudits(conversationID string, limit int) ([]agentcommand.AuditEnvelope, error) {
if limit <= 0 || limit > 500 {
limit = 100
}
var (
rows *sql.Rows
err error
)
if conversationID == "" {
rows, err = s.db.Query(s.q(`SELECT payload FROM command_audit ORDER BY time DESC LIMIT ?`), limit)
} else {
rows, err = s.db.Query(s.q(`SELECT payload FROM command_audit WHERE conversation_id=? ORDER BY time DESC LIMIT ?`), conversationID, limit)
}
if err != nil {
return nil, err
}
defer rows.Close()
out := []agentcommand.AuditEnvelope{}
for rows.Next() {
var raw string
if err := rows.Scan(&raw); err != nil {
return nil, err
}
var env agentcommand.AuditEnvelope
if err := json.Unmarshal([]byte(raw), &env); err != nil {
return nil, err
}
out = append(out, env)
}
return out, rows.Err()
}
+10
View File
@@ -66,6 +66,16 @@ func (s *Store) ListCards(owner, conversationID string) ([]core.ActivityCard, er
return out, rows.Err()
}
// CardRef 取卡的 ref 列(命令确认卡回道判定:"cmd:<id>" 前缀)。
func (s *Store) CardRef(owner, id string) (string, error) {
row := s.db.QueryRow(s.q(`SELECT ref FROM activity_cards WHERE id=? AND owner=?`), id, owner)
var ref string
if err := row.Scan(&ref); err != nil {
return "", err
}
return ref, nil
}
// CardIDByRef 按引用取卡 ID(HITL 投影联动:"hitl:<id>";最新一条)。
func (s *Store) CardIDByRef(owner, ref string) (string, error) {
row := s.db.QueryRow(s.q(`SELECT id FROM activity_cards WHERE ref=? AND owner=? ORDER BY updated_at DESC LIMIT 1`), ref, owner)
+13
View File
@@ -215,6 +215,19 @@ func init() {
`CREATE INDEX IF NOT EXISTS idx_cards_ref ON activity_cards(ref)`,
}},
down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS activity_cards`}, "postgres": {`DROP TABLE IF EXISTS activity_cards`}}},
migration{version: 7, name: "command_audit",
// 命令总线审计(agent-command.AuditEnvelope):payload 存整信封 JSON;
// 冗余列支持按会话/状态/操作号检索(提案 T3,四入口同源审计)。
up: map[string][]string{"sqlite": {
`CREATE TABLE IF NOT EXISTS command_audit (id INTEGER PRIMARY KEY AUTOINCREMENT, time TEXT NOT NULL, command_id TEXT NOT NULL, source TEXT NOT NULL DEFAULT '', entrance TEXT NOT NULL DEFAULT '', state TEXT NOT NULL, operation_id TEXT NOT NULL DEFAULT '', conversation_id TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL)`,
`CREATE INDEX IF NOT EXISTS idx_audit_conv ON command_audit(conversation_id, time)`,
`CREATE INDEX IF NOT EXISTS idx_audit_op ON command_audit(operation_id)`,
}, "postgres": {
`CREATE TABLE IF NOT EXISTS command_audit (id SERIAL PRIMARY KEY, time TEXT NOT NULL, command_id TEXT NOT NULL, source TEXT NOT NULL DEFAULT '', entrance TEXT NOT NULL DEFAULT '', state TEXT NOT NULL, operation_id TEXT NOT NULL DEFAULT '', conversation_id TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL)`,
`CREATE INDEX IF NOT EXISTS idx_audit_conv ON command_audit(conversation_id, time)`,
`CREATE INDEX IF NOT EXISTS idx_audit_op ON command_audit(operation_id)`,
}},
down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS command_audit`}, "postgres": {`DROP TABLE IF EXISTS command_audit`}}},
)
}
+2 -2
View File
@@ -56,8 +56,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if len(got) != 6 {
t.Fatalf("expected versions [1..6], got %v", got)
if len(got) != 7 {
t.Fatalf("expected versions [1..7], got %v", got)
}
// 老库形态(v2 前):events 全局自增、settings 单列主键。
if err := st.MigrateDown(1); err != nil {
+8
View File
@@ -0,0 +1,8 @@
# 模块卡:internal/cardsvc(收件箱任务卡)
- 状态:active(T2 落地,提案 D25/D33)
- 职责:ActivityCard 生命周期(创建/更新/状态机/pin/dismiss/动作+回执/ref 幂等消解)、存储(store v6 activity_cards)、card.* 事件经会话事件管道发布(persist/seq/replay)、摄取/编排/HITL 投影卡 producer 供运行时与摄取侧调用。
- 公开面:`New(st, emit)`、`Create/Get/List/Update/SetStatus/SetPinned/Dismiss/Action/ResolveByRef`。
- 依赖:core(ActivityCard)、store(自有表);被 runmgr/server/ingest/squadadapter 消费;cmdbus 的确认卡经 CardPort 消费。
- 隔离档位:L1 包(仓内);有自有存储(activity_cards 表),事件发布为注入端口。
- 升级拆仓触发条件:跨会话项目级收件箱(v1 边界外)或第二宿主出现时,连同 cmdbus 的确认卡语义一并评估。
+9
View File
@@ -0,0 +1,9 @@
# 模块卡:internal/cmdbuf(命令总线)
- 状态:active(T3 落地,提案 D27/D30/D33)
- 职责:组合 Agent-Command 生态库——命令注册(builtin/plugin/user)、四入口同源分发(button/card/slash/chat→api 归一)、params JSON-Schema 子集校验、危险命令(danger=high)的内联确认卡收口、审计信封落盘(四态回执)。
- 公开面:`New(AuditStore, CardPort, now)`、`RegisterBuiltin/RegisterPluginCommands`、`Dispatch(ctx, Request) Response`、`Confirm(owner, cardID, actionID)`、`ResolveSlash/SlashTable/HasCommand/List`。
- 依赖:agent-command(生态库,嵌套 submodule + replace);core(ActivityCard/UIAction);不依赖 server/runmgr(server 依赖它;端口注入避免反向依赖)。
- 隔离档位:L1 包(仓内);纯编排无自有存储——审计走 store 端口、确认卡走 cardsvc 端口。
- 已知限制:确认挂起态为进程内(重启后确认返回 rejected expired);params 校验为 draft-07 标量子集。
- 升级拆仓触发条件:出现第二宿主(非 backend)需要同一命令总线时,随 Agent-Command 生态位评估拆出(同 cardsvc 判据)。