fix(chat): Requeue runnable conversation work

Preserve runnable conversation state when a leased worker completes after needsRun was marked during execution. This keeps continuation and late work recovery from waiting on heartbeat repair.

Scope worker and heartbeat queue idempotency keys to a specific wake-up attempt so provider dedupe cannot suppress later legitimate recovery nudges.

Move deterministic worker, lease, mailbox, and timeout-resume coverage into component tests and document the layer boundary.

Refs GH-470

Co-Authored-By: Codex GPT-5 <noreply@openai.com>

Author: dcramer

packages/junior/package.json

@@ -48,7 +48,7 @@
     "build": "tsup && tsc -p tsconfig.build.json --emitDeclarationOnly",
     "lint": "oxlint --config .oxlintrc.json --deny-warnings src tests scripts bin tsup.config.ts",
     "lint:fix": "oxlint --config .oxlintrc.json --deny-warnings --fix src tests scripts bin tsup.config.ts",
-    "test": "pnpm run test:slack-boundary && pnpm run test:arch-boundary && vitest run",
+    "test": "pnpm run test:slack-boundary && pnpm run test:arch-boundary && vitest run --maxWorkers=4",
     "test:watch": "vitest",
     "test:slack-boundary": "node scripts/check-slack-test-boundary.mjs",
     "test:arch-boundary": "depcruise --config .dependency-cruiser.mjs src/chat",

packages/junior/src/chat/task-execution/heartbeat.ts

