diff --git a/internal/helpers/connect_approval.go b/internal/helpers/connect_approval.go index 9d228aea..a50e0475 100644 --- a/internal/helpers/connect_approval.go +++ b/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. diff --git a/internal/helpers/connect_approval_card.go b/internal/helpers/connect_approval_card.go index 5186e4e4..9d85fd7c 100644 --- a/internal/helpers/connect_approval_card.go +++ b/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. diff --git a/internal/helpers/connect_approval_card_test.go b/internal/helpers/connect_approval_card_test.go index 33bc439e..ce6671aa 100644 --- a/internal/helpers/connect_approval_card_test.go +++ b/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) {