Report honest receipts when handlers err after side effects
Why: F10 - any handler error became a no-side-effect rejected receipt and an audit-store failure dropped the generated operation ID, losing the reconciliation handle and inviting unsafe retries. Evidence: cmdbus tests pass (verify-u7b.json); post-execution-unknown, audit-failure-keeps-ID and confirmation-idempotency cases added. Impact: unknown receipts now carry the trusted operation ID via the U7A constructor; rejected stays ID-free; no auto-replay anywhere. Refs: feature/agent-stack-audit-remediation-b1 U7B
This commit is contained in:
+173
-14
@@ -26,6 +26,43 @@ type AuditStore interface {
|
||||
AppendAudit(owner string, env agentcommand.AuditEnvelope) error
|
||||
}
|
||||
|
||||
// OperationRecord 是"已记录操作"的最小对账投影（R7/KTD6）。它回答的是
|
||||
// “这个 operationID 在本 owner/project 名下到底是什么结果”，而不是重放入口。
|
||||
// 刻意保持最小：审计信封只存状态与标识，不存理由文本，所以这里没有 Reason——
|
||||
// 不为不存在的字段留位置。
|
||||
type OperationRecord struct {
|
||||
OperationID string
|
||||
CommandID string
|
||||
State agentcommand.ResultState
|
||||
Time time.Time
|
||||
}
|
||||
|
||||
// OperationQuery 是 operationID 的对账查询端口（组合根装配；nil = 不提供）。
|
||||
//
|
||||
// 授权口径：owner 与 projectID 都是可信过滤条件，实现必须把它们放进查询谓词
|
||||
// （SQL 的 owner/project 等值条件），不得由调用方参数扩大可见范围；查不到就
|
||||
// 返回 not found，不得回退为“未执行”。
|
||||
//
|
||||
// 诚实边界：本端口只解析“已经落成记录”的 operation。宿主侧的命令回执记录就是
|
||||
// 命令审计信封（command_audit 按 owner+project 归属）；handler 自己的领域操作表
|
||||
// （如文档投递 document_render_deliveries）由各自领域查询负责，不经本端口。某些
|
||||
// 命令根本不持久化 req.OperationID（内置的 builtin.help、runtime.mock、
|
||||
// models.set、conversation.profile、conversation.create 都是无操作表的即时用例），
|
||||
// 对它们的 operationID 查询恒为 not found——这类命令只能按领域资源（会话、画像、
|
||||
// 设置项）对账，不得声称可按 operationID 查询。
|
||||
type OperationQuery interface {
|
||||
LookupOperation(owner, projectID, operationID string) (OperationRecord, bool, error)
|
||||
}
|
||||
|
||||
var (
|
||||
// ErrOperationScope 缺少可信 owner/project/operation 作用域：fail-closed，无作用域即无可见性。
|
||||
ErrOperationScope = errors.New("cmdbus: operation lookup requires owner, project and operation scope")
|
||||
// ErrOperationNotRecorded 该作用域内没有此 operation 的记录（未记录 ≠ 未执行）。
|
||||
ErrOperationNotRecorded = errors.New("cmdbus: no operation record in this owner and project scope")
|
||||
// ErrOperationQueryUnavailable 组合根未装配对账查询端口。
|
||||
ErrOperationQueryUnavailable = errors.New("cmdbus: operation query unavailable")
|
||||
)
|
||||
|
||||
// CardPort 确认卡所需卡片能力（cardsvc.Service 实现）。
|
||||
type CardPort interface {
|
||||
Create(owner string, card core.ActivityCard, ref string) (core.ActivityCard, error)
|
||||
@@ -35,6 +72,39 @@ type CardPort interface {
|
||||
// Handler 宿主执行适配器：同步内置用例返回 payload；异步执行自行返回 accepted 回执。
|
||||
type Handler func(ctx context.Context, req Request) (any, error)
|
||||
|
||||
// ErrNotExecuted 标记"失败确定发生在任何副作用之前"的 handler 错误（F10/U7b）。
|
||||
// 语义口径：只有带此标记的错误才可归 rejected（确定未执行、无副作用、回执不得
|
||||
// 携带 operationID）；未标记的 handler 错误一律按"已执行随后不确定"归 unknown，
|
||||
// 并带上本次分发已签发的 trusted operation ID 交由对账收敛。宿主不得据
|
||||
// "处理方报了错"臆断未执行——那会把已写库的操作诱使成不安全重试。
|
||||
//
|
||||
// 进程内插件（pkg/extension.Handler）当前还没有等价的声明位：插件命令的
|
||||
// handler 错误一律按 unknown 对账，直到扩展面补上对应标记为止。
|
||||
var ErrNotExecuted = errors.New("cmdbus: not executed")
|
||||
|
||||
// NotExecutedError 是带"确定未执行"声明的错误包装。Error() 保留原始文案
|
||||
// （回执 reason 不因标记而变形），Unwrap/Is 让 errors.Is 同时匹配
|
||||
// ErrNotExecuted 与被包装的原始错误。
|
||||
type NotExecutedError struct{ Err error }
|
||||
|
||||
func (e *NotExecutedError) Error() string { return e.Err.Error() }
|
||||
func (e *NotExecutedError) Unwrap() error { return e.Err }
|
||||
func (e *NotExecutedError) Is(target error) bool {
|
||||
return target == ErrNotExecuted
|
||||
}
|
||||
|
||||
// NotExecuted 声明该失败确定未执行（校验/权限/确认缺失/领域事务回滚等任何
|
||||
// 副作用之前的拒绝）。nil 原样返回；重复标记幂等。
|
||||
func NotExecuted(err error) error {
|
||||
if err == nil || IsNotExecuted(err) {
|
||||
return err
|
||||
}
|
||||
return &NotExecutedError{Err: err}
|
||||
}
|
||||
|
||||
// IsNotExecuted 报告该错误是否声明了"确定未执行"。
|
||||
func IsNotExecuted(err error) bool { return errors.Is(err, ErrNotExecuted) }
|
||||
|
||||
// CommandTool EXT-14：带 exposeAsTool 声明的命令的工具投影视图（runmgr 消费）。
|
||||
// Risk 取 low|medium|high（工具权限声明口径）；ID 即工具名（命令 id）。
|
||||
type CommandTool struct {
|
||||
@@ -171,6 +241,8 @@ type Bus struct {
|
||||
strictAccountWrites bool
|
||||
allowImplicitLocal bool
|
||||
projectWorkPreflight func(context.Context, Request) error
|
||||
// operationQuery operationID 对账查询端口（nil = 不提供，见 OperationQuery）。
|
||||
operationQuery OperationQuery
|
||||
}
|
||||
|
||||
func (b *Bus) SetProjectWorkPreflight(fn func(context.Context, Request) error) {
|
||||
@@ -200,6 +272,33 @@ func (b *Bus) Registry() *agentcommand.Registry { return b.reg }
|
||||
// SetPendingStore 注入确认挂起态的持久化实现（nil = 仅进程内）。
|
||||
func (b *Bus) SetPendingStore(st PendingStore) { b.pendingStore = st }
|
||||
|
||||
// SetOperationQuery 装配 operationID 对账查询端口（组合根；启动期一次）。
|
||||
func (b *Bus) SetOperationQuery(q OperationQuery) { b.operationQuery = q }
|
||||
|
||||
// LookupOperation 按 owner/project 作用域查一个已记录 operation 的当前状态
|
||||
// （R7/KTD6）。这是 unknown 回执的对账入口：拿到 operationID 的调用方先查
|
||||
// 记录，再决定下一步，而不是重发命令。
|
||||
//
|
||||
// 三重 fail-closed：作用域任一为空即拒（不猜、不按 owner=“”放行）、未装配端口
|
||||
// 即拒、查不到即 ErrOperationNotRecorded（不推断“未执行”）。跨 owner 查询由
|
||||
// 端口的 owner 等值谓词拒绝——本层只把 owner 当可信条件往下传。
|
||||
func (b *Bus) LookupOperation(owner, projectID, operationID string) (OperationRecord, error) {
|
||||
if owner == "" || projectID == "" || operationID == "" {
|
||||
return OperationRecord{}, ErrOperationScope
|
||||
}
|
||||
if b.operationQuery == nil {
|
||||
return OperationRecord{}, ErrOperationQueryUnavailable
|
||||
}
|
||||
record, found, err := b.operationQuery.LookupOperation(owner, projectID, operationID)
|
||||
if err != nil {
|
||||
return OperationRecord{}, err
|
||||
}
|
||||
if !found || record.OperationID == "" {
|
||||
return OperationRecord{}, ErrOperationNotRecorded
|
||||
}
|
||||
return record, nil
|
||||
}
|
||||
|
||||
// SetEffectiveAccountGuard installs the composition-root check at startup.
|
||||
// Handler Store writes still need their own final transaction fence.
|
||||
func (b *Bus) SetEffectiveAccountGuard(g EffectiveAccountGuard) { b.accountGuard = g }
|
||||
@@ -405,10 +504,12 @@ func (b *Bus) Dispatch(ctx context.Context, req Request) Response {
|
||||
if b.pendingStore != nil {
|
||||
raw, _ := json.Marshal(req)
|
||||
if err := b.savePending(card.ID, accountActor(req), string(raw)); err != nil {
|
||||
return receipt(agentcommand.StateUnknown, "", "confirm pending: "+err.Error())
|
||||
// 确认卡已创建（副作用已发生），挂起态却没落库：结果不确定，且
|
||||
// 卡 ID 就是本次受理的操作标识——随 unknown 一并给出，供对账。
|
||||
return receipt(agentcommand.StateUnknown, card.ID, "confirm pending: "+err.Error())
|
||||
}
|
||||
} else if b.strictAccountWrites {
|
||||
return receipt(agentcommand.StateUnknown, "", "confirm pending: account-checked store required")
|
||||
return receipt(agentcommand.StateUnknown, card.ID, "confirm pending: account-checked store required")
|
||||
}
|
||||
b.pendMu.Lock()
|
||||
b.pending[card.ID] = pending{req: req, cardID: card.ID}
|
||||
@@ -425,7 +526,9 @@ func (b *Bus) Dispatch(ctx context.Context, req Request) Response {
|
||||
}
|
||||
// The domain handler can persist this trusted ID with its transaction/outbox.
|
||||
// Entropy, not the clock or caller Params, distinguishes concurrent operations.
|
||||
// A handler failure still writes only a rejected audit receipt without an ID.
|
||||
// A handler error keeps the ID only on the "executed then uncertain" path
|
||||
// (unknown): rejected must never carry one, and unknown without a queryable
|
||||
// handle would be a dead end for reconciliation.
|
||||
var opBytes [24]byte
|
||||
if _, err := rand.Read(opBytes[:]); err != nil {
|
||||
return receipt(agentcommand.StateUnknown, "", "operation ID unavailable")
|
||||
@@ -433,11 +536,26 @@ func (b *Bus) Dispatch(ctx context.Context, req Request) Response {
|
||||
req.OperationID = "op_" + hex.EncodeToString(opBytes[:])
|
||||
payload, err := h(ctx, req)
|
||||
if err != nil {
|
||||
return receipt(agentcommand.StateRejected, "", "rejected: "+err.Error())
|
||||
if IsNotExecuted(err) {
|
||||
// handler 声明失败发生在任何副作用之前：确定未执行，可安全归
|
||||
// rejected（回执契约禁止 rejected 携带 operationID）。
|
||||
return receipt(agentcommand.StateRejected, "", "rejected: "+err.Error())
|
||||
}
|
||||
// handler 已开始执行才报错：宿主无法断言"无副作用"（F10）。按"已执行
|
||||
// 随后不确定"归 unknown，并带上本次分发已签发的 trusted operation ID，
|
||||
// 调用方据此按记录对账，而不是盲目重试。本路径不做任何自动重放。
|
||||
return b.unknownAfterExecution(req, "handler failed after execution began: "+err.Error(), start)
|
||||
}
|
||||
return Response{Receipt: b.auditReceipt(req, agentcommand.StateCommitted, req.OperationID, "", start), Payload: payload}
|
||||
}
|
||||
|
||||
// unknownAfterExecution 记录并返回"已执行随后不确定"的回执（F10/U7b）：审计
|
||||
// 信封带 state=unknown 与已签发的 operationID（可查凭证），回执走库内受信
|
||||
// 构造器 NewUnknownAfterExecution，保证 Validate() 通过且对账句柄不丢。
|
||||
func (b *Bus) unknownAfterExecution(req Request, reason string, start time.Time) Response {
|
||||
return Response{Receipt: b.auditReceipt(req, agentcommand.StateUnknown, req.OperationID, reason, start)}
|
||||
}
|
||||
|
||||
// Confirm 确认卡回道：actionID=confirm 执行挂起请求（entrance 归一 card），其余视为取消。
|
||||
func (b *Bus) Confirm(owner, cardID, actionID string) Response {
|
||||
return b.ConfirmContext(context.Background(), owner, cardID, actionID)
|
||||
@@ -456,7 +574,7 @@ func (b *Bus) ConfirmContext(ctx context.Context, owner, cardID, actionID string
|
||||
p, ok := b.pending[cardID]
|
||||
b.pendMu.Unlock()
|
||||
if ok && p.req.Owner != "" && p.req.Owner != owner {
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: p.req.ID, State: agentcommand.StateRejected, Reason: "confirm owner mismatch", Time: b.now().UTC()}}
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: p.req.ID, State: agentcommand.StateRejected, Reason: "confirm owner mismatch", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||||
}
|
||||
if b.pendingStore != nil {
|
||||
storedOwner, payload, found, err := b.takePending(cardID, owner)
|
||||
@@ -466,7 +584,7 @@ func (b *Bus) ConfirmContext(ctx context.Context, owner, cardID, actionID string
|
||||
if found {
|
||||
var req Request
|
||||
if storedOwner != owner || json.Unmarshal([]byte(payload), &req) != nil {
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "invalid pending confirm", Time: b.now().UTC()}}
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "invalid pending confirm", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||||
}
|
||||
p = pending{req: req, cardID: cardID}
|
||||
ok = true
|
||||
@@ -477,11 +595,11 @@ func (b *Bus) ConfirmContext(ctx context.Context, owner, cardID, actionID string
|
||||
ok = false
|
||||
}
|
||||
if !ok {
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "no pending confirm (expired or already resolved)", Time: b.now().UTC()}}
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: "unknown", State: agentcommand.StateRejected, Reason: "no pending confirm (expired or already resolved)", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||||
}
|
||||
req := p.req
|
||||
if req.Owner != "" && req.Owner != owner {
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: req.ID, State: agentcommand.StateRejected, Reason: "confirm owner mismatch", Time: b.now().UTC()}}
|
||||
return Response{Receipt: agentcommand.ExecReceipt{CommandID: req.ID, State: agentcommand.StateRejected, Reason: "confirm owner mismatch", SideEffects: agentcommand.SideEffectsNone, Time: b.now().UTC()}}
|
||||
}
|
||||
b.pendMu.Lock()
|
||||
delete(b.pending, cardID)
|
||||
@@ -497,15 +615,29 @@ func (b *Bus) ConfirmContext(ctx context.Context, owner, cardID, actionID string
|
||||
}
|
||||
resp := b.Dispatch(ctx, req)
|
||||
if b.cards != nil {
|
||||
if resp.Receipt.State == agentcommand.StateCommitted {
|
||||
switch resp.Receipt.State {
|
||||
case agentcommand.StateCommitted:
|
||||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardResolved, "ok", "confirmed")
|
||||
} else {
|
||||
case agentcommand.StateUnknown:
|
||||
// 已执行随后不确定：卡片必须落到终态（悬在 action_required 会诱导
|
||||
// 二次确认，而挂起项已被消费，重放只会得到"已解决"），但不得断言
|
||||
// 失败——tone 用 warn，note 写明不确定原因与对账用 operationID。
|
||||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardFailed, "warn", uncertainNote(resp.Receipt))
|
||||
default:
|
||||
_, _ = b.setConfirmCardStatus(owner, cardID, core.CardFailed, "danger", resp.Receipt.Reason)
|
||||
}
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
// uncertainNote 组成"结果未知"卡片的说明：不确定原因 + 对账句柄。
|
||||
func uncertainNote(r agentcommand.ExecReceipt) string {
|
||||
if r.OperationID == "" {
|
||||
return r.Reason
|
||||
}
|
||||
return r.Reason + " (operation " + r.OperationID + ")"
|
||||
}
|
||||
|
||||
func accountActor(req Request) string {
|
||||
// Plugin and delegated commands use Actor for an audit label such as a
|
||||
// plugin ID; Owner is the effective account whose state controls effects.
|
||||
@@ -528,7 +660,7 @@ func (b *Bus) checkEffectiveAccount(ctx context.Context, actor string) error {
|
||||
func (b *Bus) accountRejected(commandID string, start time.Time) Response {
|
||||
return Response{Receipt: agentcommand.ExecReceipt{
|
||||
CommandID: commandID, State: agentcommand.StateRejected,
|
||||
Reason: "effective account unavailable", Time: start.UTC(),
|
||||
Reason: "effective account unavailable", SideEffects: agentcommand.SideEffectsNone, Time: start.UTC(),
|
||||
}}
|
||||
}
|
||||
|
||||
@@ -555,10 +687,37 @@ func (b *Bus) auditReceipt(req Request, state agentcommand.ResultState, op, reas
|
||||
err = b.audit.AppendAudit(req.Owner, env)
|
||||
}
|
||||
if err != nil {
|
||||
// 审计不可用：结果标记未知，不臆断成功（四态语义）。
|
||||
return agentcommand.ExecReceipt{CommandID: req.ID, State: agentcommand.StateUnknown, Reason: "audit unavailable: " + err.Error(), Time: b.now().UTC()}
|
||||
// 审计不可用：结果标记未知，不臆断成功（四态语义）。已经签发过操作标识的
|
||||
// 路径（committed/accepted/已执行后不确定）连同该标识一并返回——信封丢失
|
||||
// 不等于操作没发生，调用方仍需按 operationID 对账；确定未执行的路径不带
|
||||
// 标识（rejected 契约禁止带 ID，unknown 无 ID 时也无需对账）。
|
||||
return unknownReceipt(req.ID, state, op, "audit unavailable: "+err.Error(), b.now().UTC())
|
||||
}
|
||||
return agentcommand.ExecReceipt{CommandID: req.ID, State: state, OperationID: op, Reason: reason, Time: b.now().UTC()}
|
||||
if state == agentcommand.StateUnknown && op != "" {
|
||||
// 已执行随后不确定：只有库内受信构造器允许 unknown 携带本流程签发的 ID。
|
||||
return agentcommand.NewUnknownAfterExecution(req.ID, op, reason, b.now().UTC())
|
||||
}
|
||||
return agentcommand.ExecReceipt{CommandID: req.ID, State: state, OperationID: op, Reason: reason,
|
||||
SideEffects: sideEffectsFor(state), Time: b.now().UTC()}
|
||||
}
|
||||
|
||||
// unknownReceipt 构造"结果未知"回执：已签发过操作标识时走受信构造器携带该标识
|
||||
// （供对账），否则返回不带标识的 unknown。state=rejected 恒不带标识。
|
||||
func unknownReceipt(commandID string, state agentcommand.ResultState, op, reason string, at time.Time) agentcommand.ExecReceipt {
|
||||
if op != "" && state != agentcommand.StateRejected {
|
||||
return agentcommand.NewUnknownAfterExecution(commandID, op, reason, at)
|
||||
}
|
||||
return agentcommand.ExecReceipt{CommandID: commandID, State: agentcommand.StateUnknown, Reason: reason, Time: at}
|
||||
}
|
||||
|
||||
// sideEffectsFor 回执的副作用声明：总线自身发起的 rejected 均为任何副作用之前
|
||||
// 的拒绝（未知命令/参数校验/项目前置/账号预检/无处理器/确认取消/handler 声明的
|
||||
// not-executed），故断言 none；committed/accepted 维持今日语义（未声明）。
|
||||
func sideEffectsFor(state agentcommand.ResultState) agentcommand.SideEffects {
|
||||
if state == agentcommand.StateRejected {
|
||||
return agentcommand.SideEffectsNone
|
||||
}
|
||||
return agentcommand.SideEffectsUnset
|
||||
}
|
||||
|
||||
func digest(params map[string]any) string {
|
||||
|
||||
Reference in New Issue
Block a user