From f260195439e00fdce249cd50ab15c50ef759e8d5 Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Thu, 9 Jul 2026 19:26:57 +0800 Subject: [PATCH] fix: ack Feishu card actions before interrupt work --- hub/src/feishu/card/streaming-card.ts | 67 ++++++---- hub/src/feishu/client.ts | 9 +- hub/test/integration/trigger.test.ts | 28 ++++ hub/test/unit/feishu-card-action-ack.test.ts | 131 +++++++++++++++++++ 4 files changed, 206 insertions(+), 29 deletions(-) create mode 100644 hub/test/unit/feishu-card-action-ack.test.ts diff --git a/hub/src/feishu/card/streaming-card.ts b/hub/src/feishu/card/streaming-card.ts index 945960e..e3d5b62 100644 --- a/hub/src/feishu/card/streaming-card.ts +++ b/hub/src/feishu/card/streaming-card.ts @@ -21,7 +21,7 @@ */ import type { FeishuRuntime } from "../client.js"; -import { sendCard, patchCard } from "../client.js"; +import { sendCard, patchCard, sendText } from "../client.js"; import { DEFAULT_MAX_MESSAGE_LENGTH, splitAtBoundary } from "../textStream.js"; import { startToolUseTraceRun, @@ -106,26 +106,34 @@ export class StreamingAgentCard { async finish(fallbackText: string, options: { readonly interrupted?: boolean } = {}): Promise { await this.flushChain; this.interrupted = options.interrupted === true; - if (this.text.length > 0) { - await this.flushCard("complete", this.text); - return; + try { + let updated = true; + if (this.text.length > 0) { + updated = await this.flushCard("complete", this.text); + } else if (this.currentMessageId === null && fallbackText.length > 0) { + // No streaming text was sent. If we never created a card, send one now. + this.text = fallbackText; + updated = await this.flushCard("complete", this.text); + } else if (this.currentMessageId !== null) { + // Patch the existing card with the final text. + updated = await this.flushCard("complete", fallbackText); + } + if (!updated && this.interrupted) { + await sendText(this.rt, this.chatId, "\u5DF2\u4E2D\u65AD\u5F53\u524D\u8FD0\u884C\u3002"); + } + } finally { + clearToolUseTraceRun(this.runId); } - // No streaming text was sent. If we never created a card, send one now. - if (this.currentMessageId === null && fallbackText.length > 0) { - this.text = fallbackText; - await this.flushCard("complete", this.text); - } else if (this.currentMessageId !== null) { - // Patch the existing card with the final text - await this.flushCard("complete", fallbackText); - } - clearToolUseTraceRun(this.runId); } async fail(errorText: string): Promise { await this.flushChain; - this.text = errorText; - await this.flushCard("complete", errorText, true); - clearToolUseTraceRun(this.runId); + try { + this.text = errorText; + await this.flushCard("complete", errorText, true); + } finally { + clearToolUseTraceRun(this.runId); + } } // --- Internal flush logic --- @@ -139,23 +147,23 @@ export class StreamingAgentCard { }); } - private async flush(): Promise { + private async flush(): Promise { if (this.currentMessageId === null) { - await this.flushCard(this.currentPhase(), this.text); + const updated = await this.flushCard(this.currentPhase(), this.text); this.lastPatchAt = Date.now(); - return; + return updated; } const now = Date.now(); - if (now - this.lastPatchAt < this.patchIntervalMs) return; + if (now - this.lastPatchAt < this.patchIntervalMs) return true; this.lastPatchAt = now; - await this.flushCard(this.currentPhase(), this.text); + return this.flushCard(this.currentPhase(), this.text); } - private async flushCard(phase: CardPhase, text: string, isError = false): Promise { + private async flushCard(phase: CardPhase, text: string, isError = false): Promise { const chunks = splitAtBoundary(text, this.maxMessageLength); const firstChunk = chunks[0]; - if (firstChunk === undefined) return; + if (firstChunk === undefined) return true; const toolUseSteps = getToolUseTraceSteps(this.runId); const card = buildAgentCard({ @@ -171,6 +179,7 @@ export class StreamingAgentCard { if (this.currentMessageId === null) { this.currentMessageId = await sendCard(this.rt, this.chatId, card); + let updated = this.currentMessageId !== null; // Send overflow chunks as new messages (rare for agent output) for (const chunk of chunks.slice(1)) { const overflowCard = buildAgentCard({ @@ -183,10 +192,13 @@ export class StreamingAgentCard { interrupted: this.interrupted, runId: undefined, }); - this.currentMessageId = await sendCard(this.rt, this.chatId, overflowCard); + const overflowMessageId = await sendCard(this.rt, this.chatId, overflowCard); + updated = updated && overflowMessageId !== null; + this.currentMessageId = overflowMessageId; } + return updated; } else { - await patchCard(this.rt, this.currentMessageId, card); + let updated = await patchCard(this.rt, this.currentMessageId, card); // For overflow, create new messages for (const chunk of chunks.slice(1)) { const overflowCard = buildAgentCard({ @@ -199,8 +211,11 @@ export class StreamingAgentCard { interrupted: this.interrupted, runId: undefined, }); - this.currentMessageId = await sendCard(this.rt, this.chatId, overflowCard); + const overflowMessageId = await sendCard(this.rt, this.chatId, overflowCard); + updated = updated && overflowMessageId !== null; + this.currentMessageId = overflowMessageId; } + return updated; } } diff --git a/hub/src/feishu/client.ts b/hub/src/feishu/client.ts index f285778..9ec5e21 100644 --- a/hub/src/feishu/client.ts +++ b/hub/src/feishu/client.ts @@ -294,7 +294,7 @@ export async function sendCard(rt: FeishuRuntime, chatId: string, card: Record): Promise { +export async function patchCard(rt: FeishuRuntime, messageId: string, card: Record): Promise { const client = rt.client as unknown as { im: { v1: { message: { patch: (p: unknown) => Promise } } }; }; @@ -304,8 +304,10 @@ export async function patchCard(rt: FeishuRuntime, messageId: string, card: Reco params: { msg_type: "interactive" }, data: { content: JSON.stringify(card) }, }); + return true; } catch (e) { rt.logger.warn({ messageId, err: e instanceof Error ? e.message : String(e) }, "patchCard failed"); + return false; } } @@ -831,8 +833,9 @@ export function startFeishuListenerWithClient( }; if (onCardAction !== undefined) { handlers["card.action.trigger"] = async (data) => { - try { await onCardAction(data as CardActionEvent, rt); } - catch (e) { logger.error({ err: e }, "feishu card action handler threw"); } + void Promise.resolve() + .then(() => onCardAction(data as CardActionEvent, rt)) + .catch((e) => { logger.error({ err: e }, "feishu card action handler threw"); }); }; } void wsClient.start({ eventDispatcher: new lark.EventDispatcher({}).register(handlers) }); diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index 07bf88e..d8c3644 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -618,6 +618,34 @@ describe("trigger full lifecycle (integration)", () => { expect(rt.sentPatches.some((card) => cardHasInterruptedFooter(card))).toBe(true); }); + it("sends a fallback interrupt notice when the final card patch fails", async () => { + await seedProject("proj-int-patch-fail", "chat-int-patch-fail", { role: "MANAGE" }); + runAgent = createHangingRunAgent(); + const patch = vi.fn(async () => { + throw new Error("patch failed"); + }); + (rt.client as unknown as { im: { v1: { message: { patch: typeof patch } } } }).im.v1.message.patch = patch; + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); + + await trigger(makeEvent("chat-int-patch-fail", "@_user_1 写教案"), rt); + + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs[0]?.status).toBe("ACTIVE"); + }); + const run = await prisma.agentRun.findFirst(); + expect(run).not.toBeNull(); + + await trigger.onCardAction(makeInterruptEvent("chat-int-patch-fail", run!.id, "ou_test_user"), rt); + + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs[0]?.status).toBe("CANCELED"); + }); + expect(patch).toHaveBeenCalled(); + expect(rt.sentTexts).toContain("已中断当前运行。"); + }); + it("denies interrupt when the operator lacks agent.cancel permission", async () => { // EDIT role can trigger but cannot cancel (agent.cancel requires MANAGE). await seedProject("proj-int-deny", "chat-int-deny", { role: "EDIT" }); diff --git a/hub/test/unit/feishu-card-action-ack.test.ts b/hub/test/unit/feishu-card-action-ack.test.ts new file mode 100644 index 0000000..fae2ba7 --- /dev/null +++ b/hub/test/unit/feishu-card-action-ack.test.ts @@ -0,0 +1,131 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import type { FastifyBaseLogger } from "fastify"; +import type { CardActionEvent, FeishuRuntime } from "../../src/feishu/client.js"; +import { startFeishuListenerWithClient } from "../../src/feishu/client.js"; + +const larkMock = vi.hoisted(() => { + const starts: Array<{ readonly eventDispatcher?: unknown }> = []; + class MockEventDispatcher { + handlers: Record Promise> = {}; + + register(handlers: Record Promise>): this { + this.handlers = handlers; + return this; + } + } + class MockWSClient { + start(params: { readonly eventDispatcher?: unknown }): void { + starts.push(params); + } + } + return { starts, MockEventDispatcher, MockWSClient }; +}); + +vi.mock("@larksuiteoapi/node-sdk", () => ({ + AppType: { SelfBuild: "SelfBuild" }, + Domain: { Feishu: "Feishu" }, + LoggerLevel: { info: "info" }, + Client: vi.fn(), + EventDispatcher: larkMock.MockEventDispatcher, + WSClient: larkMock.MockWSClient, +})); + +describe("Feishu card action callbacks", () => { + beforeEach(() => { + larkMock.starts.length = 0; + }); + + it("acks card action callbacks before the business handler settles", async () => { + const action = deferred(); + const onCardAction = vi.fn(async () => action.promise); + const logger = silentLogger(); + + startFeishuListenerWithClient( + { appId: "app-id", appSecret: "app-secret", botOpenId: "bot-open-id" }, + fakeClient(), + logger, + async () => {}, + onCardAction, + ); + + const handler = cardActionHandler(); + const handlerDone = handler(makeCardActionEvent()).then(() => "done" as const); + const earlyResult = await Promise.race([handlerDone, delay(25).then(() => "timeout" as const)]); + + action.resolve(); + await handlerDone; + + expect(earlyResult).toBe("done"); + expect(onCardAction).toHaveBeenCalledTimes(1); + }); + + it("logs background card action failures after acking", async () => { + const logger = silentLogger(); + const err = new Error("handler failed"); + + startFeishuListenerWithClient( + { appId: "app-id", appSecret: "app-secret", botOpenId: "bot-open-id" }, + fakeClient(), + logger, + async () => {}, + async () => { + throw err; + }, + ); + + await cardActionHandler()(makeCardActionEvent()); + await vi.waitFor(() => { + expect(logger.error).toHaveBeenCalledWith({ err }, "feishu card action handler threw"); + }); + }); +}); + +function cardActionHandler(): (data: unknown) => Promise { + const start = larkMock.starts[0]; + if (start === undefined) throw new Error("WSClient.start was not called"); + const dispatcher = start.eventDispatcher as { readonly handlers?: Record Promise> }; + const handler = dispatcher.handlers?.["card.action.trigger"]; + if (handler === undefined) throw new Error("card.action.trigger handler was not registered"); + return handler; +} + +function makeCardActionEvent(): CardActionEvent { + return { + operator: { open_id: "ou_user" }, + action: { value: { interrupt_run: "run-1" }, tag: "button" }, + context: { open_chat_id: "chat-1", open_message_id: "message-1" }, + }; +} + +function fakeClient(): FeishuRuntime["client"] { + return {} as FeishuRuntime["client"]; +} + +function silentLogger(): FastifyBaseLogger { + return { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + fatal: vi.fn(), + child() { return this; }, + level: "silent", + } as unknown as FastifyBaseLogger; +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +interface Deferred { + readonly promise: Promise; + readonly resolve: (value: T) => void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +}