fix(chat): resume already-checkpointed running turns

packages/junior/src/chat/agent/index.ts

@@ -1141,6 +1141,10 @@ async function executeAgentRunInPrivacyContext(
             if (promptPersisted) {
               await runResume.commitInput();
             }
+          } else if (durability.onInputCommitted) {
+            // Prompt already owned by the running record. Still ack mailbox
+            // ownership when redelivery is carrying a pending inbound.
+            await runResume.commitInput();
           }
 
           /** Race one provider operation against the turn deadline and abort its owner. */

packages/junior/src/chat/agent/prompt.ts

@@ -435,11 +435,11 @@ export async function assemblePrompt(args: {
   userContentParts: UserContentPart[];
 }): Promise<PromptAssembly> {
   const source = args.routing.source;
-  const hasPromptCheckpoint =
-    args.resumedFromSessionRecord &&
-    args.existingTurnStartMessageIndex !== undefined;
-  const shouldPromptAgent =
-    !args.resumedFromSessionRecord || !hasPromptCheckpoint;
+  // The turn-start cursor is the durable ownership signal for the prompt.
+  // Resume classification can lag (still-running redelivery), but once that
+  // cursor exists the prompt is already committed and must not be rebuilt.
+  const hasPromptCheckpoint = args.existingTurnStartMessageIndex !== undefined;
+  const shouldPromptAgent = !hasPromptCheckpoint;
   const requestContentParts: UserContentPart[] = [
     ...(args.explicitSkill
       ? [

packages/junior/src/chat/services/turn-session-record.ts

@@ -85,12 +85,19 @@ export async function loadTurnSessionRecord(
     ctx.conversationId,
     ctx.sessionId,
   );
-  const hasAwaitingResumeRecord = Boolean(
-    existingSessionRecord && existingSessionRecord.state === "awaiting_resume",
+  // A still-running record with a committed turn-start cursor already owns its
+  // prompt. Treat that as resume so redelivery continues instead of rebuilding
+  // and re-checkpointing a non-identical user message.
+  const hasCommittedPrompt =
+    existingSessionRecord?.turnStartMessageIndex !== undefined;
+  const resumedFromSessionRecord = Boolean(
+    existingSessionRecord &&
+      (existingSessionRecord.state === "awaiting_resume" ||
+        (existingSessionRecord.state === "running" && hasCommittedPrompt)),
   );
   return {
-    resumedFromSessionRecord: hasAwaitingResumeRecord,
-    currentSliceId: hasAwaitingResumeRecord
+    resumedFromSessionRecord,
+    currentSliceId: resumedFromSessionRecord
       ? existingSessionRecord!.sliceId
       : 1,
     existingSessionRecord,
@@ -177,9 +184,22 @@ export async function persistRunningSessionRecord(args: {
     });
     return true;
   } catch (recordError) {
+    // Boundary mismatch is permanent for this attempt's message shape. Log it
+    // distinctly, but still return false: the poison-turn farm is stopped by
+    // treating already-checkpointed running turns as resume (no rebuild /
+    // recheckpoint), not by failing closed here. Handoff/compaction follow-ups
+    // and no-checkpoint resume still rely on false when no mailbox ack is
+    // pending.
+    const message =
+      recordError instanceof Error ? recordError.message : String(recordError);
+    const boundaryMismatch = message.includes(
+      "changed before its committed boundary",
+    );
     logSessionRecordError(
       recordError,
-      "agent.turn.running_session_record.failed",
+      boundaryMismatch
+        ? "agent.turn.running_session_record.boundary_mismatch"
+        : "agent.turn.running_session_record.failed",
       args,
       {
         "app.ai.resume_slice_id": args.sliceId,

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

@@ -60,6 +60,8 @@ export const CONVERSATION_WORK_LEASE_TTL_MS = 90_000;
 export const CONVERSATION_WORK_CHECK_IN_INTERVAL_MS = 15_000;
 export const CONVERSATION_WORK_STALE_ENQUEUE_MS = 60_000;
 export const CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS = 5;
+/** Empty continue/lost-lease wakes with no mailbox progress before fail-closed. */
+export const CONVERSATION_WORK_MAX_EMPTY_WAKES = 5;
 
 const inboundMessageSourceSchema = z.enum([
   "api",
@@ -127,6 +129,8 @@ export interface Lease {
 }
 
 export interface ConversationExecution {
+  /** Consecutive empty wakes (no mailbox attempt) that made no durable progress. */
+  emptyWakeCount?: number;
   inboundMessageIds: string[];
   lastCheckpointAtMs?: number;
   lastEnqueuedAtMs?: number;
@@ -432,6 +436,7 @@ function normalizeExecution(
     pendingCount: pendingMessages.length,
     pendingMessages,
     lease,
+    emptyWakeCount: toOptionalNumber(value.emptyWakeCount),
     lastCheckpointAtMs: toOptionalNumber(value.lastCheckpointAtMs),
     lastEnqueuedAtMs: toOptionalNumber(value.lastEnqueuedAtMs),
     runId: toOptionalString(value.runId),
@@ -1138,6 +1143,8 @@ export async function appendInboundMessage(args: {
           next,
           {
             ...current.execution,
+            // Fresh mailbox work resets the empty-wake poison budget.
+            emptyWakeCount: undefined,
             status,
             inboundMessageIds: [
               ...current.execution.inboundMessageIds,
@@ -1741,11 +1748,102 @@ export async function completeConversationWork(args: {
   });
 }
 
+/**
+ * Record one empty wake that made no mailbox progress (continue / lost-lease
+ * recovery with nothing to ack). Caps poison requeue farms that bypass the
+ * per-message delivery attempt counter.
+ */
+export async function recordEmptyWakeFailure(args: {
+  conversationId: string;
+  leaseToken: string;
+  nowMs?: number;
+  state?: StateAdapter;
+}): Promise<AttemptFailure> {
+  const nowMs = args.nowMs ?? now();
+  return await withConversationMutation(args, async (state, lock) => {
+    const current = await readConversation(state, args.conversationId);
+    if (!current || current.execution.lease?.token !== args.leaseToken) {
+      return {
+        status: "lost_lease",
+        pendingCount: 0,
+        deadLetteredMessages: [],
+      };
+    }
+    const emptyWakeCount = (current.execution.emptyWakeCount ?? 0) + 1;
+    const terminal = emptyWakeCount >= CONVERSATION_WORK_MAX_EMPTY_WAKES;
+    await writeConversation(
+      state,
+      lock,
+      withExecutionUpdate(
+        current,
+        {
+          ...current.execution,
+          emptyWakeCount,
+          ...(terminal
+            ? {
+                lease: undefined,
+                status:
+                  pendingMessages(current).length > 0 ? "pending" : "failed",
+              }
+            : {}),
+        },
+        nowMs,
+      ),
+    );
+    return {
+      status: "recorded",
+      pendingCount: current.execution.pendingMessages.length,
+      deadLetteredMessages: [],
+      ...(terminal ? { terminal: true as const } : {}),
+    };
+  });
+}
+
+/** Clear empty-wake streak after durable progress (ack, complete, new work). */
+export async function clearEmptyWakeCount(args: {
+  conversationId: string;
+  leaseToken?: string;
+  nowMs?: number;
+  state?: StateAdapter;
+}): Promise<boolean> {
+  const nowMs = args.nowMs ?? now();
+  return await withConversationMutation(args, async (state, lock) => {
+    const current = await readConversation(state, args.conversationId);
+    if (!current) {
+      return false;
+    }
+    if (
+      args.leaseToken !== undefined &&
+      current.execution.lease?.token !== args.leaseToken
+    ) {
+      return false;
+    }
+    if ((current.execution.emptyWakeCount ?? 0) === 0) {
+      return true;
+    }
+    await writeConversation(
+      state,
+      lock,
+      withExecutionUpdate(
+        current,
+        {
+          ...current.execution,
+          emptyWakeCount: undefined,
+        },
+        nowMs,
+      ),
+    );
+    return true;
+  });
+}
+
 /** Failure outcome: `lost_lease` (another owner took over), `recorded` (attempt counted), or `skipped` (durable progress was made). */
 export interface AttemptFailure {
   pendingCount: number;
   deadLetteredMessages: InboundMessage[];
   status: "lost_lease" | "recorded" | "skipped";
+  /** True when empty-wake failures hit the fail-closed ceiling. */
+  terminal?: boolean;
 }
 
 /**

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

@@ -9,6 +9,7 @@ export {
   CONVERSATION_WORK_CHECK_IN_INTERVAL_MS,
   CONVERSATION_WORK_LEASE_TTL_MS,
   CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS,
+  CONVERSATION_WORK_MAX_EMPTY_WAKES,
   CONVERSATION_WORK_STALE_ENQUEUE_MS,
   isFinalAttempt,
   isInvalidConversationRecordError,
@@ -453,6 +454,32 @@ export async function completeConversationWork(args: {
   return result;
 }
 
+/** Record one empty wake with no mailbox progress before fail-closed. */
+export async function recordEmptyWakeFailure(args: {
+  conversationId: string;
+  leaseToken: string;
+  conversationStore?: ConversationStore;
+  nowMs?: number;
+  state?: StateAdapter;
+}) {
+  const result = await workState.recordEmptyWakeFailure(args);
+  if (result.status === "recorded") {
+    await recordExecutionMetadata(args);
+  }
+  return result;
+}
+
+/** Clear empty-wake streak after durable progress. */
+export async function clearEmptyWakeCount(args: {
+  conversationId: string;
+  leaseToken?: string;
+  conversationStore?: ConversationStore;
+  nowMs?: number;
+  state?: StateAdapter;
+}) {
+  return await workState.clearEmptyWakeCount(args);
+}
+
 /** Record one failed delivery attempt and dead-letter messages at their limit. */
 export async function recordAttemptFailure(args: {
   conversationId: string;