diff --git a/server/internal/integrations/lark/client.go b/server/internal/integrations/lark/client.go index 8954b4cb5..21a4ae125 100644 --- a/server/internal/integrations/lark/client.go +++ b/server/internal/integrations/lark/client.go @@ -35,6 +35,15 @@ type APIClient interface { // performs the network call. PatchInteractiveCard(ctx context.Context, p PatchCardParams) error + // SendTextMessage posts a plain text message into a Lark chat. + // Used for the agent's chat reply — Bohan asked for replies to be + // rendered as a normal IM bubble rather than nested inside an + // interactive card, which feels more natural for free-form chat. + // Returns Lark's message_id (we don't currently persist it for the + // text path since each send is one-shot, but the API gives it back + // for consistency with SendInteractiveCard). + SendTextMessage(ctx context.Context, p SendTextParams) (string, error) + // SendBindingPromptCard is the dedicated "you need to bind" // outbound. Kept separate from SendInteractiveCard so the // abstraction stays stable when the production card template @@ -78,6 +87,15 @@ type PatchCardParams struct { CardJSON string } +// SendTextParams is the input shape for posting a plain text message. +// Text is sent verbatim to Lark; the client handles JSON encoding of +// the `{"text": "..."}` content envelope Lark requires. +type SendTextParams struct { + InstallationID InstallationCredentials + ChatID ChatID + Text string +} + // BindingPromptParams carries the data needed to render and send the // member-binding prompt card (single CTA: open the binding URL). type BindingPromptParams struct { @@ -146,6 +164,11 @@ func (s *stubAPIClient) PatchInteractiveCard(ctx context.Context, p PatchCardPar return ErrAPIClientNotConfigured } +func (s *stubAPIClient) SendTextMessage(ctx context.Context, p SendTextParams) (string, error) { + s.log.Warn("lark stub client: SendTextMessage called", "chat_id", string(p.ChatID)) + return "", ErrAPIClientNotConfigured +} + func (s *stubAPIClient) SendBindingPromptCard(ctx context.Context, p BindingPromptParams) error { s.log.Warn("lark stub client: SendBindingPromptCard called", "open_id", string(p.OpenID)) return ErrAPIClientNotConfigured diff --git a/server/internal/integrations/lark/http_client.go b/server/internal/integrations/lark/http_client.go index 3b8a858cf..718e123ca 100644 --- a/server/internal/integrations/lark/http_client.go +++ b/server/internal/integrations/lark/http_client.go @@ -238,6 +238,57 @@ func (c *httpAPIClient) SendInteractiveCard(ctx context.Context, p SendCardParam return resp.Data.MessageID, nil } +// SendTextMessage posts a plain text IM message into a Lark chat. +// This is the Patcher's primary outbound for agent chat replies — +// using a normal text bubble instead of an interactive card makes +// free-form replies feel like a native Lark conversation. The +// content envelope Lark expects is a JSON-encoded `{"text": "..."}` +// blob; we encode it here so callers pass raw text. +func (c *httpAPIClient) SendTextMessage(ctx context.Context, p SendTextParams) (string, error) { + if p.ChatID == "" { + return "", errors.New("lark http client: missing chat_id") + } + if p.Text == "" { + return "", errors.New("lark http client: missing text") + } + token, err := c.tenantAccessToken(ctx, p.InstallationID) + if err != nil { + return "", err + } + // Lark's `text` msg_type expects content = JSON-encoded {"text": "..."}. + // json.Marshal handles the escape of newlines / quotes / unicode so + // the agent's reply round-trips intact. + contentBytes, err := json.Marshal(map[string]string{"text": p.Text}) + if err != nil { + return "", fmt.Errorf("lark http client: encode text content: %w", err) + } + q := url.Values{} + q.Set("receive_id_type", "chat_id") + body := map[string]string{ + "receive_id": string(p.ChatID), + "msg_type": "text", + "content": string(contentBytes), + } + var resp struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + MessageID string `json:"message_id"` + } `json:"data"` + } + path := "/open-apis/im/v1/messages?" + q.Encode() + if err := c.doJSON(ctx, http.MethodPost, path, token, body, &resp); err != nil { + return "", fmt.Errorf("lark http client: send text message: %w", err) + } + if resp.Code != 0 || resp.Data.MessageID == "" { + if isTokenError(resp.Code) { + c.invalidateToken(p.InstallationID.AppID) + } + return "", fmt.Errorf("lark http client: send text message: code=%d msg=%q", resp.Code, resp.Msg) + } + return resp.Data.MessageID, nil +} + // PatchInteractiveCard updates an existing card's body. Lark's // message-patch endpoint replaces the whole card payload; callers // (i.e. the Patcher) render the full updated card each time. diff --git a/server/internal/integrations/lark/outbound.go b/server/internal/integrations/lark/outbound.go index 6b842547a..bc5191043 100644 --- a/server/internal/integrations/lark/outbound.go +++ b/server/internal/integrations/lark/outbound.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "log/slog" - "sync" "time" "github.com/jackc/pgx/v5" @@ -159,25 +158,19 @@ type CredentialsResolver interface { DecryptAppSecret(inst db.LarkInstallation) (string, error) } -// PatcherConfig tunes the streaming patcher. Defaults via -// withDefaults; tests typically override Renderer / Now / Logger. +// PatcherConfig tunes the outbound Patcher. Defaults via withDefaults; +// tests typically override Renderer / Now / Logger. type PatcherConfig struct { - // MinPatchInterval throttles per-card patches so a streaming run - // doesn't blow past Lark's per-card update rate-limit. Patches - // issued sooner than this since the last one are dropped (the - // final/error transition is never dropped — those bypass the - // throttle). - MinPatchInterval time.Duration - + // Renderer drives the error card template used on the EventTaskFailed + // path. The success path (EventChatDone) bypasses the renderer + // entirely — it sends the raw assistant reply as a plain text IM + // message — so this only matters for the failure branch. Renderer Renderer Now func() time.Time Logger *slog.Logger } func (c PatcherConfig) withDefaults() PatcherConfig { - if c.MinPatchInterval == 0 { - c.MinPatchInterval = 500 * time.Millisecond - } if c.Renderer == nil { c.Renderer = NewDefaultRenderer() } @@ -190,34 +183,32 @@ func (c PatcherConfig) withDefaults() PatcherConfig { return c } -// Patcher reacts to task-lifecycle events on the event bus and keeps -// the Lark interactive card for each chat-driven task in sync. It is -// the §4.5 outbound side of the design: thinking → running/streaming -// → final/error. +// Patcher reacts to task-lifecycle events on the event bus and forwards +// chat replies to Lark as plain text IM messages. It is the outbound +// side of §4.5 — but the original "thinking → streaming → final card" +// lifecycle was reduced to a single plain-text reply on EventChatDone +// after Bohan reported the card chrome made replies feel like system +// notifications. The error path is the one survivor of card rendering: +// failed runs surface as a short error card on EventTaskFailed because +// the visual distinction from a normal reply is genuinely useful. // // Scope: // // - Only tasks whose chat_session has a lark_chat_session_binding -// produce a card. Tasks born from the web UI or autopilot pass +// produce outbound. Tasks born from the web UI or autopilot pass // through unchanged. // -// - The card binding is per-task (lark_outbound_card_message.task_id), -// so a chat_session that hosts many runs gets one card per run. +// - Each EventChatDone yields one Lark text message; there is no +// streaming, no throttling, no DB row to track card-state. // -// - Throttling is per-card, in-memory. Multi-replica deployments -// are de-duplicated at the next layer up: only the replica that -// also holds the inbound WS lease for the installation will see -// the run originate locally, and the events bus is per-process. -// For the SaaS multi-node case, the same Redis fanout used by the -// UI WS layer makes a future cross-process patcher cheap to add. +// - Multi-replica safety is inherited from the inbound WS lease: at +// most one replica holds the installation lease at a time, the +// event bus is per-process, so exactly one Patcher reacts per run. type Patcher struct { queries PatcherQueries credentials CredentialsResolver client APIClient cfg PatcherConfig - - mu sync.Mutex - lastPatched map[string]time.Time // task_id -> last patch wall-clock time } // NewPatcher constructs a Patcher bound to its dependencies. The @@ -229,7 +220,6 @@ func NewPatcher(queries PatcherQueries, credentials CredentialsResolver, client credentials: credentials, client: client, cfg: cfg, - lastPatched: make(map[string]time.Time), } } @@ -239,19 +229,26 @@ func NewPatcher(queries PatcherQueries, credentials CredentialsResolver, client // during server boot (after the bus + patcher are constructed and // before HTTP traffic starts). // -// We deliberately do NOT subscribe to EventTaskCompleted here. -// TaskService publishes ChatDone (with the assistant message content) -// IMMEDIATELY BEFORE TaskCompleted (which has no content) for every -// chat task. Subscribing to both would patch the final card twice: -// the first patch shows the real reply, the second patch wipes it -// with the "Done." fallback because the TaskCompleted payload's -// `map[string]any` has no "content" key. EventChatDone is the -// canonical "agent finished replying" signal for the Patcher; -// EventTaskCompleted is left to other listeners (web UI, analytics, -// task usage rollup, etc.) where the lack of content doesn't matter. +// Subscriptions are deliberately minimal: +// +// - EventChatDone — the agent finished replying. The Patcher sends +// the reply as a plain text IM message (Lark's `msg_type=text`), +// not as an interactive card. The earlier card-based design (with +// thinking → running → final patches) made every reply look like +// a system notification nested in card chrome; flipping to plain +// text makes free-form chat feel native. +// +// - EventTaskFailed — the run failed; surface a short error card +// so the failure is visually distinct from a successful reply. +// +// We deliberately do NOT subscribe to EventTaskQueued / EventTaskRunning +// (no thinking-card lifecycle anymore — adds noise without value) or to +// EventTaskCompleted (chat tasks always emit EventChatDone first, which +// is what we care about; non-chat tasks have no Lark binding anyway and +// would early-return). Leaving EventTaskCompleted unsubscribed also +// avoids the prior "Done." overwrite regression where the no-content +// EventTaskCompleted payload would wipe the real reply. func (p *Patcher) Register(bus *events.Bus) { - bus.Subscribe(protocol.EventTaskQueued, p.handleEvent) - bus.Subscribe(protocol.EventTaskRunning, p.handleEvent) bus.Subscribe(protocol.EventTaskFailed, p.handleEvent) bus.Subscribe(protocol.EventChatDone, p.handleEvent) } @@ -311,16 +308,40 @@ func (p *Patcher) processEvent(ctx context.Context, e events.Event) error { } switch e.Type { - case protocol.EventTaskQueued, protocol.EventTaskRunning: - return p.ensureCard(ctx, creds, binding, taskID, agentName, e.Type) case protocol.EventChatDone: - return p.finalize(ctx, creds, binding, taskID, agentName, e.Payload) + return p.sendChatReply(ctx, creds, binding, e.Payload) case protocol.EventTaskFailed: return p.fail(ctx, creds, binding, taskID, agentName, e.Payload) } return nil } +// sendChatReply turns ChatDonePayload.Content into a plain text Lark +// message. No card chrome, no thinking-card lifecycle, no DB row in +// lark_outbound_card_message — each reply is a one-shot text bubble +// indistinguishable from a user typing. +// +// Empty content is silently dropped: we'd rather show nothing than +// "Done." (the prior card fallback that confused Bohan in the live +// dev env). In practice an empty Content means the daemon completed +// the task without producing visible output, which only happens for +// edge cases like a chat task that just acknowledged a system event; +// not emitting a message there is the right product call. +func (p *Patcher) sendChatReply(ctx context.Context, creds InstallationCredentials, binding db.LarkChatSessionBinding, payload any) error { + content := chatDoneContent(payload) + if content == "" { + return nil + } + if _, err := p.client.SendTextMessage(ctx, SendTextParams{ + InstallationID: creds, + ChatID: ChatID(binding.LarkChatID), + Text: content, + }); err != nil { + return fmt.Errorf("send text message: %w", err) + } + return nil +} + func (p *Patcher) installationCredentials(inst db.LarkInstallation) (InstallationCredentials, error) { if p.credentials == nil { return InstallationCredentials{}, errors.New("lark patcher: credentials resolver missing") @@ -339,117 +360,16 @@ func (p *Patcher) installationCredentials(inst db.LarkInstallation) (Installatio return creds, nil } -// ensureCard creates the initial "thinking" card on first sight of the -// task, or patches the existing card to its "running" body once the -// daemon claims the task. Both transitions respect the per-card -// throttle. -func (p *Patcher) ensureCard(ctx context.Context, creds InstallationCredentials, binding db.LarkChatSessionBinding, taskID pgtype.UUID, agentName string, eventType string) error { - card, err := p.queries.GetLarkOutboundCardByTask(ctx, taskID) - if errors.Is(err, pgx.ErrNoRows) { - // First sight: render + send the thinking card, persist the binding. - render, err := p.cfg.Renderer.Render(RenderInput{ - Kind: CardKindThinking, - AgentName: agentName, - TaskID: taskID, - }) - if err != nil { - return fmt.Errorf("render thinking card: %w", err) - } - cardMessageID, err := p.client.SendInteractiveCard(ctx, SendCardParams{ - InstallationID: creds, - ChatID: ChatID(binding.LarkChatID), - CardJSON: render.JSON, - }) - if err != nil { - return fmt.Errorf("send thinking card: %w", err) - } - if _, err := p.queries.CreateLarkOutboundCardMessage(ctx, db.CreateLarkOutboundCardMessageParams{ - ChatSessionID: binding.ChatSessionID, - LarkChatID: binding.LarkChatID, - LarkCardMessageID: cardMessageID, - Status: string(CardStatusPending), - TaskID: pgtype.UUID{Bytes: taskID.Bytes, Valid: true}, - }); err != nil { - return fmt.Errorf("persist outbound card: %w", err) - } - p.markPatched(taskID) - return nil - } - if err != nil { - return fmt.Errorf("load existing card: %w", err) - } - // Existing card. Step it forward to streaming when the task - // transitions to running; ignore duplicate queued events. - if eventType != protocol.EventTaskRunning { - return nil - } - if !p.shouldPatch(taskID) { - return nil - } - render, err := p.cfg.Renderer.Render(RenderInput{ - Kind: CardKindRunning, - AgentName: agentName, - TaskID: taskID, - }) - if err != nil { - return fmt.Errorf("render running card: %w", err) - } - if err := p.client.PatchInteractiveCard(ctx, PatchCardParams{ - InstallationID: creds, - LarkCardMessageID: card.LarkCardMessageID, - CardJSON: render.JSON, - }); err != nil { - return fmt.Errorf("patch running card: %w", err) - } - if err := p.queries.UpdateLarkOutboundCardStatus(ctx, db.UpdateLarkOutboundCardStatusParams{ - ID: card.ID, - Status: string(CardStatusStreaming), - }); err != nil { - return fmt.Errorf("update card status: %w", err) - } - p.markPatched(taskID) - return nil -} - -func (p *Patcher) finalize(ctx context.Context, creds InstallationCredentials, binding db.LarkChatSessionBinding, taskID pgtype.UUID, agentName string, payload any) error { - card, err := p.loadCardOrSkip(ctx, taskID) - if err != nil || card == nil { - return err - } - content := chatDoneContent(payload) - render, err := p.cfg.Renderer.Render(RenderInput{ - Kind: CardKindFinal, - AgentName: agentName, - TaskID: taskID, - Content: content, - }) - if err != nil { - return fmt.Errorf("render final card: %w", err) - } - // Final transitions bypass the throttle: even if a streaming - // patch fired 50ms ago, the user must see the final state. - if err := p.client.PatchInteractiveCard(ctx, PatchCardParams{ - InstallationID: creds, - LarkCardMessageID: card.LarkCardMessageID, - CardJSON: render.JSON, - }); err != nil { - return fmt.Errorf("patch final card: %w", err) - } - if err := p.queries.UpdateLarkOutboundCardStatus(ctx, db.UpdateLarkOutboundCardStatusParams{ - ID: card.ID, - Status: string(CardStatusFinal), - }); err != nil { - return fmt.Errorf("update card final status: %w", err) - } - p.markPatched(taskID) - return nil -} - +// fail surfaces a short error card on task failure. Unlike the +// success path (plain text via sendChatReply), failures stay as cards +// because the user benefits from the visual distinction — a red / +// header-styled card is much harder to miss than a regular bubble, +// and these are rare enough that the card chrome isn't noisy. +// +// One-shot send (no patching, no DB row): if the task fails a second +// time we'd just send a second card, which is fine — failure is +// usually a single terminal event. func (p *Patcher) fail(ctx context.Context, creds InstallationCredentials, binding db.LarkChatSessionBinding, taskID pgtype.UUID, agentName string, payload any) error { - card, err := p.loadCardOrSkip(ctx, taskID) - if err != nil || card == nil { - return err - } render, err := p.cfg.Renderer.Render(RenderInput{ Kind: CardKindError, AgentName: agentName, @@ -459,54 +379,16 @@ func (p *Patcher) fail(ctx context.Context, creds InstallationCredentials, bindi if err != nil { return fmt.Errorf("render error card: %w", err) } - if err := p.client.PatchInteractiveCard(ctx, PatchCardParams{ - InstallationID: creds, - LarkCardMessageID: card.LarkCardMessageID, - CardJSON: render.JSON, + if _, err := p.client.SendInteractiveCard(ctx, SendCardParams{ + InstallationID: creds, + ChatID: ChatID(binding.LarkChatID), + CardJSON: render.JSON, }); err != nil { - return fmt.Errorf("patch error card: %w", err) + return fmt.Errorf("send error card: %w", err) } - if err := p.queries.UpdateLarkOutboundCardStatus(ctx, db.UpdateLarkOutboundCardStatusParams{ - ID: card.ID, - Status: string(CardStatusError), - }); err != nil { - return fmt.Errorf("update card error status: %w", err) - } - p.markPatched(taskID) return nil } -func (p *Patcher) loadCardOrSkip(ctx context.Context, taskID pgtype.UUID) (*db.LarkOutboundCardMessage, error) { - card, err := p.queries.GetLarkOutboundCardByTask(ctx, taskID) - if err == nil { - return &card, nil - } - if errors.Is(err, pgx.ErrNoRows) { - // No card was ever sent (task didn't originate via Lark); - // quietly skip rather than spam logs. - return nil, nil - } - return nil, fmt.Errorf("load existing card: %w", err) -} - -func (p *Patcher) shouldPatch(taskID pgtype.UUID) bool { - key := uuidString(taskID) - p.mu.Lock() - defer p.mu.Unlock() - last, ok := p.lastPatched[key] - if !ok { - return true - } - return p.cfg.Now().Sub(last) >= p.cfg.MinPatchInterval -} - -func (p *Patcher) markPatched(taskID pgtype.UUID) { - key := uuidString(taskID) - p.mu.Lock() - defer p.mu.Unlock() - p.lastPatched[key] = p.cfg.Now() -} - // taskAndSessionFromEvent parses the typed-ish payload broadcastTaskEvent // publishes — a map[string]any with `task_id` (always) and // `chat_session_id` (chat tasks only). EventChatDone carries a diff --git a/server/internal/integrations/lark/outbound_test.go b/server/internal/integrations/lark/outbound_test.go index b78ac60a2..7bc30d985 100644 --- a/server/internal/integrations/lark/outbound_test.go +++ b/server/internal/integrations/lark/outbound_test.go @@ -69,13 +69,16 @@ func (f fakeCredentials) DecryptAppSecret(inst db.LarkInstallation) (string, err } type fakeAPIClient struct { - mu sync.Mutex - sent []SendCardParams - patched []PatchCardParams - sendReturn string - sendErr error - patchErr error - bindingSent []BindingPromptParams + mu sync.Mutex + sent []SendCardParams + patched []PatchCardParams + textSent []SendTextParams + sendReturn string + sendErr error + patchErr error + textSendErr error + textSendReturn string + bindingSent []BindingPromptParams } func (f *fakeAPIClient) IsConfigured() bool { return true } @@ -92,6 +95,12 @@ func (f *fakeAPIClient) PatchInteractiveCard(ctx context.Context, p PatchCardPar f.patched = append(f.patched, p) return f.patchErr } +func (f *fakeAPIClient) SendTextMessage(ctx context.Context, p SendTextParams) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.textSent = append(f.textSent, p) + return f.textSendReturn, f.textSendErr +} func (f *fakeAPIClient) SendBindingPromptCard(ctx context.Context, p BindingPromptParams) error { f.mu.Lock() defer f.mu.Unlock() @@ -106,10 +115,10 @@ func newTestPatcher(t *testing.T) (*Patcher, *fakePatcherQueries, *fakeAPIClient t.Helper() q := &fakePatcherQueries{ binding: db.LarkChatSessionBinding{ - ChatSessionID: uuidFromString(t, "cccccccc-cccc-cccc-cccc-cccccccccccc"), - InstallationID: uuidFromString(t, "1111aaaa-1111-1111-1111-111111111111"), - LarkChatID: "oc_test_chat", - LarkChatType: "p2p", + ChatSessionID: uuidFromString(t, "cccccccc-cccc-cccc-cccc-cccccccccccc"), + InstallationID: uuidFromString(t, "1111aaaa-1111-1111-1111-111111111111"), + LarkChatID: "oc_test_chat", + LarkChatType: "p2p", }, installation: db.LarkInstallation{ ID: uuidFromString(t, "1111aaaa-1111-1111-1111-111111111111"), @@ -121,83 +130,22 @@ func newTestPatcher(t *testing.T) (*Patcher, *fakePatcherQueries, *fakeAPIClient agent: db.Agent{Name: "TestAgent"}, cardErr: pgx.ErrNoRows, } - api := &fakeAPIClient{sendReturn: "lark_card_msg_1"} + api := &fakeAPIClient{sendReturn: "lark_card_msg_1", textSendReturn: "lark_text_msg_1"} p := NewPatcher(q, fakeCredentials{secret: "shh"}, api, PatcherConfig{ - MinPatchInterval: 50 * time.Millisecond, - Logger: newDiscardLogger(), - Now: time.Now, + Logger: newDiscardLogger(), + Now: time.Now, }) return p, q, api } -func TestPatcherSendsThinkingCardOnTaskQueued(t *testing.T) { - p, q, api := newTestPatcher(t) - taskID := uuidFromString(t, "ee111111-ee11-ee11-ee11-eeeeeeeeeeee") - sessionID := q.binding.ChatSessionID - - p.handleEvent(events.Event{ - Type: protocol.EventTaskQueued, - TaskID: uuidString(taskID), - ChatSessionID: uuidString(sessionID), - Payload: map[string]any{ - "task_id": uuidString(taskID), - "chat_session_id": uuidString(sessionID), - "status": "queued", - }, - }) - - api.mu.Lock() - defer api.mu.Unlock() - if len(api.sent) != 1 { - t.Fatalf("expected one SendInteractiveCard call, got %d", len(api.sent)) - } - if api.sent[0].InstallationID.AppID != "cli_test_app" { - t.Fatalf("expected app_id propagated, got %q", api.sent[0].InstallationID.AppID) - } - q.mu.Lock() - defer q.mu.Unlock() - if len(q.created) != 1 { - t.Fatalf("expected one CreateLarkOutboundCardMessage call, got %d", len(q.created)) - } - if q.created[0].Status != string(CardStatusPending) { - t.Fatalf("initial card status should be 'pending', got %q", q.created[0].Status) - } - if q.created[0].LarkCardMessageID != "lark_card_msg_1" { - t.Fatalf("expected card binding to use returned message id, got %q", q.created[0].LarkCardMessageID) - } -} - -func TestPatcherSkipsWhenNoChatSessionBinding(t *testing.T) { - p, q, api := newTestPatcher(t) - q.bindingErr = pgx.ErrNoRows - - p.handleEvent(events.Event{ - Type: protocol.EventTaskQueued, - TaskID: uuidString(uuidFromString(t, "ee222222-ee22-ee22-ee22-eeeeeeeeeeee")), - ChatSessionID: uuidString(uuidFromString(t, "cc222222-cc22-cc22-cc22-cccccccccccc")), - }) - - api.mu.Lock() - defer api.mu.Unlock() - if len(api.sent) != 0 { - t.Fatalf("web-only chat sessions must not produce Lark cards; got %d sends", len(api.sent)) - } -} - -func TestPatcherFinalCardBypassesThrottle(t *testing.T) { +// TestPatcherSendsPlainTextOnChatDone pins the new behaviour Bohan asked +// for: when the agent finishes replying, the Patcher posts the reply as +// a plain Lark IM text message (msg_type=text), not nested inside an +// interactive card. This is the load-bearing UX call — the prior card +// chrome made every reply look like a system notification. +func TestPatcherSendsPlainTextOnChatDone(t *testing.T) { p, q, api := newTestPatcher(t) taskID := uuidFromString(t, "ee333333-ee33-ee33-ee33-eeeeeeeeeeee") - cardID := uuidFromString(t, "dd111111-dd11-dd11-dd11-dddddddddddd") - q.card = db.LarkOutboundCardMessage{ - ID: cardID, - LarkCardMessageID: "lark_card_msg_existing", - ChatSessionID: q.binding.ChatSessionID, - LarkChatID: q.binding.LarkChatID, - Status: string(CardStatusStreaming), - } - q.cardErr = nil - // Pretend we just patched ms ago so the throttle would otherwise refuse. - p.markPatched(taskID) p.handleEvent(events.Event{ Type: protocol.EventChatDone, @@ -206,34 +154,88 @@ func TestPatcherFinalCardBypassesThrottle(t *testing.T) { Payload: protocol.ChatDonePayload{ TaskID: uuidString(taskID), ChatSessionID: uuidString(q.binding.ChatSessionID), - Content: "agent reply text", + Content: "Hello! I'm cc, a coding agent…", }, }) api.mu.Lock() defer api.mu.Unlock() - if len(api.patched) != 1 { - t.Fatalf("final transition must bypass throttle; got %d patches", len(api.patched)) + if len(api.textSent) != 1 { + t.Fatalf("expected one SendTextMessage call on ChatDone; got %d", len(api.textSent)) } - if api.patched[0].LarkCardMessageID != "lark_card_msg_existing" { - t.Fatalf("final patch must target existing card id, got %q", api.patched[0].LarkCardMessageID) + got := api.textSent[0] + if got.Text != "Hello! I'm cc, a coding agent…" { + t.Errorf("text mismatch: got %q", got.Text) } - q.mu.Lock() - defer q.mu.Unlock() - if len(q.statusUpdates) != 1 || q.statusUpdates[0].Status != string(CardStatusFinal) { - t.Fatalf("expected single final status update, got %+v", q.statusUpdates) + if got.ChatID != ChatID(q.binding.LarkChatID) { + t.Errorf("chat_id mismatch: got %q want %q", got.ChatID, q.binding.LarkChatID) + } + if got.InstallationID.AppID != "cli_test_app" { + t.Errorf("expected installation app_id propagated; got %q", got.InstallationID.AppID) + } + if len(api.sent) != 0 || len(api.patched) != 0 { + t.Errorf("ChatDone must NOT send / patch any card; got sent=%d patched=%d", + len(api.sent), len(api.patched)) } } -func TestPatcherFailEventTransitionsToError(t *testing.T) { +// TestPatcherDropsEmptyChatReply guards the fallback we deliberately +// removed: the previous design rendered "Done." when content was +// empty. Now an empty Content is silently dropped (no text message +// sent at all). Showing nothing is better than showing the misleading +// "Done." fallback, which Bohan reported confused him in the live env. +func TestPatcherDropsEmptyChatReply(t *testing.T) { + p, q, api := newTestPatcher(t) + taskID := uuidFromString(t, "ee777777-ee77-ee77-ee77-eeeeeeeeeeee") + + p.handleEvent(events.Event{ + Type: protocol.EventChatDone, + TaskID: uuidString(taskID), + ChatSessionID: uuidString(q.binding.ChatSessionID), + Payload: protocol.ChatDonePayload{ + TaskID: uuidString(taskID), + ChatSessionID: uuidString(q.binding.ChatSessionID), + Content: "", + }, + }) + + api.mu.Lock() + defer api.mu.Unlock() + if len(api.textSent) != 0 { + t.Errorf("empty content must drop, not render the Done. fallback; got %d text sends", len(api.textSent)) + } +} + +func TestPatcherSkipsWhenNoChatSessionBinding(t *testing.T) { + p, q, api := newTestPatcher(t) + q.bindingErr = pgx.ErrNoRows + + p.handleEvent(events.Event{ + Type: protocol.EventChatDone, + TaskID: uuidString(uuidFromString(t, "ee222222-ee22-ee22-ee22-eeeeeeeeeeee")), + ChatSessionID: uuidString(uuidFromString(t, "cc222222-cc22-cc22-cc22-cccccccccccc")), + Payload: protocol.ChatDonePayload{ + Content: "irrelevant — no binding", + }, + }) + + api.mu.Lock() + defer api.mu.Unlock() + if len(api.textSent) != 0 || len(api.sent) != 0 { + t.Fatalf("web-only chat sessions must produce no outbound; got text=%d cards=%d", + len(api.textSent), len(api.sent)) + } +} + +// TestPatcherFailEventSendsErrorCard verifies the failure path still +// surfaces a card. The visual distinction between a successful reply +// (plain text bubble) and a failure (red header card) is genuinely +// useful — and failures are rare enough that the card chrome isn't +// noisy. One-shot send (no patching of any prior thinking card, +// because there isn't one anymore). +func TestPatcherFailEventSendsErrorCard(t *testing.T) { p, q, api := newTestPatcher(t) taskID := uuidFromString(t, "ee444444-ee44-ee44-ee44-eeeeeeeeeeee") - q.card = db.LarkOutboundCardMessage{ - ID: uuidFromString(t, "dd222222-dd22-dd22-dd22-dddddddddddd"), - LarkCardMessageID: "lark_card_msg_existing", - Status: string(CardStatusStreaming), - } - q.cardErr = nil p.handleEvent(events.Event{ Type: protocol.EventTaskFailed, @@ -248,13 +250,14 @@ func TestPatcherFailEventTransitionsToError(t *testing.T) { api.mu.Lock() defer api.mu.Unlock() - if len(api.patched) != 1 { - t.Fatalf("fail event must patch the card; got %d patches", len(api.patched)) + if len(api.sent) != 1 { + t.Fatalf("fail event must send an error card; got %d card sends", len(api.sent)) } - q.mu.Lock() - defer q.mu.Unlock() - if len(q.statusUpdates) != 1 || q.statusUpdates[0].Status != string(CardStatusError) { - t.Fatalf("expected single error status update, got %+v", q.statusUpdates) + if len(api.patched) != 0 { + t.Errorf("fail event must NOT patch any card (no prior card lifecycle); got %d patches", len(api.patched)) + } + if !strings.Contains(api.sent[0].CardJSON, "boom") { + t.Errorf("error card body should embed the error message; got %s", api.sent[0].CardJSON) } } @@ -263,46 +266,37 @@ func TestPatcherSwallowsInstallationLoadErrors(t *testing.T) { q.installationErr = errors.New("db down") p.handleEvent(events.Event{ - Type: protocol.EventTaskQueued, + Type: protocol.EventChatDone, TaskID: uuidString(uuidFromString(t, "ee555555-ee55-ee55-ee55-eeeeeeeeeeee")), ChatSessionID: uuidString(q.binding.ChatSessionID), + Payload: protocol.ChatDonePayload{ + Content: "would-be reply", + }, }) - // The patcher logs but never panics; no card sent. + // The patcher logs but never panics; no outbound. api.mu.Lock() defer api.mu.Unlock() - if len(api.sent) != 0 { - t.Fatalf("DB failure must not result in a card send; got %d", len(api.sent)) + if len(api.textSent) != 0 || len(api.sent) != 0 { + t.Fatalf("DB failure must not produce outbound; got text=%d cards=%d", + len(api.textSent), len(api.sent)) } } -// TestPatcherIgnoresEventTaskCompletedForChatTasks is the regression -// for the "card shows Done. instead of the agent's reply" bug Bohan -// hit on the live env. TaskService publishes ChatDone (with content) -// IMMEDIATELY BEFORE TaskCompleted (with no content) for every chat -// task. If the Patcher subscribed to both, the second patch would -// overwrite the real reply with the "Done." fallback. The fix is to -// drop the EventTaskCompleted subscription entirely — EventChatDone -// is the canonical "agent finished" signal for the Lark card path. -// -// This test pins the contract: a Patcher that has never seen -// EventTaskCompleted does not register a subscription for it, and -// even if one is replayed (e.g. event bus replay on reconnect) the -// Patcher's processEvent switch ignores it instead of finalizing. +// TestPatcherIgnoresEventTaskCompletedForChatTasks pins the no-extra-send +// invariant. TaskService publishes ChatDone (with content) immediately +// before TaskCompleted (without content) for every chat task. The +// Patcher must NOT react to TaskCompleted — doing so would either +// re-send the same text reply (duplicate bubble) or send the "Done." +// fallback (the original bug Bohan reported). The fix is to leave +// EventTaskCompleted unsubscribed; this test asserts exactly one +// outbound text message from the sequence. func TestPatcherIgnoresEventTaskCompletedForChatTasks(t *testing.T) { p, q, api := newTestPatcher(t) taskID := uuidFromString(t, "ee666666-ee66-ee66-ee66-eeeeeeeeeeee") - q.card = db.LarkOutboundCardMessage{ - ID: uuidFromString(t, "dd333333-dd33-dd33-dd33-dddddddddddd"), - LarkCardMessageID: "lark_card_msg_existing", - ChatSessionID: q.binding.ChatSessionID, - LarkChatID: q.binding.LarkChatID, - Status: string(CardStatusStreaming), - } - q.cardErr = nil - // Step 1: ChatDone arrives with the real agent reply. The card - // should get patched to the reply text and persisted as final. + // Step 1: ChatDone arrives with the real agent reply. Plain text + // is sent to Lark. p.handleEvent(events.Event{ Type: protocol.EventChatDone, TaskID: uuidString(taskID), @@ -314,10 +308,9 @@ func TestPatcherIgnoresEventTaskCompletedForChatTasks(t *testing.T) { }, }) - // Step 2: TaskCompleted fires immediately after. The Patcher MUST - // NOT patch the card again — doing so would replay the empty - // payload through finalize and overwrite the real reply with the - // "Done." fallback. + // Step 2: TaskCompleted fires immediately after with no content. + // The Patcher MUST NOT send a second message — neither a + // duplicate of the reply nor the "Done." fallback. p.handleEvent(events.Event{ Type: protocol.EventTaskCompleted, TaskID: uuidString(taskID), @@ -331,15 +324,15 @@ func TestPatcherIgnoresEventTaskCompletedForChatTasks(t *testing.T) { api.mu.Lock() defer api.mu.Unlock() - if len(api.patched) != 1 { - t.Fatalf("exactly one patch expected (ChatDone); EventTaskCompleted must be ignored. Got %d patches", len(api.patched)) + if len(api.textSent) != 1 { + t.Fatalf("exactly one text send expected (ChatDone); EventTaskCompleted must be ignored. Got %d sends", len(api.textSent)) } - // And the surviving patch carries the actual reply text — not "Done.". - if !strings.Contains(api.patched[0].CardJSON, "Hello! I'm cc") { - t.Errorf("patched card body should contain the agent reply; got %s", api.patched[0].CardJSON) + if api.textSent[0].Text != "Hello! I'm cc, a coding agent…" { + t.Errorf("text content mismatch; got %q", api.textSent[0].Text) } - if strings.Contains(api.patched[0].CardJSON, `"content":"Done."`) { - t.Errorf("patched card body should NOT contain the Done. fallback; got %s", api.patched[0].CardJSON) + if len(api.sent) != 0 || len(api.patched) != 0 { + t.Errorf("no card outbound expected on the success path; got sent=%d patched=%d", + len(api.sent), len(api.patched)) } } diff --git a/server/internal/integrations/lark/outcome_replier_test.go b/server/internal/integrations/lark/outcome_replier_test.go index 0877260eb..f4c3de47a 100644 --- a/server/internal/integrations/lark/outcome_replier_test.go +++ b/server/internal/integrations/lark/outcome_replier_test.go @@ -41,6 +41,10 @@ func (s *stubAPIClientWithRecorder) PatchInteractiveCard(ctx context.Context, p return nil } +func (s *stubAPIClientWithRecorder) SendTextMessage(ctx context.Context, p SendTextParams) (string, error) { + return "", nil +} + func (s *stubAPIClientWithRecorder) SendBindingPromptCard(ctx context.Context, p BindingPromptParams) error { s.mu.Lock() defer s.mu.Unlock()