From 132d70bfb3c950dde0cf42980343cdc97e84daaf Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Tue, 23 Jun 2026 14:32:21 -0700 Subject: [PATCH] refactor: migrate bundled transcript target lookups (#89911) --- .../.generated/plugin-sdk-api-baseline.sha256 | 4 +- .../monitor/message-handler.process.test.ts | 61 ++-- .../src/monitor/message-handler.process.ts | 25 +- .../src/bot-message-dispatch.runtime.ts | 4 +- .../telegram/src/bot-message-dispatch.test.ts | 275 +++++++++--------- .../telegram/src/bot-message-dispatch.ts | 130 +++------ .../bot-native-commands.session-meta.test.ts | 119 ++++++-- .../telegram/src/bot-native-commands.ts | 47 ++- scripts/plugin-sdk-surface-report.mjs | 4 +- .../session-transcript-runtime.test.ts | 73 +++++ src/plugin-sdk/session-transcript-runtime.ts | 75 +++++ .../scripts/plugin-sdk-surface-report.test.ts | 24 +- 12 files changed, 496 insertions(+), 345 deletions(-) diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 63ebbb149250..4f87dfc2f9a4 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -b493f5106f64f249b83c90fd63ea3b81321d0a1ddb2df700a5aafd5cd5022a07 plugin-sdk-api-baseline.json -cc142ff92509f75517a1164930787a4954fb32fc4c69d52efb2dd29e4327afe0 plugin-sdk-api-baseline.jsonl +f7247b5bbfe3f96bffffd25a8be2f89b37999e36731f34a159ae21ded1cedd05 plugin-sdk-api-baseline.json +ce88a53dadc194ceccc63f50146aee03a1a425f551117da826a21519d5bf80db plugin-sdk-api-baseline.jsonl diff --git a/extensions/discord/src/monitor/message-handler.process.test.ts b/extensions/discord/src/monitor/message-handler.process.test.ts index 373792aa2cf1..f5c7ef648f95 100644 --- a/extensions/discord/src/monitor/message-handler.process.test.ts +++ b/extensions/discord/src/monitor/message-handler.process.test.ts @@ -204,31 +204,18 @@ const recordInboundSession = vi.hoisted(() => vi.fn<(params?: unknown) => Promise>(async () => {}), ); const configSessionsMocks = vi.hoisted(() => ({ - loadSessionStore: vi.fn<(storePath: string, opts?: unknown) => Record>( - () => ({}), - ), - readSessionUpdatedAt: vi.fn<(params?: unknown) => number | undefined>(() => undefined), - readLatestAssistantTextFromSessionTranscript: vi.fn< - (sessionFile: string) => Promise<{ text: string; timestamp?: number } | undefined> + getSessionEntry: vi.fn<(params?: unknown) => unknown>(() => undefined), + readLatestAssistantTextByIdentity: vi.fn< + (params?: unknown) => Promise<{ text: string; timestamp?: number } | undefined> >(async () => undefined), - resolveAndPersistSessionFile: vi.fn<(params?: unknown) => Promise<{ sessionFile: string }>>( - async () => ({ sessionFile: "/tmp/openclaw-discord-process-test-session.jsonl" }), - ), - resolveSessionStoreEntry: vi.fn< - (params: { store: Record; sessionKey?: string }) => { existing?: unknown } - >((params) => ({ - existing: params.sessionKey ? params.store[params.sessionKey] : undefined, - })), + readSessionUpdatedAt: vi.fn<(params?: unknown) => number | undefined>(() => undefined), resolveStorePath: vi.fn<(path?: unknown, opts?: unknown) => string>( () => "/tmp/openclaw-discord-process-test-sessions.json", ), })); -const loadSessionStore = configSessionsMocks.loadSessionStore; +const getSessionEntry = configSessionsMocks.getSessionEntry; +const readLatestAssistantTextByIdentity = configSessionsMocks.readLatestAssistantTextByIdentity; const readSessionUpdatedAt = configSessionsMocks.readSessionUpdatedAt; -const readLatestAssistantTextFromSessionTranscript = - configSessionsMocks.readLatestAssistantTextFromSessionTranscript; -const resolveAndPersistSessionFile = configSessionsMocks.resolveAndPersistSessionFile; -const resolveSessionStoreEntry = configSessionsMocks.resolveSessionStoreEntry; const resolveStorePath = configSessionsMocks.resolveStorePath; const createDiscordRestClientSpy = vi.hoisted(() => vi.fn< @@ -400,19 +387,17 @@ vi.mock("openclaw/plugin-sdk/conversation-runtime", () => ({ })); vi.mock("openclaw/plugin-sdk/session-store-runtime", () => ({ - loadSessionStore: (storePath: string, opts?: unknown) => - configSessionsMocks.loadSessionStore(storePath, opts), + getSessionEntry: (params?: unknown) => configSessionsMocks.getSessionEntry(params), readSessionUpdatedAt: (params?: unknown) => configSessionsMocks.readSessionUpdatedAt(params), - readLatestAssistantTextFromSessionTranscript: (sessionFile: string) => - configSessionsMocks.readLatestAssistantTextFromSessionTranscript(sessionFile), - resolveAndPersistSessionFile: (params?: unknown) => - configSessionsMocks.resolveAndPersistSessionFile(params), - resolveSessionStoreEntry: (params: { store: Record; sessionKey?: string }) => - configSessionsMocks.resolveSessionStoreEntry(params), resolveStorePath: (path?: unknown, opts?: unknown) => configSessionsMocks.resolveStorePath(path, opts), })); +vi.mock("openclaw/plugin-sdk/session-transcript-runtime", () => ({ + readLatestAssistantTextByIdentity: (params?: unknown) => + configSessionsMocks.readLatestAssistantTextByIdentity(params), +})); + vi.mock("../client.js", () => ({ createDiscordRuntimeAccountContext: (params: { cfg: unknown; accountId: string }) => ({ cfg: params.cfg, @@ -505,24 +490,16 @@ beforeEach(() => { createDiscordDraftStream.mockClear(); dispatchInboundMessage.mockClear(); recordInboundSession.mockClear(); - loadSessionStore.mockClear(); readSessionUpdatedAt.mockClear(); - readLatestAssistantTextFromSessionTranscript.mockClear(); - resolveAndPersistSessionFile.mockClear(); - resolveSessionStoreEntry.mockClear(); + getSessionEntry.mockClear(); + readLatestAssistantTextByIdentity.mockClear(); resolveStorePath.mockClear(); createDiscordRestClientSpy.mockClear(); dispatchInboundMessage.mockResolvedValue(createNoQueuedDispatchResult()); recordInboundSession.mockResolvedValue(undefined); - loadSessionStore.mockReturnValue({}); readSessionUpdatedAt.mockReturnValue(undefined); - readLatestAssistantTextFromSessionTranscript.mockResolvedValue(undefined); - resolveAndPersistSessionFile.mockResolvedValue({ - sessionFile: "/tmp/openclaw-discord-process-test-session.jsonl", - }); - resolveSessionStoreEntry.mockImplementation((params) => ({ - existing: params.sessionKey ? params.store[params.sessionKey] : undefined, - })); + getSessionEntry.mockReturnValue(undefined); + readLatestAssistantTextByIdentity.mockResolvedValue(undefined); resolveStorePath.mockReturnValue("/tmp/openclaw-discord-process-test-sessions.json"); threadBindingTesting.resetThreadBindingsForTests(); }); @@ -2272,10 +2249,8 @@ describe("processDiscordMessage draft streaming", () => { (_value, index) => `continuation${index}`, ).join(" ")}`; - loadSessionStore.mockReturnValue({ - "agent:main:discord:channel:c1": { sessionId: "session-1" }, - }); - readLatestAssistantTextFromSessionTranscript.mockResolvedValue({ + getSessionEntry.mockReturnValue({ sessionId: "session-1" }); + readLatestAssistantTextByIdentity.mockResolvedValue({ text: fullAnswer, timestamp: Date.now() + 60_000, }); diff --git a/extensions/discord/src/monitor/message-handler.process.ts b/extensions/discord/src/monitor/message-handler.process.ts index 79dea0ae41a1..085b1e9f8f51 100644 --- a/extensions/discord/src/monitor/message-handler.process.ts +++ b/extensions/discord/src/monitor/message-handler.process.ts @@ -1,5 +1,4 @@ // Discord plugin module implements message handler.process behavior. -import path from "node:path"; import { MessageFlags } from "discord-api-types/v10"; import { resolveAckReaction, resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; import { @@ -38,13 +37,8 @@ import { } from "openclaw/plugin-sdk/reply-payload"; import type { ReplyDispatchKind, ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; import { danger, logVerbose, shouldLogVerbose } from "openclaw/plugin-sdk/runtime-env"; -import { - loadSessionStore, - readLatestAssistantTextFromSessionTranscript, - resolveAndPersistSessionFile, - resolveSessionStoreEntry, - resolveStorePath, -} from "openclaw/plugin-sdk/session-store-runtime"; +import { getSessionEntry, resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; +import { readLatestAssistantTextByIdentity } from "openclaw/plugin-sdk/session-transcript-runtime"; import { resolveDiscordAccount, resolveDiscordMaxLinesPerMessage } from "../accounts.js"; import { createDiscordRestClient } from "../client.js"; import { beginDiscordInboundEventDeliveryCorrelation } from "../inbound-event-delivery.js"; @@ -522,21 +516,20 @@ async function processDiscordMessageInner( } try { const storePath = resolveStorePath(cfg.session?.store, { agentId: route.agentId }); - const store = loadSessionStore(storePath, { clone: false }); - const sessionEntry = resolveSessionStoreEntry({ store, sessionKey }).existing; + const sessionEntry = getSessionEntry({ + agentId: route.agentId, + sessionKey, + storePath, + }); if (!sessionEntry?.sessionId) { return undefined; } - const { sessionFile } = await resolveAndPersistSessionFile({ + const latest = await readLatestAssistantTextByIdentity({ + agentId: route.agentId, sessionId: sessionEntry.sessionId, sessionKey, - sessionStore: store, storePath, - sessionEntry, - agentId: route.agentId, - sessionsDir: path.dirname(storePath), }); - const latest = await readLatestAssistantTextFromSessionTranscript(sessionFile); if (!latest?.timestamp || latest.timestamp < dispatchStartedAt) { return undefined; } diff --git a/extensions/telegram/src/bot-message-dispatch.runtime.ts b/extensions/telegram/src/bot-message-dispatch.runtime.ts index c844d1398505..33aaa8fcde18 100644 --- a/extensions/telegram/src/bot-message-dispatch.runtime.ts +++ b/extensions/telegram/src/bot-message-dispatch.runtime.ts @@ -1,10 +1,8 @@ // Telegram plugin module implements bot message dispatch behavior. export { loadSessionStore, - readLatestAssistantTextFromSessionTranscript, - resolveAndPersistSessionFile, resolveSessionStoreEntry, - updateSessionStoreEntry, + resolveStorePath, } from "openclaw/plugin-sdk/session-store-runtime"; export { resolveMarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime"; export { getAgentScopedMediaLocalRoots } from "openclaw/plugin-sdk/media-runtime"; diff --git a/extensions/telegram/src/bot-message-dispatch.test.ts b/extensions/telegram/src/bot-message-dispatch.test.ts index a04e8a71b953..42f20b082ee9 100644 --- a/extensions/telegram/src/bot-message-dispatch.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.test.ts @@ -69,23 +69,25 @@ const createChannelMessageReplyPipeline = vi.hoisted(() => })), ); const wasSentByBot = vi.hoisted(() => vi.fn(() => false)); -const appendSessionTranscriptMessage = vi.hoisted(() => - vi.fn(async ({ message }: { message?: unknown }) => ({ +const appendAssistantMirrorMessageByIdentity = vi.hoisted(() => + vi.fn< + ( + params?: unknown, + ) => Promise< + | { ok: true; sessionFile: string; messageId: string } + | { ok: false; reason: string; code?: "blocked" | "session-rebound" } + > + >(async () => ({ + ok: true, + sessionFile: "/tmp/sessions/s1.jsonl", messageId: "m1", - message, - appended: true, })), ); -const emitSessionTranscriptUpdate = vi.hoisted(() => vi.fn()); const loadSessionStore = vi.hoisted(() => vi.fn()); -const readLatestAssistantTextFromSessionTranscript = vi.hoisted(() => vi.fn()); -const resolveStorePath = vi.hoisted(() => vi.fn(() => "/tmp/sessions.json")); -const resolveAndPersistSessionFile = vi.hoisted(() => - vi.fn(async () => ({ - sessionFile: "/tmp/session.jsonl", - sessionEntry: { sessionId: "s1", sessionFile: "/tmp/session.jsonl" }, - })), +const readLatestAssistantTextByIdentity = vi.hoisted(() => + vi.fn<() => Promise<{ text: string; timestamp?: number } | undefined>>(async () => undefined), ); +const resolveStorePath = vi.hoisted(() => vi.fn(() => "/tmp/sessions.json")); const generateTopicLabel = vi.hoisted(() => vi.fn()); const describeStickerImage = vi.hoisted(() => vi.fn(async (): Promise => null)); const loadModelCatalog = vi.hoisted(() => vi.fn(async () => ({}))); @@ -105,7 +107,6 @@ const resolveSessionStoreEntry = vi.hoisted(() => existing: store[sessionKey], })), ); -const updateSessionStoreEntry = vi.hoisted(() => vi.fn(async () => null)); vi.mock("./draft-stream.js", () => ({ createTelegramDraftStream, @@ -119,12 +120,13 @@ vi.mock("openclaw/plugin-sdk/channel-outbound", async (importOriginal) => { }; }); -vi.mock("openclaw/plugin-sdk/agent-harness-runtime", async (importOriginal) => { - const actual = await importOriginal(); +vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async (importOriginal) => { + const actual = + await importOriginal(); return { ...actual, - appendSessionTranscriptMessage, - emitSessionTranscriptUpdate, + appendAssistantMirrorMessageByIdentity, + readLatestAssistantTextByIdentity, }; }); @@ -153,14 +155,11 @@ vi.mock("./bot-message-dispatch.runtime.js", () => ({ generateTopicLabel, getAgentScopedMediaLocalRoots, loadSessionStore, - readLatestAssistantTextFromSessionTranscript, - resolveAndPersistSessionFile, resolveAutoTopicLabelConfig: resolveAutoTopicLabelConfigRuntime, resolveChunkMode, resolveMarkdownTableMode, resolveSessionStoreEntry, resolveStorePath, - updateSessionStoreEntry, })); vi.mock("./bot-message-dispatch.agent.runtime.js", () => ({ @@ -265,19 +264,15 @@ describe("dispatchTelegramMessage draft streaming", () => { listSkillCommandsForAgents.mockReset(); createChannelMessageReplyPipeline.mockReset(); wasSentByBot.mockReset(); - appendSessionTranscriptMessage.mockReset(); - emitSessionTranscriptUpdate.mockReset(); - readLatestAssistantTextFromSessionTranscript.mockReset(); + appendAssistantMirrorMessageByIdentity.mockReset(); + readLatestAssistantTextByIdentity.mockReset(); loadSessionStore.mockReset(); resolveStorePath.mockReset(); - resolveAndPersistSessionFile.mockReset(); - updateSessionStoreEntry.mockReset(); generateTopicLabel.mockReset(); getAgentScopedMediaLocalRoots.mockClear(); resolveChunkMode.mockClear(); resolveMarkdownTableMode.mockClear(); resolveSessionStoreEntry.mockClear(); - updateSessionStoreEntry.mockClear(); describeStickerImage.mockReset(); loadModelCatalog.mockReset(); findModelInCatalog.mockReset(); @@ -323,9 +318,11 @@ describe("dispatchTelegramMessage draft streaming", () => { }); wasSentByBot.mockReturnValue(false); resolveStorePath.mockReturnValue("/tmp/sessions.json"); - resolveAndPersistSessionFile.mockResolvedValue({ - sessionFile: "/tmp/session.jsonl", - sessionEntry: { sessionId: "s1", sessionFile: "/tmp/session.jsonl" }, + readLatestAssistantTextByIdentity.mockResolvedValue(undefined); + appendAssistantMirrorMessageByIdentity.mockResolvedValue({ + ok: true, + sessionFile: "/tmp/sessions/s1.jsonl", + messageId: "m1", }); loadSessionStore.mockReturnValue({}); generateTopicLabel.mockResolvedValue("Topic label"); @@ -358,6 +355,15 @@ describe("dispatchTelegramMessage draft streaming", () => { return { answerDraftStream, reasoningDraftStream }; } + function mockDefaultSessionEntry(entry: Record = { sessionId: "s1" }) { + loadSessionStore.mockReturnValue({ + "agent:default:telegram:direct:123": { + updatedAt: 1, + ...entry, + }, + }); + } + function expectRecordFields(record: unknown, expected: Record) { if (!record || typeof record !== "object") { throw new Error("Expected record"); @@ -1720,9 +1726,7 @@ describe("dispatchTelegramMessage draft streaming", () => { setupDraftStreams({ answerMessageId: 2001 }); const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, - }); + mockDefaultSessionEntry(); dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => { await dispatcherOptions.deliver({ text: "Final answer" }, { kind: "final" }); return { queuedFinal: true }; @@ -1730,29 +1734,73 @@ describe("dispatchTelegramMessage draft streaming", () => { await dispatchWithContext({ context }); - const transcriptCall = expectRecordFields(mockCallArg(appendSessionTranscriptMessage), { - transcriptPath: "/tmp/session.jsonl", - }); - expectRecordFields(transcriptCall.message, { - role: "assistant", - provider: "openclaw", - model: "delivery-mirror", - content: [{ type: "text", text: "Final answer" }], - }); - expectRecordFields(mockCallArg(emitSessionTranscriptUpdate), { - sessionFile: "/tmp/session.jsonl", + const mirrorCall = expectRecordFields(mockCallArg(appendAssistantMirrorMessageByIdentity), { + agentId: "default", + sessionId: "s1", + idempotencyKey: expect.stringContaining("telegram-final:agent:default:telegram:direct:123:"), sessionKey: "agent:default:telegram:direct:123", - messageId: "m1", + storePath: "/tmp/sessions.json", + text: "Final answer", + }); + expect(mirrorCall.deliveryMirror).toEqual({ + kind: "channel-final", + sourceMessageId: mirrorCall.idempotencyKey, }); }); - it("advances the session marker after mirroring preview-finalized finals", async () => { + it("keeps same-millisecond transcript mirror keys distinct per inbound message", async () => { + createTelegramDraftStream.mockImplementation(() => createDraftStream(2001)); + const dateNow = vi.spyOn(Date, "now").mockReturnValue(1234567890); + const firstContext = createContext({ + ctxPayload: { + MessageSid: "456", + SessionKey: "agent:default:telegram:direct:123", + } as TelegramMessageContext["ctxPayload"], + }); + const secondContext = createContext({ + ctxPayload: { + MessageSid: "457", + SessionKey: "agent:default:telegram:direct:123", + } as TelegramMessageContext["ctxPayload"], + msg: { message_id: 457 } as TelegramMessageContext["msg"], + }); + mockDefaultSessionEntry(); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => { + await dispatcherOptions.deliver({ text: "Final answer" }, { kind: "final" }); + return { queuedFinal: true }; + }); + + try { + await dispatchWithContext({ context: firstContext }); + await dispatchWithContext({ context: secondContext }); + } finally { + dateNow.mockRestore(); + } + + const firstMirrorCall = expectRecordFields( + mockCallArg(appendAssistantMirrorMessageByIdentity), + { + idempotencyKey: expect.stringContaining( + "telegram-final:agent:default:telegram:direct:123:123:456:", + ), + }, + ); + const secondMirrorCall = expectRecordFields( + mockCallArg(appendAssistantMirrorMessageByIdentity, 1), + { + idempotencyKey: expect.stringContaining( + "telegram-final:agent:default:telegram:direct:123:123:457:", + ), + }, + ); + expect(firstMirrorCall.idempotencyKey).not.toBe(secondMirrorCall.idempotencyKey); + }); + + it("skips transcript mirroring when the scoped session is absent", async () => { setupDraftStreams({ answerMessageId: 2001 }); const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, - }); + loadSessionStore.mockReturnValue({}); dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => { await dispatcherOptions.deliver({ text: "Final answer" }, { kind: "final" }); return { queuedFinal: true }; @@ -1760,21 +1808,13 @@ describe("dispatchTelegramMessage draft streaming", () => { await dispatchWithContext({ context }); - const markerUpdateCall = expectRecordFields(mockCallArg(updateSessionStoreEntry), { - storePath: "/tmp/sessions.json", - sessionKey: "agent:default:telegram:direct:123", - }); - const update = markerUpdateCall.update as (entry: { sessionId?: string }) => unknown; - expect(update({ sessionId: "s1" })).toEqual({ updatedAt: expect.any(Number) }); - expect(update({ sessionId: "new-session" })).toBeNull(); + expect(appendAssistantMirrorMessageByIdentity).not.toHaveBeenCalled(); }); it("does not mirror non-final tool progress into the session transcript", async () => { const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, - }); + mockDefaultSessionEntry(); deliverReplies.mockImplementation( async (params: { replies?: Array<{ text?: string }>; @@ -1806,15 +1846,13 @@ describe("dispatchTelegramMessage draft streaming", () => { transcriptMirror: undefined, }); expect(typeof mockCallArg(deliverReplies, 1).transcriptMirror).toBe("function"); - expect(appendSessionTranscriptMessage).toHaveBeenCalledTimes(1); - const transcriptCall = expectRecordFields(mockCallArg(appendSessionTranscriptMessage), { - transcriptPath: "/tmp/session.jsonl", - }); - expectRecordFields(transcriptCall.message, { - role: "assistant", - provider: "openclaw", - model: "delivery-mirror", - content: [{ type: "text", text: "Final answer" }], + expect(appendAssistantMirrorMessageByIdentity).toHaveBeenCalledTimes(1); + expectRecordFields(mockCallArg(appendAssistantMirrorMessageByIdentity), { + agentId: "default", + sessionId: "s1", + sessionKey: "agent:default:telegram:direct:123", + storePath: "/tmp/sessions.json", + text: "Final answer", }); }); @@ -1822,13 +1860,8 @@ describe("dispatchTelegramMessage draft streaming", () => { const repeatedText = "Final answer"; const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, - }); - readLatestAssistantTextFromSessionTranscript.mockResolvedValue({ - text: repeatedText, - timestamp: 1, - }); + mockDefaultSessionEntry(); + readLatestAssistantTextByIdentity.mockResolvedValue({ text: repeatedText, timestamp: 1 }); deliverReplies.mockImplementation( async (params: { replies?: Array<{ text?: string }>; @@ -1849,15 +1882,14 @@ describe("dispatchTelegramMessage draft streaming", () => { await dispatchWithContext({ context }); - expect(appendSessionTranscriptMessage).toHaveBeenCalledTimes(1); - const transcriptCall = expectRecordFields(mockCallArg(appendSessionTranscriptMessage), { - transcriptPath: "/tmp/session.jsonl", - }); - expectRecordFields(transcriptCall.message, { - role: "assistant", - provider: "openclaw", - model: "delivery-mirror", - content: [{ type: "text", text: repeatedText }], + expect(appendAssistantMirrorMessageByIdentity).toHaveBeenCalledTimes(1); + expectRecordFields(mockCallArg(appendAssistantMirrorMessageByIdentity), { + agentId: "default", + sessionId: "s1", + idempotencyKey: expect.stringContaining("telegram-final:agent:default:telegram:direct:123:"), + sessionKey: "agent:default:telegram:direct:123", + storePath: "/tmp/sessions.json", + text: repeatedText, }); }); @@ -1869,10 +1901,8 @@ describe("dispatchTelegramMessage draft streaming", () => { "Ja. Hier nochmal sauber Schritt fuer Schritt. Einen API Key kopiert man..."; const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, - }); - readLatestAssistantTextFromSessionTranscript.mockResolvedValue({ + mockDefaultSessionEntry(); + readLatestAssistantTextByIdentity.mockResolvedValue({ text: fullAnswer, timestamp: Date.now() + 1_000, }); @@ -1892,67 +1922,38 @@ describe("dispatchTelegramMessage draft streaming", () => { content: fullAnswer, messageId: 2001, }); - const transcriptCall = expectRecordFields(mockCallArg(appendSessionTranscriptMessage), { - transcriptPath: "/tmp/session.jsonl", - }); - expectRecordFields(transcriptCall.message, { - role: "assistant", - provider: "openclaw", - model: "delivery-mirror", - content: [{ type: "text", text: fullAnswer }], + expectRecordFields(mockCallArg(appendAssistantMirrorMessageByIdentity), { + agentId: "default", + sessionId: "s1", + sessionKey: "agent:default:telegram:direct:123", + storePath: "/tmp/sessions.json", + text: fullAnswer, }); }); - it("emits the redacted appended message in transcript updates", async () => { + it("treats session rebound mirror skips as non-fatal", async () => { setupDraftStreams({ answerMessageId: 2001 }); const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, + mockDefaultSessionEntry(); + appendAssistantMirrorMessageByIdentity.mockResolvedValueOnce({ + ok: false, + code: "session-rebound", + reason: "session rebound for sessionKey: agent:default:telegram:direct:123", }); - appendSessionTranscriptMessage.mockImplementationOnce(async ({ message }) => ({ - messageId: "m1", - appended: true, - message: { - ...(message as Record), - content: [{ type: "text", text: "Final sk-abc…0xyz" }], - }, - })); dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => { - await dispatcherOptions.deliver({ text: "Final sk-abcdef1234567890xyz" }, { kind: "final" }); + await dispatcherOptions.deliver({ text: "Final answer" }, { kind: "final" }); return { queuedFinal: true }; }); await dispatchWithContext({ context }); - expectRecordFields(mockCallArg(emitSessionTranscriptUpdate), { - sessionFile: "/tmp/session.jsonl", + expectRecordFields(mockCallArg(appendAssistantMirrorMessageByIdentity), { + agentId: "default", + sessionId: "s1", sessionKey: "agent:default:telegram:direct:123", - messageId: "m1", - message: { - role: "assistant", - content: [{ type: "text", text: "Final sk-abc…0xyz" }], - api: "openai-responses", - provider: "openclaw", - model: "delivery-mirror", - usage: { - input: 0, - output: 0, - total: 0, - prompt_tokens: 0, - completion_tokens: 0, - total_tokens: 0, - cache: { - read: 0, - write: 0, - cacheRead: 0, - cacheWrite: 0, - total: 0, - }, - }, - stopReason: "stop", - timestamp: expect.any(Number), - }, + storePath: "/tmp/sessions.json", + text: "Final answer", }); }); @@ -2901,10 +2902,8 @@ describe("dispatchTelegramMessage draft streaming", () => { "Ja. Hier nochmal sauber Schritt fuer Schritt. Einen API Key kopiert man..."; const context = createContext(); context.ctxPayload.SessionKey = "agent:default:telegram:direct:123"; - loadSessionStore.mockReturnValue({ - "agent:default:telegram:direct:123": { sessionId: "s1" }, - }); - readLatestAssistantTextFromSessionTranscript.mockResolvedValue({ + mockDefaultSessionEntry(); + readLatestAssistantTextByIdentity.mockResolvedValue({ text: fullAnswer, timestamp: Date.now() + 1_000, }); diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index e6db0392fff9..a4d64940af6a 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/extensions/telegram/src/bot-message-dispatch.ts @@ -1,10 +1,6 @@ // Telegram plugin module implements bot message dispatch behavior. import path from "node:path"; import type { Bot } from "grammy"; -import { - appendSessionTranscriptMessage, - emitSessionTranscriptUpdate, -} from "openclaw/plugin-sdk/agent-harness-runtime"; import { DEFAULT_TIMING, logAckFailure, @@ -57,6 +53,10 @@ import { logVerbose, sleepWithAbort, } from "openclaw/plugin-sdk/runtime-env"; +import { + appendAssistantMirrorMessageByIdentity, + readLatestAssistantTextByIdentity, +} from "openclaw/plugin-sdk/session-transcript-runtime"; import { resolveTelegramConfigReasoningDefault } from "./agent-config.js"; import { withTelegramApiErrorLogging } from "./api-logging.js"; import type { TelegramBotDeps } from "./bot-deps.js"; @@ -73,13 +73,10 @@ import { generateTopicLabel, getAgentScopedMediaLocalRoots, loadSessionStore, - readLatestAssistantTextFromSessionTranscript, resolveAutoTopicLabelConfig, resolveChunkMode, resolveMarkdownTableMode, - resolveAndPersistSessionFile, resolveSessionStoreEntry, - updateSessionStoreEntry, } from "./bot-message-dispatch.runtime.js"; import type { TelegramBotOptions } from "./bot.types.js"; import { deliverReplies, emitInternalMessageSentHook } from "./bot/delivery.js"; @@ -247,6 +244,7 @@ type TelegramReasoningLevel = "off" | "on" | "stream"; type TelegramTranscriptMirrorPayload = { text?: string; mediaUrls?: string[] }; type TelegramSessionStore = ReturnType; +type TelegramScopedTranscriptSession = { sessionId: string; storePath: string }; type FreshTelegramSessionStoreLoader = ((agentId: string) => { storePath: string; store: TelegramSessionStore; @@ -317,90 +315,53 @@ function resolveTelegramMirroredTranscriptText( return text ? text : null; } +function resolveTelegramScopedTranscriptSession(params: { + agentId: string; + loadFreshSessionStore: FreshTelegramSessionStoreLoader; + sessionKey: string; +}): TelegramScopedTranscriptSession | undefined { + const { store, storePath } = params.loadFreshSessionStore(params.agentId); + const entry = resolveSessionStoreEntry({ store, sessionKey: params.sessionKey }).existing; + const sessionId = entry?.sessionId?.trim(); + return sessionId ? { sessionId, storePath } : undefined; +} + async function mirrorTelegramAssistantReplyToTranscript(params: { cfg: OpenClawConfig; + idempotencyKey: string; + loadFreshSessionStore: FreshTelegramSessionStoreLoader; route: TelegramMessageContext["route"]; sessionKey: string; - loadFreshSessionStore: FreshTelegramSessionStoreLoader; payload: TelegramTranscriptMirrorPayload; }) { const text = resolveTelegramMirroredTranscriptText(params.payload); if (!text) { return; } - const { storePath, store } = params.loadFreshSessionStore(params.route.agentId); - const sessionEntry = resolveSessionStoreEntry({ - store, + const session = resolveTelegramScopedTranscriptSession({ + agentId: params.route.agentId, + loadFreshSessionStore: params.loadFreshSessionStore, sessionKey: params.sessionKey, - }).existing; - if (!sessionEntry?.sessionId) { + }); + if (!session) { return; } - const { sessionFile } = await resolveAndPersistSessionFile({ - sessionId: sessionEntry.sessionId, - sessionKey: params.sessionKey, - sessionStore: store, - storePath, - sessionEntry, + const appended = await appendAssistantMirrorMessageByIdentity({ agentId: params.route.agentId, - sessionsDir: path.dirname(storePath), - }); - const message = { - role: "assistant" as const, - content: [{ type: "text" as const, text }], - api: "openai-responses", - provider: "openclaw", - model: "delivery-mirror", - usage: { - input: 0, - output: 0, - total: 0, - prompt_tokens: 0, - completion_tokens: 0, - total_tokens: 0, - cache: { - read: 0, - write: 0, - cacheRead: 0, - cacheWrite: 0, - total: 0, - }, - }, - stopReason: "stop" as const, - timestamp: Date.now(), - }; - const { - appended, - messageId, - message: appendedMessage, - } = await appendSessionTranscriptMessage({ - transcriptPath: sessionFile, - message, config: params.cfg, - }); - if (appended) { - const transcriptMarkerUpdatedAt = Date.now(); - await updateSessionStoreEntry({ - storePath, - sessionKey: params.sessionKey, - update: (current) => - current.sessionId === sessionEntry.sessionId - ? { updatedAt: transcriptMarkerUpdatedAt } - : null, - }); - } - emitSessionTranscriptUpdate({ - sessionFile, - sessionKey: params.sessionKey, - agentId: params.route.agentId, - target: { - agentId: params.route.agentId, - sessionId: sessionEntry.sessionId, - sessionKey: params.sessionKey, + idempotencyKey: params.idempotencyKey, + deliveryMirror: { + kind: "channel-final", + sourceMessageId: params.idempotencyKey, }, - message: appendedMessage, - messageId, + sessionId: session.sessionId, + sessionKey: params.sessionKey, + storePath: session.storePath, + text, }); + if (!appended.ok && appended.code !== "session-rebound") { + logVerbose(`telegram transcript mirror append failed: ${appended.reason}`); + } } const MAX_PROGRESS_MARKDOWN_TEXT_CHARS = 300; @@ -1498,29 +1459,24 @@ export const dispatchTelegramMessage = async ({ ); const endTelegramInboundEventDeliveryCorrelation = beginDeliveryCorrelation(); const sessionKey = ctxPayload.SessionKey; + let transcriptMirrorSequence = 0; + const transcriptMirrorTurnId = `${chatId}:${ctxPayload.MessageSid ?? msg.message_id ?? dispatchStartedAt}`; const resolveCurrentTurnTranscriptFinalText = async (): Promise => { if (!sessionKey) { return undefined; } try { - const { storePath, store } = loadFreshSessionStore(route.agentId); - const sessionEntry = resolveSessionStoreEntry({ - store, - sessionKey, - }).existing; + const { store, storePath } = loadFreshSessionStore(route.agentId); + const sessionEntry = resolveSessionStoreEntry({ store, sessionKey }).existing; if (!sessionEntry?.sessionId) { return undefined; } - const { sessionFile } = await resolveAndPersistSessionFile({ + const latest = await readLatestAssistantTextByIdentity({ + agentId: route.agentId, sessionId: sessionEntry.sessionId, sessionKey, - sessionStore: store, storePath, - sessionEntry, - agentId: route.agentId, - sessionsDir: path.dirname(storePath), }); - const latest = await readLatestAssistantTextFromSessionTranscript(sessionFile); if (!latest?.timestamp || latest.timestamp < dispatchStartedAt) { return undefined; } @@ -1555,11 +1511,13 @@ export const dispatchTelegramMessage = async ({ replyQuoteByMessageId, transcriptMirror: sessionKey ? async (payload: TelegramTranscriptMirrorPayload) => { + const idempotencyKey = `telegram-final:${sessionKey}:${transcriptMirrorTurnId}:${transcriptMirrorSequence++}`; await mirrorTelegramAssistantReplyToTranscript({ cfg, + idempotencyKey, + loadFreshSessionStore, route, sessionKey, - loadFreshSessionStore, payload, }); } diff --git a/extensions/telegram/src/bot-native-commands.session-meta.test.ts b/extensions/telegram/src/bot-native-commands.session-meta.test.ts index 4757a9570e2f..fcae2b40591a 100644 --- a/extensions/telegram/src/bot-native-commands.session-meta.test.ts +++ b/extensions/telegram/src/bot-native-commands.session-meta.test.ts @@ -1,5 +1,4 @@ // Telegram tests cover bot native commands.session meta plugin behavior. -import path from "node:path"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import type { ResolvedAgentRoute } from "openclaw/plugin-sdk/routing"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -47,9 +46,10 @@ const persistentBindingMocks = vi.hoisted(() => ({ })), })); const sessionMocks = vi.hoisted(() => ({ + getSessionEntry: vi.fn(), loadSessionStore: vi.fn(), recordSessionMetaFromInbound: vi.fn(), - resolveAndPersistSessionFile: vi.fn(), + resolveSessionTranscriptLegacyFileTarget: vi.fn(), resolveStorePath: vi.fn(), })); const commandAuthMocks = vi.hoisted(() => ({ @@ -166,11 +166,20 @@ vi.mock("openclaw/plugin-sdk/session-store-runtime", async () => { ); return { ...actual, + getSessionEntry: sessionMocks.getSessionEntry, loadSessionStore: sessionMocks.loadSessionStore, - resolveAndPersistSessionFile: sessionMocks.resolveAndPersistSessionFile, resolveStorePath: sessionMocks.resolveStorePath, }; }); +vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async () => { + const actual = await vi.importActual< + typeof import("openclaw/plugin-sdk/session-transcript-runtime") + >("openclaw/plugin-sdk/session-transcript-runtime"); + return { + ...actual, + resolveSessionTranscriptLegacyFileTarget: sessionMocks.resolveSessionTranscriptLegacyFileTarget, + }; +}); vi.mock("openclaw/plugin-sdk/command-auth-native", async () => { const actual = await vi.importActual( "openclaw/plugin-sdk/command-auth-native", @@ -561,20 +570,16 @@ describe("registerTelegramNativeCommands — session metadata", () => { reasoning: true, }, ]); + sessionMocks.getSessionEntry.mockClear().mockReturnValue(undefined); sessionMocks.loadSessionStore.mockClear().mockReturnValue({}); sessionMocks.recordSessionMetaFromInbound.mockClear().mockResolvedValue(undefined); - sessionMocks.resolveAndPersistSessionFile.mockClear().mockImplementation(async (params) => { - const sessionFile = - params.fallbackSessionFile ?? `/tmp/openclaw-sessions/${params.sessionId}.jsonl`; - return { - sessionFile, - sessionEntry: { - ...params.sessionEntry, - sessionId: params.sessionId, - sessionFile, - updatedAt: Date.now(), - }, - }; + sessionMocks.resolveSessionTranscriptLegacyFileTarget.mockClear().mockResolvedValue({ + agentId: "main", + memoryKey: "transcript:main:sess-topic", + sessionId: "sess-topic", + sessionKey: "agent:main:telegram:group:-1001234567890:topic:42", + sessionFile: "/tmp/openclaw-sessions/sess-topic-topic-42.jsonl", + targetKind: "runtime-session", }); sessionMocks.resolveStorePath.mockClear().mockReturnValue("/tmp/openclaw-sessions.json"); pluginRuntimeMocks.executePluginCommand.mockClear().mockResolvedValue({ text: "ok" }); @@ -1407,6 +1412,12 @@ describe("registerTelegramNativeCommands — session metadata", () => { it("passes a persisted topic session file to plugin commands", async () => { sessionMocks.resolveStorePath.mockReturnValue("/tmp/openclaw-sessions/sessions.json"); + sessionMocks.getSessionEntry.mockReturnValue({ + authProfileOverride: "openai:owner@example.com", + sessionFile: "/tmp/openclaw-sessions/sess-topic-topic-42.jsonl", + sessionId: "sess-topic", + updatedAt: 1, + }); sessionMocks.loadSessionStore.mockReturnValue({ "agent:main:telegram:group:-1001234567890:topic:42": { authProfileOverride: "openai:owner@example.com", @@ -1445,22 +1456,25 @@ describe("registerTelegramNativeCommands — session metadata", () => { ); expectRecordFields( - firstMockArg(sessionMocks.resolveAndPersistSessionFile, "resolveAndPersistSessionFile"), + firstMockArg( + sessionMocks.resolveSessionTranscriptLegacyFileTarget, + "resolveSessionTranscriptLegacyFileTarget", + ), { + agentId: "main", sessionId: "sess-topic", sessionKey: "agent:main:telegram:group:-1001234567890:topic:42", storePath: "/tmp/openclaw-sessions/sessions.json", - sessionsDir: "/tmp/openclaw-sessions", - fallbackSessionFile: path.resolve("/tmp/openclaw-sessions", "sess-topic-topic-42.jsonl"), + threadId: 42, }, - "resolved session file params", + "resolved transcript target params", ); expectRecordFields( (pluginRuntimeMocks.executePluginCommand.mock.calls as unknown as Array<[unknown]>)[0]?.[0], { sessionKey: "agent:main:telegram:group:-1001234567890:topic:42", sessionId: "sess-topic", - sessionFile: path.resolve("/tmp/openclaw-sessions", "sess-topic-topic-42.jsonl"), + sessionFile: "/tmp/openclaw-sessions/sess-topic-topic-42.jsonl", authProfileId: "openai:owner@example.com", messageThreadId: 42, }, @@ -1468,6 +1482,71 @@ describe("registerTelegramNativeCommands — session metadata", () => { ); }); + it("passes a resolved transcript file to plugin commands when the entry has no file", async () => { + sessionMocks.resolveStorePath.mockReturnValue("/tmp/openclaw-sessions/sessions.json"); + sessionMocks.getSessionEntry.mockReturnValue({ + sessionId: "sess-main", + updatedAt: 1, + }); + sessionMocks.resolveSessionTranscriptLegacyFileTarget.mockResolvedValue({ + agentId: "main", + memoryKey: "transcript:main:sess-main", + sessionFile: "/tmp/openclaw-sessions/sess-main.jsonl", + sessionId: "sess-main", + sessionKey: "agent:main:main", + targetKind: "runtime-session", + }); + + const { handler } = registerAndResolveCommandHandler({ + commandName: "codex", + cfg: { commands: { allowFrom: { telegram: ["200"] } } } as OpenClawConfig, + useAccessGroups: false, + pluginCommandSpecs: [ + { + name: "codex", + description: "Codex", + acceptsArgs: true, + }, + ] as TelegramPluginCommandSpecs, + }); + pluginRuntimeMocks.matchPluginCommand.mockReturnValue({ + command: { + name: "codex", + description: "Codex", + handler: vi.fn(), + pluginId: "openclaw-codex-app-server", + pluginName: "Codex", + requireAuth: true, + }, + args: "status", + }); + + await handler(createTelegramPrivateCommandContext({ match: "status" })); + + expectRecordFields( + firstMockArg( + sessionMocks.resolveSessionTranscriptLegacyFileTarget, + "resolveSessionTranscriptLegacyFileTarget", + ), + { + agentId: "main", + sessionId: "sess-main", + sessionKey: "agent:main:main", + storePath: "/tmp/openclaw-sessions/sessions.json", + }, + "resolved transcript target params", + ); + expectRecordFields( + (pluginRuntimeMocks.executePluginCommand.mock.calls as unknown as Array<[unknown]>)[0]?.[0], + { + sessionKey: "agent:main:main", + sessionId: "sess-main", + sessionFile: "/tmp/openclaw-sessions/sess-main.jsonl", + }, + "plugin command params", + ); + }); + it("sends an empty-response fallback when a plugin command returns undefined", async () => { pluginRuntimeMocks.executePluginCommand.mockResolvedValue(undefined as never); diff --git a/extensions/telegram/src/bot-native-commands.ts b/extensions/telegram/src/bot-native-commands.ts index 9df547887257..e835ad2755a4 100644 --- a/extensions/telegram/src/bot-native-commands.ts +++ b/extensions/telegram/src/bot-native-commands.ts @@ -1,6 +1,5 @@ // Telegram plugin module implements bot native commands behavior. import { randomUUID } from "node:crypto"; -import path from "node:path"; import type { Bot, Context } from "grammy"; import { loadModelCatalog, @@ -40,13 +39,13 @@ import { danger, logVerbose } from "openclaw/plugin-sdk/runtime-env"; import { getChildLogger } from "openclaw/plugin-sdk/runtime-env"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import { + getSessionEntry, loadSessionStore, - resolveAndPersistSessionFile, resolveSessionStoreEntry, - resolveSessionTranscriptPathInDir, resolveStorePath, type SessionEntry, } from "openclaw/plugin-sdk/session-store-runtime"; +import { resolveSessionTranscriptLegacyFileTarget } from "openclaw/plugin-sdk/session-transcript-runtime"; import { normalizeLowercaseStringOrEmpty, normalizeOptionalString, @@ -201,7 +200,7 @@ function resolveTelegramProgressPlaceholder(command: { return text ? text : null; } -async function resolveTelegramCommandSessionFile(params: { +async function resolveTelegramCommandTranscriptContext(params: { cfg: OpenClawConfig; agentId: string; sessionKey: string; @@ -213,29 +212,23 @@ async function resolveTelegramCommandSessionFile(params: { } try { const storePath = resolveStorePath(params.cfg.session?.store, { agentId: params.agentId }); - const store = loadSessionStore(storePath); - const resolved = resolveSessionStoreEntry({ store, sessionKey }); - const sessionId = resolved.existing?.sessionId?.trim() || randomUUID(); - const authProfileId = normalizeOptionalString(resolved.existing?.authProfileOverride); - const sessionsDir = path.dirname(storePath); - const fallbackSessionFile = resolveSessionTranscriptPathInDir( - sessionId, - sessionsDir, - params.threadId, - ); - const persisted = await resolveAndPersistSessionFile({ - sessionId, - sessionKey: resolved.normalizedKey, - sessionStore: store, - storePath, - sessionEntry: resolved.existing, + const entry = getSessionEntry({ agentId: params.agentId, - sessionsDir, - fallbackSessionFile, + sessionKey, + storePath, + }); + const sessionId = entry?.sessionId?.trim() || randomUUID(); + const authProfileId = normalizeOptionalString(entry?.authProfileOverride); + const target = await resolveSessionTranscriptLegacyFileTarget({ + agentId: params.agentId, + sessionId, + sessionKey, + storePath, + ...(params.threadId !== undefined ? { threadId: params.threadId } : {}), }); return { sessionId, - sessionFile: persisted.sessionFile, + sessionFile: target.sessionFile, ...(authProfileId ? { authProfileId } : {}), }; } catch { @@ -1633,7 +1626,7 @@ export const registerTelegramNativeCommands = ({ } } - const sessionFileContext = await resolveTelegramCommandSessionFile({ + const transcriptContext = await resolveTelegramCommandTranscriptContext({ cfg: runtimeCfg, agentId: route.agentId, sessionKey: targetSessionKey, @@ -1650,10 +1643,10 @@ export const registerTelegramNativeCommands = ({ senderIsOwner, agentId: route.agentId, sessionKey: targetSessionKey, - sessionId: sessionFileContext.sessionId, - sessionFile: sessionFileContext.sessionFile, + sessionId: transcriptContext.sessionId, + sessionFile: transcriptContext.sessionFile, authProfileId: - sessionFileContext.authProfileId ?? targetSessionEntry?.authProfileOverride, + transcriptContext.authProfileId ?? targetSessionEntry?.authProfileOverride, commandBody, config: runtimeCfg, from, diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index e79feb7cea7a..e1ee5dfba481 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -202,8 +202,8 @@ let publicDeprecatedExportsByEntrypointBudget; try { budgets = { publicEntrypoints: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_ENTRYPOINTS", 322), - publicExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", 10371), - publicFunctionExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", 5202), + publicExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", 10376), + publicFunctionExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", 5205), publicDeprecatedExports: readBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_DEPRECATED_EXPORTS", 3247, diff --git a/src/plugin-sdk/session-transcript-runtime.test.ts b/src/plugin-sdk/session-transcript-runtime.test.ts index bd6f26d54e06..fb06faf59d05 100644 --- a/src/plugin-sdk/session-transcript-runtime.test.ts +++ b/src/plugin-sdk/session-transcript-runtime.test.ts @@ -7,12 +7,15 @@ import { loadSessionStore } from "../config/sessions/store.js"; import { withOwnedSessionTranscriptWrites } from "../config/sessions/transcript-write-context.js"; import * as transcriptEvents from "../sessions/transcript-events.js"; import { + appendAssistantMirrorMessageByIdentity, appendSessionTranscriptMessageByIdentity, formatSessionTranscriptMemoryHitKey, parseSessionTranscriptMemoryHitKey, publishSessionTranscriptUpdateByIdentity, + readLatestAssistantTextByIdentity, readSessionTranscriptEvents, resolveSessionTranscriptIdentity, + resolveSessionTranscriptLegacyFileTarget, resolveSessionTranscriptTarget, resolveSessionTranscriptMemoryHitKeyToSessionKeys, withSessionTranscriptWriteLock, @@ -73,6 +76,72 @@ describe("session transcript runtime SDK", () => { expect(loadSessionStore(storePath)[scope.sessionKey]?.sessionFile).toBeUndefined(); }); + it("persists and returns a file target for legacy command callers", async () => { + const scope = { + agentId: "main", + sessionId: "legacy-command-session", + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 }); + + const target = await resolveSessionTranscriptLegacyFileTarget(scope); + + expect(target).toMatchObject({ + agentId: "main", + memoryKey: "transcript:main:legacy-command-session", + sessionId: "legacy-command-session", + sessionKey: "agent:main:main", + targetKind: "runtime-session", + }); + expect(target.sessionFile).toContain("legacy-command-session"); + expect(loadSessionStore(storePath)[scope.sessionKey]?.sessionFile).toBe(target.sessionFile); + }); + + it("appends assistant mirrors through the guarded session facade", async () => { + const scope = { + agentId: "main", + sessionId: "guarded-mirror-session", + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 }); + + await expect( + appendAssistantMirrorMessageByIdentity({ + ...scope, + deliveryMirror: { kind: "channel-final", sourceMessageId: "delivery-1" }, + idempotencyKey: "delivery-1", + text: "visible assistant reply", + }), + ).resolves.toMatchObject({ ok: true, messageId: expect.any(String) }); + await expect( + appendAssistantMirrorMessageByIdentity({ + ...scope, + deliveryMirror: { kind: "channel-final", sourceMessageId: "delivery-2" }, + idempotencyKey: "delivery-2", + text: "visible assistant reply", + }), + ).resolves.toMatchObject({ ok: true, messageId: expect.any(String) }); + await expect(readLatestAssistantTextByIdentity(scope)).resolves.toBeUndefined(); + const assistantMessages = (await readSessionTranscriptEvents(scope)).filter((event) => { + const message = (event as { message?: { role?: unknown } }).message; + return message?.role === "assistant"; + }); + expect(assistantMessages).toHaveLength(2); + + await upsertSessionEntry(scope, { sessionId: "new-session", updatedAt: 20 }); + + await expect( + appendAssistantMirrorMessageByIdentity({ + ...scope, + text: "stale assistant reply", + }), + ).resolves.toMatchObject({ ok: false, code: "session-rebound" }); + }); + it("skips malformed transcript lines when reading by scoped identity", async () => { const scope = { agentId: "main", @@ -151,6 +220,10 @@ describe("session transcript runtime SDK", () => { expect(appended).toBeDefined(); expect(appended?.message).toMatchObject(message); + await expect(readLatestAssistantTextByIdentity(scope)).resolves.toMatchObject({ + text: "hello", + timestamp: 1, + }); await expect(readSessionTranscriptEvents(scope)).resolves.toEqual([ expect.objectContaining({ type: "session" }), expect.objectContaining({ message: expect.objectContaining({ role: "assistant" }) }), diff --git a/src/plugin-sdk/session-transcript-runtime.ts b/src/plugin-sdk/session-transcript-runtime.ts index 1a03d2abd4c3..396aeef94963 100644 --- a/src/plugin-sdk/session-transcript-runtime.ts +++ b/src/plugin-sdk/session-transcript-runtime.ts @@ -9,6 +9,15 @@ import { } from "../config/sessions/session-accessor.js"; import { runSessionTranscriptAppendTransaction } from "../config/sessions/transcript-append.js"; import { streamSessionTranscriptLines } from "../config/sessions/transcript-stream.js"; +import { + appendAssistantMessageToSessionTranscript, + readLatestAssistantTextFromSessionTranscript, + type LatestAssistantTranscriptText, + type SessionTranscriptAppendResult, + type SessionTranscriptDeliveryMirror, + type SessionTranscriptUpdateMode, +} from "../config/sessions/transcript.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; import { normalizeAgentId } from "../routing/session-key.js"; import { formatSessionTranscriptMemoryHitKey, @@ -51,9 +60,26 @@ export type SessionTranscriptTarget = SessionTranscriptIdentity & { targetKind: "active-session-file" | "runtime-session"; }; +export type SessionTranscriptLegacyFileTarget = SessionTranscriptTarget & { + /** + * Deprecated transitional file target for callers that still pass active + * transcript files to plugin command handlers. + */ + sessionFile: string; +}; + export type SessionTranscriptAppendMessageParams = SessionTranscriptTargetParams & TranscriptMessageAppendOptions; +export type SessionTranscriptAssistantMirrorAppendParams = SessionTranscriptReadParams & { + config?: OpenClawConfig; + deliveryMirror?: SessionTranscriptDeliveryMirror; + idempotencyKey?: string; + mediaUrls?: string[]; + text?: string; + updateMode?: SessionTranscriptUpdateMode; +}; + export type SessionTranscriptWriteLockParams = SessionTranscriptTargetParams & { config?: TranscriptMessageAppendOptions["config"]; }; @@ -97,6 +123,23 @@ export async function resolveSessionTranscriptTarget( }); } +/** + * Resolves and persists the current file-backed target for legacy plugin + * command calls that still require `sessionFile`. + */ +export async function resolveSessionTranscriptLegacyFileTarget( + params: SessionTranscriptTargetParams, +): Promise { + const target = await resolveSessionTranscriptRuntimeTarget(params); + return { + ...projectPublicTarget({ + ...target, + targetKind: params.sessionFile?.trim() ? "active-session-file" : "runtime-session", + }), + sessionFile: target.sessionFile, + }; +} + /** * Reads transcript events by public session identity instead of file path. */ @@ -115,6 +158,38 @@ export async function readSessionTranscriptEvents( return events; } +/** + * Reads the latest visible assistant text by scoped identity using the + * bounded reverse transcript reader. + */ +export async function readLatestAssistantTextByIdentity( + params: SessionTranscriptTargetParams, +): Promise { + const target = await resolveSessionTranscriptRuntimeReadTarget(params); + return await readLatestAssistantTextFromSessionTranscript(target.sessionFile); +} + +/** + * Appends a delivery-mirror assistant message through the guarded session + * append facade. + */ +export async function appendAssistantMirrorMessageByIdentity( + params: SessionTranscriptAssistantMirrorAppendParams, +): Promise { + return await appendAssistantMessageToSessionTranscript({ + agentId: params.agentId, + sessionKey: params.sessionKey, + expectedSessionId: params.sessionId, + ...(params.text !== undefined ? { text: params.text } : {}), + ...(params.mediaUrls !== undefined ? { mediaUrls: params.mediaUrls } : {}), + ...(params.idempotencyKey !== undefined ? { idempotencyKey: params.idempotencyKey } : {}), + ...(params.deliveryMirror !== undefined ? { deliveryMirror: params.deliveryMirror } : {}), + ...(params.storePath !== undefined ? { storePath: params.storePath } : {}), + ...(params.updateMode !== undefined ? { updateMode: params.updateMode } : {}), + ...(params.config !== undefined ? { config: params.config } : {}), + }); +} + /** * Appends a transcript message by scoped transcript target. */ diff --git a/test/scripts/plugin-sdk-surface-report.test.ts b/test/scripts/plugin-sdk-surface-report.test.ts index b448459dfc69..31b5c5778dca 100644 --- a/test/scripts/plugin-sdk-surface-report.test.ts +++ b/test/scripts/plugin-sdk-surface-report.test.ts @@ -28,10 +28,14 @@ function readDefaultPublicFunctionExportBudget() { describe("plugin SDK surface report", () => { it("rejects unknown CLI options before collecting SDK stats", () => { - const result = spawnSync(process.execPath, ["scripts/plugin-sdk-surface-report.mjs", "--chekc"], { - cwd: process.cwd(), - encoding: "utf8", - }); + const result = spawnSync( + process.execPath, + ["scripts/plugin-sdk-surface-report.mjs", "--chekc"], + { + cwd: process.cwd(), + encoding: "utf8", + }, + ); expect(result.status).toBe(1); expect(result.stdout).toBe(""); @@ -40,10 +44,14 @@ describe("plugin SDK surface report", () => { }); it("prints help before collecting SDK stats", () => { - const result = spawnSync(process.execPath, ["scripts/plugin-sdk-surface-report.mjs", "--help"], { - cwd: process.cwd(), - encoding: "utf8", - }); + const result = spawnSync( + process.execPath, + ["scripts/plugin-sdk-surface-report.mjs", "--help"], + { + cwd: process.cwd(), + encoding: "utf8", + }, + ); expect(result.status).toBe(0); expect(result.stdout).toContain("Usage: node scripts/plugin-sdk-surface-report.mjs");