diff --git a/src/gateway/channels/lark-card.test.ts b/src/gateway/channels/lark-card.test.ts index 5fa4f845..28f415eb 100644 --- a/src/gateway/channels/lark-card.test.ts +++ b/src/gateway/channels/lark-card.test.ts @@ -308,7 +308,7 @@ describe("feedback buttons", () => { const session = { cardId: "CARD-FB-1", elementId: "md_main", sequence: 0 }; const ok = await finalizeCard(client as any, session, "done", { - ctx: { sessionId: "sess-1", channelId: "ch-1" }, + ctx: { sessionId: "sess-1", channelId: "ch-1", messageId: "msg-assistant-1" }, locale: "zh-CN", }); expect(ok).toBe(true); @@ -328,7 +328,7 @@ describe("feedback buttons", () => { expect(row.element_id).toBe("fb_row"); const buttons = row.columns.map((c: any) => c.elements[0]); expect(buttons.map((b: any) => b.element_id)).toEqual(["fb_up", "fb_down"]); - // The value payload is self-contained — persistence needs no message-id map. + // The value payload is self-contained — persistence needs no card-to-message map. expect(buttons[0].behaviors[0]).toEqual({ type: "callback", value: { @@ -337,6 +337,7 @@ describe("feedback buttons", () => { session_id: "sess-1", card_id: "CARD-FB-1", channel_id: "ch-1", + message_id: "msg-assistant-1", locale: "zh-CN", }, }); diff --git a/src/gateway/channels/lark-card.ts b/src/gateway/channels/lark-card.ts index f6d89951..3f161c23 100644 --- a/src/gateway/channels/lark-card.ts +++ b/src/gateway/channels/lark-card.ts @@ -110,8 +110,9 @@ export interface CardSession { // // The final answer card carries a feedback row. Clicks arrive as a // `card.action.trigger` callback over the SAME long connection as messages; -// the button's `value` payload is self-contained (session/card/channel), so -// persistence never needs a Feishu-message-id → session mapping. +// the button's `value` payload is self-contained (session/card/channel plus +// the persisted assistant message), so persistence never needs an in-memory +// Feishu-card → chat-message mapping. /** Discriminator inside `action.value` so unrelated card actions are ignored. */ export const FEEDBACK_ACTION_KIND = "siclaw_feedback"; @@ -124,6 +125,8 @@ export type FeedbackRating = "up" | "down"; export interface FeedbackContext { sessionId: string; channelId: string; + /** Exact persisted assistant reply rated by this card. Absent on legacy cards. */ + messageId?: string; } /** Payload embedded in each button; comes back verbatim in the callback. */ @@ -133,6 +136,8 @@ export interface FeedbackActionValue { session_id: string; card_id: string; channel_id: string; + /** Added after channel audit persistence exposed the final assistant id. */ + message_id?: string; locale: LarkLocale; } @@ -162,6 +167,7 @@ function buildFeedbackRow( session_id: ctx.sessionId, card_id: cardId, channel_id: ctx.channelId, + ...(ctx.messageId ? { message_id: ctx.messageId } : {}), locale, }; return { @@ -359,7 +365,7 @@ async function appendFeedbackRow( */ export async function applyFeedbackSelection( larkClient: any, - value: Pick, + value: Pick, rating: FeedbackRating, ): Promise { const session = feedbackEchoSessions.get(value.card_id); @@ -371,7 +377,7 @@ export async function applyFeedbackSelection( data: { element: JSON.stringify(buildFeedbackRow( value.card_id, - { sessionId: value.session_id, channelId: value.channel_id }, + { sessionId: value.session_id, channelId: value.channel_id, messageId: value.message_id }, locale, rating, )), diff --git a/src/gateway/channels/lark.test.ts b/src/gateway/channels/lark.test.ts index b97376fb..0f794404 100644 --- a/src/gateway/channels/lark.test.ts +++ b/src/gateway/channels/lark.test.ts @@ -1086,6 +1086,7 @@ describe("handleLarkCardAction — 👍/👎 feedback", () => { session_id: "sess-1", card_id: "CARD-1", channel_id: "lark", + message_id: "msg-assistant-1", locale: "zh-CN", }, }, @@ -1102,6 +1103,25 @@ describe("handleLarkCardAction — 👍/👎 feedback", () => { const result = handleLarkCardAction(makeCardAction(), makeLarkClient()); expect(result).toEqual({ toast: { type: "success", content: expect.stringContaining("反馈") } }); + await flush(); + expect(recordChannelFeedbackMock).toHaveBeenCalledWith({ + sessionId: "sess-1", + messageRef: "CARD-1", + messageId: "msg-assistant-1", + rating: "up", + senderExternalId: "ou_clicker", + channelId: "lark", + source: "lark", + }); + }); + + it("keeps legacy cards without message_id compatible", async () => { + recordChannelFeedbackMock.mockResolvedValue({ success: true }); + const data = makeCardAction(); + delete (data.action as any).value.message_id; + + const result = handleLarkCardAction(data, makeLarkClient()); + expect(result).toEqual({ toast: { type: "success", content: expect.any(String) } }); await flush(); expect(recordChannelFeedbackMock).toHaveBeenCalledWith({ sessionId: "sess-1", @@ -1355,6 +1375,7 @@ describe("handleLarkMessage — streaming card flow", () => { }, cardElement: { content: vi.fn().mockResolvedValue({ code: 0 }), + create: vi.fn().mockResolvedValue({ code: 0 }), }, }, }, @@ -1373,6 +1394,9 @@ describe("handleLarkMessage — streaming card flow", () => { }, }; }); + appendMessageMock.mockImplementation(async (message: { role: string }) => + message.role === "assistant" ? "msg-assistant-final" : "msg-user", + ); const lark = makeCardAwareLarkClient(); await handleLarkMessage( @@ -1400,9 +1424,71 @@ describe("handleLarkMessage — streaming card flow", () => { expect(lark.cardkit.v1.card.settings).toHaveBeenCalledTimes(1); const settingsPayload = JSON.parse(lark.cardkit.v1.card.settings.mock.calls[0][0].data.settings); expect(settingsPayload.config.streaming_mode).toBe(false); + const feedbackAppend = lark.cardkit.v1.cardElement.create.mock.calls[0][0]; + const [feedbackRow] = JSON.parse(feedbackAppend.data.elements); + expect(feedbackRow.columns[0].elements[0].behaviors[0].value.message_id).toBe("msg-assistant-final"); expect(lark.im.image.create).not.toHaveBeenCalled(); }); + it("persists a delta-only reply and appends feedback linked to that message", async () => { + resolveBindingMock.mockResolvedValue(makeBinding()); + promptMock.mockResolvedValue({ sessionId: "s-delta-only" }); + streamEventsMock.mockImplementation(async function* () { + yield { type: "content_block_delta", delta: { text: "delta-only " } }; + yield { type: "content_block_delta", delta: { text: "answer" } }; + }); + appendMessageMock.mockImplementation(async (message: { role: string }) => + message.role === "assistant" ? "msg-assistant-delta" : "msg-user", + ); + const lark = makeCardAwareLarkClient(); + + await handleLarkMessage( + makeTextEvent("hello"), + lark, + "lark", + makeAgentBoxManager("a1") as any, + undefined, + {} as any, + ); + + expect(lark.cardkit.v1.cardElement.content.mock.calls[0][0].data.content).toContain("delta-only answer"); + expect(appendMessageMock).toHaveBeenCalledWith(expect.objectContaining({ + sessionId: "s-delta-only", + role: "assistant", + content: "delta-only answer", + })); + const feedbackAppend = lark.cardkit.v1.cardElement.create.mock.calls[0][0]; + const [feedbackRow] = JSON.parse(feedbackAppend.data.elements); + expect(feedbackRow.columns[0].elements[0].behaviors[0].value.message_id).toBe("msg-assistant-delta"); + }); + + it("does not append feedback buttons when the final assistant row fails to persist", async () => { + resolveBindingMock.mockResolvedValue(makeBinding()); + promptMock.mockResolvedValue({ sessionId: "s-persist-fail" }); + streamEventsMock.mockImplementation(async function* () { + yield { + type: "message_end", + message: { role: "assistant", content: [{ type: "text", text: "answer still delivered" }] }, + }; + }); + appendMessageMock + .mockResolvedValueOnce("msg-user") + .mockRejectedValueOnce(new Error("db down")); + const lark = makeCardAwareLarkClient(); + + await handleLarkMessage( + makeTextEvent("hello"), + lark, + "lark", + makeAgentBoxManager("a1") as any, + undefined, + {} as any, + ); + + expect(lark.cardkit.v1.cardElement.content.mock.calls[0][0].data.content).toContain("answer still delivered"); + expect(lark.cardkit.v1.cardElement.create).not.toHaveBeenCalled(); + }); + it("updates the Lark card when a background channel report arrives after the first SSE turn", async () => { resolveBindingMock.mockResolvedValue(makeBinding()); promptMock.mockResolvedValue({ sessionId: "s-background" }); @@ -2420,6 +2506,10 @@ describe("collectChannelResponse — audit persistence", () => { } it("persists every assistant turn + each tool call when persist is set", async () => { + appendMessageMock + .mockResolvedValueOnce("msg-assistant-intermediate") + .mockResolvedValueOnce("msg-tool") + .mockResolvedValueOnce("msg-assistant-final"); const events = [ { type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "Checking nodes" }] } }, { type: "tool_execution_start", toolName: "bash", args: { command: "kubectl get nodes" } }, @@ -2431,6 +2521,7 @@ describe("collectChannelResponse — audit persistence", () => { }); // Reply text is still the final assistant turn. expect(collected.text).toBe("All healthy."); + expect(collected.assistantMessageId).toBe("msg-assistant-final"); const calls = appendMessageMock.mock.calls.map((c) => c[0] as any); expect(calls.filter((m) => m.role === "assistant").map((m) => m.content)).toEqual(["Checking nodes", "All healthy."]); @@ -2442,6 +2533,29 @@ describe("collectChannelResponse — audit persistence", () => { expect(toolRows[0].content).toBe("node ok"); }); + it("persists the synthesized assistant reply when the stream is delta-only", async () => { + appendMessageMock.mockResolvedValueOnce("msg-assistant-delta"); + const events = [ + { type: "content_block_delta", delta: { text: "Hello" } }, + { type: "content_block_delta", delta: { text: " world" } }, + ]; + + const collected = await collectChannelResponse(fakeClient(events), "s-delta", "lark", { + persist: { agentId: "a1" }, + }); + + expect(collected).toMatchObject({ + text: "Hello world", + assistantMessageId: "msg-assistant-delta", + }); + expect(appendMessageMock).toHaveBeenCalledTimes(1); + expect(appendMessageMock).toHaveBeenCalledWith(expect.objectContaining({ + sessionId: "s-delta", + role: "assistant", + content: "Hello world", + })); + }); + it("does NOT persist anything when persist is omitted (reply-only path)", async () => { const events = [ { type: "tool_execution_start", toolName: "bash", args: {} }, @@ -2471,6 +2585,7 @@ describe("collectChannelResponse — audit persistence", () => { ]; const collected = await collectChannelResponse(fakeClient(events), "s-fail", "lark", { persist: { agentId: "a1" } }); expect(collected.text).toBe("still replies"); + expect(collected.assistantMessageId).toBeNull(); }); }); diff --git a/src/gateway/channels/lark.ts b/src/gateway/channels/lark.ts index 4849cc83..9c2104da 100644 --- a/src/gateway/channels/lark.ts +++ b/src/gateway/channels/lark.ts @@ -476,6 +476,7 @@ export function handleLarkCardAction( const result = await recordChannelFeedback({ sessionId, messageRef: cardId, + ...(typeof fb.message_id === "string" && fb.message_id.trim() ? { messageId: fb.message_id.trim() } : {}), rating, senderExternalId: operatorOpenId, channelId: channelId ?? null, @@ -1256,6 +1257,7 @@ async function processQueuedLarkMessage(ctx: QueuedLarkMessageContext): Promise< }; let resultText = ""; let replyImages: RenderedReplyImage[] = []; + let assistantMessageId: string | null = null; let agentError: Error | null = null; let sessionBusy = false; try { @@ -1275,6 +1277,7 @@ async function processQueuedLarkMessage(ctx: QueuedLarkMessageContext): Promise< }); resultText = collected.text; replyImages = collected.images; + assistantMessageId = collected.assistantMessageId; } catch (err) { if (isSessionBusyError(err)) { // Still busy after the retry window — surface a friendly notice, don't clobber. @@ -1312,7 +1315,9 @@ async function processQueuedLarkMessage(ctx: QueuedLarkMessageContext): Promise< // non-answer and skew the feedback signal Metrics aggregates. const isAnswer = !agentError && resultText.trim().length > 0; const ok = await finalizeCard(larkClient, cardSession, finalCardBody, - isAnswer ? { ctx: { sessionId, channelId }, locale } : undefined); + isAnswer && assistantMessageId + ? { ctx: { sessionId, channelId, messageId: assistantMessageId }, locale } + : undefined); deliveredTextChars = finalCardBody.length; if (!ok) { // Partial-failure path: the card is visible but stuck in streaming @@ -1534,6 +1539,8 @@ async function replyVisualImages(larkClient: any, messageId: string, images: Ren export interface CollectedChannelResponse { text: string; images: RenderedReplyImage[]; + /** Persisted id of the final assistant turn, or null when audit persistence failed/was disabled. */ + assistantMessageId: string | null; } export async function collectResponse( @@ -1572,6 +1579,7 @@ export async function collectChannelResponse( // (tool-use turns emit intermediate message_end events that aren't meant // for the user). pi-agent's agent_end signals the last turn is complete. let lastAssistantText = ""; + let lastAssistantMessageId: string | null = null; // ── Audit persistence (opt-in) ────────────────────────────────────────── // Mirrors the field mapping in sse-consumer.ts so a channel transcript looks @@ -1589,9 +1597,12 @@ export async function collectChannelResponse( const toolStarts = new Map(); const pushQ = (m: Map, k: string, v: T): void => { const a = m.get(k) ?? []; a.push(v); m.set(k, a); }; const shiftQ = (m: Map, k: string): T | undefined => m.get(k)?.shift(); - const persistRow = async (msg: Parameters[0]): Promise => { - try { await appendMessage({ ...msg, traceId: persist?.traceId ?? msg.traceId }); } - catch (err) { console.warn(`[${logPrefix}] audit persist failed session=${sessionId}:`, err); } + const persistRow = async (msg: Parameters[0]): Promise => { + try { return await appendMessage({ ...msg, traceId: persist?.traceId ?? msg.traceId }); } + catch (err) { + console.warn(`[${logPrefix}] audit persist failed session=${sessionId}:`, err); + return null; + } }; try { @@ -1683,7 +1694,11 @@ export async function collectChannelResponse( // Persist every assistant turn (intermediate narration + final answer), // mirroring sse-consumer. Awaited so its created_at precedes the next // tool row in the transcript. - if (persist) await persistRow({ sessionId, role: "assistant", content: redact(turnText) }); + // Replace the id even when this write fails: retaining an earlier + // narration id would link the final card to the wrong assistant turn. + lastAssistantMessageId = persist + ? await persistRow({ sessionId, role: "assistant", content: redact(turnText) }) + : null; } } } @@ -1693,7 +1708,18 @@ export async function collectChannelResponse( // Prefer the last full assistant turn; fall back to streamed deltas if the // brain only emits content_block_delta events. const text = lastAssistantText || parts.join(""); - return { text, images }; + // A delta-only stream has no assistant message_end, so persist the exact + // synthesized reply once the stream finishes. This gives feedback cards the + // same precise message linkage as the normal message_end path without + // duplicating assistant rows when a full turn was already persisted. + if (!lastAssistantText && text.trim() && persist) { + lastAssistantMessageId = await persistRow({ + sessionId, + role: "assistant", + content: redact(text), + }); + } + return { text, images, assistantMessageId: lastAssistantMessageId }; } /** diff --git a/src/gateway/chat-repo.ts b/src/gateway/chat-repo.ts index 71fe6686..2f0ff827 100644 --- a/src/gateway/chat-repo.ts +++ b/src/gateway/chat-repo.ts @@ -202,6 +202,8 @@ export async function bindMessageTraceId( export async function recordChannelFeedback(params: { sessionId: string; messageRef: string; + /** Exact assistant chat_messages id. Missing only for legacy cards. */ + messageId?: string; rating: "up" | "down"; senderExternalId: string; channelId?: string | null; @@ -210,6 +212,7 @@ export async function recordChannelFeedback(params: { return getClient().request("chat.recordFeedback", { session_id: params.sessionId, message_ref: params.messageRef, + ...(params.messageId ? { message_id: params.messageId } : {}), rating: params.rating, sender_external_id: params.senderExternalId, channel_id: params.channelId ?? null,