diff --git a/go.mod b/go.mod index b84eccb..e18f2d5 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 535e370..15b9529 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/internal/app/app.go b/internal/app/app.go index 04bc612..b2908fb 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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 diff --git a/internal/cmdbus/cmdbus.go b/internal/cmdbus/cmdbus.go new file mode 100644 index 0000000..b9a5bf1 --- /dev/null +++ b/internal/cmdbus/cmdbus.go @@ -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 +} diff --git a/internal/cmdbus/cmdbus_test.go b/internal/cmdbus/cmdbus_test.go new file mode 100644 index 0000000..8feb8cf --- /dev/null +++ b/internal/cmdbus/cmdbus_test.go @@ -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) + } +} diff --git a/internal/plugins/host.go b/internal/plugins/host.go index 4c431a5..bc08506 100644 --- a/internal/plugins/host.go +++ b/internal/plugins/host.go @@ -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[""] = err + } + } if onIntents != nil { if err := onIntents(overlay); err != nil { results[""] = 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 { diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index a6b9723..e004afc 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -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) diff --git a/internal/server/server.go b/internal/server/server.go index e763afe..9510d9b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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 切换画像 · /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 切换画像 · /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 ", 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 ", 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 ") + } + return map[string]any{"model": strings.Join(args, " ")}, nil + case "conversation.profile": + if len(args) < 1 { + return nil, errors.New("usage: /profile ") + } + 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:)走总线回道: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()) diff --git a/internal/store/audit.go b/internal/store/audit.go new file mode 100644 index 0000000..1bff805 --- /dev/null +++ b/internal/store/audit.go @@ -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() +} diff --git a/internal/store/cards.go b/internal/store/cards.go index e6eae5e..274242a 100644 --- a/internal/store/cards.go +++ b/internal/store/cards.go @@ -66,6 +66,16 @@ func (s *Store) ListCards(owner, conversationID string) ([]core.ActivityCard, er return out, rows.Err() } +// CardRef 取卡的 ref 列(命令确认卡回道判定:"cmd:" 前缀)。 +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:";最新一条)。 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) diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 383db3f..f702db2 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -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`}}}, ) } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 1ebf07d..b80abc3 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -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 { diff --git a/policy/modules/cardsvc.md b/policy/modules/cardsvc.md new file mode 100644 index 0000000..35f31cb --- /dev/null +++ b/policy/modules/cardsvc.md @@ -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 的确认卡语义一并评估。 diff --git a/policy/modules/cmdbus.md b/policy/modules/cmdbus.md new file mode 100644 index 0000000..adbf761 --- /dev/null +++ b/policy/modules/cmdbus.md @@ -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 判据)。