feat(connect): deferred execution — don't lose a task when the twin can't act yet

When an approved action can't execute right now (most often the connector's
dws login is not the bot owner, so an owner-scoped action like creating a
personal todo fails), the request is no longer dropped with an opaque error.
Instead it is held and retried after the owner recovers.

- new 'deferred' state: an approved request whose execution didn't complete;
  persisted like any other, distinct from terminal 'failed'
- on execution failure (auto-approve path and owner-approve path), mark
  deferred instead of failed, and privately DM the owner what went wrong and
  how to recover (log dws in as the bot owner, then reply 「重试」). The
  requester is NOT told it failed — their task is held, not lost
- 「重试/恢复」owner keyword flushes the backlog: re-execute each deferred
  request oldest-first; on success mark executed and send the requester the
  outcome, on failure keep it deferred
- markDeferred / allDeferred on the gate; deferred counts as approved() so the
  retry can still mark it executed
- tests: deferred-hold-then-retry end to end, isRetryWord

This is the graceful-degradation half of the digital twin: a request survives
an owner-offline / wrong-identity window and completes once the owner is back.

internal/helpers/connect_approval.go

@@ -20,6 +20,7 @@ import (
 	"os"
 	"path/filepath"
 	"regexp"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -51,6 +52,12 @@ const (
 	approvalRejected approvalState = "rejected"
 	approvalExecuted approvalState = "executed"
 	approvalFailed   approvalState = "failed"
+	// approvalDeferred is an approved request whose execution could not complete
+	// right now — typically because the connector's dws login is not (yet) the
+	// bot's owner, so an owner-scoped action can't run. The request is NOT lost:
+	// it stays on disk in this state, the owner is told to recover (log in), and
+	// a later "retry" flushes it. Distinct from Failed, which is terminal.
+	approvalDeferred approvalState = "deferred"
 )
 
 // plannedAction is the structured command the gate will run once approved. It
@@ -96,16 +103,18 @@ type ApprovalRequest struct {
 // regardless of any later execution outcome.
 func (r *ApprovalRequest) decided() bool {
 	switch r.State {
-	case approvalApproved, approvalRejected, approvalExecuted, approvalFailed:
+	case approvalApproved, approvalRejected, approvalExecuted, approvalFailed, approvalDeferred:
 		return true
 	}
 	return false
 }
 
-// approved reports whether the owner approved (in any post-approval state).
+// approved reports whether the owner approved (in any post-approval state,
+// including deferred — a deferred request WAS approved, its execution just
+// hasn't completed yet).
 func (r *ApprovalRequest) approved() bool {
 	switch r.State {
-	case approvalApproved, approvalExecuted, approvalFailed:
+	case approvalApproved, approvalExecuted, approvalFailed, approvalDeferred:
 		return true
 	}
 	return false
@@ -304,6 +313,36 @@ func (g *approvalGate) markFailed(id string, cause string) {
 	}
 }
 
+// markDeferred records that an approved request could not execute now and is
+// being held for a later retry (it is NOT lost). The cause is kept so the owner
+// and the audit trail can see why. Persisted so a connector restart still
+// remembers the backlog.
+func (g *approvalGate) markDeferred(id string, cause string) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if r, ok := g.reqs[id]; ok && r.approved() {
+		r.State = approvalDeferred
+		r.ExecErr = truncateRunes(strings.TrimSpace(cause), 500)
+		g.persist(r)
+	}
+}
+
+// allDeferred returns snapshots of every request awaiting retry, oldest first
+// (so a flush replays them in the order they were asked).
+func (g *approvalGate) allDeferred() []*ApprovalRequest {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	out := make([]*ApprovalRequest, 0)
+	for _, r := range g.reqs {
+		if r.State == approvalDeferred {
+			cp := *r
+			out = append(out, &cp)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].CreatedAt.Before(out[j].CreatedAt) })
+	return out
+}
+
 // Get returns a snapshot copy of a request, or nil if unknown.
 func (g *approvalGate) Get(id string) *ApprovalRequest {
 	g.mu.Lock()
@@ -554,6 +593,21 @@ var (
 	}
 )
 
+// retryWords are the whole-message replies that flush the deferred backlog
+// (the owner has recovered — e.g. logged dws in as themselves — and wants the
+// held requests replayed).
+var retryWords = map[string]struct{}{
+	"重试": {}, "恢复": {}, "补做": {}, "重新执行": {}, "继续": {},
+	"retry": {}, "resume": {},
+}
+
+// isRetryWord reports whether a whole message is a bare "flush the backlog"
+// command.
+func isRetryWord(msg string) bool {
+	_, ok := retryWords[strings.ToLower(strings.TrimSpace(msg))]
+	return ok
+}
+
 // parseDecisionWord classifies a whole message as an approve/reject decision.
 // It returns (approve, ok): ok is false when the message is not a bare decision
 // keyword, so the caller forwards it to the agent as an ordinary message.

internal/helpers/connect_approval_card.go

@@ -299,10 +299,14 @@ func (o *approvalOrchestrator) autoApproveAndExecute(ctx context.Context, req *A
 		decided.ID, o.ownerUserID, decided.Summary)
 	out, execErr := o.execute(ctx, decided)
 	if execErr != nil {
-		o.gate.markFailed(decided.ID, execErr.Error())
+		// 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, "执行失败："+execErr.Error())
+			_ = reply(ctx, convID, "收到，但现在没能直接完成（已记下，稍后会补做）。")
 		}
 		return "", true
 	}
