forked from EduCraft/curriculum-project-hub
fix: ack Feishu card actions before interrupt work
This commit is contained in:
@@ -21,7 +21,7 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import type { FeishuRuntime } from "../client.js";
|
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 { DEFAULT_MAX_MESSAGE_LENGTH, splitAtBoundary } from "../textStream.js";
|
||||||
import {
|
import {
|
||||||
startToolUseTraceRun,
|
startToolUseTraceRun,
|
||||||
@@ -106,26 +106,34 @@ export class StreamingAgentCard {
|
|||||||
async finish(fallbackText: string, options: { readonly interrupted?: boolean } = {}): Promise<void> {
|
async finish(fallbackText: string, options: { readonly interrupted?: boolean } = {}): Promise<void> {
|
||||||
await this.flushChain;
|
await this.flushChain;
|
||||||
this.interrupted = options.interrupted === true;
|
this.interrupted = options.interrupted === true;
|
||||||
if (this.text.length > 0) {
|
try {
|
||||||
await this.flushCard("complete", this.text);
|
let updated = true;
|
||||||
return;
|
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<void> {
|
async fail(errorText: string): Promise<void> {
|
||||||
await this.flushChain;
|
await this.flushChain;
|
||||||
this.text = errorText;
|
try {
|
||||||
await this.flushCard("complete", errorText, true);
|
this.text = errorText;
|
||||||
clearToolUseTraceRun(this.runId);
|
await this.flushCard("complete", errorText, true);
|
||||||
|
} finally {
|
||||||
|
clearToolUseTraceRun(this.runId);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// --- Internal flush logic ---
|
// --- Internal flush logic ---
|
||||||
@@ -139,23 +147,23 @@ export class StreamingAgentCard {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private async flush(): Promise<void> {
|
private async flush(): Promise<boolean> {
|
||||||
if (this.currentMessageId === null) {
|
if (this.currentMessageId === null) {
|
||||||
await this.flushCard(this.currentPhase(), this.text);
|
const updated = await this.flushCard(this.currentPhase(), this.text);
|
||||||
this.lastPatchAt = Date.now();
|
this.lastPatchAt = Date.now();
|
||||||
return;
|
return updated;
|
||||||
}
|
}
|
||||||
|
|
||||||
const now = Date.now();
|
const now = Date.now();
|
||||||
if (now - this.lastPatchAt < this.patchIntervalMs) return;
|
if (now - this.lastPatchAt < this.patchIntervalMs) return true;
|
||||||
this.lastPatchAt = now;
|
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<void> {
|
private async flushCard(phase: CardPhase, text: string, isError = false): Promise<boolean> {
|
||||||
const chunks = splitAtBoundary(text, this.maxMessageLength);
|
const chunks = splitAtBoundary(text, this.maxMessageLength);
|
||||||
const firstChunk = chunks[0];
|
const firstChunk = chunks[0];
|
||||||
if (firstChunk === undefined) return;
|
if (firstChunk === undefined) return true;
|
||||||
|
|
||||||
const toolUseSteps = getToolUseTraceSteps(this.runId);
|
const toolUseSteps = getToolUseTraceSteps(this.runId);
|
||||||
const card = buildAgentCard({
|
const card = buildAgentCard({
|
||||||
@@ -171,6 +179,7 @@ export class StreamingAgentCard {
|
|||||||
|
|
||||||
if (this.currentMessageId === null) {
|
if (this.currentMessageId === null) {
|
||||||
this.currentMessageId = await sendCard(this.rt, this.chatId, card);
|
this.currentMessageId = await sendCard(this.rt, this.chatId, card);
|
||||||
|
let updated = this.currentMessageId !== null;
|
||||||
// Send overflow chunks as new messages (rare for agent output)
|
// Send overflow chunks as new messages (rare for agent output)
|
||||||
for (const chunk of chunks.slice(1)) {
|
for (const chunk of chunks.slice(1)) {
|
||||||
const overflowCard = buildAgentCard({
|
const overflowCard = buildAgentCard({
|
||||||
@@ -183,10 +192,13 @@ export class StreamingAgentCard {
|
|||||||
interrupted: this.interrupted,
|
interrupted: this.interrupted,
|
||||||
runId: undefined,
|
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 {
|
} else {
|
||||||
await patchCard(this.rt, this.currentMessageId, card);
|
let updated = await patchCard(this.rt, this.currentMessageId, card);
|
||||||
// For overflow, create new messages
|
// For overflow, create new messages
|
||||||
for (const chunk of chunks.slice(1)) {
|
for (const chunk of chunks.slice(1)) {
|
||||||
const overflowCard = buildAgentCard({
|
const overflowCard = buildAgentCard({
|
||||||
@@ -199,8 +211,11 @@ export class StreamingAgentCard {
|
|||||||
interrupted: this.interrupted,
|
interrupted: this.interrupted,
|
||||||
runId: undefined,
|
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;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -294,7 +294,7 @@ export async function sendCard(rt: FeishuRuntime, chatId: string, card: Record<s
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Patch an interactive card in-place with raw JSON. */
|
/** Patch an interactive card in-place with raw JSON. */
|
||||||
export async function patchCard(rt: FeishuRuntime, messageId: string, card: Record<string, unknown>): Promise<void> {
|
export async function patchCard(rt: FeishuRuntime, messageId: string, card: Record<string, unknown>): Promise<boolean> {
|
||||||
const client = rt.client as unknown as {
|
const client = rt.client as unknown as {
|
||||||
im: { v1: { message: { patch: (p: unknown) => Promise<unknown> } } };
|
im: { v1: { message: { patch: (p: unknown) => Promise<unknown> } } };
|
||||||
};
|
};
|
||||||
@@ -304,8 +304,10 @@ export async function patchCard(rt: FeishuRuntime, messageId: string, card: Reco
|
|||||||
params: { msg_type: "interactive" },
|
params: { msg_type: "interactive" },
|
||||||
data: { content: JSON.stringify(card) },
|
data: { content: JSON.stringify(card) },
|
||||||
});
|
});
|
||||||
|
return true;
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
rt.logger.warn({ messageId, err: e instanceof Error ? e.message : String(e) }, "patchCard failed");
|
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) {
|
if (onCardAction !== undefined) {
|
||||||
handlers["card.action.trigger"] = async (data) => {
|
handlers["card.action.trigger"] = async (data) => {
|
||||||
try { await onCardAction(data as CardActionEvent, rt); }
|
void Promise.resolve()
|
||||||
catch (e) { logger.error({ err: e }, "feishu card action handler threw"); }
|
.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) });
|
void wsClient.start({ eventDispatcher: new lark.EventDispatcher({}).register(handlers) });
|
||||||
|
|||||||
@@ -618,6 +618,34 @@ describe("trigger full lifecycle (integration)", () => {
|
|||||||
expect(rt.sentPatches.some((card) => cardHasInterruptedFooter(card))).toBe(true);
|
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 () => {
|
it("denies interrupt when the operator lacks agent.cancel permission", async () => {
|
||||||
// EDIT role can trigger but cannot cancel (agent.cancel requires MANAGE).
|
// EDIT role can trigger but cannot cancel (agent.cancel requires MANAGE).
|
||||||
await seedProject("proj-int-deny", "chat-int-deny", { role: "EDIT" });
|
await seedProject("proj-int-deny", "chat-int-deny", { role: "EDIT" });
|
||||||
|
|||||||
@@ -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<string, (data: unknown) => Promise<void>> = {};
|
||||||
|
|
||||||
|
register(handlers: Record<string, (data: unknown) => Promise<void>>): 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<void>();
|
||||||
|
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<void> {
|
||||||
|
const start = larkMock.starts[0];
|
||||||
|
if (start === undefined) throw new Error("WSClient.start was not called");
|
||||||
|
const dispatcher = start.eventDispatcher as { readonly handlers?: Record<string, (data: unknown) => Promise<void>> };
|
||||||
|
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<void> {
|
||||||
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||||
|
}
|
||||||
|
|
||||||
|
interface Deferred<T> {
|
||||||
|
readonly promise: Promise<T>;
|
||||||
|
readonly resolve: (value: T) => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
function deferred<T>(): Deferred<T> {
|
||||||
|
let resolve!: (value: T) => void;
|
||||||
|
const promise = new Promise<T>((res) => {
|
||||||
|
resolve = res;
|
||||||
|
});
|
||||||
|
return { promise, resolve };
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user