diff --git a/src/bot-registry.ts b/src/bot-registry.ts
index 2d4cb7f8b0..81e50677db 100644
--- a/src/bot-registry.ts
+++ b/src/bot-registry.ts
@@ -39,6 +39,7 @@ import {
   normalizeSessionOwnerReminderConfig,
   type SessionOwnerReminderConfig,
 } from './core/session-owner-reminder.js';
+import { normalizeMessageListenerCleanupConfig } from './services/message-listener-session-cleanup.js';
 import type {
   VcMeetingConsumerAgentConfig,
   VcMeetingConsumerConfig,
@@ -237,6 +238,12 @@ export interface MessageListenerConfig {
     /** V1 starts one session per matched message. */
     sessionMode?: 'per_message';
   };
+  cleanup?: {
+    /** Default true. */
+    enabled?: boolean;
+    /** Default 168 hours. */
+    retentionHours?: number;
+  };
 }
 
 export interface SummaryRangeConfig {
@@ -1060,6 +1067,7 @@ function normalizeMessageListenerConfig(raw: unknown, botIndex: number, chatId:
   const includeMsgTypes = normalizeMessageListenerStringList(messageRaw.includeMsgTypes);
   if (includeMsgTypes) messagePolicy.includeMsgTypes = includeMsgTypes;
   messagePolicy.scope = 'top_level';
+  const cleanup = normalizeMessageListenerCleanupConfig(entry.cleanup);
 
   const contentRaw = entry.contentPolicy && typeof entry.contentPolicy === 'object' && !Array.isArray(entry.contentPolicy)
     ? entry.contentPolicy as Record<string, unknown>
@@ -1089,6 +1097,7 @@ function normalizeMessageListenerConfig(raw: unknown, botIndex: number, chatId:
     ...(Object.keys(messagePolicy).length > 0 ? { messagePolicy } : {}),
     ...(contentPolicy ? { contentPolicy } : {}),
     replyPolicy: { mode: 'thread', sessionMode: 'per_message' },
+    cleanup,
   };
 }
 
diff --git a/src/config.ts b/src/config.ts
index c0740082da..5de1dea854 100644
--- a/src/config.ts
+++ b/src/config.ts
@@ -184,12 +184,11 @@ export const config = {
     cliId: (process.env.CLI_ID ?? 'claude-code') as import('./adapters/cli/types.js').CliId,
     cliPathOverride: process.env.CLI_PATH,
     backendType: (process.env.BACKEND_TYPE ?? detectDefaultBackend()) as BackendType,
-    /** Auto-recovery throttle: on restart every surviving persistent-backend
-     *  session is eagerly re-forked to re-attach its pane. With dozens of
-     *  sessions per daemon (and many daemons on one box) firing them all at
-     *  once spikes CPU/IO, so the re-fork is staggered: spawn `batchSize`
-     *  workers, wait `delayMs`, repeat. Tune via BOTMUX_RECOVERY_FORK_BATCH /
-     *  BOTMUX_RECOVERY_FORK_DELAY_MS. */
+    /** Auto-recovery re-attach is opt-in: keeping restored persistent sessions
+     *  lazy protects daemon startup paths like message listeners from thousands
+     *  of worker re-forks. Tune via BOTMUX_RECOVERY_FORK_ENABLED,
+     *  BOTMUX_RECOVERY_FORK_BATCH, and BOTMUX_RECOVERY_FORK_DELAY_MS. */
+    recoveryForkEnabled: (process.env.BOTMUX_RECOVERY_FORK_ENABLED ?? 'false').toLowerCase() === 'true',
     recoveryForkBatchSize: Math.max(1, Number(process.env.BOTMUX_RECOVERY_FORK_BATCH) || 5),
     recoveryForkDelayMs: Math.max(0, Number(process.env.BOTMUX_RECOVERY_FORK_DELAY_MS ?? 250)),
     forwardFollowupWaitMs: resolveForwardFollowupWaitMs(),
diff --git a/src/core/dashboard-ipc-server.ts b/src/core/dashboard-ipc-server.ts
index 985cf6e3b5..80f9bd4f81 100644
--- a/src/core/dashboard-ipc-server.ts
+++ b/src/core/dashboard-ipc-server.ts
@@ -967,16 +967,18 @@ export { composeRowFromActive, composeRowFromClosed, composeRowFromPersistedActi
 // holder.
 export function setBotName(name: string): void { setRowsBotName(name); }
 
+const DASHBOARD_SNAPSHOT_ROW_OPTS = { lightweight: true } as const;
+
 function composeDashboardSessionRows(): SessionRow[] {
-  const active = listActiveSessions().map((ds) => composeRowFromActive(ds));
+  const active = listActiveSessions().map(ds => composeRowFromActive(ds, DASHBOARD_SNAPSHOT_ROW_OPTS));
   const activeIds = new Set(active.map(row => row.sessionId));
   const persisted = sessionStore.listSessions();
   const unregisteredActive = persisted
     .filter(session => session.status === 'active' && !activeIds.has(session.sessionId))
-    .map(composeRowFromPersistedActive);
+    .map(session => composeRowFromPersistedActive(session, DASHBOARD_SNAPSHOT_ROW_OPTS));
   const closed = persisted
     .filter(session => session.status === 'closed' && !activeIds.has(session.sessionId))
-    .map(composeRowFromClosed);
+    .map(session => composeRowFromClosed(session, DASHBOARD_SNAPSHOT_ROW_OPTS));
   return [...active, ...unregisteredActive, ...closed];
 }
 
@@ -3519,6 +3521,7 @@ async function collectMessageListenerPreviewMatches(
     prompt: listener.prompt,
     ...(listener.senderPolicy && Object.keys(listener.senderPolicy).length > 0 ? { senderPolicy: listener.senderPolicy } : {}),
     ...(listener.messagePolicy ? { messagePolicy: { ...listener.messagePolicy, scope: 'top_level' } } : { messagePolicy: { scope: 'top_level' } }),
+    ...(listener.cleanup ? { cleanup: listener.cleanup } : {}),
     replyPolicy: { mode: 'thread', sessionMode: 'per_message' },
   };
   const previewBot = {
@@ -5593,7 +5596,7 @@ ipcRoute('GET', '/api/events', (_req, res) => {
     const activeIds = new Set<string>();
     for (const ds of listActiveSessions()) {
       activeIds.add(ds.session.sessionId);
-      res.write(`event: session.spawned\ndata: ${JSON.stringify({ session: composeRowFromActive(ds) })}\n\n`);
+      res.write(`event: session.spawned\ndata: ${JSON.stringify({ session: composeRowFromActive(ds, DASHBOARD_SNAPSHOT_ROW_OPTS) })}\n\n`);
     }
     // Persisted active rows may be intentionally absent from the runtime Map
     // after an inconclusive exact-backend teardown. Replay them as dormant
@@ -5601,7 +5604,7 @@ ipcRoute('GET', '/api/events', (_req, res) => {
     // /api/sessions and never synthesize a closed row.
     for (const s of sessionStore.listSessions()) {
       if (s.status !== 'active' || activeIds.has(s.sessionId)) continue;
-      res.write(`event: session.spawned\ndata: ${JSON.stringify({ session: composeRowFromPersistedActive(s) })}\n\n`);
+      res.write(`event: session.spawned\ndata: ${JSON.stringify({ session: composeRowFromPersistedActive(s, DASHBOARD_SNAPSHOT_ROW_OPTS) })}\n\n`);
     }
     // Also replay sessions CLOSED during this run as `session.spawned` carrying a
     // closed row. The active-only replay above can't cover a restore-time zombie:
@@ -5621,7 +5624,7 @@ ipcRoute('GET', '/api/events', (_req, res) => {
       if (s.status !== 'closed' || activeIds.has(s.sessionId)) continue;
       const closedMs = s.closedAt ? Date.parse(s.closedAt) : NaN;
       if (!Number.isFinite(closedMs) || closedMs < PROCESS_START_MS) continue;
-      res.write(`event: session.spawned\ndata: ${JSON.stringify({ session: composeRowFromClosed(s) })}\n\n`);
+      res.write(`event: session.spawned\ndata: ${JSON.stringify({ session: composeRowFromClosed(s, DASHBOARD_SNAPSHOT_ROW_OPTS) })}\n\n`);
     }
   } catch (err) {
     logger.warn(`[dashboard-ipc] /api/events snapshot replay failed: ${err}`);
diff --git a/src/core/dashboard-rows.ts b/src/core/dashboard-rows.ts
index 5132480e3b..a14041150e 100644
--- a/src/core/dashboard-rows.ts
+++ b/src/core/dashboard-rows.ts
@@ -142,6 +142,11 @@ let cachedBotName = '';
 export function setBotName(name: string): void { cachedBotName = name; }
 export function getBotName(): string { return cachedBotName; }
 
+export interface ComposeSessionRowOptions {
+  fresh?: boolean;
+  lightweight?: boolean;
+}
+
 function parseSessionTime(iso: string | undefined): number | undefined {
   if (!iso) return undefined;
   const ms = Date.parse(iso);
@@ -212,10 +217,13 @@ function sessionRuntimeFields(s: Session): Pick<SessionRow, 'runtimeId' | 'runti
   return {};
 }
 
-export function composeRowFromActive(ds: DaemonSession, opts?: { fresh?: boolean }): SessionRow {
+export function composeRowFromActive(
+  ds: DaemonSession,
+  opts: ComposeSessionRowOptions = {},
+): SessionRow {
   const brand = getBotBrand(ds.larkAppId);
   const topicLink = sessionThreadLink(ds.session, brand);
-  return {
+  const row: SessionRow = {
     sessionId: ds.session.sessionId,
     larkAppId: ds.larkAppId,
     botName: cachedBotName,
@@ -273,18 +281,21 @@ export function composeRowFromActive(ds: DaemonSession, opts?: { fresh?: boolean
     agentAttention: ds.agentAttention
       ? { kind: ds.agentAttention.kind, reason: ds.agentAttention.reason, at: ds.agentAttention.at }
       : undefined,
-    tokenUsage: sessionTokenUsage(ds.session, ds.workingDir),
-    openTodos: sessionOpenTodos(ds.session, ds.workingDir, opts?.fresh),
     ...(ds.worker?.pid !== undefined ? { workerPid: ds.worker.pid } : {}),
     ...(ds.adoptedFrom?.originalCliPid !== undefined ? { adoptCliPid: ds.adoptedFrom.originalCliPid } : {}),
-    ...buildSessionMessagePreview(ds.session),
   };
+  if (!opts.lightweight) {
+    row.tokenUsage = sessionTokenUsage(ds.session, ds.workingDir);
+    row.openTodos = sessionOpenTodos(ds.session, ds.workingDir, opts.fresh);
+    Object.assign(row, buildSessionMessagePreview(ds.session));
+  }
+  return row;
 }
 
-export function composeRowFromClosed(s: Session): SessionRow {
+export function composeRowFromClosed(s: Session, opts: ComposeSessionRowOptions = {}): SessionRow {
   const brand = getBotBrand(s.larkAppId ?? '');
   const topicLink = sessionThreadLink(s, brand);
-  return {
+  const row: SessionRow = {
     sessionId: s.sessionId,
     larkAppId: s.larkAppId ?? '',
     botName: cachedBotName,
@@ -315,9 +326,12 @@ export function composeRowFromClosed(s: Session): SessionRow {
     previewTarget: safeSessionPreviewTarget(s.previewTarget),
     feishuChatLink: feishuChatLink(s.chatId, brand),
     ...(topicLink ? { feishuThreadLink: topicLink } : {}),
-    tokenUsage: sessionTokenUsage(s),
-    ...buildSessionMessagePreview(s),
   };
+  if (!opts.lightweight) {
+    row.tokenUsage = sessionTokenUsage(s);
+    Object.assign(row, buildSessionMessagePreview(s));
+  }
+  return row;
 }
 
 /**
@@ -328,10 +342,10 @@ export function composeRowFromClosed(s: Session): SessionRow {
  * dashboard presents it as dormant, with no terminal port, so operators can
  * see and explicitly retry closing it without an unsafe resume affordance.
  */
-export function composeRowFromPersistedActive(s: Session): SessionRow {
+export function composeRowFromPersistedActive(s: Session, opts: ComposeSessionRowOptions = {}): SessionRow {
   const brand = getBotBrand(s.larkAppId ?? '');
   const topicLink = sessionThreadLink(s, brand);
-  return {
+  const row: SessionRow = {
     sessionId: s.sessionId,
     larkAppId: s.larkAppId ?? '',
     botName: cachedBotName,
@@ -363,7 +377,10 @@ export function composeRowFromPersistedActive(s: Session): SessionRow {
     queued: !!s.queued,
     hasHistory: !!(s.cliId || s.lastCliInput || s.backendType || s.adoptedFrom),
     quarantined: !!s.restoreQuarantinedAt,
-    tokenUsage: sessionTokenUsage(s),
-    ...buildSessionMessagePreview(s),
   };
+  if (!opts.lightweight) {
+    row.tokenUsage = sessionTokenUsage(s);
+    Object.assign(row, buildSessionMessagePreview(s));
+  }
+  return row;
 }
diff --git a/src/core/session-activity.ts b/src/core/session-activity.ts
index 258550e5f2..6c424c5be1 100644
--- a/src/core/session-activity.ts
+++ b/src/core/session-activity.ts
@@ -142,10 +142,10 @@ export function clearAgentAttention(ds: DaemonSession): boolean {
   return true;
 }
 
-export function announceSessionRow(ds: DaemonSession): void {
+export function announceSessionRow(ds: DaemonSession, opts?: { lightweight?: boolean }): void {
   dashboardEventBus.publish({
     type: 'session.spawned',
-    body: { session: composeRowFromActive(ds) },
+    body: { session: composeRowFromActive(ds, opts) },
   });
 }
 
diff --git a/src/core/session-manager.ts b/src/core/session-manager.ts
index b055458026..699f4c6c06 100644
--- a/src/core/session-manager.ts
+++ b/src/core/session-manager.ts
@@ -1729,19 +1729,16 @@ export function rememberLastCliInput(
 
 /**
  * Whether daemon restore should eagerly re-fork a worker to re-attach a
- * surviving backing pane. True for every persistent backend (tmux/herdr/zellij/zmx);
- * the pty backend has nothing to re-attach to, so it stays lazy.
- *
- * Eager re-attach is what makes a session actually come back after a restart —
- * otherwise a killed worker leaves the session dead until its next message, and
- * a pane whose CLI died in the meantime never gets healed, so the transcript
- * fallback can't fire. The old `BOTMUX_QUIET_RESTART` gate that suppressed this
- * (to avoid re-pushing cards on dev restarts) is gone: restored sessions now
- * carry `suppressRecoveryCard`, so the recovery re-fork stays silent in the
- * Lark thread without having to skip recovery altogether.
+ * surviving backing pane. This is opt-in for persistent backends: on long-lived
+ * hosts with thousands of active rows, eager recovery can saturate the daemon's
+ * event loop and starve message listeners. Lazy recovery still re-attaches on
+ * the next user message or terminal access.
  */
-export function shouldAutoForkOnRestore(backendType: BackendType): boolean {
-  return backendType !== 'pty';
+export function shouldAutoForkOnRestore(
+  backendType: BackendType,
+  enabled: boolean = config.daemon.recoveryForkEnabled === true,
+): boolean {
+  return enabled && backendType !== 'pty';
 }
 
 const RECOVERY_FORK_BATCH_SIZE = config.daemon.recoveryForkBatchSize ?? 5;
@@ -1788,6 +1785,22 @@ export async function staggeredRecoveryFork(
   }
 }
 
+export function scheduleStaggeredRecoveryFork(
+  sessions: readonly DaemonSession[],
+  fork: (ds: DaemonSession) => void,
+  batchSize: number = RECOVERY_FORK_BATCH_SIZE,
+  delayMs: number = RECOVERY_FORK_DELAY_MS,
+  stillOwned: (ds: DaemonSession) => boolean = ds => ds.session.status === 'active',
+): void {
+  void staggeredRecoveryFork(sessions, fork, batchSize, delayMs, stillOwned)
+    .catch(err => {
+      logger.error(
+        `[restore] background recovery re-attach failed: `
+        + `${err instanceof Error ? err.message : String(err)}`,
+      );
+    });
+}
+
 export async function restoreActiveSessions(
   activeSessions: Map<string, DaemonSession>,
   quarantinedSessionIds: ReadonlySet<string> = new Set(),
@@ -2139,7 +2152,7 @@ export async function restoreActiveSessions(
           continue;
         }
         restoredByThisInvocation.push(ds);
-        announceSessionRow(ds);
+        announceSessionRow(ds, { lightweight: true });
         forkAdoptWorker(ds, { restoredFromMetadata: true });
         logger.info(`[${session.sessionId.substring(0, 8)}] Restored adopt session (target: ${adoptTargetLabel(adopted)}, scope: ${scope})`);
         continue;
@@ -2245,7 +2258,7 @@ export async function restoreActiveSessions(
       restoredByThisInvocation.push(ds);
       // 重启后把待办池卡片重新广播给 dashboard，否则会从看板消失（#277 同款修复，
       // 我这条 queued 分支提前 continue 绕过了下面的 announceSessionRow，要自己补）。
-      announceSessionRow(ds);
+      announceSessionRow(ds, { lightweight: true });
       if (restoredPendingRepo) {
         try {
           await resumeRestoredPendingRepoSetup(ds, activeSessions);
@@ -2474,7 +2487,7 @@ export async function restoreActiveSessions(
       continue;
     }
     restoredByThisInvocation.push(ds);
-    announceSessionRow(ds);
+    announceSessionRow(ds, { lightweight: true });
 
     if (session.initialUserTurnPending) {
       // `hasHistory: true` above means "there may be a CLI process/transcript to
@@ -2518,6 +2531,7 @@ export async function restoreActiveSessions(
     backendName: string;
   }> = [];
   const namesByBackend = new Map<PersistentBackendType, Set<string>>();
+  const skippedReattachByBackend = new Map<PersistentBackendType, number>();
   for (const ds of restoredByThisInvocation) {
     // A later restore CAS awaited after this row was registered. During that
     // yield the user may have closed/resumed/replaced it; never carry the stale
@@ -2541,7 +2555,10 @@ export async function restoreActiveSessions(
       }
       continue;
     }
-    if (!shouldAutoForkOnRestore(backendType)) continue;
+    if (!shouldAutoForkOnRestore(backendType)) {
+      skippedReattachByBackend.set(backendType, (skippedReattachByBackend.get(backendType) ?? 0) + 1);
+      continue;
+    }
     // Honour the worker-selected target (Herdr may own an agent inside a shared
     // host session) rather than assuming the deterministic whole-session name.
     const backendTarget = persistentBackendTargetForSession(ds)!;
@@ -2556,6 +2573,16 @@ export async function restoreActiveSessions(
     names.add(backendTarget.sessionName);
     namesByBackend.set(backendType, names);
   }
+  if (skippedReattachByBackend.size > 0) {
+    const total = [...skippedReattachByBackend.values()].reduce((sum, count) => sum + count, 0);
+    const detail = [...skippedReattachByBackend.entries()]
+      .map(([backendType, count]) => `${backendType}=${count}`)
+      .join(', ');
+    logger.info(
+      `[restore] skipped eager re-attach for ${total} persistent session(s) `
+      + `(${detail}); lazy recovery remains available`,
+    );
+  }
   // ZMX/Zellij can classify every requested name from one control-plane list.
   // This is both a consistent restore snapshot and avoids an O(N²) ZMX restart
   // when each per-row probe would otherwise scan every per-session daemon.
@@ -2658,11 +2685,7 @@ export async function restoreActiveSessions(
     toReattach.push(ds);
   }
 
-  // Staggered re-fork (see staggeredRecoveryFork): empty prompt = re-attach
-  // only, no new turn — same as the old per-session eager fork.
-  await staggeredRecoveryFork(
-    toReattach,
-    (ds) => {
+  const reattach = (ds: DaemonSession): void => {
       // A quarantined tail-only owner (restore promotion failed transiently) is
       // handled by the CENTRAL guard inside forkWorker: this blank fork retries
       // the old head's promotion first and, if it still fails, refuses to fork
@@ -2686,11 +2709,24 @@ export async function restoreActiveSessions(
             }
           : true,
       );
-    },
-    RECOVERY_FORK_BATCH_SIZE,
-    RECOVERY_FORK_DELAY_MS,
-    ds => activeSessions.get(activeSessionKey(ds)) === ds,
-  );
+  };
+
+  // Staggered re-fork (see staggeredRecoveryFork): empty prompt = re-attach
+  // only, no new turn — same as the old per-session eager fork. Keep it off the
+  // restore critical path: on long-lived installations thousands of restored
+  // active rows can otherwise hold daemon readiness and message-listener
+  // backfill behind worker re-attach for minutes, even though the sessions are
+  // already registered and can lazy cold-resume on demand.
+  if (toReattach.length > 0) {
+    logger.info(`[restore] scheduling ${toReattach.length} persistent session(s) for background re-attach`);
+    scheduleStaggeredRecoveryFork(
+      toReattach,
+      reattach,
+      RECOVERY_FORK_BATCH_SIZE,
+      RECOVERY_FORK_DELAY_MS,
+      ds => activeSessions.get(activeSessionKey(ds)) === ds,
+    );
+  }
 
   const hasPersistentBackend = [...activeSessions.values()].some(ds => !!getSessionPersistentBackendType(ds));
   logger.info(`Restored ${active.length} session(s)${hasPersistentBackend ? '' : ', waiting for messages to resume'}`);
diff --git a/src/daemon.ts b/src/daemon.ts
index 412f674170..aa30b4bcea 100644
--- a/src/daemon.ts
+++ b/src/daemon.ts
@@ -192,6 +192,7 @@ import {
   readableTerminalUrlFor,
   findActiveBySessionId,
   withActiveSessionKeyLock,
+  destroyUnregisteredPersistentBacking,
   getDaemonBootId,
   getDaemonStreamingCardUsageSnapshot,
   postTurnStartingCard,
@@ -283,6 +284,10 @@ import {
 } from './core/session-title.js';
 import { settleDeferredScheduleRun } from './core/deferred-schedule-settlement.js';
 import { renderMessageListenerPrompt, refreshListenerCardTextFromResolved } from './services/message-listener.js';
+import {
+  MESSAGE_LISTENER_CLEANUP_INTERVAL_MS,
+  selectExpiredMessageListenerSessions,
+} from './services/message-listener-session-cleanup.js';
 import { sweepOrphanSandboxes } from './adapters/backend/sandbox.js';
 import { TmuxBackend } from './adapters/backend/tmux-backend.js';
 import { HerdrBackend } from './adapters/backend/herdr-backend.js';
@@ -3359,6 +3364,74 @@ function startMemoryDiagnostics(): ReturnType<typeof setInterval> | undefined {
   return timer;
 }
 
+function createMessageListenerSessionCleanupRunner(larkAppId: string): {
+  run(reason: 'startup' | 'interval'): Promise<void>;
+  start(): ReturnType<typeof setInterval>;
+} {
+  let running = false;
+  const run = async (reason: 'startup' | 'interval'): Promise<void> => {
+    if (running) return;
+    running = true;
+    try {
+      const bot = getBot(larkAppId);
+      const expired = selectExpiredMessageListenerSessions({
+        sessions: sessionStore.listSessions(),
+        listeners: bot.config.messageListeners,
+      });
+      if (expired.length === 0) return;
+      let closed = 0;
+      let failed = 0;
+      const inactive: string[] = [];
+      logger.info(`[message-listener-cleanup] ${reason}: closing ${expired.length} expired listener session(s) for ${larkAppId}`);
+      for (const session of expired) {
+        if (!findActiveBySessionId(session.sessionId)) {
+          try {
+            destroyUnregisteredPersistentBacking(session);
+            inactive.push(session.sessionId);
+          } catch (err) {
+            failed += 1;
+            logger.warn(
+              `[message-listener-cleanup] backing cleanup failed session=${session.sessionId}: `
+              + `${err instanceof Error ? err.message : String(err)}`,
+            );
+          }
+          continue;
+        }
+        try {
+          const result = await closeSessionHelper(session.sessionId);
+          if (result.ok && result.known) closed += 1;
+          else failed += 1;
+        } catch (err) {
+          failed += 1;
+          logger.warn(
+            `[message-listener-cleanup] close failed session=${session.sessionId}: `
+            + `${err instanceof Error ? err.message : String(err)}`,
+          );
+        }
+      }
+      if (inactive.length > 0) {
+        const inactiveIds = new Set(inactive);
+        closed += sessionStore.closeSessionsMatching(session => inactiveIds.has(session.sessionId));
+      }
+      logger.info(`[message-listener-cleanup] ${reason}: closed=${closed} failed=${failed} bot=${larkAppId}`);
+    } catch (err) {
+      logger.warn(`[message-listener-cleanup] ${reason} failed: ${err instanceof Error ? err.message : String(err)}`);
+    } finally {
+      running = false;
+    }
+  };
+  return {
+    run,
+    start() {
+      const timer = setInterval(() => {
+        void run('interval');
+      }, MESSAGE_LISTENER_CLEANUP_INTERVAL_MS);
+      timer.unref?.();
+      return timer;
+    },
+  };
+}
+
 /**
  * Reply into a session — scope-aware.
  *
@@ -18199,6 +18272,7 @@ async function handleNewTopicAdmitted(data: any, ctx: RoutingContext): Promise<v
   session.ownerUnionId = ownerUnionIdForSession;
   session.creatorOpenId = senderOpenId;
   session.lastCallerOpenId = senderOpenId;
+  if (messageListener) session.messageListener = { chatId };
   // First turn of a brand-new topic: seed quoteTarget* so the very first
   // `botmux send` can --mention-back / 引用 the triggering message (chat scope).
   // Without this the first reply hits hasQuoteTargetSender=false (exit 2) and
@@ -21488,6 +21562,7 @@ export async function startDaemon(botIndex?: number): Promise<void> {
   sessionStore.init(cfg.larkAppId);
   chatFirstSeenStore.init(cfg.larkAppId);
   initSessionGroups(cfg.larkAppId);
+  const messageListenerCleanup = createMessageListenerSessionCleanupRunner(cfg.larkAppId);
   const ambiguousOnBoot = reconcileVcMeetingDeliveriesOnBoot(
     config.session.dataDir,
     { receiverBootId: getDaemonBootId(), agentAppId: cfg.larkAppId },
@@ -22307,7 +22382,7 @@ export async function startDaemon(botIndex?: number): Promise<void> {
   reapOrphanWorkers();
 
   // Restore active sessions from previous run
-  // Restore active sessions from previous run
+  await messageListenerCleanup.run('startup');
   await restoreActiveSessions(activeSessions, idempotencyQuarantinedSessionIds);
   // Restore complete → /api/asks may now safely 403 unknown sessions again; a
   // reconnecting ask hook that raced the restore got retryable 503s until here.
@@ -22465,6 +22540,7 @@ export async function startDaemon(botIndex?: number): Promise<void> {
     enforceLiveSessionCap('periodic');
   }, 60_000);
   idleWorkerSweepTimer.unref?.();
+  const messageListenerCleanupTimer = messageListenerCleanup.start();
 
   const sessionOwnerReminder = new SessionOwnerReminderController({
     load: () => loadSessionOwnerReminderRecords(config.session.dataDir, cfg.larkAppId),
@@ -22862,6 +22938,7 @@ export async function startDaemon(botIndex?: number): Promise<void> {
     clearInterval(descriptorHeartbeat);
     clearInterval(idleWorkerSweepTimer);
     clearInterval(sessionOwnerReminderTimer);
+    clearInterval(messageListenerCleanupTimer);
     if (memoryDiagnostics) clearInterval(memoryDiagnostics);
     removeDaemonDescriptor(cfg.larkAppId);
     ipcHandle.close().catch(() => { /* swallow */ });
@@ -23074,6 +23151,7 @@ export async function startDaemon(botIndex?: number): Promise<void> {
     clearInterval(descriptorHeartbeat);
     clearInterval(idleWorkerSweepTimer);
     clearInterval(sessionOwnerReminderTimer);
+    clearInterval(messageListenerCleanupTimer);
     clearInterval(docCommentPollTimer);
     if (memoryDiagnostics) clearInterval(memoryDiagnostics);
     removeDaemonDescriptor(cfg.larkAppId);
diff --git a/src/dashboard/web/i18n.ts b/src/dashboard/web/i18n.ts
index 677a972d35..eaccc6c7f9 100644
--- a/src/dashboard/web/i18n.ts
+++ b/src/dashboard/web/i18n.ts
@@ -2435,6 +2435,8 @@ const zh = {
   'roles.listenerReplyCardTitlePlaceholder': '留空使用默认标题',
   'roles.listenerWorkingDir': '工作目录',
   'roles.listenerWorkingDirPlaceholder': '留空使用 Bot 默认工作目录',
+  'roles.listenerCleanupEnabled': '自动清理监听会话',
+  'roles.listenerCleanupRetentionHours': '保留小时数',
   'roles.listenerSenderTypes': '发送者类型',
   'roles.listenerSenderUser': '用户',
   'roles.listenerSenderBot': '机器人',
@@ -4991,6 +4993,8 @@ const en: Record<keyof typeof zh, string> = {
   'roles.listenerReplyCardTitlePlaceholder': 'Leave empty to use the default title',
   'roles.listenerWorkingDir': 'Working directory',
   'roles.listenerWorkingDirPlaceholder': 'Leave empty to use the bot default',
+  'roles.listenerCleanupEnabled': 'Auto-clean listener sessions',
+  'roles.listenerCleanupRetentionHours': 'Retention hours',
   'roles.listenerSenderTypes': 'Sender types',
   'roles.listenerSenderUser': 'Users',
   'roles.listenerSenderBot': 'Bots',
diff --git a/src/dashboard/web/roles-page.tsx b/src/dashboard/web/roles-page.tsx
index c43a710025..b3ab348854 100644
--- a/src/dashboard/web/roles-page.tsx
+++ b/src/dashboard/web/roles-page.tsx
@@ -47,6 +47,7 @@ import {
   MAX_MESSAGE_LISTENER_PROMPT_BYTES,
   MAX_ROLE_BYTES,
   MESSAGE_LISTENER_WARN_BYTES,
+  DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS,
   DEFAULT_MESSAGE_LISTENER_PREVIEW_LIMIT,
   MAX_MESSAGE_LISTENER_PREVIEW_LIMIT,
   roleKey,
@@ -116,6 +117,10 @@ const DEFAULT_LISTENER: MessageListenerData = {
     includeMsgTypes: [...LISTENER_MESSAGE_TYPES],
     scope: 'top_level',
   },
+  cleanup: {
+    enabled: true,
+    retentionHours: DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS,
+  },
 };
 
 function cloneListener(listener: MessageListenerData | null | undefined): MessageListenerData {
@@ -152,9 +157,24 @@ function cloneListener(listener: MessageListenerData | null | undefined): Messag
         ...(listener.contentPolicy.matchMode ? { matchMode: listener.contentPolicy.matchMode } : {}),
       },
     } : {}),
+    cleanup: {
+      enabled: listener?.cleanup?.enabled !== false,
+      retentionHours: normalizeCleanupRetentionHours(listener?.cleanup?.retentionHours),
+    },
   };
 }
 
+function normalizeCleanupRetentionHours(raw: unknown): number {
+  const value = typeof raw === 'string' && raw.trim()
+    ? Number(raw.trim())
+    : raw;
+  if (typeof value !== 'number' || !Number.isFinite(value)) {
+    return DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS;
+  }
+  const normalized = Math.floor(value);
+  return normalized > 0 ? normalized : DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS;
+}
+
 function listenerHasConfig(listener: MessageListenerData | null): boolean {
   // A persisted listener is worth loading into the editor whenever it carries a
   // prompt — INCLUDING a disabled draft (enabled:false + non-empty prompt). The
@@ -599,6 +619,17 @@ function RolesPage(props: { tab: RolesTab }) {
     }));
   }
 
+  function updateListenerCleanup(patch: Partial<NonNullable<MessageListenerData['cleanup']>>): void {
+    setEditingListener(prev => ({
+      ...prev,
+      cleanup: {
+        enabled: prev.cleanup?.enabled !== false,
+        retentionHours: normalizeCleanupRetentionHours(prev.cleanup?.retentionHours),
+        ...patch,
+      },
+    }));
+  }
+
   function toggleListenerSenderType(type: SenderTypeOption, checked: boolean): void {
     setEditingListener(prev => {
       const current = new Set(prev.senderPolicy?.includeSenderTypes ?? []);
@@ -720,6 +751,10 @@ function RolesPage(props: { tab: RolesTab }) {
         ...(raw.matchMode === 'all' ? { matchMode: 'all' as const } : {}),
       };
     })();
+    const cleanup = {
+      enabled: editingListener.cleanup?.enabled !== false,
+      retentionHours: normalizeCleanupRetentionHours(editingListener.cleanup?.retentionHours),
+    };
     return {
       enabled: editingListener.enabled,
       ...(editingListener.name?.trim() ? { name: editingListener.name.trim() } : {}),
@@ -741,6 +776,7 @@ function RolesPage(props: { tab: RolesTab }) {
         scope: 'top_level',
       },
       ...(contentPolicy ? { contentPolicy } : {}),
+      cleanup,
     };
   }
 
@@ -1203,6 +1239,7 @@ function RolesPage(props: { tab: RolesTab }) {
                   onSenderPolicyPatch={updateListenerSenderPolicy}
                   onMessagePolicyPatch={updateListenerMessagePolicy}
                   onContentPolicyPatch={updateListenerContentPolicy}
+                  onCleanupPatch={updateListenerCleanup}
                   onToggleSenderType={toggleListenerSenderType}
                   onToggleMsgType={toggleListenerMsgType}
                   onSetTargetPolicy={setListenerTargetPolicy}
@@ -1545,6 +1582,7 @@ function MessageListenerEditor(props: {
   onSenderPolicyPatch(patch: NonNullable<MessageListenerData['senderPolicy']>): void;
   onMessagePolicyPatch(patch: NonNullable<MessageListenerData['messagePolicy']>): void;
   onContentPolicyPatch(patch: NonNullable<MessageListenerData['contentPolicy']>): void;
+  onCleanupPatch(patch: Partial<NonNullable<MessageListenerData['cleanup']>>): void;
   onToggleSenderType(type: SenderTypeOption, checked: boolean): void;
   onToggleMsgType(msgType: string, checked: boolean): void;
   onSetTargetPolicy(openId: string, listening: boolean): void;
@@ -1754,6 +1792,27 @@ function MessageListenerEditor(props: {
           </small>
         </div>
       </div>
+      <div className="roles-listener-policy-row">
+        <label className="filter-toggle roles-listener-enabled">
+          <input
+            type="checkbox"
+            checked={listener.cleanup?.enabled !== false}
+            onChange={ev => props.onCleanupPatch({ enabled: ev.currentTarget.checked })}
+          />
+          <span className="filter-toggle-switch" aria-hidden="true"></span>
+          <span className="filter-toggle-label">{tr('roles.listenerCleanupEnabled')}</span>
+        </label>
+        <label className="roles-listener-field">
+          <span className="roles-field-label">{tr('roles.listenerCleanupRetentionHours')}</span>
+          <input
+            type="number"
+            min={1}
+            step={1}
+            value={normalizeCleanupRetentionHours(listener.cleanup?.retentionHours)}
+            onChange={ev => props.onCleanupPatch({ retentionHours: normalizeCleanupRetentionHours(ev.currentTarget.value) })}
+          />
+        </label>
+      </div>
       <div className="roles-listener-content-policy">
         <div className="roles-field-label">{tr('roles.listenerContentPolicy')}</div>
         <label className="roles-listener-field">
diff --git a/src/dashboard/web/roles.ts b/src/dashboard/web/roles.ts
index 23debbb72d..cfb3313f2d 100644
--- a/src/dashboard/web/roles.ts
+++ b/src/dashboard/web/roles.ts
@@ -69,6 +69,10 @@ export interface MessageListenerData {
     includeKeywords?: string[];
     matchMode?: 'any' | 'all';
   };
+  cleanup?: {
+    enabled: boolean;
+    retentionHours: number;
+  };
 }
 
 export interface MessageListenerPreviewItem {
@@ -165,6 +169,7 @@ export const MAX_MESSAGE_LISTENER_PROMPT_BYTES = 32768;
 export const MESSAGE_LISTENER_WARN_BYTES = Math.floor(MAX_MESSAGE_LISTENER_PROMPT_BYTES * 0.95);
 export const DEFAULT_MESSAGE_LISTENER_PREVIEW_LIMIT = 5;
 export const MAX_MESSAGE_LISTENER_PREVIEW_LIMIT = 20;
+export const DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS = 168;
 
 const PROFILE_ID_RE = /^[A-Za-z0-9._-]{1,64}$/;
 
@@ -311,7 +316,12 @@ export async function deleteRole(larkAppId: string, chatId: string): Promise<boo
 export async function loadMessageListener(larkAppId: string, chatId: string): Promise<MessageListenerData> {
   const r = await fetch(`/api/message-listeners/${encodeURIComponent(larkAppId)}/${encodeURIComponent(chatId)}`);
   const data = await readJson(r);
-  return data.listener ?? { enabled: false, prompt: '', messagePolicy: { scope: 'top_level' } };
+  return data.listener ?? {
+    enabled: false,
+    prompt: '',
+    messagePolicy: { scope: 'top_level' },
+    cleanup: { enabled: true, retentionHours: DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS },
+  };
 }
 
 export async function saveMessageListener(larkAppId: string, chatId: string, listener: MessageListenerData): Promise<boolean> {
diff --git a/src/im/lark/client.ts b/src/im/lark/client.ts
index a038e9406b..14b09018f4 100644
--- a/src/im/lark/client.ts
+++ b/src/im/lark/client.ts
@@ -1611,6 +1611,10 @@ export async function listChatMessages(
 export interface ChatMessageScanOptions {
   /** Lark page size per request. Clamped to the API max of 50. */
   pageSize?: number;
+  /** Optional HTTP timeout for each page request. Falls back to the client default. */
+  timeoutMs?: number;
+  /** Optional cancellation signal for the underlying page requests. */
+  signal?: AbortSignal;
   /**
    * Called while scanning newest -> oldest. Returning true stops after the
    * current message has been included in the returned chronological list.
@@ -1639,7 +1643,7 @@ export async function listChatMessagesUntil(
       sort_type: 'ByCreateTimeDesc',
       with_sender_name: 'true',
       ...(pageToken ? { page_token: pageToken } : {}),
-    });
+    }, { timeoutMs: options.timeoutMs, signal: options.signal });
 
     if (res.code !== 0) {
       throw new Error(`Failed to list chat messages: ${res.msg} (code: ${res.code})`);
diff --git a/src/im/lark/event-dispatcher.ts b/src/im/lark/event-dispatcher.ts
index fb3263cadf..327d4690f9 100644
--- a/src/im/lark/event-dispatcher.ts
+++ b/src/im/lark/event-dispatcher.ts
@@ -137,6 +137,16 @@ export function writeBotInfoFile(dataDir: string): void {
 /** Per-app in-flight open_id probe, so a startup burst of events shares one probe. */
 const inflightOpenIdProbes = new Map<string, Promise<void>>();
 
+async function fetchWithTimeout(input: RequestInfo | URL, init: RequestInit, timeoutMs: number): Promise<Response> {
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), timeoutMs);
+  try {
+    return await fetch(input, { ...init, signal: controller.signal });
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
 /**
  * Ensure the bot's own open_id is resolved before @-detection. `probeBotOpenId`
  * is fired fire-and-forget at daemon startup, so events can arrive while
@@ -163,19 +173,19 @@ export async function probeBotOpenId(larkAppId: string): Promise<void> {
   const openApi = larkHosts(normalizeBrand(bot.config.brand)).openApi;
 
   // Call /bot/v3/info to get the bot's open_id using tenant_access_token
-  const tokenRes = await fetch(`${openApi}/open-apis/auth/v3/tenant_access_token/internal`, {
+  const tokenRes = await fetchWithTimeout(`${openApi}/open-apis/auth/v3/tenant_access_token/internal`, {
     method: 'POST',
     headers: { 'Content-Type': 'application/json' },
     body: JSON.stringify({ app_id: bot.config.larkAppId, app_secret: bot.config.larkAppSecret }),
-  });
+  }, 10_000);
   const tokenData = await tokenRes.json() as any;
   if (tokenData.code !== 0) {
     throw new Error(`Failed to get tenant_access_token: ${tokenData.msg}`);
   }
 
-  const botRes = await fetch(`${openApi}/open-apis/bot/v3/info/`, {
+  const botRes = await fetchWithTimeout(`${openApi}/open-apis/bot/v3/info/`, {
     headers: { Authorization: `Bearer ${tokenData.tenant_access_token}` },
-  });
+  }, 10_000);
   const botData = await botRes.json() as any;
   if (botData.code !== 0) {
     throw new Error(`Failed to get bot info: ${botData.msg}`);
@@ -2146,6 +2156,40 @@ const MESSAGE_LISTENER_BACKFILL_PAGE_SIZE = Math.min(50, Math.max(
   1,
   Number(process.env.BOTMUX_MESSAGE_LISTENER_BACKFILL_PAGE_SIZE) || 50,
 ));
+const MESSAGE_LISTENER_BACKFILL_TIMEOUT_MS = Math.max(
+  1_000,
+  Number(process.env.BOTMUX_MESSAGE_LISTENER_BACKFILL_TIMEOUT_MS) || 15_000,
+);
+const MESSAGE_LISTENER_POLL_RUN_TIMEOUT_MS = Math.max(
+  MESSAGE_LISTENER_BACKFILL_TIMEOUT_MS + 5_000,
+  Number(process.env.BOTMUX_MESSAGE_LISTENER_POLL_RUN_TIMEOUT_MS) || 30_000,
+);
+
+function withMessageListenerTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  return Promise.race([
+    promise,
+    new Promise<T>((_, reject) => {
+      timer = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs);
+    }),
+  ]).finally(() => {
+    if (timer) clearTimeout(timer);
+  });
+}
+
+async function withAbortableMessageListenerTimeout<T>(
+  run: (signal: AbortSignal) => Promise<T>,
+  timeoutMs: number,
+  label: string,
+): Promise<T> {
+  const controller = new AbortController();
+  try {
+    return await withMessageListenerTimeout(run(controller.signal), timeoutMs, label);
+  } catch (err) {
+    controller.abort();
+    throw err;
+  }
+}
 
 function enabledMessageListenerChatIds(bot: BotState): string[] {
   return Object.entries(bot.config.messageListeners ?? {})
@@ -2198,10 +2242,14 @@ function larkReceiveEventFromHistoryMessage(message: any, chatId: string): any {
   const isBotSenderType = senderIdType === 'app_id' || senderTypeRaw === 'app' || senderTypeRaw === 'bot';
   const isOpenIdDomain = senderIdType === 'open_id'
     || (typeof senderOpenId === 'string' && senderOpenId.startsWith('ou_'));
+  const content = typeof message?.content === 'string'
+    ? message.content
+    : typeof message?.body?.content === 'string' ? message.body.content : '';
   return {
     message: {
       ...message,
       message_type: message?.message_type ?? message?.msg_type,
+      content,
       chat_id: message?.chat_id ?? chatId,
       chat_type: message?.chat_type ?? 'group',
     },
@@ -2264,19 +2312,26 @@ async function pollMessageListenersOnce(larkAppId: string, handlers: EventHandle
   if (chatIds.length === 0) return;
 
   const cutoff = now - MESSAGE_LISTENER_BACKFILL_WINDOW_MS;
-  await ensureBotOpenId(larkAppId).catch(() => { /* degrade; heartbeat retries */ });
+  await withMessageListenerTimeout(ensureBotOpenId(larkAppId), 10_000, 'ensure bot open_id')
+    .catch(err => logger.warn(`[message-listener:${larkAppId}] bot open_id probe skipped: ${err instanceof Error ? err.message : String(err)}`));
 
   for (const chatId of chatIds) {
     let messages: any[];
     try {
-      messages = await listChatMessagesUntil(larkAppId, chatId, {
-        pageSize: MESSAGE_LISTENER_BACKFILL_PAGE_SIZE,
-        stopAfter: (message, seenCount) => {
-          const createdAt = messageCreateTimeMs(message);
-          return seenCount >= MESSAGE_LISTENER_BACKFILL_SCAN_LIMIT ||
-            (Number.isFinite(createdAt) && (createdAt as number) < cutoff);
-        },
-      });
+      messages = await withAbortableMessageListenerTimeout(
+        signal => listChatMessagesUntil(larkAppId, chatId, {
+          pageSize: MESSAGE_LISTENER_BACKFILL_PAGE_SIZE,
+          timeoutMs: MESSAGE_LISTENER_BACKFILL_TIMEOUT_MS,
+          signal,
+          stopAfter: (message, seenCount) => {
+            const createdAt = messageCreateTimeMs(message);
+            return seenCount >= MESSAGE_LISTENER_BACKFILL_SCAN_LIMIT ||
+              (Number.isFinite(createdAt) && (createdAt as number) < cutoff);
+          },
+        }),
+        MESSAGE_LISTENER_BACKFILL_TIMEOUT_MS,
+        `poll chat ${chatId.substring(0, 12)}`,
+      );
     } catch (err) {
       logger.warn(`[message-listener:${larkAppId}] failed to poll chat=${chatId.substring(0, 12)}: ${err instanceof Error ? err.message : String(err)}`);
       continue;
@@ -2324,6 +2379,14 @@ export async function __pollMessageListenersOnceForTest(larkAppId: string, handl
   await pollMessageListenersOnce(larkAppId, handlers, now);
 }
 
+function runMessageListenerPoll(larkAppId: string, handlers: EventHandlers, label: string): Promise<void> {
+  return withMessageListenerTimeout(
+    pollMessageListenersOnce(larkAppId, handlers),
+    MESSAGE_LISTENER_POLL_RUN_TIMEOUT_MS,
+    `${label} message-listener poll`,
+  );
+}
+
 function usesForwardFollowupDelay(mentionMode: GroupMentionMode): boolean {
   return mentionMode === 'never' || mentionMode === 'ambient';
 }
@@ -4174,7 +4237,7 @@ export function startLarkEventDispatcher(larkAppId: string, larkAppSecret: strin
   const listenerPollTimer = setInterval(() => {
     if (listenerPollInFlight) return;
     listenerPollInFlight = true;
-    void pollMessageListenersOnce(larkAppId, handlers)
+    void runMessageListenerPoll(larkAppId, handlers, 'scheduled')
       .catch(err => logger.error(`[message-listener:${larkAppId}] poll failed: ${err instanceof Error ? err.message : String(err)}`))
       .finally(() => { listenerPollInFlight = false; });
   }, MESSAGE_LISTENER_POLL_INTERVAL_MS);
@@ -4184,7 +4247,7 @@ export function startLarkEventDispatcher(larkAppId: string, larkAppSecret: strin
     setTimeout(() => {
       if (listenerPollInFlight) return;
       listenerPollInFlight = true;
-      void pollMessageListenersOnce(larkAppId, handlers)
+      void runMessageListenerPoll(larkAppId, handlers, 'initial')
         .catch(err => logger.error(`[message-listener:${larkAppId}] initial poll failed: ${err instanceof Error ? err.message : String(err)}`))
         .finally(() => { listenerPollInFlight = false; });
     }, 2_000).unref();
diff --git a/src/im/lark/message-parser.ts b/src/im/lark/message-parser.ts
index 6450a4984e..d5eedfd3fc 100644
--- a/src/im/lark/message-parser.ts
+++ b/src/im/lark/message-parser.ts
@@ -25,7 +25,8 @@ interface RawEventData {
     thread_id?: string;
     parent_id?: string;
     message_type: string; // NOT msg_type
-    content: string;
+    content?: string;
+    body?: { content?: string };
     chat_id: string;
     chat_type: string;
     create_time: string;
@@ -123,6 +124,11 @@ export function mentionIdType(m: any): string | undefined {
   return undefined;
 }
 
+function rawEventMessageContent(message: RawEventData['message']): string {
+  const raw = message.content ?? message.body?.content;
+  return typeof raw === 'string' ? raw : '';
+}
+
 /**
  * Extract a bot mention's app_id across ALL shapes Lark has been observed to
  * use. app_id-form bot mentions must never flow through mentionOpenId() (which
@@ -335,7 +341,7 @@ export function unwrapUserDslContent(rawContent: string): string | null {
 }
 
 function unwrapUserDsl(data: RawEventData): boolean {
-  const unwrapped = unwrapUserDslContent(data.message.content);
+  const unwrapped = unwrapUserDslContent(rawEventMessageContent(data.message));
   if (unwrapped === null) return false;
   data.message.content = unwrapped;
   logger.info(`[parser] Unwrapped user_dsl for ${data.message.message_id}`);
@@ -497,12 +503,14 @@ export function parseEventMessage(
   numberer: ImgNumberer = createImgNumberer(),
 ): { parsed: LarkMessage; resources: MessageResource[] } {
   const { sender, message } = data;
+  const rawContent = rawEventMessageContent(message);
+  const normalizedContent = normalizeApiMessageContent(message.message_type, rawContent);
 
   // Trace non-text messages at debug only (DEBUG=1 gated). The raw card/post
   // content can be many KB and may include attachment metadata that's
   // noisy or sensitive — truncate to ~500 chars so DEBUG logs stay scannable.
   if (message.message_type !== 'text' && logger.isDebug()) {
-    const raw = message.content ?? '';
+    const raw = rawContent;
     const trimmed = raw.length > 500 ? raw.slice(0, 500) + `…(+${raw.length - 500}b)` : raw;
     logger.debug(`[parser] type=${message.message_type} content=${trimmed} keys=${Object.keys(message).join(',')}`);
   }
@@ -510,7 +518,7 @@ export function parseEventMessage(
   // Share numberer so in-body [图片 N] placeholders use the same numbers as
   // the attachment list. Resources first → numbers assigned; text second →
   // reuses them.
-  const resources = extractResources(message.message_type, message.content, numberer);
+  const resources = extractResources(message.message_type, rawContent, numberer);
 
   // Extract structured mentions
   const mentions: LarkMention[] | undefined =
@@ -535,7 +543,7 @@ export function parseEventMessage(
     senderUnionId: sender.sender_id?.union_id,
     senderType: sender.sender_type,
     msgType: message.message_type,
-    content: extractTextContent(message.message_type, message.content, message.mentions, numberer),
+    content: extractTextContent(message.message_type, normalizedContent, message.mentions, numberer),
     createTime: message.create_time,
     mentions,
   };
diff --git a/src/services/message-listener-session-cleanup.ts b/src/services/message-listener-session-cleanup.ts
new file mode 100644
index 0000000000..c49948f44f
--- /dev/null
+++ b/src/services/message-listener-session-cleanup.ts
@@ -0,0 +1,73 @@
+import type { MessageListenerConfig } from '../bot-registry.js';
+import type { Session } from '../types.js';
+
+export const DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS = 168;
+export const MESSAGE_LISTENER_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
+
+export interface MessageListenerCleanupConfig {
+  enabled: boolean;
+  retentionHours: number;
+}
+
+export function normalizeMessageListenerCleanupConfig(raw: unknown): MessageListenerCleanupConfig {
+  const entry = raw && typeof raw === 'object' && !Array.isArray(raw)
+    ? raw as Record<string, unknown>
+    : {};
+  const enabled = entry.enabled === false ? false : true;
+  const retentionHours = normalizePositiveHour(entry.retentionHours)
+    ?? DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS;
+  return { enabled, retentionHours };
+}
+
+function normalizePositiveHour(raw: unknown): number | undefined {
+  const value = typeof raw === 'string' && raw.trim()
+    ? Number(raw.trim())
+    : raw;
+  if (typeof value !== 'number' || !Number.isFinite(value)) return undefined;
+  const normalized = Math.floor(value);
+  return normalized > 0 ? normalized : undefined;
+}
+
+function sessionText(session: Session): string {
+  return [
+    session.title,
+    session.nativeSessionTitle,
+    session.lastUserPrompt,
+    session.lastCliInput,
+  ].filter((value): value is string => typeof value === 'string').join('\n');
+}
+
+export function isMessageListenerSession(session: Session): boolean {
+  if (session.messageListener) return true;
+  return sessionText(session).includes('<message_listener>');
+}
+
+function activityTimeMs(session: Session): number | undefined {
+  const candidates = [session.lastMessageAt, session.createdAt];
+  for (const value of candidates) {
+    if (!value) continue;
+    const ms = Date.parse(value);
+    if (Number.isFinite(ms)) return ms;
+  }
+  return undefined;
+}
+
+export function selectExpiredMessageListenerSessions(input: {
+  sessions: Session[];
+  listeners: Record<string, MessageListenerConfig> | undefined;
+  nowMs?: number;
+}): Session[] {
+  const nowMs = input.nowMs ?? Date.now();
+  const listeners = input.listeners ?? {};
+  return input.sessions.filter(session => {
+    if (session.status !== 'active') return false;
+    if (!isMessageListenerSession(session)) return false;
+    const listener = listeners[session.messageListener?.chatId ?? session.chatId];
+    if (!listener?.enabled) return false;
+    const cleanup = normalizeMessageListenerCleanupConfig(listener.cleanup);
+    if (!cleanup.enabled) return false;
+    const lastActivityMs = activityTimeMs(session);
+    if (lastActivityMs === undefined) return false;
+    return nowMs - lastActivityMs >= cleanup.retentionHours * 60 * 60 * 1000;
+  });
+}
diff --git a/src/services/message-listener-store.ts b/src/services/message-listener-store.ts
index 43c9371963..528adab387 100644
--- a/src/services/message-listener-store.ts
+++ b/src/services/message-listener-store.ts
@@ -1,5 +1,11 @@
 import { getBot, type MessageListenerConfig } from '../bot-registry.js';
 import { rmwBotEntry } from './config-store.js';
+import {
+  DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS,
+  normalizeMessageListenerCleanupConfig,
+} from './message-listener-session-cleanup.js';
+
+export { DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS };
 
 export type MessageListenerUpdate = {
   enabled: boolean;
@@ -10,6 +16,7 @@ export type MessageListenerUpdate = {
   senderPolicy?: MessageListenerConfig['senderPolicy'];
   messagePolicy?: MessageListenerConfig['messagePolicy'];
   contentPolicy?: MessageListenerConfig['contentPolicy'];
+  cleanup?: MessageListenerConfig['cleanup'];
 };
 
 function stringList(raw: unknown): string[] | undefined {
@@ -77,6 +84,7 @@ export function sanitizeMessageListenerUpdate(raw: unknown): MessageListenerUpda
   const messagePolicy: MessageListenerConfig['messagePolicy'] = { scope: 'top_level' };
   const includeMsgTypes = stringList(rawMessage.includeMsgTypes);
   if (includeMsgTypes) messagePolicy.includeMsgTypes = includeMsgTypes;
+  const cleanup = normalizeMessageListenerCleanupConfig(entry.cleanup);
 
   const rawContent = entry.contentPolicy && typeof entry.contentPolicy === 'object' && !Array.isArray(entry.contentPolicy)
     ? entry.contentPolicy as Record<string, unknown>
@@ -103,6 +111,7 @@ export function sanitizeMessageListenerUpdate(raw: unknown): MessageListenerUpda
     ...(Object.keys(senderPolicy).length > 0 ? { senderPolicy } : {}),
     messagePolicy,
     ...(contentPolicy ? { contentPolicy } : {}),
+    cleanup,
   };
 }
 
@@ -153,6 +162,7 @@ export function messageListenerConfigFromUpdate(patch: MessageListenerUpdate): M
     ...(patch.senderPolicy && Object.keys(patch.senderPolicy).length > 0 ? { senderPolicy: patch.senderPolicy } : {}),
     ...(patch.messagePolicy ? { messagePolicy: { ...patch.messagePolicy, scope: 'top_level' } } : { messagePolicy: { scope: 'top_level' } }),
     ...(patch.contentPolicy ? { contentPolicy: patch.contentPolicy } : {}),
+    cleanup: normalizeMessageListenerCleanupConfig(patch.cleanup),
     replyPolicy: { mode: 'thread', sessionMode: 'per_message' },
   };
 }
@@ -166,7 +176,7 @@ export async function updateMessageListenerConfig(
   try { bot = getBot(larkAppId); } catch { return { ok: false, reason: 'bot_not_registered' }; }
 
   const validation = validateMessageListenerUpdate(patch);
-  if (!validation.ok) return { ok: false, reason: validation.reason };
+  if (validation.ok === false) return { ok: false, reason: validation.reason };
 
   // A disabled update with a non-empty prompt persists as an off DRAFT (kept for
   // the editor, never matched at runtime); only a blank-prompt disabled update
@@ -187,7 +197,7 @@ export async function updateMessageListenerConfig(
     entry.messageListeners[chatId] = normalized;
     return { write: true, result: normalized };
   });
-  if (!result.ok) return { ok: false, reason: result.reason };
+  if (result.ok === false) return { ok: false, reason: result.reason };
 
   if (!bot.config.messageListeners) bot.config.messageListeners = {};
   if (normalized) bot.config.messageListeners[chatId] = normalized;
diff --git a/src/services/session-store.ts b/src/services/session-store.ts
index 704d309fbb..bb748a11af 100644
--- a/src/services/session-store.ts
+++ b/src/services/session-store.ts
@@ -1035,6 +1035,39 @@ export function reactivateClosedSession(
   return { ok: true, session };
 }
 
+export function closeSessionsMatching(
+  predicate: (session: Session) => boolean,
+  opts: { cleanupBridgeMarkers?: boolean; closedAt?: string } = {},
+): number {
+  load();
+  const closedAt = opts.closedAt ?? new Date().toISOString();
+  const closedSessionIds: string[] = [];
+  for (const [sessionId, session] of sessions) {
+    if (session.status !== 'active') continue;
+    if (!predicate(session)) continue;
+    if (session.larkAppId && session.dashboardAttachments?.length) {
+      try {
+        cleanupMaterializedDashboardImages(session.larkAppId, session.dashboardAttachments);
+        session.dashboardAttachments = undefined;
+        session.queuedAttachments = undefined;
+      } catch (error: any) {
+        logger.warn(`Failed to clean Dashboard images for session ${sessionId}: ${error?.message ?? error}`);
+      }
+    }
+    session.status = 'closed';
+    session.closedAt = closedAt;
+    closedSessionIds.push(sessionId);
+  }
+  if (closedSessionIds.length === 0) return 0;
+  save();
+  if (opts.cleanupBridgeMarkers !== false) {
+    for (const sessionId of closedSessionIds) cleanupSessionBridgeSendMarkers(sessionId);
+  }
+  for (const sessionId of closedSessionIds) deleteFrozenCards(sessionId);
+  logger.info(`Closed ${closedSessionIds.length} sessions in batch`);
+  return closedSessionIds.length;
+}
+
 export function updateSessionPid(sessionId: string, pid: number | null): void {
   loadForWrite();
   const session = sessions.get(sessionId);
diff --git a/src/types.ts b/src/types.ts
index b4e96b7f24..e3adb1cbea 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -415,6 +415,10 @@ export interface Session {
    *  绝不扫描子目录乱带仓库。任何其它选仓路径都会清除本 stamp。 */
   riffRepoDirs?: string[];
   larkAppId?: string;
+  /** Created by a configured message listener; used by retention cleanup. */
+  messageListener?: {
+    chatId: string;
+  };
   /** Daemon-selected, app-scoped session owner. Frozen for the worker lifetime;
    *  not the current-turn sender. Absent for ownerless/foreign-bot sessions. */
   ownerOpenId?: string;       // receives owner-only replies/mentions
diff --git a/test/bot-registry.test.ts b/test/bot-registry.test.ts
index 898bf281d5..f78d2ed6ec 100644
--- a/test/bot-registry.test.ts
+++ b/test/bot-registry.test.ts
@@ -459,6 +459,7 @@ describe('parseBotConfigsFromText — brand', () => {
         scope: 'top_level',
       },
       replyPolicy: { mode: 'thread', sessionMode: 'per_message' },
+      cleanup: { enabled: true, retentionHours: 168 },
     });
     expect(cfg.messageListeners?.oc_disabled.enabled).toBe(false);
     expect(cfg.messageListeners?.oc_bad).toBeUndefined();
diff --git a/test/dashboard-attention-signals.test.ts b/test/dashboard-attention-signals.test.ts
index ef70694635..c44f0c140f 100644
--- a/test/dashboard-attention-signals.test.ts
+++ b/test/dashboard-attention-signals.test.ts
@@ -125,6 +125,14 @@ describe('attention signals', () => {
     expect(composeRowFromActive(queued).status).toBe('idle');
   });
 
+  it('composeRowFromActive can skip restore-time expensive enrichment', () => {
+    const row = composeRowFromActive(makeDs(), { lightweight: true });
+
+    expect(row.tokenUsage).toBeUndefined();
+    expect(row.previewUserText).toBeUndefined();
+    expect(row.previewBotText).toBeUndefined();
+  });
+
   it('composeRowFromActive carries the agent raise-hand signal with its reason', () => {
     const raised = composeRowFromActive(makeDs({
       agentAttention: { kind: 'authz', reason: '需要 prod 部署授权', at: 1234 },
diff --git a/test/dashboard-row-lightweight-source.test.ts b/test/dashboard-row-lightweight-source.test.ts
new file mode 100644
index 0000000000..4014f1573d
--- /dev/null
+++ b/test/dashboard-row-lightweight-source.test.ts
@@ -0,0 +1,33 @@
+import { readFileSync } from 'node:fs';
+import { resolve } from 'node:path';
+import { describe, expect, it } from 'vitest';
+
+const source = readFileSync(resolve(process.cwd(), 'src/core/dashboard-ipc-server.ts'), 'utf8');
+const rowsSource = readFileSync(resolve(process.cwd(), 'src/core/dashboard-rows.ts'), 'utf8');
+
+describe('dashboard bulk session snapshots', () => {
+  it('uses lightweight rows for /api/sessions snapshots', () => {
+    expect(source).toContain('const DASHBOARD_SNAPSHOT_ROW_OPTS = { lightweight: true } as const;');
+    expect(source).toContain('composeRowFromActive(ds, DASHBOARD_SNAPSHOT_ROW_OPTS)');
+    expect(source).toContain('composeRowFromPersistedActive(session, DASHBOARD_SNAPSHOT_ROW_OPTS)');
+    expect(source).toContain('composeRowFromClosed(session, DASHBOARD_SNAPSHOT_ROW_OPTS)');
+  });
+
+  it('uses lightweight rows for /api/events replay snapshots', () => {
+    expect(source).toContain('composeRowFromActive(ds, DASHBOARD_SNAPSHOT_ROW_OPTS) })}\\n\\n`);');
+    expect(source).toContain('composeRowFromPersistedActive(s, DASHBOARD_SNAPSHOT_ROW_OPTS) })}\\n\\n`);');
+    expect(source).toContain('composeRowFromClosed(s, DASHBOARD_SNAPSHOT_ROW_OPTS) })}\\n\\n`);');
+  });
+
+  it('keeps open todo extraction behind the lightweight guard', () => {
+    const start = rowsSource.indexOf('export function composeRowFromActive');
+    const end = rowsSource.indexOf('export function composeRowFromClosed');
+    const activeComposer = rowsSource.slice(start, end);
+    const guard = 'if (!opts.lightweight) {';
+    const beforeGuard = activeComposer.slice(0, activeComposer.indexOf(guard));
+
+    expect(activeComposer).toContain(guard);
+    expect(activeComposer).toContain('row.openTodos = sessionOpenTodos(ds.session, ds.workingDir, opts.fresh);');
+    expect(beforeGuard).not.toContain('sessionOpenTodos(');
+  });
+});
diff --git a/test/event-dispatcher.test.ts b/test/event-dispatcher.test.ts
index 028f74f5cd..66bbc1185f 100644
--- a/test/event-dispatcher.test.ts
+++ b/test/event-dispatcher.test.ts
@@ -1606,7 +1606,7 @@ describe('message listener polling backfill', () => {
     await flushEventWork();
 
     expect(mockListChatMessagesUntil).toHaveBeenCalledWith(MY_APP_ID, 'chat_listener', expect.objectContaining({
-      pageSize: expect.any(Number),
+      pageSize: 50,
       stopAfter: expect.any(Function),
     }));
     expect(handlers.handleNewTopic).toHaveBeenCalledWith(
@@ -1625,6 +1625,30 @@ describe('message listener polling backfill', () => {
     );
   });
 
+  it('passes an abort signal to polled chat-history requests', async () => {
+    const card = makeHistoryMessage({
+      senderAppId: OTHER_BOT_APP_ID,
+      senderType: 'app',
+      messageType: 'interactive',
+      messageId: 'msg-polled-abort-signal',
+      chatId: 'chat_listener',
+      content: JSON.stringify({ title: 'Argos平台报警', elements: [[{ tag: 'text', text: '需要可取消的轮询' }]] }),
+      createTime: String(Date.now()),
+    });
+    let signal: AbortSignal | undefined;
+    mockListChatMessagesUntil.mockImplementationOnce(async (_appId, _chatId, options) => {
+      signal = options?.signal;
+      return [card];
+    });
+
+    await __pollMessageListenersOnceForTest(MY_APP_ID, handlers);
+    await flushEventWork();
+
+    expect(signal).toBeDefined();
+    expect(typeof signal?.addEventListener).toBe('function');
+    expect(signal?.aborted).toBe(false);
+  });
+
   it('does not replay a polled listener message after the message_id is claimed', async () => {
     const card = makeHistoryMessage({
       senderAppId: OTHER_BOT_APP_ID,
@@ -1644,6 +1668,35 @@ describe('message listener polling backfill', () => {
     expect(handlers.handleNewTopic).toHaveBeenCalledTimes(1);
   });
 
+  it('normalizes REST body.content before dispatching a polled listener match', async () => {
+    const cardContent = JSON.stringify({
+      title: 'Argos平台报警',
+      elements: [[{ tag: 'text', text: 'body-only 告警详情' }]],
+    });
+    const card = {
+      message_id: 'msg-polled-body-only',
+      chat_id: 'chat_listener',
+      chat_type: 'group',
+      msg_type: 'interactive',
+      body: { content: cardContent },
+      create_time: String(Date.now()),
+      sender: {
+        id: OTHER_BOT_APP_ID,
+        id_type: 'app_id',
+        sender_type: 'app',
+      },
+    };
+    mockListChatMessagesUntil.mockResolvedValueOnce([card]);
+    handlers.handleNewTopic.mockImplementationOnce(async (data) => {
+      expect((data as any).message.content).toBe(cardContent);
+    });
+
+    await __pollMessageListenersOnceForTest(MY_APP_ID, handlers);
+    await flushEventWork();
+
+    expect(handlers.handleNewTopic).toHaveBeenCalledTimes(1);
+  });
+
   it('resolves a sibling bot app_id to open_id so an open_id include list matches on the polled path', async () => {
     // Realistic config: the sender filter stores the peer bot's OPEN_ID (what the
     // dashboard member picker saves), while chat history reports the bot by app_id.
diff --git a/test/message-listener-session-cleanup.test.ts b/test/message-listener-session-cleanup.test.ts
new file mode 100644
index 0000000000..f86e0cb998
--- /dev/null
+++ b/test/message-listener-session-cleanup.test.ts
@@ -0,0 +1,112 @@
+import { describe, expect, it } from 'vitest';
+import { selectExpiredMessageListenerSessions } from '../src/services/message-listener-session-cleanup.js';
+import type { MessageListenerConfig } from '../src/bot-registry.js';
+import type { Session } from '../src/types.js';
+
+const NOW = Date.UTC(2026, 7, 26, 12, 0, 0);
+const hour = 60 * 60 * 1000;
+
+function session(id: string, patch: Partial<Session> = {}): Session {
+  return {
+    sessionId: id,
+    chatId: 'oc_chat',
+    chatType: 'group',
+    rootMessageId: `om_${id}`,
+    title: '<message_listener> alert',
+    status: 'active',
+    createdAt: new Date(NOW - 200 * hour).toISOString(),
+    lastUserPrompt: '<message_listener> alert',
+    larkAppId: 'cli_listener',
+    scope: 'thread',
+    ...patch,
+  };
+}
+
+describe('message listener session cleanup', () => {
+  it('selects active listener sessions older than the configured retention', () => {
+    const listeners: Record<string, MessageListenerConfig> = {
+      oc_chat: {
+        enabled: true,
+        prompt: 'listen',
+        cleanup: { enabled: true, retentionHours: 168 },
+      },
+    };
+
+    const selected = selectExpiredMessageListenerSessions({
+      sessions: [
+        session('old-listener'),
+        session('fresh-listener', { createdAt: new Date(NOW - 2 * hour).toISOString() }),
+        session('manual-old', { lastUserPrompt: '@bot help', title: '@bot help' }),
+        session('other-chat', { chatId: 'oc_other' }),
+        session('closed-old', { status: 'closed' }),
+      ],
+      listeners,
+      nowMs: NOW,
+    });
+
+    expect(selected.map(s => s.sessionId)).toEqual(['old-listener']);
+  });
+
+  it('uses enabled seven-day cleanup by default and honors disabled cleanup', () => {
+    expect(selectExpiredMessageListenerSessions({
+      sessions: [session('eight-days', { createdAt: new Date(NOW - 8 * 24 * hour).toISOString() })],
+      listeners: { oc_chat: { enabled: true, prompt: 'listen' } },
+      nowMs: NOW,
+    }).map(s => s.sessionId)).toEqual(['eight-days']);
+
+    expect(selectExpiredMessageListenerSessions({
+      sessions: [session('old-but-disabled')],
+      listeners: { oc_chat: { enabled: true, prompt: 'listen', cleanup: { enabled: false, retentionHours: 24 } } },
+      nowMs: NOW,
+    })).toEqual([]);
+  });
+
+  it('bases age on last activity when available', () => {
+    const listeners: Record<string, MessageListenerConfig> = {
+      oc_chat: { enabled: true, prompt: 'listen', cleanup: { enabled: true, retentionHours: 24 } },
+    };
+
+    expect(selectExpiredMessageListenerSessions({
+      sessions: [
+        session('recent-activity', {
+          createdAt: new Date(NOW - 72 * hour).toISOString(),
+          lastMessageAt: new Date(NOW - 2 * hour).toISOString(),
+        }),
+        session('old-activity', {
+          createdAt: new Date(NOW - 72 * hour).toISOString(),
+          lastMessageAt: new Date(NOW - 25 * hour).toISOString(),
+        }),
+      ],
+      listeners,
+      nowMs: NOW,
+    }).map(s => s.sessionId)).toEqual(['old-activity']);
+  });
+
+  it('recognizes explicitly stamped listener sessions without marker text', () => {
+    expect(selectExpiredMessageListenerSessions({
+      sessions: [
+        session('stamped', {
+          title: '告警自动分析',
+          lastUserPrompt: '分析原因',
+          messageListener: { chatId: 'oc_chat' },
+        }),
+      ],
+      listeners: { oc_chat: { enabled: true, prompt: 'listen', cleanup: { enabled: true, retentionHours: 24 } } },
+      nowMs: NOW,
+    }).map(s => s.sessionId)).toEqual(['stamped']);
+  });
+
+  it('recognizes legacy listener sessions renamed by later user activity', () => {
+    expect(selectExpiredMessageListenerSessions({
+      sessions: [
+        session('renamed-legacy-listener', {
+          title: '告警分析',
+          nativeSessionTitle: '<message_listener> alert',
+          lastUserPrompt: '@bot follow up',
+        }),
+      ],
+      listeners: { oc_chat: { enabled: true, prompt: 'listen', cleanup: { enabled: true, retentionHours: 24 } } },
+      nowMs: NOW,
+    }).map(s => s.sessionId)).toEqual(['renamed-legacy-listener']);
+  });
+});
diff --git a/test/message-listener-store.test.ts b/test/message-listener-store.test.ts
index 47cc414c1b..2bc1e7ed77 100644
--- a/test/message-listener-store.test.ts
+++ b/test/message-listener-store.test.ts
@@ -1,5 +1,10 @@
 import { describe, expect, it } from 'vitest';
-import { messageListenerConfigFromUpdate, sanitizeMessageListenerUpdate, validateMessageListenerUpdate } from '../src/services/message-listener-store.js';
+import {
+  DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS,
+  messageListenerConfigFromUpdate,
+  sanitizeMessageListenerUpdate,
+  validateMessageListenerUpdate,
+} from '../src/services/message-listener-store.js';
 
 describe('message listener store', () => {
   it('keeps the custom reply card title from dashboard updates', () => {
@@ -30,6 +35,56 @@ describe('message listener store', () => {
     })).not.toHaveProperty('replyCardTitle');
   });
 
+  it('defaults listener session cleanup to enabled with a seven-day retention', () => {
+    expect(sanitizeMessageListenerUpdate({
+      enabled: true,
+      prompt: '分析命中的告警消息',
+    })?.cleanup).toEqual({
+      enabled: true,
+      retentionHours: DEFAULT_MESSAGE_LISTENER_CLEANUP_RETENTION_HOURS,
+    });
+  });
+
+  it('accepts dashboard listener cleanup overrides', () => {
+    expect(sanitizeMessageListenerUpdate({
+      enabled: true,
+      prompt: '分析命中的告警消息',
+      cleanup: {
+        enabled: false,
+        retentionHours: '24',
+      },
+    })?.cleanup).toEqual({
+      enabled: false,
+      retentionHours: 24,
+    });
+  });
+
+  it('preserves selected sender whitelist when saving listener cleanup settings', () => {
+    expect(sanitizeMessageListenerUpdate({
+      enabled: true,
+      prompt: '分析命中的告警消息',
+      senderPolicy: {
+        mode: 'include_only',
+        includeSenderOpenIds: ['ou_argos', 'ou_locator'],
+        includeSenderTypes: ['bot'],
+      },
+      cleanup: {
+        enabled: true,
+        retentionHours: 24,
+      },
+    })).toMatchObject({
+      senderPolicy: {
+        mode: 'include_only',
+        includeSenderOpenIds: ['ou_argos', 'ou_locator'],
+        includeSenderTypes: ['bot'],
+      },
+      cleanup: {
+        enabled: true,
+        retentionHours: 24,
+      },
+    });
+  });
+
   it('rejects enabled include-only listeners without selected senders', () => {
     const update = sanitizeMessageListenerUpdate({
       enabled: true,
diff --git a/test/message-parser.test.ts b/test/message-parser.test.ts
index dee5d1c6d4..2230ba594c 100644
--- a/test/message-parser.test.ts
+++ b/test/message-parser.test.ts
@@ -59,6 +59,32 @@ describe('parseApiMessage metadata', () => {
   });
 });
 
+describe('parseEventMessage compatibility', () => {
+  it('parses REST history messages that carry content in body.content', () => {
+    const result = parseEventMessage({
+      sender: {
+        sender_type: 'app',
+        sender_id: { app_id: 'cli_argos' },
+      },
+      message: {
+        message_id: 'om_body_only',
+        chat_id: 'oc_alerts',
+        chat_type: 'group',
+        message_type: 'interactive',
+        create_time: '1000',
+        body: {
+          content: JSON.stringify({
+            title: 'Argos平台报警',
+            elements: [[{ tag: 'text', text: 'body-only 告警详情' }]],
+          }),
+        },
+      },
+    } as any);
+
+    expect(result.parsed.content).toContain('body-only 告警详情');
+  });
+});
+
 // ─── Interactive card: Format A (Lark API simplified) ─────────────────────
 
 describe('Interactive card parsing: Format A (API simplified)', () => {
diff --git a/test/session-manager-auto-recover.test.ts b/test/session-manager-auto-recover.test.ts
index 055f4d6022..7f865a2893 100644
--- a/test/session-manager-auto-recover.test.ts
+++ b/test/session-manager-auto-recover.test.ts
@@ -1,12 +1,10 @@
 /**
  * Auto-recovery on daemon restart.
  *
- * On restart every surviving persistent-backend session is eagerly re-forked to
- * re-attach its pane, so the session actually comes back instead of sitting dead
- * until its next message (and a pane whose CLI died gets healed, keeping the
- * transcript fallback working). The old `BOTMUX_QUIET_RESTART` gate that
- * suppressed this is gone — card silence is now handled by `suppressRecoveryCard`
- * on restored sessions, not by skipping recovery.
+ * On restart surviving persistent-backend sessions are kept lazy by default:
+ * restoring thousands of active rows must not starve message listeners by
+ * eagerly re-forking every worker. Operators can explicitly opt in to eager
+ * re-attach when terminal readiness is more important than startup latency.
  *
  * `staggeredRecoveryFork` spaces the re-forks out (batch + delay) so a box with
  * dozens of surviving sessions doesn't spike on restart, and skips any session
@@ -21,24 +19,38 @@ vi.mock('../src/bot-registry.js', () => ({
 
 vi.mock('../src/config.js', () => ({
   config: {
-    daemon: { workingDir: '~', workingDirs: ['~'], recoveryForkBatchSize: 5, recoveryForkDelayMs: 0 },
+    daemon: {
+      workingDir: '~',
+      workingDirs: ['~'],
+      recoveryForkBatchSize: 5,
+      recoveryForkDelayMs: 0,
+      recoveryForkEnabled: false,
+    },
     session: { dataDir: '/tmp/botmux-test' },
   },
 }));
 
-import { shouldAutoForkOnRestore, staggeredRecoveryFork } from '../src/core/session-manager.js';
+import { shouldAutoForkOnRestore, staggeredRecoveryFork, scheduleStaggeredRecoveryFork } from '../src/core/session-manager.js';
 import type { DaemonSession } from '../src/core/types.js';
 
 describe('shouldAutoForkOnRestore', () => {
-  it('eagerly re-forks every persistent backend (tmux/herdr/zellij/zmx)', () => {
-    expect(shouldAutoForkOnRestore('tmux')).toBe(true);
-    expect(shouldAutoForkOnRestore('herdr')).toBe(true);
-    expect(shouldAutoForkOnRestore('zellij')).toBe(true);
-    expect(shouldAutoForkOnRestore('zmx')).toBe(true);
+  it('keeps persistent backends lazy by default so startup does not starve listeners', () => {
+    expect(shouldAutoForkOnRestore('tmux')).toBe(false);
+    expect(shouldAutoForkOnRestore('herdr')).toBe(false);
+    expect(shouldAutoForkOnRestore('zellij')).toBe(false);
+    expect(shouldAutoForkOnRestore('zmx')).toBe(false);
+  });
+
+  it('can explicitly eager re-attach persistent backends', () => {
+    expect(shouldAutoForkOnRestore('tmux', true)).toBe(true);
+    expect(shouldAutoForkOnRestore('herdr', true)).toBe(true);
+    expect(shouldAutoForkOnRestore('zellij', true)).toBe(true);
+    expect(shouldAutoForkOnRestore('zmx', true)).toBe(true);
   });
 
   it('never eagerly forks the pty backend — it has no pane to re-attach', () => {
     expect(shouldAutoForkOnRestore('pty')).toBe(false);
+    expect(shouldAutoForkOnRestore('pty', true)).toBe(false);
   });
 });
 
@@ -112,4 +124,24 @@ describe('staggeredRecoveryFork', () => {
 
     expect(forked).toEqual(['a']);
   });
+
+  it('can be scheduled without blocking restore on delayed batches', async () => {
+    vi.useFakeTimers();
+    try {
+      const sessions = Array.from({ length: 5 }, (_, i) => ds(`s${i}`));
+      const forked: string[] = [];
+
+      scheduleStaggeredRecoveryFork(sessions, (d) => forked.push(d.session.sessionId), 2, 20);
+
+      expect(forked).toEqual(['s0', 's1']);
+      await vi.advanceTimersByTimeAsync(19);
+      expect(forked).toEqual(['s0', 's1']);
+      await vi.advanceTimersByTimeAsync(1);
+      expect(forked).toEqual(['s0', 's1', 's2', 's3']);
+      await vi.advanceTimersByTimeAsync(20);
+      expect(forked).toEqual(['s0', 's1', 's2', 's3', 's4']);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
 });
diff --git a/test/session-store.test.ts b/test/session-store.test.ts
index 31f8f9e7cc..cf490818b4 100644
--- a/test/session-store.test.ts
+++ b/test/session-store.test.ts
@@ -74,6 +74,7 @@ import {
   finishMojoCloseAbort,
   closeSession,
   reactivateClosedSession,
+  closeSessionsMatching,
   updateSession,
   updateSessionPid,
   persistActiveRemoteLineageExact,
@@ -903,6 +904,35 @@ describe('reactivateClosedSession()', () => {
   });
 });
 
+// ─── closeSessionsMatching() ──────────────────────────────────────────────
+
+describe('closeSessionsMatching()', () => {
+  it('closes matching active sessions in one persisted batch', () => {
+    const keep = createSession('chat1', 'root1', 'Keep');
+    const closeA = createSession('chat1', 'root2', 'Close A');
+    const closeB = createSession('chat1', 'root3', 'Close B');
+    const alreadyClosed = createSession('chat1', 'root4', 'Already Closed');
+    closeSession(alreadyClosed.sessionId);
+
+    const closedAt = '2026-08-26T04:00:00.000Z';
+    const count = closeSessionsMatching(
+      session => session.title.startsWith('Close') || session.title === 'Already Closed',
+      { closedAt },
+    );
+
+    expect(count).toBe(2);
+    expect(getSession(keep.sessionId)!.status).toBe('active');
+    expect(getSession(closeA.sessionId)!.status).toBe('closed');
+    expect(getSession(closeA.sessionId)!.closedAt).toBe(closedAt);
+    expect(getSession(closeB.sessionId)!.status).toBe('closed');
+    expect(getSession(closeB.sessionId)!.closedAt).toBe(closedAt);
+
+    init();
+    expect(getSession(closeA.sessionId)!.status).toBe('closed');
+    expect(getSession(closeB.sessionId)!.closedAt).toBe(closedAt);
+  });
+});
+
 // ─── updateSession() ─────────────────────────────────────────────────────
 
 describe('updateSession()', () => {