@@ -314,6 +318,23 @@ func (o *approvalOrchestrator) autoApproveAndExecute(ctx context.Context, req *A
 	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
@@ -351,6 +372,10 @@ func (o *approvalOrchestrator) handleOwnerDecision(ctx context.Context, senderSt
 	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
@@ -373,10 +398,11 @@ func (o *approvalOrchestrator) handleOwnerDecision(ctx context.Context, senderSt
 	}
 	out, execErr := o.execute(ctx, decided)
 	if execErr != nil {
-		o.gate.markFailed(decided.ID, execErr.Error())
+		// 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.notifier.sendOTOText(ctx, []string{o.ownerUserID}, "已同意，但执行失败："+execErr.Error())
-		o.notifyRequester(ctx, decided, "你的请求已获批准，但执行失败："+execErr.Error())
+		o.notifyOwnerDeferred(ctx, decided, execErr)
 		return true
 	}
 	o.gate.markExecuted(decided.ID)
@@ -386,6 +412,38 @@ func (o *approvalOrchestrator) handleOwnerDecision(ctx context.Context, senderSt
 	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 := 0, 0
+	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++
+	}
+	summary := fmt.Sprintf("已补做 %d 个积压请求。", done)
+	if stuck > 0 {
+		summary += fmt.Sprintf("仍有 %d 个未成功（身份可能还没对上，确认 dws 登录后再回复「重试」）。", stuck)
+	}
+	_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, summary)
+	return true
+}
+
 // 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.

internal/helpers/connect_approval_card_test.go

@@ -58,16 +58,22 @@ func (f *fakeCardSender) UpdateApprovalCard(_ context.Context, _, text string) e
 type fakeRunner struct {
 	mu   sync.Mutex
 	runs []executor.Invocation
+	fail bool // when true, Run returns an error (simulates identity not ready)
 }
 
 func (r *fakeRunner) Run(_ context.Context, inv executor.Invocation) (executor.Result, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.runs = append(r.runs, inv)
+	if r.fail {
+		return executor.Result{}, context.DeadlineExceeded
+	}
 	inv.Implemented = true
 	return executor.Result{Invocation: inv}, nil
 }
 
+func (r *fakeRunner) setFail(v bool) { r.mu.Lock(); defer r.mu.Unlock(); r.fail = v }
+
 func (r *fakeRunner) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.runs) }
 
 // fakeNotifier records the proactive 1:1 messages text-mode approval sends to
@@ -507,6 +513,66 @@ func TestAudit_RecordsTerminalOutcomes(t *testing.T) {
 	}
 }
 
+// TestDeferredExecution_HoldsAndRetries is the graceful-degradation path: when
+// execution fails (identity not ready), the request is DEFERRED (not lost, not
+// failed), the owner is privately notified; after the owner recovers and replies
+// "重试", the backlog is replayed, executes, and the requester gets the result.
+func TestDeferredExecution_HoldsAndRetries(t *testing.T) {
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	gate := newApprovalGate("ctdef")
+	runner := &fakeRunner{fail: true} // identity not ready → execution fails
+	notifier := &fakeNotifier{}
+	o := newTextApprovalOrchestrator(gate, runner, "owner", notifier)
+
+	// A non-owner request, approved by the owner, but execution fails → deferred.
+	o.handleReply(context.Background(), "requester", "conv-g", `[[ACTION:todo.create title="待恢复"]]`, nil)
+	if !o.handleOwnerDecision(context.Background(), "owner", "同意") {
+		t.Fatal("owner approval must be consumed")
+	}
+	rec := gate.list()[0]
+	if rec.State != approvalDeferred {
+		t.Fatalf("failed execution must DEFER (not fail), got state=%s", rec.State)
+	}
+	// Owner was told to recover; requester was NOT told it failed (task not lost).
+	if !notifier.toUser("owner") {
+		t.Fatal("owner should be notified of the deferred action")
+	}
+	if notifier.toUser("requester") {
+		t.Fatal("requester must NOT be told it failed — their task is held, not lost")
+	}
+
+	// A bare non-retry word from the owner should pass through (not consumed).
+	if o.handleOwnerDecision(context.Background(), "owner", "在吗") {
+		t.Fatal("ordinary owner message must not be consumed")
+	}
+
+	// Owner recovers (identity now works) and replies "重试" → backlog flushes.
+	runner.setFail(false)
+	if !o.handleOwnerDecision(context.Background(), "owner", "重试") {
+		t.Fatal("retry word must be consumed")
+	}
+	if gate.list()[0].State != approvalExecuted {
+		t.Fatalf("after retry the deferred request must execute, got %s", gate.list()[0].State)
+	}
+	if !notifier.toUser("requester") {
+		t.Fatal("requester should get the outcome after the backlog is flushed")
+	}
+}
+
+// TestIsRetryWord checks whole-message retry-keyword matching.
+func TestIsRetryWord(t *testing.T) {
+	for _, w := range []string{"重试", " 恢复 ", "retry", "继续"} {
+		if !isRetryWord(w) {
+			t.Fatalf("%q should be a retry word", w)
+		}
+	}
+	for _, w := range []string{"重试一下他的方案", "", "你好"} {
+		if isRetryWord(w) {
+			t.Fatalf("%q should NOT be a retry word", w)
+		}
+	}
+}
+
 // TestHandleReply_TextMode_OwnerRejects: the owner's "拒绝" declines without
 // executing.
 func TestHandleReply_TextMode_OwnerRejects(t *testing.T) {