fix(telegram): prevent multi-agent startup migration false positives (#122877)

* fix(telegram): resolve legacy state owners explicitly * docs(changelog): note Telegram migration ownership fix * Revert "docs(changelog): note Telegram migration ownership fix" This reverts commit e02962eb4b2e2475854adf8c2a2915bcfd1cf65a.
2026-08-12 21:53:00 -06:00 · 2026-08-12 18:13:45 -07:00
parent 469be48967
commit c2d8b3be4d
2 changed files with 282 additions and 60 deletions
@@ -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 };
@@ -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<typeof topicNameCacheImportSource> | 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",