fix: give reconnect-in-progress submissions a bounded retry window

src/bot/active-runs.ts

@@ -10,6 +10,7 @@ export class ActiveRuns {
   private readonly reservations = new Set<string>();
   private pauseDepth = 0;
   private pauseReason: string | undefined;
+  private resumeWaiters: Array<() => void> = [];
 
   reserve(chatId: string): (() => void) | undefined {
     if (this.handles.has(chatId) || this.reservations.has(chatId)) return undefined;
@@ -40,7 +41,11 @@ export class ActiveRuns {
       if (released) return;
       released = true;
       this.pauseDepth = Math.max(0, this.pauseDepth - 1);
-      if (this.pauseDepth === 0) this.pauseReason = undefined;
+      if (this.pauseDepth === 0) {
+        this.pauseReason = undefined;
+        const waiters = this.resumeWaiters.splice(0);
+        for (const waiter of waiters) waiter();
+      }
     };
   }
 
@@ -52,6 +57,35 @@ export class ActiveRuns {
     return this.pauseReason;
   }
 
+  /**
+   * Resolves `true` as soon as new runs are no longer paused, or `false`
+   * once `timeoutMs` elapses first. Intended for smoothing over short-lived
+   * pauses (a reconnect that clears in a few seconds) without making callers
+   * that need an immediate answer (e.g. `nowait` submissions) wait at all —
+   * they should check `newRunsPaused()` directly instead.
+   */
+  waitForResume(timeoutMs: number): Promise<boolean> {
+    if (!this.newRunsPaused()) return Promise.resolve(true);
+    if (timeoutMs <= 0) return Promise.resolve(false);
+    return new Promise((resolve) => {
+      let settled = false;
+      const timer = setTimeout(() => {
+        if (settled) return;
+        settled = true;
+        const idx = this.resumeWaiters.indexOf(onResume);
+        if (idx >= 0) this.resumeWaiters.splice(idx, 1);
+        resolve(false);
+      }, timeoutMs);
+      const onResume = (): void => {
+        if (settled) return;
+        settled = true;
+        clearTimeout(timer);
+        resolve(true);
+      };
+      this.resumeWaiters.push(onResume);
+    });
+  }
+
   get(chatId: string): RunHandle | undefined {
     return this.handles.get(chatId);
   }

src/bot/channel.ts

@@ -75,6 +75,12 @@ import {
 const DEBOUNCE_MS = 600;
 const STREAM_TERMINAL_GRACE_MS = 3000;
 const REACTION_CLEANUP_GRACE_MS = 1000;
+// Most WS reconnects (ping timeout, brief keepalive blip) clear in a few
+// seconds — see keepalive's 15s ping-timeout window. Give a paused
+// submission this long to ride it out before falling back to the
+// "reconnect-in-progress" rejection, instead of dropping the message
+// on every reconnect regardless of how long it actually lasts.
+const RECONNECT_WAIT_MS = 8000;
 
 const BRIDGE_AGENT_INSTRUCTIONS = [
   '你在 bridge 进程中运行，普通 lark-cli 会继承 LARK_CHANNEL=1 并进入 bridge-bound 模式。',
@@ -187,7 +193,12 @@ export async function startChannel(deps: StartChannelDeps): Promise<BridgeChanne
   // Concurrency cap — reads `preferences.maxConcurrentRuns` on each acquire,
   // so /config bumps take effect for the next run.
   const pool = new ProcessPool(() => getMaxConcurrentRuns(controls.cfg));
-  const executor = new RunExecutor({ agent, pool, activeRuns });
+  const executor = new RunExecutor({
+    agent,
+    pool,
+    activeRuns,
+    reconnectWaitMs: RECONNECT_WAIT_MS,
+  });
 
   // Resolve the App Secret to plaintext. The config field can be a literal
   // string, a "${VAR}" template, or a {source, id} SecretRef referencing

src/runtime/run-executor.ts

@@ -13,6 +13,13 @@ export interface RunExecutorDeps {
   createRunId?: () => string;
   now?: () => number;
   postDoneExitGraceMs?: number;
+  /**
+   * How long a submission may wait for an in-progress pause (reconnect,
+   * bridge restart, etc.) to clear before rejecting with
+   * `reconnect-in-progress`. Defaults to `0` (reject immediately, the prior
+   * behavior) so existing callers are unaffected unless they opt in.
+   */
+  reconnectWaitMs?: number;
 }
 
 export interface SubmitRunInput {
@@ -42,6 +49,7 @@ export interface RunExecution {
 }
 
 const DEFAULT_POST_DONE_EXIT_GRACE_MS = 2000;
+const DEFAULT_RECONNECT_WAIT_MS = 0;
 
 export class RunExecutor {
   private readonly agent: AgentAdapter;
@@ -50,6 +58,7 @@ export class RunExecutor {
   private readonly createRunId: () => string;
   private readonly now: () => number;
   private readonly postDoneExitGraceMs: number;
+  private readonly reconnectWaitMs: number;
 
   constructor(deps: RunExecutorDeps) {
     this.agent = deps.agent;
@@ -58,6 +67,7 @@ export class RunExecutor {
     this.createRunId = deps.createRunId ?? randomUUID;
     this.now = deps.now ?? Date.now;
     this.postDoneExitGraceMs = deps.postDoneExitGraceMs ?? DEFAULT_POST_DONE_EXIT_GRACE_MS;
+    this.reconnectWaitMs = deps.reconnectWaitMs ?? DEFAULT_RECONNECT_WAIT_MS;
   }
 
   async submit(input: SubmitRunInput): Promise<RunExecution> {
@@ -66,10 +76,20 @@ export class RunExecutor {
       throw new RunRejected('policy-expired', 'run policy expired before spawn');
     }
     if (this.activeRuns.newRunsPaused()) {
-      throw new RunRejected(
-        'reconnect-in-progress',
-        this.activeRuns.newRunsPauseReason() ?? 'new runs are temporarily paused',
-      );
+      // Short reconnect blips (a WS ping timeout that self-heals in a
+      // second or two) are common and otherwise silently drop the user's
+      // message — see the "reconnect-in-progress" rejection below, which
+      // used to fire immediately with no retry. Give the pause a bounded
+      // window to clear before giving up; `nowait` callers opt out since
+      // they've already asked not to wait for anything.
+      const shouldWait = this.reconnectWaitMs > 0 && !input.nowait;
+      const resumed = shouldWait ? await this.activeRuns.waitForResume(this.reconnectWaitMs) : false;
+      if (!resumed && this.activeRuns.newRunsPaused()) {
+        throw new RunRejected(
+          'reconnect-in-progress',
+          this.activeRuns.newRunsPauseReason() ?? 'new runs are temporarily paused',
+        );
+      }
     }
     const releaseScope = this.activeRuns.reserve(input.scopeId);
     if (!releaseScope) {

tests/integration/executor/run-executor.test.ts

@@ -128,6 +128,80 @@ describe('RunExecutor', () => {
     }
   });
 
+  it('with reconnectWaitMs configured, holds a submission until a short reconnect clears and then runs it', async () => {
+    const h = await createHarness({
+      events: [{ type: 'done', terminationReason: 'normal' }],
+      reconnectWaitMs: 5000,
+    });
+    const resume = h.activeRuns.pauseNewRuns('reconnect');
+
+    const submitPromise = h.executor.submit({
+      scopeId: 'scope-1',
+      policy: policy(h.tmp.workspace),
+    });
+    // Give submit() a tick to hit the paused check and start waiting.
+    await new Promise((r) => setTimeout(r, 10));
+    expect(h.agent.runs).toHaveLength(0);
+
+    resume();
+
+    const execution = await submitPromise;
+    expect(execution.runId).toBe('run-1');
+    expect(h.agent.runs).toHaveLength(1);
+    await collect(execution.subscribe());
+  });
+
+  it('with reconnectWaitMs configured, still rejects once the wait window elapses without a resume', async () => {
+    const h = await createHarness({ events: [], reconnectWaitMs: 20 });
+    const resume = h.activeRuns.pauseNewRuns('reconnect');
+    try {
+      await expect(
+        h.executor.submit({
+          scopeId: 'scope-1',
+          policy: policy(h.tmp.workspace),
+        }),
+      ).rejects.toMatchObject({ code: 'reconnect-in-progress' });
+      expect(h.agent.runs).toHaveLength(0);
+    } finally {
+      resume();
+    }
+  });
+
+  it('with reconnectWaitMs configured, nowait submissions still fail fast instead of waiting', async () => {
+    const h = await createHarness({ events: [], reconnectWaitMs: 5000 });
+    const resume = h.activeRuns.pauseNewRuns('reconnect');
+    try {
+      const start = Date.now();
+      await expect(
+        h.executor.submit({
+          scopeId: 'scope-1',
+          policy: policy(h.tmp.workspace),
+          nowait: true,
+        }),
+      ).rejects.toMatchObject({ code: 'reconnect-in-progress' });
+      expect(Date.now() - start).toBeLessThan(1000);
+    } finally {
+      resume();
+    }
+  });
+
+  it('without reconnectWaitMs configured, keeps the original immediate-reject behavior', async () => {
+    const h = await createHarness({ events: [] });
+    const resume = h.activeRuns.pauseNewRuns('reconnect');
+    try {
+      const start = Date.now();
+      await expect(
+        h.executor.submit({
+          scopeId: 'scope-1',
+          policy: policy(h.tmp.workspace),
+        }),
+      ).rejects.toMatchObject({ code: 'reconnect-in-progress' });
+      expect(Date.now() - start).toBeLessThan(1000);
+    } finally {
+      resume();
+    }
+  });
+
   it('rejects duplicate submissions for a scope that already has a run', async () => {
     const h = await createHarness({ events: [{ type: 'done', terminationReason: 'normal' }] });
 
@@ -271,6 +345,7 @@ async function createHarness(options: {
   waitForExit?: boolean | readonly boolean[];
   poolCap?: number;
   agent?: AgentAdapter;
+  reconnectWaitMs?: number;
 }): Promise<{
   tmp: TmpProfile;
   agent: FakeAgentAdapter;
@@ -301,6 +376,7 @@ async function createHarness(options: {
       createRunId: () => `run-${nextRun++}`,
       now: () => 1000,
       postDoneExitGraceMs: 10,
+      reconnectWaitMs: options.reconnectWaitMs,
     }),
   };
 }

tests/unit/bot/active-runs.test.ts

@@ -0,0 +1,72 @@
+import { describe, expect, it } from 'vitest';
+import { ActiveRuns } from '../../../src/bot/active-runs';
+
+describe('ActiveRuns.waitForResume', () => {
+  it('resolves true immediately when new runs are not paused', async () => {
+    const activeRuns = new ActiveRuns();
+    await expect(activeRuns.waitForResume(1000)).resolves.toBe(true);
+  });
+
+  it('resolves true as soon as the pause is released, without waiting out the timeout', async () => {
+    const activeRuns = new ActiveRuns();
+    const resume = activeRuns.pauseNewRuns('reconnect');
+
+    const waiting = activeRuns.waitForResume(5000);
+    setTimeout(() => resume(), 10);
+
+    const start = Date.now();
+    await expect(waiting).resolves.toBe(true);
+    expect(Date.now() - start).toBeLessThan(5000);
+  });
+
+  it('resolves false once the timeout elapses while still paused', async () => {
+    const activeRuns = new ActiveRuns();
+    const resume = activeRuns.pauseNewRuns('reconnect');
+    try {
+      await expect(activeRuns.waitForResume(20)).resolves.toBe(false);
+      expect(activeRuns.newRunsPaused()).toBe(true);
+    } finally {
+      resume();
+    }
+  });
+
+  it('resolves false immediately for a non-positive timeout while paused', async () => {
+    const activeRuns = new ActiveRuns();
+    const resume = activeRuns.pauseNewRuns('reconnect');
+    try {
+      await expect(activeRuns.waitForResume(0)).resolves.toBe(false);
+    } finally {
+      resume();
+    }
+  });
+
+  it('only resolves waiters once the last of several nested pauses is released', async () => {
+    const activeRuns = new ActiveRuns();
+    const resumeA = activeRuns.pauseNewRuns('reconnect-a');
+    const resumeB = activeRuns.pauseNewRuns('reconnect-b');
+
+    const waiting = activeRuns.waitForResume(5000);
+    resumeA();
+    // still paused by B — waiter must not have resolved yet.
+    await new Promise((r) => setTimeout(r, 10));
+    expect(activeRuns.newRunsPaused()).toBe(true);
+
+    resumeB();
+    await expect(waiting).resolves.toBe(true);
+  });
+
+  it('does not leak a waiter entry after it resolves via resume', async () => {
+    const activeRuns = new ActiveRuns();
+    const resume = activeRuns.pauseNewRuns('reconnect');
+    const waiting = activeRuns.waitForResume(5000);
+    resume();
+    await waiting;
+
+    // A fresh pause/waitForResume cycle should behave normally — proves the
+    // earlier waiter was cleaned up rather than firing again or throwing.
+    const resumeAgain = activeRuns.pauseNewRuns('reconnect-again');
+    const secondWait = activeRuns.waitForResume(20);
+    await expect(secondWait).resolves.toBe(false);
+    resumeAgain();
+  });
+});