613 lines
27 KiB
Go
613 lines
27 KiB
Go
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(owner string, 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" {
|
||
// 参数级拒绝发生在任何副作用之前：显式声明 not-executed（F10）。
|
||
return nil, NotExecuted(errors.New("handler rejected"))
|
||
}
|
||
return map[string]any{"did": t}, nil
|
||
})
|
||
return b
|
||
}
|
||
|
||
func TestDispatchFourEntrancesSameAudit(t *testing.T) {
|
||
a := &memAudit{}
|
||
b := newTestBus(a, &memCards{})
|
||
b.SetProjectWorkPreflight(func(context.Context, Request) error { return nil })
|
||
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 TestHandlerReceivesCommittedOperationIDBeforeDomainWrite(t *testing.T) {
|
||
audit := &memAudit{}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 123).UTC() })
|
||
var seen string
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "rdm.confirm", Title: "Confirm"}, "", func(_ context.Context, req Request) (any, error) {
|
||
seen = req.OperationID
|
||
return "committed-domain-event", nil
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "rdm.confirm", Actor: "alice", Owner: "alice", Params: map[string]any{"operationId": "forged-client-id"}})
|
||
if seen == "" || seen == "forged-client-id" || resp.Receipt.OperationID != seen || len(audit.logs) != 1 || audit.logs[0].OperationID != seen {
|
||
t.Fatalf("operation correlation missing: handler=%q receipt=%+v audit=%+v", seen, resp.Receipt, audit.logs)
|
||
}
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "rdm.fail", Title: "Fail"}, "", func(_ context.Context, req Request) (any, error) {
|
||
if req.OperationID == "" {
|
||
t.Fatal("handler must receive a tentative host operation ID")
|
||
}
|
||
// 领域事务回滚 = 确定未执行：声明 not-executed 才是 rejected 的依据。
|
||
return nil, NotExecuted(errors.New("domain transaction rejected"))
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
failed := bus.Dispatch(context.Background(), Request{ID: "rdm.fail", Actor: "alice", Owner: "alice"})
|
||
if failed.Receipt.State != agentcommand.StateRejected || failed.Receipt.OperationID != "" || len(audit.logs) != 2 || audit.logs[1].OperationID != "" {
|
||
t.Fatalf("not-executed handler was not rejected: %+v audits=%+v", failed.Receipt, audit.logs)
|
||
}
|
||
if err := failed.Receipt.Validate(); err != nil {
|
||
t.Fatalf("rejected receipt must satisfy the library contract: %v", err)
|
||
}
|
||
// 上游 agent-command main 移除了 ExecReceipt.SideEffects（“确定无副作用”
|
||
// 的显式声明），rejected 不得携带 operationID 这条库契约即是等价断言。
|
||
}
|
||
|
||
func TestOperationIDsRemainUniqueWithFixedClockAndConcurrentDispatch(t *testing.T) {
|
||
audit := &memAudit{}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 123).UTC() })
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "rdm.confirm", Title: "Confirm"}, "", func(_ context.Context, req Request) (any, error) {
|
||
return req.OperationID, nil
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
const count = 64
|
||
var wg sync.WaitGroup
|
||
ids := make([]string, count)
|
||
for i := range ids {
|
||
wg.Add(1)
|
||
go func(i int) {
|
||
defer wg.Done()
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "rdm.confirm", Actor: "alice", Owner: "alice"})
|
||
ids[i] = resp.Receipt.OperationID
|
||
}(i)
|
||
}
|
||
wg.Wait()
|
||
seen := map[string]bool{}
|
||
for _, id := range ids {
|
||
if id == "" || seen[id] {
|
||
t.Fatalf("fixed clock produced a missing or duplicate trusted operation ID: %q", id)
|
||
}
|
||
seen[id] = true
|
||
}
|
||
if len(audit.logs) != count {
|
||
t.Fatalf("audit rows=%d want=%d", len(audit.logs), count)
|
||
}
|
||
}
|
||
|
||
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)
|
||
}
|
||
}
|
||
|
||
// ---- EXT-1：进程内插件命令处理器 ----
|
||
|
||
func TestRegisterPluginCommandDispatch(t *testing.T) {
|
||
a := &memAudit{}
|
||
b := newTestBus(a, &memCards{})
|
||
b.SetProjectWorkPreflight(func(context.Context, Request) error { return nil })
|
||
err := b.RegisterPluginCommand(agentcommand.CommandSpec{ID: "rdm.ingest", Title: "入库"}, func(ctx context.Context, req Request) (any, error) {
|
||
return map[string]any{"echo": req.Params["ref"], "owner": req.Owner}, nil
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := b.Dispatch(context.Background(), Request{ID: "rdm.ingest", Params: map[string]any{"ref": "doc-1"}, Owner: "alice"})
|
||
if resp.Receipt.State != agentcommand.StateCommitted {
|
||
t.Fatalf("dispatch: %+v", resp.Receipt)
|
||
}
|
||
m, _ := resp.Payload.(map[string]any)
|
||
if m["echo"] != "doc-1" || m["owner"] != "alice" {
|
||
t.Fatalf("payload: %+v", resp.Payload)
|
||
}
|
||
if len(a.logs) != 1 || a.logs[0].Source != agentcommand.SourcePlugin {
|
||
t.Fatalf("audit: %+v", a.logs)
|
||
}
|
||
// 重复注册同一 ID：fail-closed。
|
||
if err := b.RegisterPluginCommand(agentcommand.CommandSpec{ID: "rdm.ingest", Title: "重复"}, nil); err == nil {
|
||
t.Fatal("duplicate plugin command must be rejected")
|
||
}
|
||
}
|
||
|
||
func TestPluginCommandWithoutHandlerRejected(t *testing.T) {
|
||
b := newTestBus(&memAudit{}, &memCards{})
|
||
// 声明式路径只登记规格（manifest provides.commands）。
|
||
if err := b.RegisterPluginCommands([]json.RawMessage{json.RawMessage(`{"id":"plug.only","title":"仅规格"}`)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := b.Dispatch(context.Background(), Request{ID: "plug.only"})
|
||
if resp.Receipt.State != agentcommand.StateRejected || !strings.Contains(resp.Receipt.Reason, "no host handler bound") {
|
||
t.Fatalf("unbound: %+v", resp.Receipt)
|
||
}
|
||
}
|
||
|
||
func TestHighDangerPluginCommandConfirmsFirst(t *testing.T) {
|
||
a := &memAudit{}
|
||
cards := &memCards{}
|
||
b := newTestBus(a, cards)
|
||
executed := 0
|
||
err := b.RegisterPluginCommand(agentcommand.CommandSpec{
|
||
ID: "rdm.purge", Title: "清库", Danger: agentcommand.DangerHigh,
|
||
}, func(ctx context.Context, req Request) (any, error) {
|
||
executed++
|
||
return "purged", nil
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 未确认：先出内联确认卡，处理器不得执行。
|
||
resp := b.Dispatch(context.Background(), Request{ID: "rdm.purge", ConversationID: "cnv_9", Owner: "alice"})
|
||
if resp.Receipt.State != agentcommand.StateAccepted || resp.Card == nil || resp.Card.Kind != core.CardKindConfirm {
|
||
t.Fatalf("pre-confirm: %+v card=%+v", resp.Receipt, resp.Card)
|
||
}
|
||
if executed != 0 {
|
||
t.Fatal("handler must not run before confirm")
|
||
}
|
||
// 确认后执行 committed。
|
||
resp2 := b.Confirm("alice", "card_x1", "confirm")
|
||
if resp2.Receipt.State != agentcommand.StateCommitted || resp2.Payload != "purged" || executed != 1 {
|
||
t.Fatalf("confirm: %+v executed=%d", resp2.Receipt, executed)
|
||
}
|
||
}
|
||
|
||
// ---- F10/U7b：确定未执行 vs 已执行后不确定 + operationID 对账 ----
|
||
|
||
// memOperationQuery 是按 owner/project 等值谓词过滤的操作记录替身（模拟 SQL
|
||
// 作用域）：跨 owner 或跨 project 的 operationID 一律查不到。
|
||
type memOperationQuery struct {
|
||
records map[string]OperationRecord // key: owner|projectID|operationID
|
||
}
|
||
|
||
func operationKey(owner, projectID, operationID string) string {
|
||
return owner + "|" + projectID + "|" + operationID
|
||
}
|
||
|
||
func (q *memOperationQuery) LookupOperation(owner, projectID, operationID string) (OperationRecord, bool, error) {
|
||
rec, ok := q.records[operationKey(owner, projectID, operationID)]
|
||
return rec, ok, nil
|
||
}
|
||
|
||
// recordAudit 把总线真实落下的审计信封登记为可查操作记录（对齐"审计信封就是
|
||
// 宿主侧操作记录"的口径），用于验证 unknown 回执给出的对账句柄真的查得到。
|
||
func (q *memOperationQuery) recordAudit(owner, projectID string, env agentcommand.AuditEnvelope) {
|
||
if env.OperationID == "" {
|
||
return
|
||
}
|
||
q.records[operationKey(owner, projectID, env.OperationID)] = OperationRecord{
|
||
OperationID: env.OperationID, CommandID: env.CommandID, State: env.State, Time: env.Time,
|
||
}
|
||
}
|
||
|
||
// operationHandle 取回 unknown 回执给出的对账句柄。agent-command main 契约禁止
|
||
// StateUnknown 携带 OperationID，句柄改写在 Reason 里（"(operation <id>)"），
|
||
// 这里统一读出来供断言与按记录对账使用。
|
||
func operationHandle(r agentcommand.ExecReceipt) string {
|
||
if r.OperationID != "" {
|
||
return r.OperationID
|
||
}
|
||
const marker = " (operation "
|
||
i := strings.Index(r.Reason, marker)
|
||
if i < 0 || !strings.HasSuffix(r.Reason, ")") {
|
||
return ""
|
||
}
|
||
return r.Reason[i+len(marker) : len(r.Reason)-1]
|
||
}
|
||
|
||
// TestHandlerErrorAfterStoreWriteIsUnknownWithOperationID 是 F10 的核心回归：
|
||
// handler 已经写库、之后才失败时，回执必须是 unknown 且携带它拿到的
|
||
// operationID，理由点明"执行开始后失败"，绝不退化成 rejected（会诱发不安全重试）。
|
||
//
|
||
// 上游 agent-command main 移除了 ExecReceipt.SideEffects 与 NewUnknownAfterExecution，
|
||
// 且 Validate 禁止 unknown 携带 OperationID；对账句柄改由 Reason 承载（见
|
||
// cmdbus.unknownReceipt），断言随之改写到新等价行为。
|
||
func TestHandlerErrorAfterStoreWriteIsUnknownWithOperationID(t *testing.T) {
|
||
audit := &memAudit{}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() })
|
||
var writes []string
|
||
var seen string
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.write", Title: "写入"}, "", func(_ context.Context, req Request) (any, error) {
|
||
seen = req.OperationID
|
||
writes = append(writes, req.OperationID) // 副作用已发生
|
||
return nil, errors.New("commit acknowledgement lost")
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "demo.write", Actor: "alice", Owner: "alice"})
|
||
if resp.Receipt.State != agentcommand.StateUnknown {
|
||
t.Fatalf("post-execution failure must not be reported as rejected: %+v", resp.Receipt)
|
||
}
|
||
if seen == "" || resp.Receipt.OperationID != "" || !strings.Contains(resp.Receipt.Reason, seen) {
|
||
t.Fatalf("unknown receipt lost the reconciliation handle: receipt=%+v handler=%q", resp.Receipt, seen)
|
||
}
|
||
if !strings.Contains(resp.Receipt.Reason, "after execution began") {
|
||
t.Fatalf("reason must name post-execution uncertainty: %q", resp.Receipt.Reason)
|
||
}
|
||
if err := resp.Receipt.Validate(); err != nil {
|
||
t.Fatalf("unknown receipt must satisfy the library contract: %v", err)
|
||
}
|
||
if len(writes) != 1 {
|
||
t.Fatalf("handler ran %d times, want 1", len(writes))
|
||
}
|
||
// 对账凭证：审计信封落 unknown + 同一个 operationID。
|
||
if len(audit.logs) != 1 || audit.logs[0].State != agentcommand.StateUnknown || audit.logs[0].OperationID != seen {
|
||
t.Fatalf("audit envelope is not reconcilable: %+v", audit.logs)
|
||
}
|
||
// 宿主不得自动重放：再次分发是一次新操作（新的 operationID），旧的不被复用。
|
||
again := bus.Dispatch(context.Background(), Request{ID: "demo.write", Actor: "alice", Owner: "alice"})
|
||
if again.Receipt.State != agentcommand.StateUnknown || operationHandle(again.Receipt) == operationHandle(resp.Receipt) || len(writes) != 2 {
|
||
t.Fatalf("re-dispatch must be a new operation, not a replay: %+v writes=%d", again.Receipt, len(writes))
|
||
}
|
||
}
|
||
|
||
// TestHandlerNotExecutedErrorStaysRejected：handler 显式声明未执行才是
|
||
// rejected 的依据；回执不得携带 operationID（库契约）。
|
||
func TestHandlerNotExecutedErrorStaysRejected(t *testing.T) {
|
||
audit := &memAudit{}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() })
|
||
calls := 0
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.guard", Title: "守卫"}, "", func(context.Context, Request) (any, error) {
|
||
calls++
|
||
boom := errors.New("permission denied")
|
||
return nil, NotExecuted(boom)
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "demo.guard", Owner: "alice", Actor: "alice"})
|
||
if resp.Receipt.State != agentcommand.StateRejected || resp.Receipt.OperationID != "" {
|
||
t.Fatalf("not-executed failure must stay rejected without an ID: %+v", resp.Receipt)
|
||
}
|
||
if resp.Receipt.Reason != "rejected: permission denied" {
|
||
t.Fatalf("not-executed marking must not distort the reason: %q", resp.Receipt.Reason)
|
||
}
|
||
if err := resp.Receipt.Validate(); err != nil {
|
||
t.Fatalf("rejected receipt must satisfy the library contract: %v", err)
|
||
}
|
||
if calls != 1 || len(audit.logs) != 1 || audit.logs[0].State != agentcommand.StateRejected || audit.logs[0].OperationID != "" {
|
||
t.Fatalf("unexpected effects: calls=%d audits=%+v", calls, audit.logs)
|
||
}
|
||
// 标记语义：errors.Is 同时命中 ErrNotExecuted 与原始错误；nil 幂等。
|
||
wrapped := NotExecuted(errors.New("boom"))
|
||
if !IsNotExecuted(wrapped) || !errors.Is(wrapped, ErrNotExecuted) || wrapped.Error() != "boom" {
|
||
t.Fatalf("not-executed wrapper semantics: %v / %v", wrapped, IsNotExecuted(wrapped))
|
||
}
|
||
if NotExecuted(nil) != nil || !IsNotExecuted(NotExecuted(wrapped)) {
|
||
t.Fatal("not-executed marking must be nil-safe and idempotent")
|
||
}
|
||
}
|
||
|
||
// TestCommittedWithAuditFailureKeepsOperationID 是 F10 Bug 2 的回归：审计
|
||
// 落盘失败时回执仍是 unknown，但必须带着已交给 handler 的 operationID，
|
||
// 理由点名审计不可用，调用方才能按记录对账而不是重发。
|
||
func TestCommittedWithAuditFailureKeepsOperationID(t *testing.T) {
|
||
audit := &memAudit{fail: true}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() })
|
||
var seen string
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.ping", Title: "Ping"}, "", func(_ context.Context, req Request) (any, error) {
|
||
seen = req.OperationID
|
||
return map[string]any{"pong": true}, nil
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "demo.ping", Owner: "alice", Actor: "alice"})
|
||
if resp.Receipt.State != agentcommand.StateUnknown || !strings.Contains(resp.Receipt.Reason, "audit unavailable") {
|
||
t.Fatalf("audit loss must stay unknown and name the audit: %+v", resp.Receipt)
|
||
}
|
||
if seen == "" || resp.Receipt.OperationID != "" || !strings.Contains(resp.Receipt.Reason, seen) {
|
||
t.Fatalf("audit loss dropped the reconciliation handle: receipt=%+v handler=%q", resp.Receipt, seen)
|
||
}
|
||
if err := resp.Receipt.Validate(); err != nil {
|
||
t.Fatalf("audit-loss receipt must satisfy the library contract: %v", err)
|
||
}
|
||
if len(audit.logs) != 0 {
|
||
t.Fatalf("audit rows written while failing: %+v", audit.logs)
|
||
}
|
||
// 校验：未执行路径（rejected）在审计不可用时不得凭空出现 operationID。
|
||
rejected := bus.Dispatch(context.Background(), Request{ID: "demo.absent"})
|
||
if rejected.Receipt.OperationID != "" || rejected.Receipt.State != agentcommand.StateUnknown {
|
||
t.Fatalf("audit loss must not invent an operation ID: %+v", rejected.Receipt)
|
||
}
|
||
}
|
||
|
||
// TestValidationRejectionNeverRunsHandler：显式校验拒绝仍是 rejected、无 ID、
|
||
// 无副作用，且 handler 一次都不许跑。
|
||
func TestValidationRejectionNeverRunsHandler(t *testing.T) {
|
||
audit := &memAudit{}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() })
|
||
calls := 0
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{
|
||
ID: "demo.needs", Title: "需要参数",
|
||
ParamsSchema: json.RawMessage(`{"type":"object","required":["target"],"properties":{"target":{"type":"string"}}}`),
|
||
}, "", func(context.Context, Request) (any, error) {
|
||
calls++
|
||
return "ran", nil
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "demo.needs", Owner: "alice", Params: map[string]any{}})
|
||
if resp.Receipt.State != agentcommand.StateRejected || resp.Receipt.OperationID != "" || resp.Receipt.Validate() != nil {
|
||
t.Fatalf("validation rejection: %+v", resp.Receipt)
|
||
}
|
||
if calls != 0 {
|
||
t.Fatalf("validation rejection ran the handler: calls=%d %+v", calls, resp.Receipt)
|
||
}
|
||
if len(audit.logs) != 1 || audit.logs[0].State != agentcommand.StateRejected || audit.logs[0].OperationID != "" {
|
||
t.Fatalf("validation audit row: %+v", audit.logs)
|
||
}
|
||
}
|
||
|
||
// TestRepeatedConfirmExecutesOnce：确认卡重复确认不得二次执行（F10 分类改动
|
||
// 不得放宽既有收口语义）；unknown 卡面也不得悬在待确认。
|
||
func TestRepeatedConfirmExecutesOnce(t *testing.T) {
|
||
audit, cards := &memAudit{}, &memCards{}
|
||
bus := New(audit, cards, func() time.Time { return time.Unix(1700000000, 0).UTC() })
|
||
executed := 0
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{
|
||
ID: "demo.once", Title: "只执行一次", Danger: agentcommand.DangerHigh,
|
||
}, "/once", func(context.Context, Request) (any, error) {
|
||
executed++
|
||
return "done", nil
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if r := bus.Dispatch(context.Background(), Request{ID: "demo.once", Owner: "alice", ConversationID: "cnv_1"}); r.Receipt.State != agentcommand.StateAccepted {
|
||
t.Fatalf("pre-confirm: %+v", r.Receipt)
|
||
}
|
||
first := bus.Confirm("alice", "card_x1", "confirm")
|
||
second := bus.Confirm("alice", "card_x1", "confirm")
|
||
if first.Receipt.State != agentcommand.StateCommitted || executed != 1 {
|
||
t.Fatalf("confirm: %+v executed=%d", first.Receipt, executed)
|
||
}
|
||
if second.Receipt.State != agentcommand.StateRejected || second.Receipt.OperationID != "" || executed != 1 {
|
||
t.Fatalf("re-confirm executed twice: %+v executed=%d", second.Receipt, executed)
|
||
}
|
||
}
|
||
|
||
// TestLookupOperationIsOwnerProjectScoped（R7/KTD6）：unknown 回执的对账查询
|
||
// 复用 owner/project 等值作用域；跨 owner、跨 project、缺作用域、未装配端口
|
||
// 一律 fail-closed，查不到绝不推断"未执行"。
|
||
func TestLookupOperationIsOwnerProjectScoped(t *testing.T) {
|
||
audit := &memAudit{}
|
||
query := &memOperationQuery{records: map[string]OperationRecord{}}
|
||
bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() })
|
||
if _, err := bus.LookupOperation("alice", "proj_1", "op_1"); !errors.Is(err, ErrOperationQueryUnavailable) {
|
||
t.Fatalf("unwired query must fail closed: %v", err)
|
||
}
|
||
bus.SetOperationQuery(query)
|
||
if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.write", Title: "写入"}, "", func(context.Context, Request) (any, error) {
|
||
// 副作用之后失败 → unknown + operationID（对账句柄）。
|
||
return nil, errors.New("ack lost")
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp := bus.Dispatch(context.Background(), Request{ID: "demo.write", Owner: "alice", Actor: "alice"})
|
||
if resp.Receipt.State != agentcommand.StateUnknown || operationHandle(resp.Receipt) == "" {
|
||
t.Fatalf("setup dispatch: %+v", resp.Receipt)
|
||
}
|
||
// 对账记录＝总线真实落下的审计信封（同一 owner/project 快照）。
|
||
for _, env := range audit.logs {
|
||
query.recordAudit("alice", "proj_1", env)
|
||
}
|
||
op := operationHandle(resp.Receipt)
|
||
record, err := bus.LookupOperation("alice", "proj_1", op)
|
||
if err != nil {
|
||
t.Fatalf("owner-scoped lookup: %v", err)
|
||
}
|
||
if record.OperationID != op || record.CommandID != "demo.write" || record.State != agentcommand.StateUnknown {
|
||
t.Fatalf("lookup returned the wrong record: %+v", record)
|
||
}
|
||
if _, err := bus.LookupOperation("bob", "proj_1", op); !errors.Is(err, ErrOperationNotRecorded) {
|
||
t.Fatalf("cross-owner lookup must not resolve: %v", err)
|
||
}
|
||
if _, err := bus.LookupOperation("alice", "proj_other", op); !errors.Is(err, ErrOperationNotRecorded) {
|
||
t.Fatalf("cross-project lookup must not resolve: %v", err)
|
||
}
|
||
for _, scope := range [][3]string{{"", "proj_1", op}, {"alice", "", op}, {"alice", "proj_1", ""}} {
|
||
if _, err := bus.LookupOperation(scope[0], scope[1], scope[2]); !errors.Is(err, ErrOperationScope) {
|
||
t.Fatalf("scope %v must fail closed: %v", scope, err)
|
||
}
|
||
}
|
||
if _, err := bus.LookupOperation("bob", "proj_1", "op_absent"); !errors.Is(err, ErrOperationNotRecorded) {
|
||
t.Fatalf("unknown operation must not be reported as executed: %v", err)
|
||
}
|
||
}
|