fix(listener): prevent daemon stalls during group polling

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,
   };
 }
 

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(),

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,15 +5596,15 @@ 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
     // upserts so SSE reconnects retain the same truthful state as GET
     // /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}`);

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;
 }

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) },
   });
 }
 

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'}`);