Files
dingtalk-workspace-cli/internal/helpers/connect_approval_card.go
修雨 ee2fa735b2 feat(connect): confirm_policy — manual / auto / remember for others' requests
RoleConfig.confirm_policy was parsed but never acted on. Wire the three
strategies for how OTHERS' action requests are confirmed (the owner's own
requests always auto-run regardless):

- manual (default): ask the owner every time
- auto: run without asking — full trust — still fully audited
- remember: ask once per action verb, then reuse that decision (approve →
  auto-run same verb; reject → auto-decline same verb)

- gateDecision(requester, verb) returns auto / ask / reject
- ApprovalRequest.Verb carries the action verb (remember key + audit)
- handleOwnerDecision caches the decision per verb under the remember policy
- applyRoleConfig fills opts.ConfirmPolicy; launchConnector wires + logs it

Tests: auto runs without asking, manual asks, remember reuses the 1st decision.
2026-06-15 17:22:45 +08:00

858 lines
35 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package helpers
import (
"context"
"fmt"
"net/http"
"os"
"strings"
"sync"
"time"
"github.com/open-dingtalk/dingtalk-stream-sdk-go/card"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/executor"
)
// This file wires the confirmation gate to DingTalk:
// - approvalCardSender: the dependency-injected boundary that delivers the
// [Approve]/[Reject] card to the owner and updates it with the outcome.
// Tests inject a fake; production uses dingtalkApprovalCardSender (HTTP).
// - approvalOrchestrator: the connector-side glue. It owns the gate + sender +
// runner, decides whether an agent reply is an action request, drives the
// Submit → card → Await → execute → reply loop, and routes button callbacks
// into gate.Decide. connect_stream.go calls into it; it never imports stream
// internals back, so it stays unit-testable without a live connection.
// Button action constants. The card's two buttons each carry a fixed action
// id; the approval id and the decision travel as private action params so a
// callback can be associated and resolved without any server-side lookup table.
const (
approvalActionApprove = "dws_approval_approve"
approvalActionReject = "dws_approval_reject"
approvalParamID = "dwsApprovalId" // approval request id
approvalParamDecision = "dwsDecision" // "approve" | "reject"
approvalDecisionApprove = "approve"
approvalDecisionReject = "reject"
)
// approvalCardSender delivers and updates the owner-facing confirmation card.
// It is an interface so the orchestrator and its tests never touch the network:
// the real sender talks to the DingTalk card API, the fake records calls.
type approvalCardSender interface {
// SendApprovalCard delivers an interactive card with [Approve]/[Reject]
// buttons (each carrying req.ID + its decision in the action params) to the
// owner's 1:1 chat with the bot, returning the delivered card instance id
// (outTrackId) for later result updates and callback association.
SendApprovalCard(ctx context.Context, ownerUserID string, req *ApprovalRequest) (string, error)
// UpdateApprovalCard replaces the card body with the final outcome text
// (best-effort; an error must not fail the surrounding flow).
UpdateApprovalCard(ctx context.Context, outTrackID, text string) error
}
// approvalRunner is the subset of executor.Runner the orchestrator needs. Kept
// as a named alias so the dependency is explicit and easy to fake.
type approvalRunner interface {
Run(context.Context, executor.Invocation) (executor.Result, error)
}
// approvalReplier abstracts "send a plain text line back into the conversation"
// (the group reply after approve/reject). connect_stream.go supplies a closure
// over the chatbot replier + sessionWebhook; tests supply a recorder.
type approvalReplier func(ctx context.Context, convID, text string) error
// ownerNotifier sends a proactive 1:1 message to specific users (the owner, the
// requester), independent of any inbound sessionWebhook. Text approval needs it
// because the approval conversation (owner's 1:1 with the bot) is not the
// conversation the request arrived on, and because the decision can land minutes
// later when the original webhook is stale. *aiCardClient implements it.
type ownerNotifier interface {
sendOTOText(ctx context.Context, userIDs []string, text string) error
}
// auditSink records a terminal-state request to a durable, reviewable place
// (e.g. a DingTalk online sheet) so every action the twin takes is auditable
// beyond the local approvals JSON. Best-effort: a sink failure must never block
// or fail the action.
type auditSink interface {
record(ctx context.Context, req *ApprovalRequest)
}
// sheetAuditSink appends one row per action to a DingTalk online sheet (axls)
// via the sheet `append_rows` tool, run under the connector's (bot) identity.
// Columns: 时间 | 摘要 | 请求人 | 状态 | 批准人 | 是否自动 | 失败原因 | 单据ID.
type sheetAuditSink struct {
runner approvalRunner
nodeID string // axls doc id / URL
sheetID string // worksheet id or name (e.g. "Sheet1")
}
func (s *sheetAuditSink) record(ctx context.Context, req *ApprovalRequest) {
if s == nil || s.runner == nil || req == nil {
return
}
auto := ""
if req.AutoApproved {
auto = "自动"
}
decidedAt := ""
if !req.DecidedAt.IsZero() {
decidedAt = req.DecidedAt.Format("2006-01-02 15:04:05")
}
row := []any{
req.CreatedAt.Format("2006-01-02 15:04:05"),
req.Summary,
req.Requester,
string(req.State),
req.DecidedBy,
auto,
req.ExecErr,
req.ID,
decidedAt,
}
inv := executor.NewHelperInvocation("sheet append", "sheet", "append_rows", map[string]any{
"nodeId": s.nodeID,
"sheetId": s.sheetID,
"values": [][]any{row},
})
// The online-sheet API throttles (e.g. THREADPOOL_BUSY) under bursts, dropping
// an audit row on a transient error. Retry a few times with backoff so the
// trail stays complete; still best-effort — give up (log) after the last try.
var err error
for attempt := 1; attempt <= 3; attempt++ {
if _, err = s.runner.Run(ctx, inv); err == nil {
return
}
if attempt < 3 && isTransientSheetErr(err) {
time.Sleep(time.Duration(attempt) * 600 * time.Millisecond)
continue
}
break
}
if err != nil {
fmt.Fprintf(os.Stderr, "[connect][approval][audit] 写审计表格失败(已重试)approvalId=%s: %v\n", req.ID, err)
}
}
// isTransientSheetErr reports whether a sheet-append error is a transient
// throttle/timeout worth retrying (vs. a permanent error like a bad node id).
func isTransientSheetErr(err error) bool {
if err == nil {
return false
}
s := strings.ToUpper(err.Error())
for _, sig := range []string{"BUSY", "THREADPOOL", "TIMEOUT", "RATE", "LIMIT", "TOO MANY", "503", "429"} {
if strings.Contains(s, sig) {
return true
}
}
return false
}
// approvalOrchestrator is the connector-side controller for the gate. It is
// constructed per connector with the owner's userId, the gate, the card sender
// and the runner. Zero owner / nil sender disables the gate (the connector then
// behaves exactly as before — plain Q&A).
type approvalOrchestrator struct {
gate *approvalGate
sender approvalCardSender
runner approvalRunner
ownerUserID string
awaitWindow time.Duration
// textMode swaps the interactive [Approve]/[Reject] card for a private
// text confirmation: the bot DMs the OWNER "X 请求执行……" and the owner
// replies "同意"/"拒绝" in their own 1:1 chat with the bot. The requester
// never sees the approval — they only get the final result. Used when no
// approval-card template is configured, so the gate works with zero
// card-platform setup. The decision is captured asynchronously (see
// handleOwnerDecision), not by blocking on Await.
textMode bool
// notifier delivers the proactive owner/requester 1:1 messages text mode
// needs (nil in card mode).
notifier ownerNotifier
// audit, when set, records every terminal-state request to a durable sink
// (e.g. a DingTalk online sheet) on top of the local approvals JSON. nil =
// local-file audit only.
audit auditSink
// allowedScopes is the role's capability allowlist (product names). When
// non-empty, an action whose product is not listed is refused before it ever
// reaches the gate, keeping the role in its lane. Empty = no restriction.
allowedScopes []string
// confirmPolicy governs how OTHERS' requests are confirmed: "manual" (ask
// every time, the default), "auto" (run without asking, still audited),
// "remember" (ask once per action verb then reuse). The owner's own requests
// always auto-run regardless.
confirmPolicy string
// remembered caches the owner's decision per action verb for the "remember"
// policy (verb → approved). Guarded by rememberMu.
rememberMu sync.Mutex
remembered map[string]bool
}
// gateDecision decides how to handle a write action: "auto" (run now), "ask"
// (route to the owner), or "reject" (a remembered rejection). The owner's own
// requests always auto-run. For others it follows confirmPolicy.
func (o *approvalOrchestrator) gateDecision(requester, verb string) string {
if strings.TrimSpace(requester) != "" && strings.TrimSpace(requester) == o.ownerUserID {
return "auto"
}
switch strings.ToLower(strings.TrimSpace(o.confirmPolicy)) {
case "auto":
return "auto"
case "remember":
o.rememberMu.Lock()
decided, ok := o.remembered[verb]
o.rememberMu.Unlock()
if !ok {
return "ask"
}
if decided {
return "auto"
}
return "reject"
default: // manual / empty
return "ask"
}
}
// rememberDecision records the owner's decision for an action verb so the
// "remember" policy can reuse it. No-op unless the policy is "remember".
func (o *approvalOrchestrator) rememberDecision(verb string, approved bool) {
if strings.ToLower(strings.TrimSpace(o.confirmPolicy)) != "remember" || strings.TrimSpace(verb) == "" {
return
}
o.rememberMu.Lock()
if o.remembered == nil {
o.remembered = make(map[string]bool)
}
o.remembered[verb] = approved
o.rememberMu.Unlock()
}
// scopeAllows reports whether the role may use the given product. An empty
// allowlist means no restriction (allow all).
func (o *approvalOrchestrator) scopeAllows(product string) bool {
if len(o.allowedScopes) == 0 {
return true
}
product = strings.ToLower(strings.TrimSpace(product))
for _, s := range o.allowedScopes {
if strings.ToLower(strings.TrimSpace(s)) == product {
return true
}
}
return false
}
// recordAudit writes the request's current (terminal) state to the audit sink,
// fetching the fresh snapshot so state/exec_err reflect the final outcome.
// No-op when no sink is configured; best-effort.
func (o *approvalOrchestrator) recordAudit(ctx context.Context, id string) {
if o == nil || o.audit == nil {
return
}
if r := o.gate.Get(id); r != nil {
o.audit.record(ctx, r)
}
}
// newApprovalOrchestrator builds the card-mode controller. awaitWindow caps how
// long the connector blocks for the owner's decision before giving up (the
// request stays Pending on disk, so a late tap still records and executes
// nothing twice).
func newApprovalOrchestrator(gate *approvalGate, sender approvalCardSender, runner approvalRunner, ownerUserID string) *approvalOrchestrator {
return &approvalOrchestrator{
gate: gate,
sender: sender,
runner: runner,
ownerUserID: strings.TrimSpace(ownerUserID),
awaitWindow: 10 * time.Minute,
}
}
// newTextApprovalOrchestrator builds the text-mode controller (no card sender).
// notifier delivers the private owner/requester messages. The owner confirms by
// replying "同意"/"拒绝" in their 1:1 chat with the bot; the decision is captured
// by handleOwnerDecision on the next inbound message, so handleReply must NOT
// block waiting for it.
func newTextApprovalOrchestrator(gate *approvalGate, runner approvalRunner, ownerUserID string, notifier ownerNotifier) *approvalOrchestrator {
o := newApprovalOrchestrator(gate, nil, runner, ownerUserID)
o.textMode = true
o.notifier = notifier
return o
}
// enabled reports whether the gate is active. Card mode needs an owner + a card
// sender; text mode needs an owner + a notifier to reach them privately.
func (o *approvalOrchestrator) enabled() bool {
if o == nil || o.ownerUserID == "" || o.gate == nil {
return false
}
if o.textMode {
return o.notifier != nil
}
return o.sender != nil
}
// agentSystemHint is appended to the forwarded prompt so the agent emits the
// structured action marker (see parseActionMarker) instead of executing
// anything itself. It is the simplified "intent detection" for the slice.
const agentSystemHint = "\n\n[系统提示] 如果用户的请求是要执行一个动作(而不是单纯提问)," +
"不要假装已经完成,而是在回复末尾用下面其中一个标记声明这个动作,由主人确认后再执行:\n" +
`· 建待办:[[ACTION:todo.create title="待办标题" due="2026-06-14T18:00:00+08:00"]](due 可省略)` + "\n" +
`· 建日程:[[ACTION:calendar.create title="日程标题" start="2026-06-20T10:00:00+08:00" end="2026-06-20T11:00:00+08:00"]]` + "\n" +
`· 建文档:[[ACTION:doc.create name="文档名"]]` + "\n" +
"时间一律用 ISO-8601(带时区)。如果只是普通提问,正常回答即可,不要输出任何标记。"
// decorateForActionDetection appends the action-marker instruction to a prompt
// when the gate is enabled. A no-op when disabled, so plain Q&A is untouched.
func (o *approvalOrchestrator) decorateForActionDetection(prompt string) string {
if !o.enabled() {
return prompt
}
return prompt + agentSystemHint
}
// handleReply is the orchestration hook the connector calls with the agent's
// raw reply. It returns (finalReply, handled): when handled is true the gate
// took over (it has already sent the card / will reply asynchronously via
// reply) and the connector must NOT send finalReply itself. When false the
// connector replies normally with finalReply (the marker, if any, is stripped).
//
// requester is the staffId who asked; convID is where the group answer goes;
// reply is the function used to post the approve/reject outcome back.
func (o *approvalOrchestrator) handleReply(ctx context.Context, requester, convID, agentReply string, reply approvalReplier) (string, bool) {
if !o.enabled() {
return agentReply, false
}
act, cleaned, found := parseActionMarker(agentReply)
if !found {
return agentReply, false // plain Q&A
}
pa, summary, ok := toPlannedAction(act, o.ownerUserID)
if !ok {
// Unknown/blank action: degrade to a plain reply (marker stripped) so a
// malformed marker never silently swallows the answer.
return cleaned, false
}
// Role scope: an action outside the role's capability allowlist is refused
// up front (never reaches the gate or the owner). The requester is told it is
// out of lane — this is a capability boundary, not an approval, so it is fine
// to surface to them.
if !o.scopeAllows(pa.Product) {
fmt.Fprintf(os.Stderr, "[connect][approval] 越权拦截:角色无 %q 能力,拒绝动作 %s\n", pa.Product, summary)
return fmt.Sprintf("(该数字员工没有「%s」能力,无法执行此操作:%s)", pa.Product, summary), false
}
// 写类才拦:route the gate-or-not decision through the read/write classifier
// instead of gating every detected marker. A definitively read-class action
// is safe and bypasses the gate (the marker is stripped and the reply goes
// out normally). Write — and Unknown, per the CmdClass safety contract — keep
// the owner's sign-off requirement.
if classifyPlannedAction(pa) == CmdClassRead {
return cleaned, false
}
// Confirmation strategy: owner-self always auto-runs; others follow
// confirmPolicy (manual=ask / auto=run / remember=reuse last decision).
decision := o.gateDecision(requester, act.Verb)
req := o.gate.Submit(ApprovalRequest{
Requester: requester,
ConvID: convID,
Summary: summary,
Verb: act.Verb,
Action: pa,
AutoApproved: decision == "auto",
})
switch decision {
case "auto":
// Asking IS the authorization (owner-self), or the policy pre-authorized
// it (auto / remembered-approve). Execute directly; still fully audited.
return o.autoApproveAndExecute(ctx, req, convID, reply)
case "reject":
// Remembered rejection for this action kind — decline without bothering
// the owner, and tell the requester.
o.gate.Decide(req.ID, false, o.ownerUserID)
o.recordAudit(ctx, req.ID)
fmt.Fprintf(os.Stderr, "[connect][approval] 记忆策略:%q 此前被拒,自动拒绝 approvalId=%s\n", act.Verb, req.ID)
return fmt.Sprintf("(主人此前已拒绝这类操作,本次未执行:%s)", summary), false
default: // "ask"
// Text mode: privately DM the owner and return immediately — the owner's
// "同意/拒绝" reply is captured asynchronously by handleOwnerDecision
// (blocking here would deadlock the per-conversation worker that must also
// process that reply). The requester (this conversation) sees nothing.
if o.textMode {
return o.handleReplyText(ctx, req)
}
return o.handleReplyCard(ctx, req, convID, reply)
}
}
// autoApproveAndExecute runs an owner's own request without a second
// confirmation, while still recording the full lifecycle (Decide by the owner →
// execute) so the on-disk approvals log audits it like any other action. It
// replies into the conversation the owner asked in (the inbound webhook is still
// fresh — the owner just messaged). Used for both card and text mode.
func (o *approvalOrchestrator) autoApproveAndExecute(ctx context.Context, req *ApprovalRequest, convID string, reply approvalReplier) (string, bool) {
decided, _ := o.gate.Decide(req.ID, true, o.ownerUserID)
if decided == nil {
decided = req
}
fmt.Fprintf(os.Stderr, "[connect][approval][audit] 主人自助操作,自动执行 approvalId=%s owner=%s: %s\n",
decided.ID, o.ownerUserID, decided.Summary)
out, execErr := o.execute(ctx, decided)
if execErr != nil {
// Do not lose it: hold for retry and tell the owner to recover. Most
// likely the connector's dws login is not the bot owner, so an
// owner-scoped action can't run yet.
o.gate.markDeferred(decided.ID, execErr.Error())
o.recordAudit(ctx, decided.ID)
o.notifyOwnerDeferred(ctx, decided, execErr)
if reply != nil {
_ = reply(ctx, convID, "收到,但现在没能直接完成(已记下,稍后会补做)。")
}
return "", true
}
o.gate.markExecuted(decided.ID)
o.recordAudit(ctx, decided.ID)
if reply != nil {
_ = reply(ctx, convID, "已为你完成:"+decided.Summary+"\n"+out)
}
return "", true
}
// notifyOwnerDeferred privately tells the owner that an action could not run now
// and is being held, with the likely cause and how to recover (log dws in as the
// bot owner, then reply "重试"). Best-effort.
func (o *approvalOrchestrator) notifyOwnerDeferred(ctx context.Context, req *ApprovalRequest, execErr error) {
if o.notifier == nil {
return
}
who := strings.TrimSpace(req.Requester)
if who == "" || who == o.ownerUserID {
who = "你"
}
msg := fmt.Sprintf("⚠️ %s 请求执行:%s\n但我现在没能以你的身份完成(很可能这台连接器的 dws 没有登录成你本人的账号)。\n"+
"已先记下,不会丢。把 dws 登录成机器人主人账号后会自动补做;想立即补做可回复「重试」。\n(错误:%s)",
who, req.Summary, truncateRunes(execErr.Error(), 120))
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, msg)
}
// handleReplyText privately DMs the OWNER the approval request and hands control
// back to the connector (handled=true). The requester sees nothing — the
// approval happens only between the owner and the bot. It never blocks: the
// owner's decision arrives later as an inbound message routed through
// handleOwnerDecision.
func (o *approvalOrchestrator) handleReplyText(ctx context.Context, req *ApprovalRequest) (string, bool) {
who := strings.TrimSpace(req.Requester)
if who == "" {
who = "有人"
}
prompt := fmt.Sprintf("🔔 %s 请求执行一个操作,需要你确认:\n%s\n回复「同意」执行,或「拒绝」取消。", who, req.Summary)
if err := o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, prompt); err != nil {
// Cannot reach the owner: the action does not run. Do NOT leak the
// approval to the requester — text mode's whole point is privacy — so
// just log and swallow (handled), leaving the request Pending on disk.
fmt.Fprintf(os.Stderr, "[connect][approval] 私聊主人 %s 失败,本次未执行: %v\n", o.ownerUserID, err)
return "", true
}
fmt.Fprintf(os.Stderr, "[connect][approval] 已私聊主人 %s 待确认 approvalId=%s requester=%s: %s\n",
o.ownerUserID, req.ID, req.Requester, req.Summary)
return "", true
}
// handleOwnerDecision intercepts an inbound message that is the OWNER replying
// "同意/拒绝" (in their own 1:1 chat with the bot) to the pending request. It
// decides, executes-or-declines, privately acks the owner, and sends the
// outcome to the original requester. Returns true when it consumed the message
// (so the connector skips forwarding it to the agent). A non-owner sender, a
// non-decision message, or no pending request all return false (ordinary
// message). Only active in text mode.
func (o *approvalOrchestrator) handleOwnerDecision(ctx context.Context, senderStaffID, text string) bool {
if o == nil || !o.textMode || !o.enabled() {
return false
}
if senderStaffID == "" || senderStaffID != o.ownerUserID {
return false
}
// "重试/恢复": the owner has recovered — flush the deferred backlog.
if isRetryWord(text) {
return o.flushDeferred(ctx)
}
approve, ok := parseDecisionWord(text)
if !ok {
return false
}
req := o.gate.latestPending()
if req == nil {
return false
}
decided, deciding := o.gate.Decide(req.ID, approve, senderStaffID)
if !deciding {
// Already decided (e.g. a double reply): consume the keyword, do not
// re-execute.
return decided != nil
}
// "remember" policy: cache this decision for the action kind so the next
// same-verb request reuses it without asking again.
o.rememberDecision(decided.Verb, approve)
if !decided.approved() {
o.recordAudit(ctx, decided.ID)
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, "好的,已拒绝,未执行。")
o.notifyRequester(ctx, decided, "你的请求未获主人批准,本次未执行。")
return true
}
out, execErr := o.execute(ctx, decided)
if execErr != nil {
// Approved but could not run now → hold for retry, tell the owner how to
// recover. The requester is NOT told it failed (their task is not lost).
o.gate.markDeferred(decided.ID, execErr.Error())
o.recordAudit(ctx, decided.ID)
o.notifyOwnerDeferred(ctx, decided, execErr)
return true
}
o.gate.markExecuted(decided.ID)
o.recordAudit(ctx, decided.ID)
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, "已同意并执行完成:"+decided.Summary)
o.notifyRequester(ctx, decided, "已为你完成:"+decided.Summary+"\n"+out)
return true
}
// flushDeferred replays the deferred backlog after the owner recovers. Each
// request is re-executed; on success it is marked executed and the original
// requester gets the outcome, on failure it stays deferred (still not lost).
// The owner gets a summary. Returns true (the "重试" message is always consumed).
func (o *approvalOrchestrator) flushDeferred(ctx context.Context) bool {
pending := o.gate.allDeferred()
if len(pending) == 0 {
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, "当前没有积压的请求。")
return true
}
done, stuck := o.flushDeferredOnce(ctx, pending)
// Manual "重试" always reports back (the owner asked), naming what was done.
var b strings.Builder
fmt.Fprintf(&b, "已补做 %d 个积压请求", len(done))
if len(done) > 0 {
b.WriteString(":\n")
for i, s := range done {
fmt.Fprintf(&b, "%d. %s\n", i+1, s)
}
} else {
b.WriteString("。")
}
if stuck > 0 {
fmt.Fprintf(&b, "仍有 %d 个未成功(身份可能还没对上,确认 dws 登录后再回复「重试」)。", stuck)
}
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, strings.TrimRight(b.String(), "\n"))
return true
}
// flushDeferredOnce re-executes each given deferred request once. Successful
// ones are marked executed and their requester is notified; failed ones stay
// deferred. Returns the completed summaries and the still-stuck count. Shared by
// the manual "重试" path and the background auto-retry.
func (o *approvalOrchestrator) flushDeferredOnce(ctx context.Context, pending []*ApprovalRequest) (done []string, stuck int) {
for _, req := range pending {
out, execErr := o.execute(ctx, req)
if execErr != nil {
o.gate.markDeferred(req.ID, execErr.Error())
o.recordAudit(ctx, req.ID)
stuck++
continue
}
o.gate.markExecuted(req.ID)
o.recordAudit(ctx, req.ID)
o.notifyRequester(ctx, req, "(已恢复)已为你完成:"+req.Summary+"\n"+out)
done = append(done, req.Summary)
}
return done, stuck
}
// autoFlushDeferred is the background-retry pass: it replays the backlog and
// messages the owner ONLY when something actually completed (so a periodic tick
// while the identity is still wrong stays silent — no spam). The requester is
// still notified per completed item inside flushDeferredOnce.
func (o *approvalOrchestrator) autoFlushDeferred(ctx context.Context) {
pending := o.gate.allDeferred()
if len(pending) == 0 {
return
}
done, _ := o.flushDeferredOnce(ctx, pending)
if len(done) == 0 {
return
}
var b strings.Builder
fmt.Fprintf(&b, "(已自动恢复)补做了 %d 个积压请求:\n", len(done))
for i, s := range done {
fmt.Fprintf(&b, "%d. %s\n", i+1, s)
}
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, strings.TrimRight(b.String(), "\n"))
}
// startAutoRetry runs autoFlushDeferred on an interval until ctx is cancelled,
// so a deferred backlog drains by itself once the owner's identity comes back —
// no manual "重试" needed. A no-op when there is no notifier (card mode) or a
// non-positive interval.
func (o *approvalOrchestrator) startAutoRetry(ctx context.Context, interval time.Duration) {
if o == nil || o.notifier == nil || interval <= 0 {
return
}
go func() {
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
o.autoFlushDeferred(ctx)
}
}
}()
}
// notifyRequester sends the outcome to the original requester, unless they are
// the owner themselves (who already got the owner-side ack — no need to double
// message). Best-effort.
func (o *approvalOrchestrator) notifyRequester(ctx context.Context, req *ApprovalRequest, text string) {
who := strings.TrimSpace(req.Requester)
if who == "" || who == o.ownerUserID {
return
}
_ = o.notifier.sendOTOText(ctx, []string{who}, text)
}
// handleReplyCard runs the interactive-card flow: deliver the [Approve]/[Reject]
// card to the owner, block for the decision, then execute-or-decline and report
// the outcome into the conversation.
func (o *approvalOrchestrator) handleReplyCard(ctx context.Context, req *ApprovalRequest, convID string, reply approvalReplier) (string, bool) {
outTrackID, err := o.sender.SendApprovalCard(ctx, o.ownerUserID, req)
if err != nil {
fmt.Fprintf(os.Stderr, "[connect][approval] 确认卡片投放失败: %v\n", err)
// Cannot ask the owner → tell the requester the gate could not engage,
// rather than silently executing or silently dropping.
return "(需要主人确认,但确认卡片发送失败,本次未执行)", false
}
o.gate.setOutTrackID(req.ID, outTrackID)
fmt.Fprintf(os.Stderr, "[connect][approval] 已向主人 %s 发送确认卡片 approvalId=%s outTrackId=%s: %s\n",
o.ownerUserID, req.ID, outTrackID, req.Summary)
// Block for the owner's decision off the connector's per-conversation
// worker is fine: the worker already serializes one conversation, and the
// gate persists so a restart mid-wait loses only the in-memory block.
decided, gotDecision := o.gate.Await(ctx, req.ID, o.awaitWindow)
if !gotDecision || decided == nil {
_ = reply(ctx, convID, "(等待主人确认超时,本次未执行)")
return "", true
}
if !decided.approved() {
o.recordAudit(ctx, decided.ID)
_ = o.sender.UpdateApprovalCard(ctx, decided.OutTrackID, "已拒绝:未执行。")
_ = reply(ctx, convID, "主人暂时没批准。")
return "", true
}
// Approved → the orchestrator runs the planned command directly (no agent
// round-trip), then reports the outcome to the group and the card.
out, execErr := o.execute(ctx, decided)
if execErr != nil {
o.gate.markFailed(decided.ID, execErr.Error())
o.recordAudit(ctx, decided.ID)
_ = o.sender.UpdateApprovalCard(ctx, decided.OutTrackID, "已同意,但执行失败:"+execErr.Error())
_ = reply(ctx, convID, "主人已同意,但执行失败:"+execErr.Error())
return "", true
}
o.gate.markExecuted(decided.ID)
o.recordAudit(ctx, decided.ID)
_ = o.sender.UpdateApprovalCard(ctx, decided.OutTrackID, "已同意并执行完成。")
_ = reply(ctx, convID, "主人已同意,已执行:"+decided.Summary+"\n"+out)
return "", true
}
// execute runs the request's planned action via the runner and returns a short
// human-readable result line. It is the only place the gate touches the
// executor, so adding a new approvable verb is a matter of toPlannedAction +
// the runner already supporting that tool.
func (o *approvalOrchestrator) execute(ctx context.Context, req *ApprovalRequest) (string, error) {
if o.runner == nil {
return "", fmt.Errorf("no runner configured")
}
inv := executor.NewHelperInvocation(req.Action.LegacyPath, req.Action.Product, req.Action.Tool, req.Action.Params)
res, err := o.runner.Run(ctx, inv)
if err != nil {
return "", err
}
if !res.Invocation.Implemented {
// The helper override was not wired at runtime (e.g. dry-run / discovery
// gap): surface it rather than claim success.
return "(动作已受理,但运行时未实际执行——请检查命令是否在当前环境可用)", nil
}
return "(待办已创建)", nil
}
// handleCardCallback maps a button tap to gate.Decide. It pulls the approval id
// and decision from the action params (primary association), falling back to the
// card instance id (OutTrackId) when the params are absent. Returns a card
// response that flips the buttons to a decided state. Safe to call repeatedly:
// a second tap after a decision is a no-op (Decide is idempotent).
func (o *approvalOrchestrator) handleCardCallback(ctx context.Context, req *card.CardRequest) (*card.CardResponse, error) {
if o == nil || o.gate == nil || req == nil {
return &card.CardResponse{}, nil
}
id, approve, ok := decodeCardAction(req)
if !ok {
fmt.Fprintf(os.Stderr, "[connect][approval] 卡片回调无法解析动作 outTrackId=%s\n", req.OutTrackId)
return &card.CardResponse{}, nil
}
if id == "" {
// No approval id in the params: associate by the card instance id.
if found := o.gate.findByOutTrackID(req.OutTrackId); found != nil {
id = found.ID
}
}
if id == "" {
fmt.Fprintf(os.Stderr, "[connect][approval] 卡片回调无法关联审批 outTrackId=%s\n", req.OutTrackId)
return &card.CardResponse{}, nil
}
decided, deciding := o.gate.Decide(id, approve, strings.TrimSpace(req.UserId))
verb := "同意"
if !approve {
verb = "拒绝"
}
if deciding {
fmt.Fprintf(os.Stderr, "[connect][approval] 主人 %s 点了[%s] approvalId=%s\n", req.UserId, verb, id)
} else if decided != nil {
fmt.Fprintf(os.Stderr, "[connect][approval] 审批 %s 已是终态(%s),忽略重复点击\n", id, decided.State)
}
// Echo the decision back into the card's private data so the renderer can
// show a decided state. Best-effort; the gate state is the source of truth.
return &card.CardResponse{
UserPrivateData: &card.CardDataDto{
CardParamMap: map[string]string{"dwsDecision": verb},
},
}, nil
}
// decodeCardAction extracts (approvalId, approve, ok) from a card callback. It
// prefers the explicit decision param, then the tapped action id, so the card
// can carry the decision either way. ok is false when neither identifies a
// decision (a non-approval card, or a malformed payload).
func decodeCardAction(req *card.CardRequest) (id string, approve bool, ok bool) {
id = strings.TrimSpace(req.GetActionString(approvalParamID))
decision := strings.TrimSpace(req.GetActionString(approvalParamDecision))
switch decision {
case approvalDecisionApprove:
return id, true, true
case approvalDecisionReject:
return id, false, true
}
// Fall back to which button was pressed (its action id).
for _, a := range req.CardActionData.CardPrivateData.ActionIdList {
switch strings.TrimSpace(a) {
case approvalActionApprove:
return id, true, true
case approvalActionReject:
return id, false, true
}
}
return id, false, false
}
// ---- Real (HTTP) card sender ----
// dingtalkApprovalCardSender delivers the interactive approval card via the
// DingTalk card API, reusing aiCardClient for auth + the create/deliver/update
// HTTP plumbing. The card template must define two buttons whose action params
// the connector fills in (approval id + decision); see the docstring on
// SendApprovalCard for the contract.
type dingtalkApprovalCardSender struct {
cli *aiCardClient
templateID string
}
// newDingtalkApprovalCardSender builds the real sender. An empty templateID
// returns nil: without an interactive-card template the connector cannot render
// buttons, so the caller must keep the gate disabled rather than deliver a card
// the owner cannot act on.
func newDingtalkApprovalCardSender(clientID, clientSecret, templateID string) approvalCardSender {
templateID = strings.TrimSpace(templateID)
if templateID == "" {
return nil
}
return &dingtalkApprovalCardSender{
cli: newAICardClient(clientID, clientSecret, templateID),
templateID: templateID,
}
}
// SendApprovalCard creates an interactive card instance and delivers it to the
// owner's 1:1 chat with the bot. The approval id and per-button decision are
// passed in cardData.cardParamMap so the template's [Approve]/[Reject] buttons
// echo them back as action params on tap (see decodeCardAction). Best-effort
// result-check via callChecked surfaces a per-target deliver failure that the
// card API hides inside an HTTP 200.
func (s *dingtalkApprovalCardSender) SendApprovalCard(ctx context.Context, ownerUserID string, req *ApprovalRequest) (string, error) {
outTrackID := "dws_approval_" + req.ID
create := map[string]any{
"cardTemplateId": s.templateID,
"outTrackId": outTrackID,
"cardData": map[string]any{
"cardParamMap": map[string]any{
"title": "需要你确认",
"summary": req.Summary,
"requester": req.Requester,
approvalParamID: req.ID,
"approveDecision": approvalDecisionApprove,
"rejectDecision": approvalDecisionReject,
"approveActionId": approvalActionApprove,
"rejectActionId": approvalActionReject,
"config": `{"autoLayout":true}`,
},
},
"callbackType": "STREAM",
"imRobotOpenSpaceModel": map[string]any{"supportForward": true},
}
if err := s.cli.call(ctx, http.MethodPost, "/v1.0/card/instances", create); err != nil {
return "", err
}
deliver := map[string]any{
"outTrackId": outTrackID,
"userIdType": 1,
"openSpaceId": "dtv1.card//IM_ROBOT." + ownerUserID,
"imRobotOpenDeliverModel": map[string]any{
"spaceType": "IM_ROBOT",
"robotCode": s.cli.clientID,
},
}
if err := s.cli.callChecked(ctx, http.MethodPost, "/v1.0/card/instances/deliver", deliver); err != nil {
return "", err
}
return outTrackID, nil
}
// UpdateApprovalCard rewrites the card's result line after a decision. It is
// best-effort: a stuck card is bad UX but must never fail the gate flow, so
// callers ignore the error.
func (s *dingtalkApprovalCardSender) UpdateApprovalCard(ctx context.Context, outTrackID, text string) error {
return s.cli.call(ctx, http.MethodPut, "/v1.0/card/instances", map[string]any{
"outTrackId": outTrackID,
"cardData": map[string]any{"cardParamMap": map[string]any{"result": text, "summary": text}},
"cardUpdateOptions": map[string]any{"updateCardDataByKey": true},
})
}