fix(worker): 避免慢启动误报消息未接收

src/core/worker-pool.ts

@@ -6292,7 +6292,11 @@ const transferInputGates = new WeakMap<DaemonSession, TransferInputGate>();
 // cannot forge an option that bypasses the transfer gate.
 const transferReplacementForkBypass = new WeakSet<DaemonSession>();
 
-const ORDINARY_IM_RECEIPT_TIMEOUT_MS = 2_000;
+// IPC transport and worker acknowledgement are separate stages. A transport
+// timeout may retry because the parent never confirmed enqueue; an ACK timeout
+// is only a delayed/ambiguous state because the child may still execute later.
+const ORDINARY_IM_TRANSPORT_TIMEOUT_MS = 2_000;
+const ORDINARY_IM_ACK_SETTLEMENT_TIMEOUT_MS = 2_000;
 const ORDINARY_IM_MAX_ATTEMPTS = 2;
 
 type OrdinaryImDelivery = {
@@ -6303,6 +6307,9 @@ type OrdinaryImDelivery = {
   message: Extract<DaemonToWorker, { type: 'message' | 'init' }>;
   turnId: string;
   attempt: number;
+  received: boolean;
+  transportConfirmed: boolean;
+  delayNotified: boolean;
   timer?: ReturnType<typeof setTimeout>;
 };
 
@@ -6328,7 +6335,12 @@ function clearOrdinaryImDeliveryTimer(record: OrdinaryImDelivery): void {
   record.timer = undefined;
 }
 
-function failOrdinaryImDelivery(record: OrdinaryImDelivery, reason: string): void {
+function failOrdinaryImDelivery(
+  record: OrdinaryImDelivery,
+  reason: string,
+  messageKey: 'worker.input_delivery_failed' | 'worker.input_retired_unconfirmed'
+    = 'worker.input_delivery_failed',
+): void {
   if (pendingOrdinaryImDeliveries.get(record.key) !== record) return;
   clearOrdinaryImDelivery(record);
   logger.error(
@@ -6355,7 +6367,7 @@ function failOrdinaryImDelivery(record: OrdinaryImDelivery, reason: string): voi
   const loc = botLocale(getBot(record.ds.larkAppId).config);
   void requireCallbacks().sessionReply(
     sessionAnchorId(record.ds),
-    tr('worker.input_delivery_failed', { turnId: record.turnId.substring(0, 16) }, loc),
+    tr(messageKey, { turnId: record.turnId.substring(0, 16) }, loc),
     'text',
     record.ds.larkAppId,
     record.turnId,
@@ -6365,6 +6377,41 @@ function failOrdinaryImDelivery(record: OrdinaryImDelivery, reason: string): voi
   ));
 }
 
+function delayOrdinaryImDelivery(record: OrdinaryImDelivery): void {
+  if (pendingOrdinaryImDeliveries.get(record.key) !== record) return;
+  // A delayed notice is only an intermediate status. Keep the delivery record
+  // so a later explicit rejection or worker exit can still produce the real
+  // terminal outcome instead of silently dropping the turn after telling the
+  // user not to resend it.
+  clearOrdinaryImDeliveryTimer(record);
+  if (record.delayNotified) return;
+  record.delayNotified = true;
+  logger.warn(
+    `[${tag(record.ds)}] Ordinary IM input is still waiting for the worker after IPC enqueue `
+    + `turn=${record.turnId.substring(0, 16)} generation=${record.workerGeneration} `
+    + `attempt=${record.attempt}`,
+  );
+  if (
+    record.turnId.startsWith('bmx-recovery-')
+    || isMeetingDrivenTurn(record.ds, record.turnId)
+    || isSilentScheduledTurn(record.ds, record.turnId)
+  ) return;
+  const loc = botLocale(getBot(record.ds.larkAppId).config);
+  const messageKey = record.received
+    ? 'worker.input_commit_delayed'
+    : 'worker.input_delivery_delayed';
+  void requireCallbacks().sessionReply(
+    sessionAnchorId(record.ds),
+    tr(messageKey, { turnId: record.turnId.substring(0, 16) }, loc),
+    'text',
+    record.ds.larkAppId,
+    record.turnId,
+  ).catch(err => logger.error(
+    `[${tag(record.ds)}] Failed to report delayed ordinary IM worker delivery: `
+    + `${err instanceof Error ? err.message : String(err)}`,
+  ));
+}
+
 function retryOrFailOrdinaryImDelivery(record: OrdinaryImDelivery, reason: string): void {
   if (pendingOrdinaryImDeliveries.get(record.key) !== record) return;
   if (
@@ -6399,18 +6446,31 @@ function sendOrdinaryImDeliveryAttempt(record: OrdinaryImDelivery): boolean {
 
   if (record.timer) clearTimeout(record.timer);
   record.timer = undefined;
+  record.received = false;
+  record.transportConfirmed = false;
   const attempt = ++record.attempt;
+  record.timer = setTimeout(() => {
+    retryOrFailOrdinaryImDelivery(record, 'ipc_callback_timeout');
+  }, ORDINARY_IM_TRANSPORT_TIMEOUT_MS);
+  record.timer.unref?.();
   try {
     record.worker.send(record.message, (err) => {
       if (pendingOrdinaryImDeliveries.get(record.key) !== record || record.attempt !== attempt) return;
+      if (record.received) return;
       if (err) {
         retryOrFailOrdinaryImDelivery(record, `ipc_callback:${err.message}`);
         return;
       }
+      record.transportConfirmed = true;
+      clearOrdinaryImDeliveryTimer(record);
       logger.info(
         `[${tag(record.ds)}] Ordinary IM input enqueued to worker IPC `
         + `turn=${record.turnId.substring(0, 16)} generation=${record.workerGeneration} attempt=${attempt}`,
       );
+      record.timer = setTimeout(() => {
+        delayOrdinaryImDelivery(record);
+      }, ORDINARY_IM_ACK_SETTLEMENT_TIMEOUT_MS);
+      record.timer.unref?.();
     });
   } catch (err) {
     queueMicrotask(() => retryOrFailOrdinaryImDelivery(
@@ -6419,16 +6479,6 @@ function sendOrdinaryImDeliveryAttempt(record: OrdinaryImDelivery): boolean {
     ));
     return true;
   }
-
-  // The worker ACKs synchronously when its IPC handler claims the exact turn.
-  // Slow CLI startup/processing therefore does not extend this transport-only
-  // timeout; the later committed ACK retains input-queue semantics.
-  if (pendingOrdinaryImDeliveries.get(record.key) === record) {
-    record.timer = setTimeout(() => {
-      retryOrFailOrdinaryImDelivery(record, 'receipt_timeout');
-    }, ORDINARY_IM_RECEIPT_TIMEOUT_MS);
-    record.timer.unref?.();
-  }
   return true;
 }
 
@@ -6452,6 +6502,9 @@ function sendOrdinaryImDeliveryTracked(
     message,
     turnId,
     attempt: 0,
+    received: false,
+    transportConfirmed: false,
+    delayNotified: false,
   };
   pendingOrdinaryImDeliveries.set(key, record);
   return sendOrdinaryImDeliveryAttempt(record);
@@ -6485,7 +6538,14 @@ function acknowledgeOrdinaryImDeliveryReceipt(
   const key = ordinaryImDeliveryKey(ds, turnId, workerGeneration);
   const record = pendingOrdinaryImDeliveries.get(key);
   if (!record) return;
-  clearOrdinaryImDeliveryTimer(record);
+  if (!record.received) {
+    record.received = true;
+    clearOrdinaryImDeliveryTimer(record);
+    record.timer = setTimeout(() => {
+      delayOrdinaryImDelivery(record);
+    }, ORDINARY_IM_ACK_SETTLEMENT_TIMEOUT_MS);
+    record.timer.unref?.();
+  }
   logger.info(
     `[${tag(ds)}] Ordinary IM input received by worker `
     + `turn=${turnId.substring(0, 16)} generation=${workerGeneration} attempt=${record.attempt}`,
@@ -6502,6 +6562,21 @@ function completeOrdinaryImDelivery(
   if (record) clearOrdinaryImDelivery(record);
 }
 
+/** A retiring worker's late COMMIT ACK settles only the deliveries tracked
+ * against that exact worker object. The record's own worker identity is the
+ * authority here — deliberately NOT ds.worker/ds.workerGeneration, which have
+ * already moved on (suspend nulls the worker; a replacement fork advances the
+ * generation) by the time the ACK drains from the old child. Receipt ACKs are
+ * deliberately excluded: a stale receipt is not settlement-grade — the turn
+ * can still die unexecuted with the old worker, and swallowing the pending
+ * timers on it would silence the original generation's visible failure. */
+function completeStaleWorkerOrdinaryImDelivery(worker: ChildProcess, turnId: string): void {
+  for (const record of pendingOrdinaryImDeliveries.values()) {
+    if (record.worker !== worker || record.turnId !== turnId) continue;
+    clearOrdinaryImDelivery(record);
+  }
+}
+
 function rejectOrdinaryImDelivery(
   ds: DaemonSession,
   turnId: string,
@@ -6514,9 +6589,38 @@ function rejectOrdinaryImDelivery(
   retryOrFailOrdinaryImDelivery(record, `worker_rejected:${reason}`);
 }
 
-function abandonOrdinaryImDeliveriesForWorker(worker: ChildProcess): void {
+function settleOrdinaryImDeliveriesForWorker(
+  worker: ChildProcess,
+  options: {
+    suppressAllFailures: boolean;
+    startupOwnedTurnId?: string;
+    retiredBeforeCommit?: boolean;
+  },
+): void {
   for (const record of pendingOrdinaryImDeliveries.values()) {
-    if (record.worker === worker) clearOrdinaryImDelivery(record);
+    if (record.worker !== worker) continue;
+    if (options.suppressAllFailures || record.turnId === options.startupOwnedTurnId) {
+      // A record still pending here means the daemon never observed the
+      // commit ACK — but a fire-and-forget ACK can also be lost when the old
+      // child exits right after sending it, so this does NOT prove the turn
+      // never entered the CLI. A deliberate lifecycle retirement (suspend /
+      // worker replacement) must not turn that into silence, and must not
+      // claim certainty either: report an honest unconfirmed outcome that
+      // asks the user to check the session before resending. Transfer, close
+      // and plain kill keep silent settling.
+      if (options.retiredBeforeCommit) {
+        failOrdinaryImDelivery(record, 'worker_retired_before_commit', 'worker.input_retired_unconfirmed');
+        continue;
+      }
+      clearOrdinaryImDelivery(record);
+      continue;
+    }
+    const reason = record.received
+      ? 'worker_exited_after_receipt'
+      : record.transportConfirmed
+        ? 'worker_exited_after_ipc_enqueue'
+        : 'worker_exited_before_ipc_enqueue';
+    failOrdinaryImDelivery(record, reason);
   }
 }
 
@@ -10246,6 +10350,17 @@ function setupWorkerHandlers(
     // installed; never let those stale events mutate the replacement's cards,
     // tokens, readiness, transcript metadata, or durable turn state.
     if (ds.worker !== worker) {
+      // A retiring worker's own COMMIT ACK still settles the ordinary
+      // deliveries tracked against THAT worker object: suspend detaches
+      // ds.worker and a replacement advances the generation BEFORE the old
+      // child's fire-and-forget ACKs drain, so without this the exit
+      // settlement would report an already-committed turn as unconfirmed.
+      // Only the commit ACK is settlement-grade; a stale receipt proves
+      // nothing about execution and must keep the visible-failure timers
+      // running. Stale workers gain no other authority.
+      if (msg.type === 'turn_input_committed') {
+        completeStaleWorkerOrdinaryImDelivery(worker, msg.turnId);
+      }
       logger.debug(`[${t}] Ignored stale worker message: ${msg.type}`);
       return;
     }
@@ -12748,16 +12863,43 @@ function setupWorkerHandlers(
   });
 
   worker.on('exit', (code, signal) => {
-    abandonOrdinaryImDeliveriesForWorker(worker);
     const transferRetirement = transferRetiringWorkers.has(worker);
+    const lifecycleRetirement = lifecycleRetiringWorkers.get(ds)?.has(worker) === true;
+    const preReadyExit = !startupState.ready;
+    const suppressDeliveryFailure = transferRetirement
+      || lifecycleRetirement
+      || worker.killed
+      || ds.session.status === 'closed';
+    settleOrdinaryImDeliveriesForWorker(worker, {
+      suppressAllFailures: suppressDeliveryFailure,
+      // The startup failure path owns only the initial cold-start turn. Any
+      // concurrent follow-up has its own user-visible delivery contract and
+      // must not disappear behind the init turn's single failure notice.
+      startupOwnedTurnId: !suppressDeliveryFailure && preReadyExit
+        ? startupState.initTurnId
+        : undefined,
+      // A deliberate retirement suppresses the misleading crash/ambiguity
+      // notices, but a tracked turn that never committed still owes the user
+      // a terminal outcome: it will never run and nothing redelivers it.
+      retiredBeforeCommit: lifecycleRetirement
+        && !transferRetirement
+        && ds.session.status !== 'closed',
+    });
     transferRetiringWorkers.delete(worker);
     clearLifecycleRetirement(ds, worker);
     logger.info(`[${t}] Worker process exited (code: ${code})`);
     // Last-resort startup guard: syntax/import crashes and abrupt exits can
     // happen before the worker sends either ready or a structured error.  Do
     // not leave the originating Lark message unanswered. Intentional close /
     // replacement kills are excluded to avoid noisy false alarms.
-    if (!transferRetirement && !startupState.ready && !startupState.failureNotified && !worker.killed && ds.session.status !== 'closed') {
+    if (
+      !transferRetirement
+      && !lifecycleRetirement
+      && preReadyExit
+      && !startupState.failureNotified
+      && !worker.killed
+      && ds.session.status !== 'closed'
+    ) {
       const reason = tr('worker.start_exited_early', { code: code ?? 'null' }, loc);
       // Carry the frozen init attribution so an abrupt pre-ready exit of a
       // durable VC delivery is fenced to the receipt/lease chain, not replied

src/i18n/en.ts

@@ -859,7 +859,10 @@ export const messages: Record<string, string> = {
   'worker.mojo_lineage_quarantined': '⚠️ This session was created before botmux recorded which mojo control plane (endpoint / workspace) it ran on, so its earlier remote session cannot be verified.\nIt has been parked rather than discarded — the previous context will NOT continue, and your next message starts a fresh mojo session on the current configuration. The parked id is kept on the session for manual cleanup: {lineage}',
   'worker.mojo_legacy_pinned': '⚠️ This mojo session predates the host-execution upgrade, so it is pinned to the legacy sandbox-fallback mode — tools and replies will mostly NOT work here. This is deliberate (an upgrade must never silently move a live session onto the host).\nPlease close this session (❌ button or /close) and send a new message to start a fresh session on the new behaviour.',
   'worker.start_failed': '⚠️ The {cliName} session failed to start: {reason}\nCheck the Agent/backend settings in Dashboard and the installation environment on the daemon host, then resend your message to retry.',
-  'worker.input_delivery_failed': '⚠️ The Worker could not receive this message. Botmux retried on the same Worker but delivery still did not complete; it stopped before a cross-process retry to avoid duplicate execution. Please resend the message.\nturn: {turnId}',
+  'worker.input_delivery_failed': '⚠️ Botmux could not confirm whether this message entered the Worker execution queue. It stopped delivery to avoid duplicate execution. Check the session status first; do not resend immediately.\nturn: {turnId}',
+  'worker.input_delivery_delayed': '⏳ The message entered the Worker IPC queue, but the Worker has not acknowledged it yet. The machine may be busy; the message can still execute later, so do not resend it.\nturn: {turnId}',
+  'worker.input_commit_delayed': '⏳ The Worker received this message, but has not confirmed that it entered the execution queue yet. The machine may be busy; the message can still execute later, so do not resend it.\nturn: {turnId}',
+  'worker.input_retired_unconfirmed': '⚠️ The session was deliberately suspended or replaced while this message was in flight, and Botmux could not confirm whether it entered the execution queue. Check the session history first; resend the message only if it did not run.\nturn: {turnId}',
   'worker.start_exited_early': 'The worker exited before becoming ready (exit code: {code}); see the Botmux logs for details.',
   'worker.tui_submit_failed': '⚠️ The TUI answer could not be confirmed as delivered to {cliName}. The CLI may still be waiting for input; open the local terminal or send a new message to recover.',
   'worker.raw_input_failed': '⚠️ The slash command could not be confirmed as delivered to {cliName}, so the follow-up text in the same message was not submitted. Check the terminal state, then resend.',

src/i18n/zh.ts

@@ -860,7 +860,10 @@ export const messages: Record<string, string> = {
   'worker.mojo_lineage_quarantined': '⚠️ 这个会话创建于 botmux 记录 mojo 控制面（endpoint / workspace）之前，因此无法确认它此前的远端会话跑在哪里。\n该远端会话已被暂存而非丢弃：原有上下文不会延续，你的下一条消息将在当前配置上新建 mojo 会话。暂存的 id 保留在会话上以便人工清理：{lineage}',
   'worker.mojo_legacy_pinned': '⚠️ 本 mojo 会话创建于「本机执行」升级之前，已被固定在旧的沙箱回退模式——这里的工具和回复基本不可用。这是刻意为之（升级绝不能把活跃会话悄悄切到本机执行）。\n请关闭本会话（❌ 按钮或 /close），再发一条新消息即可用新行为开启全新会话。',
   'worker.start_failed': '⚠️ {cliName} 会话启动失败：{reason}\n请检查 Dashboard 的 Agent / 后端配置和 daemon 所在机器的安装环境，修复后重发消息即可重试。',
-  'worker.input_delivery_failed': '⚠️ Worker 未能接收这条消息。Botmux 已在同一 Worker 上自动重试，但仍未完成接收；为避免跨进程重复执行，没有继续重投。请重发本条消息。\nturn: {turnId}',
+  'worker.input_delivery_failed': '⚠️ Botmux 无法确认这条消息是否已进入 Worker 的执行队列。已停止继续投递以避免重复执行。请先查看会话状态，不要直接重发。\nturn: {turnId}',
+  'worker.input_delivery_delayed': '⏳ 消息已进入 Worker 的 IPC 队列，但 Worker 暂未确认接收。机器可能较忙；消息仍可能继续执行，请勿重发。\nturn: {turnId}',
+  'worker.input_commit_delayed': '⏳ Worker 已收到这条消息，但暂未确认它已进入执行队列。机器可能较忙；消息仍可能继续执行，请勿重发。\nturn: {turnId}',
+  'worker.input_retired_unconfirmed': '⚠️ 会话在处理这条消息期间被主动休眠或更换，Botmux 未能确认它是否已进入执行队列。请先查看会话记录确认结果；若未执行，再重新发送这条消息。\nturn: {turnId}',
   'worker.start_exited_early': 'worker 在就绪前退出（exit code: {code}）；详细错误可查看 Botmux 日志。',
   'worker.tui_submit_failed': '⚠️ TUI 答案未能确认送达 {cliName}。CLI 可能仍在等待输入；请打开本机终端处理，或发送一条新消息解除并继续。',
   'worker.raw_input_failed': '⚠️ Slash 命令未能确认送达 {cliName}，同一条消息中紧随其后的正文没有继续提交。请检查当前终端状态后重发。',