diff --git a/extensions/telegram/src/state-migrations.test.ts b/extensions/telegram/src/state-migrations.test.ts index 5d5b1cb9a08e..c02b145a1664 100644 --- a/extensions/telegram/src/state-migrations.test.ts +++ b/extensions/telegram/src/state-migrations.test.ts @@ -11,7 +11,10 @@ import { resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; import { afterEach, describe, expect, it, vi } from "vitest"; import { stateMigrations } from "../doctor-contract-api.js"; import { resolveTelegramBotInfoCachePath } from "./bot-info-cache.js"; -import { resolveTelegramMessageCachePath } from "./message-cache-persistence.js"; +import { + resolveTelegramMessageCachePath, + resolveTelegramMessageCachePersistentScopeKey, +} from "./message-cache-persistence.js"; import { detectTelegramLegacyStateMigrations } from "./state-migrations.js"; import { resolveTopicNameCacheNamespace, @@ -47,6 +50,181 @@ afterEach(() => { }); describe("telegram state migrations", () => { + it("does not require a migration owner when multi-agent startup has no legacy artifacts", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + try { + const cfg = { + agents: { + ownership: "explicit", + entries: { main: {}, ops: {}, research: {} }, + }, + } as OpenClawConfig; + + await expect(detectTelegramLegacyStateMigrations({ cfg, env })).resolves.toEqual([]); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it("uses the materialized Telegram binding as legacy-state owner after H2 normalization", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + const legacyStorePath = path.join(dir, "sessions", "sessions.json"); + const messageCachePath = resolveTelegramMessageCachePath(legacyStorePath); + const sentMessagePath = `${legacyStorePath}.telegram-sent-messages.json`; + const topicNamePath = resolveTopicNameCachePath(legacyStorePath); + const ownerStorePath = resolveStorePath(undefined, { env, agentId: "main" }); + const ownerMessagePath = resolveTelegramMessageCachePath(ownerStorePath); + const ownerTopicNamespace = resolveTopicNameCacheNamespace( + resolveTopicNameCacheScope(ownerStorePath), + ); + try { + await mkdir(path.dirname(legacyStorePath), { recursive: true }); + await writeFile(messageCachePath, JSON.stringify([persistedCacheEntry(51, "bound owner")])); + await writeFile(sentMessagePath, JSON.stringify({ 7: { 51: Date.now() } })); + await writeFile( + topicNamePath, + JSON.stringify({ "7:51": { name: "Bound owner", updatedAt: Date.now() } }), + ); + + const cfg = { + agents: { + ownership: "explicit", + entries: { main: {}, ops: {}, research: {} }, + }, + bindings: [{ agentId: "main", match: { channel: "telegram", accountId: "*" } }], + } as OpenClawConfig; + const plans = await detectTelegramLegacyStateMigrations({ cfg, env }); + const messagePlan = plans.find((plan) => plan.sourcePath === messageCachePath); + const sentPlan = plans.find((plan) => plan.sourcePath === sentMessagePath); + const topicPlan = plans.find((plan) => plan.sourcePath === topicNamePath); + + expect(messagePlan).toMatchObject({ + kind: "plugin-state-import", + scopeKey: resolveTelegramMessageCachePersistentScopeKey(ownerMessagePath), + }); + expect(topicPlan).toMatchObject({ + kind: "plugin-state-import", + namespace: ownerTopicNamespace, + }); + if (!sentPlan || sentPlan.kind !== "plugin-state-import") { + throw new Error("expected Telegram sent-message import plan"); + } + expect((await sentPlan.readEntries())[0]?.value).toMatchObject({ + scopeKey: createHash("sha256").update(ownerStorePath, "utf8").digest("hex").slice(0, 24), + }); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it("retains raw legacy default-marker ownership during rollback compatibility", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + const legacyStorePath = path.join(dir, "sessions", "sessions.json"); + const sentMessagePath = `${legacyStorePath}.telegram-sent-messages.json`; + const ownerStorePath = resolveStorePath(undefined, { env, agentId: "ops" }); + try { + await mkdir(path.dirname(legacyStorePath), { recursive: true }); + await writeFile(sentMessagePath, JSON.stringify({ 7: { 52: Date.now() } })); + + const cfg = { + agents: { list: [{ id: "main" }, { id: "ops", default: true }] }, + } as OpenClawConfig; + const plans = await detectTelegramLegacyStateMigrations({ cfg, env }); + const sentPlan = plans.find((plan) => plan.sourcePath === sentMessagePath); + if (!sentPlan || sentPlan.kind !== "plugin-state-import") { + throw new Error("expected Telegram sent-message import plan"); + } + expect((await sentPlan.readEntries())[0]?.value).toMatchObject({ + scopeKey: createHash("sha256").update(ownerStorePath, "utf8").digest("hex").slice(0, 24), + }); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it("fails closed when legacy Telegram state has no explicit multi-agent owner", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + const legacyStorePath = path.join(dir, "sessions", "sessions.json"); + const sentMessagePath = `${legacyStorePath}.telegram-sent-messages.json`; + try { + await mkdir(path.dirname(legacyStorePath), { recursive: true }); + await writeFile(sentMessagePath, JSON.stringify({ 7: { 53: Date.now() } })); + + const cfg = { + agents: { ownership: "explicit", entries: { main: {}, ops: {}, research: {} } }, + } as OpenClawConfig; + await expect(detectTelegramLegacyStateMigrations({ cfg, env })).rejects.toMatchObject({ + name: "AgentSelectionRequiredError", + code: "AGENT_SELECTION_REQUIRED", + }); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it("fails closed when global legacy state spans multiple Telegram route owners", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + const legacyStorePath = path.join(dir, "sessions", "sessions.json"); + const sentMessagePath = `${legacyStorePath}.telegram-sent-messages.json`; + try { + await mkdir(path.dirname(legacyStorePath), { recursive: true }); + await writeFile(sentMessagePath, JSON.stringify({ 7: { 54: Date.now() } })); + + const cfg = { + agents: { ownership: "explicit", entries: { main: {}, ops: {} } }, + channels: { + telegram: { + accounts: { + primary: { botToken: "123456:primary" }, + alerts: { botToken: "123456:alerts" }, + }, + }, + }, + bindings: [ + { agentId: "main", match: { channel: "telegram", accountId: "primary" } }, + { agentId: "ops", match: { channel: "telegram", accountId: "alerts" } }, + ], + } as OpenClawConfig; + await expect(detectTelegramLegacyStateMigrations({ cfg, env })).rejects.toThrow( + /^Legacy Telegram state has multiple routed owners \((?:main, ops|ops, main)\)/, + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it("imports an account-scoped topic cache without requiring a global migration owner", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + const opsStorePath = resolveStorePath(undefined, { env, agentId: "ops" }); + const topicNamePath = resolveTopicNameCachePath(opsStorePath); + try { + await mkdir(path.dirname(topicNamePath), { recursive: true }); + await writeFile( + topicNamePath, + JSON.stringify({ "7:55": { name: "Ops", updatedAt: Date.now() } }), + ); + + const cfg = { + agents: { ownership: "explicit", entries: { main: {}, ops: {} } }, + channels: { telegram: { accounts: { ops: { botToken: "123456:ops" } } } }, + } as OpenClawConfig; + const plans = await detectTelegramLegacyStateMigrations({ cfg, env }); + + expect(plans.find((plan) => plan.sourcePath === topicNamePath)).toMatchObject({ + kind: "plugin-state-import", + namespace: resolveTopicNameCacheNamespace(resolveTopicNameCacheScope(opsStorePath)), + }); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + it("detects legacy bot-info cache import", async () => { const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; diff --git a/extensions/telegram/src/state-migrations.ts b/extensions/telegram/src/state-migrations.ts index 396c85dd2ae0..3e824397eab9 100644 --- a/extensions/telegram/src/state-migrations.ts +++ b/extensions/telegram/src/state-migrations.ts @@ -1,9 +1,10 @@ // Telegram plugin module implements state migrations behavior. import fs from "node:fs"; import path from "node:path"; -import { resolveDefaultAgentId } from "openclaw/plugin-sdk/agent-scope-runtime"; +import { listAgentIds } from "openclaw/plugin-sdk/agent-scope-runtime"; import type { ChannelLegacyStateMigrationPlan } from "openclaw/plugin-sdk/channel-contract"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { resolveAgentRoute } from "openclaw/plugin-sdk/routing"; import { fileExists } from "openclaw/plugin-sdk/security-runtime"; import { resolveStorePath } from "openclaw/plugin-sdk/session-store-paths"; import { isRecord, uniqueStrings } from "openclaw/plugin-sdk/string-coerce-runtime"; @@ -73,6 +74,37 @@ function resolveAgentSessionStorePath(params: { }); } +function listLegacyAgentSessionStorePaths(params: { + cfg: OpenClawConfig; + env: NodeJS.ProcessEnv; + stateDir?: string; +}): string[] { + return uniqueStrings([ + ...listAgentIds(params.cfg).map((agentId) => + resolveAgentSessionStorePath({ ...params, agentId }), + ), + resolveAgentSessionStorePath({ ...params, agentId: "main" }), + resolveLegacySessionStorePath(params), + ]); +} + +function resolveTelegramLegacyStateOwnerAgentId(cfg: OpenClawConfig): string { + const configuredAccountIds = listTelegramAccountIds(cfg); + const accountIds = + configuredAccountIds.length > 0 ? configuredAccountIds : [resolveDefaultTelegramAccountId(cfg)]; + const ownerAgentIds = uniqueStrings( + accountIds.map( + (accountId) => resolveAgentRoute({ cfg, channel: "telegram", accountId }).agentId, + ), + ); + if (ownerAgentIds.length === 1) { + return ownerAgentIds[0]!; + } + throw new Error( + `Legacy Telegram state has multiple routed owners (${ownerAgentIds.join(", ")}); preserve it until one migration owner is configured.`, + ); +} + function resolveMigrationStateDir(params: { env: NodeJS.ProcessEnv; stateDir?: string }): string { return ( params.stateDir ?? @@ -195,23 +227,20 @@ function detectTelegramMessageCacheLegacyStateMigration(params: { env: NodeJS.ProcessEnv; stateDir?: string; }): ChannelLegacyStateMigrationPlan[] { - const storePath = resolveAgentSessionStorePath({ + const persistedPaths = listLegacyAgentSessionStorePaths(params) + .map(resolveTelegramMessageCachePath) + .filter(fileExists); + if (persistedPaths.length === 0) { + return []; + } + const ownerStorePath = resolveAgentSessionStorePath({ ...params, - agentId: resolveDefaultAgentId(params.cfg), + agentId: resolveTelegramLegacyStateOwnerAgentId(params.cfg), }); - const legacyMainStorePath = resolveAgentSessionStorePath({ ...params, agentId: "main" }); - const runtimePersistedPath = resolveTelegramMessageCachePath(storePath); - const legacyStorePath = resolveLegacySessionStorePath(params); - const legacyPersistedPath = resolveTelegramMessageCachePath(legacyStorePath); - const scopeKey = resolveTelegramMessageCachePersistentScopeKey(runtimePersistedPath); - return uniqueStrings([ - runtimePersistedPath, - resolveTelegramMessageCachePath(legacyMainStorePath), - legacyPersistedPath, - ]).flatMap((persistedPath) => { - if (!fileExists(persistedPath)) { - return []; - } + const scopeKey = resolveTelegramMessageCachePersistentScopeKey( + resolveTelegramMessageCachePath(ownerStorePath), + ); + return persistedPaths.map((persistedPath) => { return { kind: "plugin-state-import", label: "Telegram prompt-context message cache", @@ -341,24 +370,22 @@ function detectTelegramSentMessageCacheLegacyStateMigration(params: { env: NodeJS.ProcessEnv; stateDir?: string; }): ChannelLegacyStateMigrationPlan[] { - const defaultAgentId = resolveDefaultAgentId(params.cfg); - const storePath = resolveAgentSessionStorePath({ ...params, agentId: defaultAgentId }); - const legacyMainStorePath = resolveAgentSessionStorePath({ ...params, agentId: "main" }); - const legacyStorePath = resolveLegacySessionStorePath(params); - const sources = uniqueStrings([storePath, legacyMainStorePath, legacyStorePath]).map( - (sourceStorePath) => ({ - targetStorePath: storePath, - sourcePath: `${sourceStorePath}.telegram-sent-messages.json`, - }), - ); - return sources.flatMap((source) => { - if (!fileExists(source.sourcePath)) { - return []; - } + const sourcePaths = listLegacyAgentSessionStorePaths(params) + .map((storePath) => `${storePath}.telegram-sent-messages.json`) + .filter(fileExists); + if (sourcePaths.length === 0) { + return []; + } + const ownerAgentId = resolveTelegramLegacyStateOwnerAgentId(params.cfg); + const targetStorePath = resolveAgentSessionStorePath({ + ...params, + agentId: ownerAgentId, + }); + return sourcePaths.map((sourcePath) => { return { kind: "plugin-state-import", label: "Telegram sent-message cache", - sourcePath: source.sourcePath, + sourcePath, targetPath: `plugin state:${TELEGRAM_SENT_MESSAGE_CACHE_NAMESPACE}`, pluginId: "telegram", namespace: TELEGRAM_SENT_MESSAGE_CACHE_NAMESPACE, @@ -366,13 +393,13 @@ function detectTelegramSentMessageCacheLegacyStateMigration(params: { scopeKey: "", cleanupSource: "rename", cleanupWhenEmpty: true, - preview: `- Telegram sent-message cache: ${source.sourcePath} → plugin state (${TELEGRAM_SENT_MESSAGE_CACHE_NAMESPACE})`, + preview: `- Telegram sent-message cache: ${sourcePath} → plugin state (${TELEGRAM_SENT_MESSAGE_CACHE_NAMESPACE})`, readEntries: () => listTelegramLegacySentMessageCacheEntries({ cfg: params.cfg, - agentId: defaultAgentId, - persistedPath: source.sourcePath, - targetStorePath: source.targetStorePath, + agentId: ownerAgentId, + persistedPath: sourcePath, + targetStorePath, }), }; }); @@ -434,31 +461,48 @@ function detectTelegramTopicNameCacheLegacyStateMigration(params: { }); return topicNameCacheImportSource({ sourceStorePath: storePath }); }); - const defaultStorePath = resolveAgentSessionStorePath({ - ...params, - agentId: resolveDefaultAgentId(params.cfg), - }); - const legacyMainStorePath = resolveAgentSessionStorePath({ ...params, agentId: "main" }); - const defaultAccountStorePath = resolveStorePath(params.cfg.session?.store, { - env: params.env, - agentId: resolveDefaultTelegramAccountId(params.cfg), - }); - const legacyStorePath = resolveLegacySessionStorePath(params); - const sourcesByKey = new Map( - [ - ...accountSources, - topicNameCacheImportSource({ sourceStorePath: defaultStorePath }), - topicNameCacheImportSource({ sourceStorePath: legacyMainStorePath }), - topicNameCacheImportSource({ - sourceStorePath: legacyStorePath, - targetStorePath: defaultAccountStorePath, - }), - ].map((source) => [`${source.sourcePath}\0${source.namespace}`, source] as const), + const agentSources = listAgentIds(params.cfg).map((agentId) => + topicNameCacheImportSource({ + sourceStorePath: resolveAgentSessionStorePath({ ...params, agentId }), + }), ); - return [...sourcesByKey.values()].flatMap((source) => { - if (!fileExists(source.sourcePath)) { - return []; - } + const legacyMainStorePath = resolveAgentSessionStorePath({ ...params, agentId: "main" }); + const legacyStorePath = resolveLegacySessionStorePath(params); + const legacySourcePath = resolveTopicNameCachePath(legacyStorePath); + const fixedSources = [ + ...accountSources, + ...agentSources, + topicNameCacheImportSource({ sourceStorePath: legacyMainStorePath }), + ].filter((source) => fileExists(source.sourcePath)); + if (fixedSources.length === 0 && !fileExists(legacySourcePath)) { + return []; + } + let legacySource: ReturnType | undefined; + if (fileExists(legacySourcePath)) { + const ownerStorePath = resolveAgentSessionStorePath({ + ...params, + agentId: resolveTelegramLegacyStateOwnerAgentId(params.cfg), + }); + // Pre-roster Telegram scoped this legacy cache by account id. Once an agent roster exists, + // routing owns the migration target just as it owns new Telegram conversations. + const legacyTargetStorePath = + params.cfg.agents?.entries !== undefined || params.cfg.agents?.list !== undefined + ? ownerStorePath + : resolveStorePath(params.cfg.session?.store, { + env: params.env, + agentId: resolveDefaultTelegramAccountId(params.cfg), + }); + legacySource = topicNameCacheImportSource({ + sourceStorePath: legacyStorePath, + targetStorePath: legacyTargetStorePath, + }); + } + const sourcesByKey = new Map( + [...fixedSources, ...(legacySource ? [legacySource] : [])].map( + (source) => [`${source.sourcePath}\0${source.namespace}`, source] as const, + ), + ); + return [...sourcesByKey.values()].map((source) => { return { kind: "plugin-state-import", label: "Telegram forum topic-name cache",