@@ -20,8 +20,9 @@ export interface ConversationWorkRecoveryResult {
 function heartbeatIdempotencyKey(
   reason: string,
   conversationId: string,
+  nowMs: number,
 ): string {
-  return `heartbeat:${reason}:${conversationId}`;
+  return `heartbeat:${reason}:${conversationId}:${nowMs}`;
 }
 
 async function sendRecoveryNudge(args: {
@@ -79,7 +80,11 @@ export async function recoverConversationWork(args: {
         }
         await sendRecoveryNudge({
           conversationId,
-          idempotencyKey: heartbeatIdempotencyKey("lease", conversationId),
+          idempotencyKey: heartbeatIdempotencyKey(
+            "lease",
+            conversationId,
+            args.nowMs,
+          ),
           nowMs: args.nowMs,
           queue: args.queue,
           state: args.state,
@@ -106,7 +111,11 @@ export async function recoverConversationWork(args: {
 
       await sendRecoveryNudge({
         conversationId,
-        idempotencyKey: heartbeatIdempotencyKey("pending", conversationId),
+        idempotencyKey: heartbeatIdempotencyKey(
+          "pending",
+          conversationId,
+          args.nowMs,
+        ),
         nowMs: args.nowMs,
         queue: args.queue,
         state: args.state,

packages/junior/src/chat/task-execution/store.ts

@@ -753,7 +753,7 @@ export async function releaseConversationWork(args: {
   });
 }
 
-/** Finish a leased conversation when no pending mailbox work remains. */
+/** Finish a leased conversation and report whether runnable work remains. */
 export async function completeConversationWork(args: {
   conversationId: string;
   leaseToken: string;
@@ -767,13 +767,14 @@ export async function completeConversationWork(args: {
       return "lost_lease";
     }
     const hasPending = pendingMessages(current).length > 0;
+    const hasRunnableWork = current.needsRun || hasPending;
     await writeWorkState(state, {
       ...current,
       lease: undefined,
-      needsRun: current.needsRun || hasPending,
+      needsRun: hasRunnableWork,
       updatedAtMs: nowMs,
     });
-    return hasPending ? "pending" : "completed";
+    return hasRunnableWork ? "pending" : "completed";
   });
 }
 

packages/junior/src/chat/task-execution/worker.ts

@@ -55,8 +55,12 @@ function now(options: ProcessConversationWorkOptions): number {
   return options.nowMs?.() ?? Date.now();
 }
 
-function nudgeIdempotencyKey(reason: string, conversationId: string): string {
-  return `${reason}:${conversationId}`;
+function nudgeIdempotencyKey(
+  reason: string,
+  conversationId: string,
+  nowMs: number,
+): string {
+  return `${reason}:${conversationId}:${nowMs}`;
 }
 
 async function sendWakeNudge(args: {
@@ -145,11 +149,12 @@ export async function processConversationWork(
     return { status: "no_work" };
   }
   if (lease.status === "active") {
+    const nudgeNowMs = now(options);
     await sendWakeNudge({
       conversationId,
       delayMs: CONVERSATION_WORK_DEFER_DELAY_MS,
-      idempotencyKey: nudgeIdempotencyKey("active", conversationId),
-      nowMs: now(options),
+      idempotencyKey: nudgeIdempotencyKey("active", conversationId, nudgeNowMs),
+      nowMs: nudgeNowMs,
       options,
     });
     logInfo(
@@ -206,25 +211,30 @@ export async function processConversationWork(
   try {
     const result = await options.run(workerContext);
     if (result.status === "yielded") {
+      const yieldNowMs = now(options);
       const continuationMarked = await requestConversationContinuation({
         conversationId,
         leaseToken: lease.leaseToken,
-        nowMs: now(options),
+        nowMs: yieldNowMs,
         state: options.state,
       });
       if (!continuationMarked) {
         return { status: "lost_lease" };
       }
       await sendWakeNudge({
         conversationId,
-        idempotencyKey: nudgeIdempotencyKey("yield", conversationId),
-        nowMs: now(options),
+        idempotencyKey: nudgeIdempotencyKey(
+          "yield",
+          conversationId,
+          yieldNowMs,
+        ),
+        nowMs: yieldNowMs,
         options,
       });
       await releaseConversationWork({
         conversationId,
         leaseToken: lease.leaseToken,
-        nowMs: now(options),
+        nowMs: yieldNowMs,
         state: options.state,
       });
       logInfo(
@@ -249,10 +259,15 @@ export async function processConversationWork(
       return { status: "lost_lease" };
     }
     if (completion === "pending") {
+      const nudgeNowMs = now(options);
       await sendWakeNudge({
         conversationId,
-        idempotencyKey: nudgeIdempotencyKey("pending", conversationId),
-        nowMs: now(options),
+        idempotencyKey: nudgeIdempotencyKey(
+          "pending",
+          conversationId,
+          nudgeNowMs,
+        ),
+        nowMs: nudgeNowMs,
         options,
       });
       return { status: "pending_requeued" };

…ests/unit/runtime/timeout-resume.test.ts …component/runtime/timeout-resume.test.tspackages/junior/tests/unit/runtime/timeout-resume.test.ts renamed to packages/junior/tests/component/runtime/timeout-resume.test.ts packages/junior/tests/unit/runtime/timeout-resume.test.ts renamed to packages/junior/tests/component/runtime/timeout-resume.test.ts

@@ -8,6 +8,17 @@ import type { ConversationWorkQueue } from "@/chat/task-execution/queue";
 import { getConversationWorkState } from "@/chat/task-execution/store";
 import { disconnectStateAdapter } from "@/chat/state/adapter";
 
+const ORIGINAL_ENV = vi.hoisted(() => {
+  const original = {
+    JUNIOR_SECRET: process.env.JUNIOR_SECRET,
+    JUNIOR_STATE_ADAPTER: process.env.JUNIOR_STATE_ADAPTER,
+    SLACK_SIGNING_SECRET: process.env.SLACK_SIGNING_SECRET,
+  };
+  process.env.JUNIOR_STATE_ADAPTER = "memory";
+  process.env.JUNIOR_SECRET = "resume-secret";
+  return original;
+});
+
 class FakeQueue implements ConversationWorkQueue {
   sent: Array<{
     conversationId: string;
@@ -45,9 +56,15 @@ function makeSignedResumeRequest(body: Record<string, unknown>): Request {
   });
 }
 
-describe("timeout resume callback signing", () => {
-  const originalSlackSigningSecret = process.env.SLACK_SIGNING_SECRET;
+function restoreEnv(name: string, value: string | undefined): void {
+  if (value === undefined) {
+    delete process.env[name];
+    return;
+  }
+  process.env[name] = value;
+}
 
+describe("timeout resume callback signing", () => {
   beforeEach(async () => {
     process.env.JUNIOR_STATE_ADAPTER = "memory";
     process.env.JUNIOR_SECRET = "resume-secret";
@@ -56,13 +73,9 @@ describe("timeout resume callback signing", () => {
 
   afterEach(async () => {
     await disconnectStateAdapter();
-    delete process.env.JUNIOR_STATE_ADAPTER;
-    delete process.env.JUNIOR_SECRET;
-    if (originalSlackSigningSecret === undefined) {
-      delete process.env.SLACK_SIGNING_SECRET;
-    } else {
-      process.env.SLACK_SIGNING_SECRET = originalSlackSigningSecret;
-    }
+    restoreEnv("JUNIOR_STATE_ADAPTER", ORIGINAL_ENV.JUNIOR_STATE_ADAPTER);
+    restoreEnv("JUNIOR_SECRET", ORIGINAL_ENV.JUNIOR_SECRET);
+    restoreEnv("SLACK_SIGNING_SECRET", ORIGINAL_ENV.SLACK_SIGNING_SECRET);
     vi.restoreAllMocks();
   });
 

…ts/integration/conversation-work.test.ts …task-execution/conversation-work.test.tspackages/junior/tests/integration/conversation-work.test.ts renamed to packages/junior/tests/component/task-execution/conversation-work.test.ts packages/junior/tests/integration/conversation-work.test.ts renamed to packages/junior/tests/component/task-execution/conversation-work.test.ts

@@ -198,7 +198,7 @@ describe("conversation work execution", () => {
     expect(queue.sent).toEqual([
       {
         conversationId: CONVERSATION_ID,
-        idempotencyKey: `heartbeat:pending:${CONVERSATION_ID}`,
+        idempotencyKey: `heartbeat:pending:${CONVERSATION_ID}:62000`,
       },
     ]);
   });
@@ -243,30 +243,73 @@ describe("conversation work execution", () => {
     await expect(first).resolves.toEqual({ status: "completed" });
   });
 
-  it("preserves work requested while a lease is running", async () => {
+  it("requeues work requested while a lease is running", async () => {
     const queue = new FakeQueue();
+    let currentNowMs = 1_000;
     await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 });
 
     await expect(
       processConversationWork(CONVERSATION_ID, {
+        nowMs: () => currentNowMs,
         queue,
         run: async (context) => {
           await context.drainMailbox(async () => {});
+          currentNowMs = 2_000;
           await requestConversationWork({
             conversationId: context.conversationId,
-            nowMs: 2_000,
+            nowMs: currentNowMs,
           });
           return { status: "completed" };
         },
       }),
-    ).resolves.toEqual({ status: "completed" });
+    ).resolves.toEqual({ status: "pending_requeued" });
 
     const state = await getConversationWorkState({
       conversationId: CONVERSATION_ID,
     });
     expect(state?.lease).toBeUndefined();
     expect(state?.needsRun).toBe(true);
     expect(state ? countPendingConversationMessages(state) : 0).toBe(0);
+    expect(queue.sent).toMatchObject([
+      {
+        conversationId: CONVERSATION_ID,
+        idempotencyKey: `pending:${CONVERSATION_ID}:2000`,
+      },
+    ]);
+  });
+
+  it("uses fresh queue idempotency keys for repeated worker requeues", async () => {
+    const queue = new FakeQueue();
+    let currentNowMs = 1_000;
+    await requestConversationWork({
+      conversationId: CONVERSATION_ID,
+      nowMs: currentNowMs,
+    });
+
+    async function runSlice(nowMs: number): Promise<void> {
+      currentNowMs = nowMs;
+      await expect(
+        processConversationWork(CONVERSATION_ID, {
+          nowMs: () => currentNowMs,
+          queue,
+          run: async (context) => {
+            await requestConversationWork({
+              conversationId: context.conversationId,
+              nowMs: currentNowMs,
+            });
+            return { status: "completed" };
+          },
+        }),
+      ).resolves.toEqual({ status: "pending_requeued" });
+    }
+
+    await runSlice(2_000);
+    await runSlice(63_000);
+
+    expect(queue.sent.map((send) => send.idempotencyKey)).toEqual([
+      `pending:${CONVERSATION_ID}:2000`,
+      `pending:${CONVERSATION_ID}:63000`,
+    ]);
   });
 
   it("drains pending messages and completes the leased conversation", async () => {
@@ -353,7 +396,7 @@ describe("conversation work execution", () => {
     expect(queue.sent).toMatchObject([
       {
         conversationId: CONVERSATION_ID,
-        idempotencyKey: `heartbeat:lease:${CONVERSATION_ID}`,
+        idempotencyKey: `heartbeat:lease:${CONVERSATION_ID}:92000`,
       },
     ]);
   });
@@ -371,6 +414,29 @@ describe("conversation work execution", () => {
     expect(queue.sent).toHaveLength(1);
   });
 
+  it("uses fresh queue idempotency keys for repeated heartbeat recovery", async () => {
+    const queue = new FakeQueue();
+    await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 });
+
+    await expect(
+      recoverConversationWork({
+        nowMs: 62_000,
+        queue,
+      }),
+    ).resolves.toEqual({ expiredLeaseCount: 0, pendingCount: 1 });
+    await expect(
+      recoverConversationWork({
+        nowMs: 122_001,
+        queue,
+      }),
+    ).resolves.toEqual({ expiredLeaseCount: 0, pendingCount: 1 });
+
+    expect(queue.sent.map((send) => send.idempotencyKey)).toEqual([
+      `heartbeat:pending:${CONVERSATION_ID}:62000`,
+      `heartbeat:pending:${CONVERSATION_ID}:122001`,
+    ]);
+  });
+
   it("runs conversation work recovery from the core heartbeat", async () => {
     const queue = new FakeQueue();
     await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 });
@@ -383,7 +449,7 @@ describe("conversation work execution", () => {
     expect(queue.sent).toEqual([
       {
         conversationId: CONVERSATION_ID,
-        idempotencyKey: `heartbeat:pending:${CONVERSATION_ID}`,
+        idempotencyKey: `heartbeat:pending:${CONVERSATION_ID}:62000`,
       },
     ]);
   });
@@ -447,19 +513,22 @@ describe("conversation work execution", () => {
 
   it("requeues instead of completing when final mailbox work remains", async () => {
     const queue = new FakeQueue();
+    let currentNowMs = 1_000;
     await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 });
 
     await expect(
       processConversationWork(CONVERSATION_ID, {
+        nowMs: () => currentNowMs,
         queue,
         run: async (context) => {
           await context.drainMailbox(async () => {});
+          currentNowMs = 2_100;
           await appendInboundMessage({
             message: inboundMessage("m2", {
               createdAtMs: 2_000,
               receivedAtMs: 2_100,
             }),
-            nowMs: 2_100,
+            nowMs: currentNowMs,
           });
           return { status: "completed" };
         },
@@ -468,7 +537,7 @@ describe("conversation work execution", () => {
     expect(queue.sent).toMatchObject([
       {
         conversationId: CONVERSATION_ID,
-        idempotencyKey: `pending:${CONVERSATION_ID}`,
+        idempotencyKey: `pending:${CONVERSATION_ID}:2100`,
       },
     ]);
   });
@@ -499,7 +568,7 @@ describe("conversation work execution", () => {
     expect(queue.sent).toMatchObject([
       {
         conversationId: CONVERSATION_ID,
-        idempotencyKey: `yield:${CONVERSATION_ID}`,
+        idempotencyKey: `yield:${CONVERSATION_ID}:242000`,
       },
     ]);
   });

packages/junior/vitest.config.ts

@@ -23,6 +23,7 @@ for (const envRoot of [workspaceRoot, packageRoot]) {
 }
 
 process.env.JUNIOR_SECRET = "junior-test-secret";
+process.env.JUNIOR_STATE_ADAPTER = "memory";
 process.env.JUNIOR_STATE_KEY_PREFIX ??= `junior:test:${process.pid}`;
 
 export default defineConfig({