From e2b46faa058b8d9848c143436f976ae307f0bdba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Sun, 20 Sep 2026 18:15:04 +0800 Subject: [PATCH] feat(dingtalk): add card terminal-write order modes (refresh, close-refresh) A DingTalk AI card replays streamed content with a typewriter animation the client drives itself, so a finished answer can still be typed out character by character - and a card that is not in the foreground pauses that replay. DINGTALK_CARD_FINAL_WRITE now sequences the two terminal writes explicitly: streaming (default) streaming finalize only (unchanged) update finalize, then a plain write (observed to blank the card) refresh plain write of the finished answer, then a zero-delta finalize carrying the same text close-refresh zero-delta finalize on the text the card already shows, then a plain write of the finished answer The sender remembers the text each live card shows so a terminal write can close the stream with a frame the client has nothing to animate; a failing plain write is logged and never fails the delivery. --- cmd/dobotai/card_stream.go | 9 +- internal/channels/dingtalk/card.go | 114 ++++++++++-- .../channels/dingtalk/card_delivery_mode.go | 40 ++++- .../dingtalk/card_final_write_test.go | 168 ++++++++++++++++++ 4 files changed, 311 insertions(+), 20 deletions(-) diff --git a/cmd/dobotai/card_stream.go b/cmd/dobotai/card_stream.go index fe74828..f203bfe 100644 --- a/cmd/dobotai/card_stream.go +++ b/cmd/dobotai/card_stream.go @@ -29,8 +29,13 @@ func cardPartialUpdatesMode() dingtalk.CardPartialUpdatesMode { // unset / streaming → the finished answer is written with the card streaming // API and isFinalize=true (historical behaviour) // update / plain → the same answer is written once more through the plain -// card update API (PUT /v1.0/card/instances), so the -// terminal content carries no client-side typewriter queue +// card update API (PUT /v1.0/card/instances) AFTER the +// streaming finalize (observed to blank the delivered card) +// refresh → the answer goes through the plain card update API FIRST +// and the stream is then closed with the very same text, +// so the client has no delta left to animate +// close-refresh → the stream is closed on the text the card already shows +// (zero delta) and the answer is refreshed afterwards // // DingTalk renders streamed card content with a typewriter animation the client // drives itself, so an answer that is already complete on the server can still be diff --git a/internal/channels/dingtalk/card.go b/internal/channels/dingtalk/card.go index 9e8bdc5..4e939c1 100644 --- a/internal/channels/dingtalk/card.go +++ b/internal/channels/dingtalk/card.go @@ -67,6 +67,13 @@ type CardSender struct { mu sync.Mutex token string tokenTill time.Time + + // streamed keeps the text every live card currently shows, so a terminal + // write can close the stream with a frame the client has nothing to animate + // (see CardFinalWriteMode). Bounded by the number of cards in flight in one + // process; entries are dropped once the card is finalized. + contentMu sync.Mutex + streamed map[string]string } // CardSenderOption configures optional CardSender behavior. @@ -247,23 +254,100 @@ func (s *CardSender) UpdateCard(ctx context.Context, outTrackID, content string, return err } - if err := s.finalizeCard(ctx, outTrackID, content, token, isFinalize, isError); err != nil { - return err - } - // Optionally let the terminal content land through the plain card update API - // as well: content written through the streaming endpoint is replayed by the - // client with a typewriter animation, while a plain write has no streaming - // semantics. The streaming finalize above already closed the card, so a - // failing follow-up write is logged and swallowed - it must never turn a - // delivered answer into an error. - if isFinalize && !isError && s.finalWrite.alsoPlainUpdates() { - if err := s.replaceCardContent(ctx, outTrackID, content, token); err != nil { - log.Printf("DingTalk card final write: plain update failed error=%v", err) - } else { - log.Printf("DingTalk card final write: plain update ok bytes=%d", len(content)) + if !isFinalize || isError { + if err := s.finalizeCard(ctx, outTrackID, content, token, isFinalize, isError); err != nil { + return err } + if !isFinalize { + // Remember what the card now shows: a terminal write can close the + // stream with a frame that carries no new text. + s.rememberStreamedContent(outTrackID, content) + } + return nil } - return nil + + // Terminal write. The order of the two available writes decides whether the + // client replays the finished answer with its own typewriter animation, so + // each mode sequences them explicitly - see CardFinalWriteMode. + defer s.forgetStreamedContent(outTrackID) + switch { + case s.finalWrite.refreshBeforeFinalize(): + // Show the finished answer first (plain write, no streaming semantics) + // and close the stream with the very same text afterwards: the client + // has no delta left to animate, so the answer is simply there. + s.refreshCardPlain(ctx, outTrackID, content, token) + return s.finalizeCard(ctx, outTrackID, content, token, true, false) + case s.finalWrite.closeThenRefresh(): + shown := s.lastStreamedContent(outTrackID) + if shown == "" { + // Nothing was streamed into this card: there is no zero-delta frame + // to close on, so refresh first and finalize after. + s.refreshCardPlain(ctx, outTrackID, content, token) + return s.finalizeCard(ctx, outTrackID, content, token, true, false) + } + // Close the stream on the text the card already shows (zero delta, + // nothing to animate), then refresh with the finished answer. + if err := s.finalizeCard(ctx, outTrackID, shown, token, true, false); err != nil { + return err + } + s.refreshCardPlain(ctx, outTrackID, content, token) + return nil + default: + if err := s.finalizeCard(ctx, outTrackID, content, token, isFinalize, isError); err != nil { + return err + } + // Optionally let the terminal content land through the plain card update + // API as well: content written through the streaming endpoint is replayed + // by the client with a typewriter animation, while a plain write has no + // streaming semantics. The streaming finalize above already closed the + // card, so a failing follow-up write is logged and swallowed - it must + // never turn a delivered answer into an error. + if s.finalWrite.alsoPlainUpdates() { + if err := s.replaceCardContent(ctx, outTrackID, content, token); err != nil { + log.Printf("DingTalk card final write: plain update failed error=%v", err) + } else { + log.Printf("DingTalk card final write: plain update ok bytes=%d", len(content)) + } + } + return nil + } +} + +// refreshCardPlain writes a finished answer through the plain card update API +// and logs the outcome. The write is best effort: the answer is already on its +// way through the streaming channel, so a failure here is never turned into an +// error for the caller. +func (s *CardSender) refreshCardPlain(ctx context.Context, outTrackID, content, token string) { + if err := s.replaceCardContent(ctx, outTrackID, content, token); err != nil { + log.Printf("DingTalk card final write: plain refresh failed error=%v", err) + return + } + log.Printf("DingTalk card final write: plain refresh ok bytes=%d", len(content)) +} + +// rememberStreamedContent records the text a card currently shows. +func (s *CardSender) rememberStreamedContent(outTrackID, content string) { + s.contentMu.Lock() + defer s.contentMu.Unlock() + if s.streamed == nil { + s.streamed = make(map[string]string) + } + s.streamed[outTrackID] = content +} + +// lastStreamedContent returns the text a card currently shows, or "" when the +// card was created but never received a partial frame. +func (s *CardSender) lastStreamedContent(outTrackID string) string { + s.contentMu.Lock() + defer s.contentMu.Unlock() + return s.streamed[outTrackID] +} + +// forgetStreamedContent drops the remembered text of a finalized card. +func (s *CardSender) forgetStreamedContent(outTrackID string) { + s.contentMu.Lock() + defer s.contentMu.Unlock() + delete(s.streamed, outTrackID) } // cardParamMap builds the template variables for a given answer. The plain card diff --git a/internal/channels/dingtalk/card_delivery_mode.go b/internal/channels/dingtalk/card_delivery_mode.go index e0d73ac..e8b12a9 100644 --- a/internal/channels/dingtalk/card_delivery_mode.go +++ b/internal/channels/dingtalk/card_delivery_mode.go @@ -69,14 +69,31 @@ func (m CardPartialUpdatesMode) placeholderCardRequested() bool { // (PUT /v1.0/card/instances). That write carries no // streaming semantics, so the terminal state is applied to // the card without leaving a typewriter queue behind. +// NOTE: this order is the one that blanked a live card on +// 2026-09-20 (the finalize frame starts the client's +// typewriter and the plain write lands in the middle of it); +// kept only to reproduce that regression. +// refresh → the finished answer is written through the plain card +// update API FIRST, so the card shows it without any +// typewriter queue, and only then is the stream closed by a +// finalize frame that carries exactly the same text. The +// client has no delta left to animate, so the answer is +// simply there when it arrives. +// close-refresh → the stream is closed with a frame carrying the text the +// card already shows (zero delta, nothing to animate) and +// the finished answer is written through the plain update +// API afterwards. Used when the plain write must land on a +// stream that is already closed and quiet. // // The default stays "streaming" so an environment without the variable behaves // exactly as before; DINGTALK_CARD_FINAL_WRITE selects the other mode. type CardFinalWriteMode string const ( - CardFinalWriteStreaming CardFinalWriteMode = "streaming" - CardFinalWriteUpdate CardFinalWriteMode = "update" + CardFinalWriteStreaming CardFinalWriteMode = "streaming" + CardFinalWriteUpdate CardFinalWriteMode = "update" + CardFinalWriteRefresh CardFinalWriteMode = "refresh" + CardFinalWriteCloseRefresh CardFinalWriteMode = "close-refresh" ) // ParseCardFinalWriteMode maps DINGTALK_CARD_FINAL_WRITE onto a mode. Unset or @@ -85,13 +102,30 @@ func ParseCardFinalWriteMode(raw string) CardFinalWriteMode { switch strings.ToLower(strings.TrimSpace(raw)) { case "update", "plain", "plain_update", "on", "true", "1", "yes": return CardFinalWriteUpdate + case "refresh", "refresh_first", "plain_first", "refresh_then_finalize", "refresh-first", "plain-first": + return CardFinalWriteRefresh + case "close-refresh", "close_refresh", "finalize_then_refresh", "refresh_after_finalize", "close-then-refresh": + return CardFinalWriteCloseRefresh default: return CardFinalWriteStreaming } } // alsoPlainUpdates reports whether the finished answer must additionally be -// written through the plain (non-streaming) card update endpoint. +// written through the plain (non-streaming) card update endpoint after the +// streaming finalize. func (m CardFinalWriteMode) alsoPlainUpdates() bool { return m == CardFinalWriteUpdate } + +// refreshBeforeFinalize reports whether the finished answer goes through the +// plain card update endpoint before the stream is closed. +func (m CardFinalWriteMode) refreshBeforeFinalize() bool { + return m == CardFinalWriteRefresh +} + +// closeThenRefresh reports whether the stream is closed on the text already on +// the card and the finished answer is refreshed afterwards. +func (m CardFinalWriteMode) closeThenRefresh() bool { + return m == CardFinalWriteCloseRefresh +} diff --git a/internal/channels/dingtalk/card_final_write_test.go b/internal/channels/dingtalk/card_final_write_test.go index bca3735..4367076 100644 --- a/internal/channels/dingtalk/card_final_write_test.go +++ b/internal/channels/dingtalk/card_final_write_test.go @@ -28,6 +28,11 @@ type finalWriteTestServer struct { lastUpdateMethod string lastUpdateHeader string + // callSeq records the endpoint order ("create"/"stream"/"update") so the + // write ORDER of a terminal answer can be asserted, not just the counts. + callSeq []string + streamBodies []map[string]any + streamStatus int streamResponse any updateStatus int @@ -50,13 +55,16 @@ func newFinalWriteTestServer() *finalWriteTestServer { json.NewEncoder(w).Encode(map[string]any{"accessToken": "test-access-token", "expireIn": 3600}) case "/create": ts.createCalls++ + ts.callSeq = append(ts.callSeq, "create") ts.lastCreateBody = make(map[string]any) json.NewDecoder(r.Body).Decode(&ts.lastCreateBody) json.NewEncoder(w).Encode(ts.createResponse) case "/stream": ts.streamCalls++ + ts.callSeq = append(ts.callSeq, "stream") ts.lastStreamBody = make(map[string]any) json.NewDecoder(r.Body).Decode(&ts.lastStreamBody) + ts.streamBodies = append(ts.streamBodies, ts.lastStreamBody) if ts.streamStatus != http.StatusOK { w.WriteHeader(ts.streamStatus) return @@ -64,6 +72,7 @@ func newFinalWriteTestServer() *finalWriteTestServer { json.NewEncoder(w).Encode(ts.streamResponse) case "/update": ts.updateCalls++ + ts.callSeq = append(ts.callSeq, "update") ts.lastUpdateMethod = r.Method ts.lastUpdateHeader = r.Header.Get("x-acs-dingtalk-access-token") ts.lastUpdateBody = make(map[string]any) @@ -291,3 +300,162 @@ func TestCardFinalWriteUpdateModeAppliesToSendFinalCard(t *testing.T) { t.Fatalf("plain update content=%v", ts.paramMap(t)["content"]) } } + +// --------------------------------------------------------------------------- +// Order modes: refresh (plain first) and close-refresh (zero-delta finalize +// first). Both exist because the client replays streamed text with its own +// typewriter animation, so the ORDER of the two terminal writes decides what +// the user sees when the answer is ready. +// --------------------------------------------------------------------------- + +func wantCallSeq(t *testing.T, got, want []string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("call sequence=%v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("call sequence=%v, want %v", got, want) + } + } +} + +func TestCardFinalWriteRefreshModeRefreshesBeforeFinalize(t *testing.T) { + ts := newFinalWriteTestServer() + defer ts.Close() + sender := ts.sender(t, CardFinalWriteRefresh) + + if _, err := sender.StartCard(context.Background(), CardRequest{ + ConversationType: core.ConversationDirect, + ExternalUserID: "user-1", + Content: "占位", + }); err != nil { + t.Fatalf("StartCard: %v", err) + } + if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil { + t.Fatalf("partial UpdateCard: %v", err) + } + if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil { + t.Fatalf("terminal UpdateCard: %v", err) + } + + // The finished answer goes out through the plain update BEFORE the frame + // that closes the stream. + wantCallSeq(t, ts.callSeq, []string{"create", "stream", "update", "stream"}) + if got := ts.paramMap(t)["content"]; got != "完整答案" { + t.Fatalf("plain refresh content=%v, want the finished answer", got) + } + if len(ts.streamBodies) != 2 { + t.Fatalf("stream frames=%d, want 2", len(ts.streamBodies)) + } + closing := ts.streamBodies[1] + if closing["content"] != "完整答案" || closing["isFinalize"] != true { + t.Fatalf("closing frame=%v, want the finished answer with isFinalize=true", closing) + } +} + +func TestCardFinalWriteRefreshModeKeepsDeliveryWhenPlainRefreshFails(t *testing.T) { + ts := newFinalWriteTestServer() + defer ts.Close() + ts.updateStatus = http.StatusInternalServerError + sender := ts.sender(t, CardFinalWriteRefresh) + + if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil { + t.Fatalf("a failed plain refresh must not fail the delivery, got %v", err) + } + if ts.updateCalls != 1 || ts.streamCalls != 1 { + t.Fatalf("update/stream=%d/%d, want 1/1 (the closing frame still goes out)", ts.updateCalls, ts.streamCalls) + } + if ts.lastStreamBody["isFinalize"] != true || ts.lastStreamBody["content"] != "完整答案" { + t.Fatalf("closing frame=%v", ts.lastStreamBody) + } +} + +func TestCardFinalWriteCloseRefreshModeClosesOnShownTextThenRefreshes(t *testing.T) { + ts := newFinalWriteTestServer() + defer ts.Close() + sender := ts.sender(t, CardFinalWriteCloseRefresh) + + if _, err := sender.StartCard(context.Background(), CardRequest{ + ConversationType: core.ConversationDirect, + ExternalUserID: "user-1", + Content: "占位", + }); err != nil { + t.Fatalf("StartCard: %v", err) + } + if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil { + t.Fatalf("partial UpdateCard: %v", err) + } + if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil { + t.Fatalf("terminal UpdateCard: %v", err) + } + + // The closing frame carries the text the card already shows (zero delta, so + // the client has nothing to animate) and the finished answer follows through + // the plain update. + wantCallSeq(t, ts.callSeq, []string{"create", "stream", "stream", "update"}) + if len(ts.streamBodies) != 2 { + t.Fatalf("stream frames=%d, want 2", len(ts.streamBodies)) + } + closing := ts.streamBodies[1] + if closing["content"] != "部分" || closing["isFinalize"] != true { + t.Fatalf("closing frame=%v, want the previously shown text with isFinalize=true", closing) + } + if got := ts.paramMap(t)["content"]; got != "完整答案" { + t.Fatalf("plain refresh content=%v, want the finished answer", got) + } +} + +func TestCardFinalWriteCloseRefreshModeRefreshesFirstWhenNothingStreamed(t *testing.T) { + ts := newFinalWriteTestServer() + defer ts.Close() + sender := ts.sender(t, CardFinalWriteCloseRefresh) + + // A card that never received a partial frame has no zero-delta frame to + // close on, so the refresh has to go first. + if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil { + t.Fatalf("terminal UpdateCard: %v", err) + } + wantCallSeq(t, ts.callSeq, []string{"update", "stream"}) + if ts.lastStreamBody["isFinalize"] != true || ts.lastStreamBody["content"] != "完整答案" { + t.Fatalf("closing frame=%v", ts.lastStreamBody) + } +} + +func TestCardFinalWriteCloseRefreshModeKeepsDeliveryWhenPlainRefreshFails(t *testing.T) { + ts := newFinalWriteTestServer() + defer ts.Close() + ts.updateStatus = http.StatusInternalServerError + sender := ts.sender(t, CardFinalWriteCloseRefresh) + + if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil { + t.Fatalf("partial UpdateCard: %v", err) + } + if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil { + t.Fatalf("a failed plain refresh must not fail the delivery, got %v", err) + } + wantCallSeq(t, ts.callSeq, []string{"stream", "stream", "update"}) + if ts.lastStreamBody["isFinalize"] != true || ts.lastStreamBody["content"] != "部分" { + t.Fatalf("closing frame=%v", ts.lastStreamBody) + } +} + +func TestCardFinalWriteParserAcceptsOrderValues(t *testing.T) { + for raw, want := range map[string]CardFinalWriteMode{ + "": CardFinalWriteStreaming, + "streaming": CardFinalWriteStreaming, + "off": CardFinalWriteStreaming, + "nonsense": CardFinalWriteStreaming, + "update": CardFinalWriteUpdate, + "refresh": CardFinalWriteRefresh, + "plain_first": CardFinalWriteRefresh, + "refresh-first": CardFinalWriteRefresh, + "close_refresh": CardFinalWriteCloseRefresh, + "close-refresh": CardFinalWriteCloseRefresh, + "CLOSE-REFRESH ": CardFinalWriteCloseRefresh, + } { + if got := ParseCardFinalWriteMode(raw); got != want { + t.Fatalf("ParseCardFinalWriteMode(%q)=%q, want %q", raw, got, want) + } + } +}