mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-10 00:11:19 +00:00
refactor: migrate bundled transcript target lookups (#89911)
This commit is contained in:
Josh Lehman
GitHub
parent
009d6b261a
commit
132d70bfb3
12 changed files with 496 additions and 345 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -204,31 +204,18 @@ const recordInboundSession = vi.hoisted(() =>
|
|||
vi.fn<(params?: unknown) => Promise<void>>(async () => {}),
|
||||
);
|
||||
const configSessionsMocks = vi.hoisted(() => ({
|
||||
loadSessionStore: vi.fn<(storePath: string, opts?: unknown) => Record<string, unknown>>(
|
||||
() => ({}),
|
||||
),
|
||||
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<string, unknown>; 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<string, unknown>; 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,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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<string | null> => 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<typeof import("openclaw/plugin-sdk/agent-harness-runtime")>();
|
||||
vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async (importOriginal) => {
|
||||
const actual =
|
||||
await importOriginal<typeof import("openclaw/plugin-sdk/session-transcript-runtime")>();
|
||||
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<string, unknown> = { sessionId: "s1" }) {
|
||||
loadSessionStore.mockReturnValue({
|
||||
"agent:default:telegram:direct:123": {
|
||||
updatedAt: 1,
|
||||
...entry,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function expectRecordFields(record: unknown, expected: Record<string, unknown>) {
|
||||
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<string, unknown>),
|
||||
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,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<typeof loadSessionStore>;
|
||||
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<string | undefined> => {
|
||||
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,
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<typeof import("openclaw/plugin-sdk/command-auth-native")>(
|
||||
"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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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" }) }),
|
||||
|
|
|
|||
|
|
@ -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<TMessage> = SessionTranscriptTargetParams &
|
||||
TranscriptMessageAppendOptions<TMessage>;
|
||||
|
||||
export type SessionTranscriptAssistantMirrorAppendParams = SessionTranscriptReadParams & {
|
||||
config?: OpenClawConfig;
|
||||
deliveryMirror?: SessionTranscriptDeliveryMirror;
|
||||
idempotencyKey?: string;
|
||||
mediaUrls?: string[];
|
||||
text?: string;
|
||||
updateMode?: SessionTranscriptUpdateMode;
|
||||
};
|
||||
|
||||
export type SessionTranscriptWriteLockParams = SessionTranscriptTargetParams & {
|
||||
config?: TranscriptMessageAppendOptions<unknown>["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<SessionTranscriptLegacyFileTarget> {
|
||||
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<LatestAssistantTranscriptText | undefined> {
|
||||
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<SessionTranscriptAppendResult> {
|
||||
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.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue