Integrate current master through work-folder PR 13341

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/issue-queued-comments-routes.test.ts

@@ -314,6 +314,75 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
     expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3);
   });
 
+  it.each(["user", "system", "manual_receipt_other_actor", "rejected_admission", "consumed_first", "consumed_last", "deleted_first", "agent_first", "cancelled_queued"])("delivers saved user messages on a %s queue after automatic recovery stopped, without another click", async (actorType) => {
+    const seeded = await seedQueue();
+    await db.update(agentWakeupRequests).set({ requestedByActorType: actorType === "user" ? "user" : "system" })
+      .where(eq(agentWakeupRequests.id, seeded.wakeId));
+    if (actorType === "manual_receipt_other_actor") {
+      const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
+      await db.update(agentWakeupRequests).set({ requestedByActorType: "user", requestedByActorId: "other-operator",
+        payload: { ...saved.payload, manualUserWake: true },
+      }).where(eq(agentWakeupRequests.id, seeded.wakeId));
+    }
+    await db.update(agents).set({ adapterType: "claude_local",
+      runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
+    }).where(eq(agents.id, seeded.agentId));
+    await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "failed",
+      processPid: process.pid, errorCode: "process_lost", finishedAt: new Date("2026-08-22T15:03:00.000Z"),
+    }).where(eq(heartbeatRuns.id, seeded.runId));
+    await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
+    await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId,
+      kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: seeded.runId,
+      status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
+      evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
+    });
+    // Leave the agent's only slot occupied on another task so dispatch stays queued.
+    await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
+      status: "running", contextSnapshot: { issueId: randomUUID() },
+    });
+    await heartbeatService(db).resumeQueuedRuns();
+    expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status)
+      .toBe("deferred_issue_execution");
+    await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, seeded.runId));
+    const excludedIndex = actorType === "consumed_last" ? 1 : 0;
+    const filtersInput = ["consumed_first", "consumed_last", "deleted_first", "agent_first"].includes(actorType);
+    if (actorType.startsWith("consumed_")) await db.update(heartbeatRuns).set({
+      contextSnapshot: { issueId: seeded.issueId, wakeCommentIds: [seeded.commentIds[excludedIndex]] },
+    }).where(eq(heartbeatRuns.id, seeded.runId));
+    if (actorType === "deleted_first") await db.update(issueComments).set({ deletedAt: new Date() })
+      .where(eq(issueComments.id, seeded.commentIds[0]));
+    if (actorType === "agent_first") await db.update(issueComments).set({
+      authorType: "agent", authorUserId: null, authorAgentId: seeded.agentId,
+    }).where(eq(issueComments.id, seeded.commentIds[0]));
+    const expectedIds = seeded.commentIds.filter((_, index) => !filtersInput || index !== excludedIndex);
+    if (actorType === "cancelled_queued") await db.insert(heartbeatRuns).values({
+      companyId: seeded.companyId, agentId: seeded.agentId, status: "cancelled", runtimeMode: "legacy",
+      errorCode: "agent_paused", createdAt: new Date(0), finishedAt: new Date(1),
+      contextSnapshot: { issueId: seeded.issueId, wakeCommentIds: [seeded.commentIds[0]] },
+    });
+    if (actorType === "rejected_admission") {
+      await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
+        status: "cancelled", runtimeMode: "legacy", errorCode: "execution_reconciliation_required",
+        contextSnapshot: { issueId: seeded.issueId }, finishedAt: new Date(),
+      });
+    }
+    await Promise.all([heartbeatService(db).resumeQueuedRuns(), heartbeatService(db).resumeQueuedRuns()]);
+    const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
+    expect(delivered.status).toBe("coalesced");
+    const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!));
+    expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds,
+      previousRunId: seeded.runId, forceFreshSession: true });
+    if (actorType === "manual_receipt_other_actor") {
+      const [dispatch] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, successor.wakeupRequestId!));
+      expect(dispatch.requestedByActorId).toBe("queue-owner");
+      expect(dispatch.payload?.manualUserWake).toBeUndefined();
+      expect(successor.responsibleUserId).toBe("queue-owner");
+    }
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(["rejected_admission", "cancelled_queued"].includes(actorType) ? 4 : 3);
+    const [recovery] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId));
+    expect(recovery.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown" });
+  });
+
   it("recovers a message deferred after legacy finalization released the task lock", async () => {
     const seeded = await seedQueue();
     await db.update(agents).set({ adapterType: "claude_local",

server/src/services/execution-continuation.ts

@@ -118,7 +118,10 @@ export async function buildExecutionContinuation(input: {
   const triggerInteraction = interactions.find(
     (row) => row.id === input.context.interactionId,
   );
+  const explicitContinuation = object(input.context.explicitUserContinuation);
+  const explicitUserSource = string(explicitContinuation.previousRunId);
   const sourceRunId =
+    explicitUserSource ??
     triggerInteraction?.sourceRunId ??
     string(input.context.retryOfRunId) ??
     string(input.context.previousRunId);
@@ -137,7 +140,7 @@ export async function buildExecutionContinuation(input: {
       )[0]
     : null;
   if (sourceRunId && !sourceRun)
-    throw new Error("continuation_source_context_missing");
+    throw new Error(explicitUserSource ? "continuation_user_authorization_missing" : "continuation_source_context_missing");
   const originCommentIds = [
     ...new Set([
       ...continuationOriginCommentIds(input.context),
@@ -259,8 +262,6 @@ export async function buildExecutionContinuation(input: {
     ["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(run.status) &&
     !(run.status === "cancelled" && run.errorCode === "execution_reconciliation_required"),
   );
-  const explicitContinuation = object(input.context.explicitUserContinuation);
-  const explicitUserSource = string(explicitContinuation.previousRunId);
   if (explicitUserSource) {
     const predecessor = priorRuns.find(run => run.id === explicitUserSource &&
       ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status));
@@ -289,6 +290,7 @@ export async function buildExecutionContinuation(input: {
         eq(agentWakeupRequests.status, "coalesced"),
         sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
         sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`,
+        input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
       )) : [];
     const authorization = continuationAuthorizations.find(value => failedRunId
           ? value.failedRunId === failedRunId && retryWakes.some(wake =>

server/src/services/explicit-native-continuation.test.ts

@@ -180,6 +180,90 @@ const support = await getEmbeddedPostgresTestSupport();
     } }).where(eq(issueRecoveryActions.id, action.id));
     await expect(dispatch()).rejects.toThrow("continuation_user_authorization_missing");
   });
+  it.each(["valid", "stale_retry_context", "wrong_actor", "wrong_run", "wrong_company", "discarded"])("verifies another author's queued Interrupt at execution setup: %s", async kind => {
+    const f = await seed();
+    await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
+    await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+    await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null })
+      .where(eq(heartbeatRuns.id, f.sourceRunId));
+    await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
+      .where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
+    await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") })
+      .where(eq(issueComments.id, f.commentId));
+    await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
+      status: "running", contextSnapshot: { issueId: randomUUID() } });
+    const queueId = randomUUID();
+    const payload = { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] },
+      queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } };
+    await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
+      source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
+      requestedByActorType: "system", payload });
+    await heartbeatService(db).resumeQueuedCommentInterrupt(f.companyId, queueId);
+    const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId));
+    expect(wake.status).toBe("coalesced");
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
+    if (kind === "wrong_actor") await db.update(agentWakeupRequests).set({
+      payload: { ...payload, queuedCommentInterrupt: { ...payload.queuedCommentInterrupt, actorId: "forged" } },
+    }).where(eq(agentWakeupRequests.id, queueId));
+    if (kind === "wrong_run") await db.update(agentWakeupRequests).set({ runId: f.sourceRunId })
+      .where(eq(agentWakeupRequests.id, queueId));
+    if (kind === "wrong_company") await db.update(agentWakeupRequests).set({ companyId: (await seed()).companyId })
+      .where(eq(agentWakeupRequests.id, queueId));
+    if (kind === "discarded") await db.update(issueComments).set({ deletedAt: new Date() })
+      .where(eq(issueComments.id, f.commentId));
+    const result = buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
+      agentId: f.agentId, runId: run.id, context: {
+        ...run.contextSnapshot, ...(kind === "stale_retry_context" ? { retryOfRunId: randomUUID() } : {}),
+      }, summary: null, exposeLowTrustRaw: false });
+    if (kind === "valid" || kind === "stale_retry_context") await expect(result).resolves.toMatchObject({ interruptedRunId: f.sourceRunId });
+    else await expect(result).rejects.toThrow("continuation_user_authorization_missing");
+  });
+
+  it.each(["valid", "wrong_actor", "consumed", "discarded", "operator_stop", "already_delivered", "earlier_delivered", "unstarted_cancelled", "unstarted_cancelled_metadata", "foreign_queue"])("validates automatic saved-message delivery: %s", async kind => {
+    const f = await seed();
+    await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
+    await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+    await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null,
+      status: kind === "operator_stop" ? "cancelled" : "failed",
+      contextSnapshot: { issueId: f.issueId, ...(kind === "already_delivered" ? { wakeCommentIds: [f.commentId] } : {}) },
+    }).where(eq(heartbeatRuns.id, f.sourceRunId));
+    await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
+      .where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
+    await db.update(issueComments).set({ createdAt: new Date("2026-09-11T09:00:00Z"),
+      ...(kind === "discarded" ? { deletedAt: new Date() } : {}),
+    }).where(eq(issueComments.id, f.commentId));
+    const queueId = randomUUID();
+    await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
+      source: "automation", reason: "issue_commented", status: kind === "consumed" ? "coalesced" : "deferred_issue_execution",
+      requestedByActorType: "system", payload: { issueId: kind === "foreign_queue" ? randomUUID() : f.issueId,
+        _paperclipWakeContext: { wakeCommentIds: [f.commentId] } },
+    });
+    if (kind.startsWith("unstarted_cancelled")) {
+      await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
+        status: "cancelled", runtimeMode: "legacy", errorCode: "agent_paused", finishedAt: new Date(),
+        ...(kind === "unstarted_cancelled_metadata" ? { nativeIssueId: f.issueId, processPid: 999999999, processGroupId: 999999999 } : {}),
+        contextSnapshot: { issueId: f.issueId, wakeCommentIds: [f.commentId] },
+      });
+    }
+    if (kind === "earlier_delivered") {
+      const earlierId = randomUUID();
+      await db.insert(issueComments).values({ id: earlierId, companyId: f.companyId, issueId: f.issueId,
+        authorType: "user", authorUserId: f.actorId, body: "Already handled" });
+      await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeCommentIds: [earlierId] } })
+        .where(eq(heartbeatRuns.id, f.sourceRunId));
+      await db.update(agentWakeupRequests).set({ payload: { issueId: f.issueId,
+        _paperclipWakeContext: { wakeCommentIds: [earlierId, f.commentId] } } })
+        .where(eq(agentWakeupRequests.id, queueId));
+    }
+    const result = await db.transaction(async tx => {
+      await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
+      return admitExplicitNativeContinuation({ ...f, actorId: kind === "wrong_actor" ? "someone-else" : f.actorId,
+        db: tx as unknown as typeof db, queuedCommentRequestId: queueId, dryRun: true });
+    });
+    if (kind === "valid" || kind.startsWith("unstarted_cancelled")) expect(result).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId });
+    else expect(result).toBeNull();
+  });
+
   const admit = (f: Fixture, dryRun = false) => db.transaction(async tx => {
     await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
     const result = await admitExplicitNativeContinuation({ ...f, dryRun, db: tx as unknown as typeof db });

server/src/services/explicit-native-continuation.ts

@@ -25,6 +25,43 @@ function processStopped(pid: number): boolean {
   catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; }
 }
 
+/** Validate the whole saved queue, preserving order and original authors.
+ * Call again under the task lock before adopting IDs into a new run.
+ */
+export async function undeliveredLegacyUserCommentIds(
+  db: Db, companyId: string, issueId: string, agentId: string, commentIds: string[],
+): Promise<string[]> {
+  if (!commentIds.length) return [];
+  const comments = await db.select({ id: issueComments.id }).from(issueComments).where(and(
+    eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
+    inArray(issueComments.id, commentIds), eq(issueComments.authorType, "user"),
+    isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt),
+    sql`nullif(trim(${issueComments.body}), '') is not null`,
+    sql`nullif(trim(${issueComments.authorUserId}), '') is not null`,
+  ));
+  const valid = new Set(comments.map(comment => comment.id));
+  const previous = await db.select({ context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(and(
+    eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId),
+    sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
+    // A queued turn cancelled before dispatch has not consumed input, whatever
+    // cancelled it (pause, stale assignment, or rejected admission). Reserved
+    // native identity/process metadata is not proof that its prompt was sent.
+    // Admission separately verifies process termination before a new turn.
+    sql`not (${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.startedAt} is null)`,
+    or(...commentIds.map(id => or(
+      sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`,
+      sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`,
+    ))),
+  ));
+  for (const { context } of previous) {
+    if (typeof context?.wakeCommentId === "string") valid.delete(context.wakeCommentId);
+    if (Array.isArray(context?.wakeCommentIds)) {
+      for (const id of context.wakeCommentIds) if (typeof id === "string") valid.delete(id);
+    }
+  }
+  return commentIds.filter(id => valid.has(id));
+}
+
 /** Called under the issue lock, in the transaction that creates the new turn.
  * A user request authorizes a new conversation, not replay of the failed run.
  * Unknown action outcomes and all prior records stay intact.
@@ -36,6 +73,8 @@ export async function admitExplicitNativeContinuation(input: {
   failedRunId?: string | null;
   /** Server-recorded board intent to send an existing legacy message queue. */
   queuedCommentInterruptId?: string;
+  /** Internal delivery of an unconsumed, user-authored legacy queue entry. */
+  queuedCommentRequestId?: string;
   dryRun?: boolean;
   resumingSavedMessage?: boolean;
   onBlocked?: (reason: string, message: string) => void;
@@ -61,6 +100,21 @@ export async function admitExplicitNativeContinuation(input: {
   const queuedInterrupt = Boolean(interruptQueue && commentId &&
     queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId));
   if (input.queuedCommentInterruptId && !queuedInterrupt) return null;
+  const [savedQueue] = input.queuedCommentRequestId ? await db.select().from(agentWakeupRequests).where(and(
+    eq(agentWakeupRequests.id, input.queuedCommentRequestId),
+    eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId),
+    eq(agentWakeupRequests.status, "deferred_issue_execution"),
+    sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
+  )) : [];
+  const queuedRequest = Boolean(savedQueue && commentId &&
+    !savedQueue.idempotencyKey?.startsWith("chat-inbound:") &&
+    queuedCommentIdsFromWakePayload(savedQueue.payload).includes(commentId));
+  if (input.queuedCommentRequestId && !queuedRequest) return null;
+  if (queuedRequest) {
+    const ids = queuedCommentIdsFromWakePayload(savedQueue!.payload);
+    const undelivered = await undeliveredLegacyUserCommentIds(db, companyId, issueId, agentId, ids);
+    if (undelivered.length !== ids.length) return null;
+  }
   const [comment] = retry ? [] : await db.select().from(issueComments).where(and(
     eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
     eq(issueComments.id, commentId!), eq(issueComments.authorType, "user"),
@@ -104,7 +158,7 @@ export async function admitExplicitNativeContinuation(input: {
     if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
         (run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId ||
         !run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved.");
-    if (!queuedInterrupt && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
+    if (!queuedInterrupt && !queuedRequest && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
     if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically.");
     const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
       run.errorCode === "execution_reconciliation_required" &&
@@ -112,7 +166,12 @@ export async function admitExplicitNativeContinuation(input: {
     const legacyUserTurn = run.runtimeMode === "legacy" &&
       action.cause === "legacy_execution_requires_reconciliation" &&
       isConversationAdapter(agent.adapterType);
-    if (queuedInterrupt && !legacyUserTurn) return null;
+    if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission) return null;
+    // Saved input is a request for a new turn, never permission to undo an
+    // operator Stop or redeliver a message already consumed by this run.
+    if (queuedRequest && !queuedInterrupt && ((run.status === "cancelled" && !unusedAdmission) ||
+        run.contextSnapshot?.wakeCommentId === commentId ||
+        (Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(commentId)))) return null;
     if (legacyUserTurn) {
       const historicalAdapter = await historicalAdapterType(db, run);
       // A settings change never converts a known process/webhook execution into
@@ -181,7 +240,8 @@ export async function admitExplicitNativeContinuation(input: {
     summary: null, exposeLowTrustRaw: false });
   if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
   const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}),
-    ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), runId: input.successorRunId,
+    ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}),
+    ...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId,
     previousRunId: previous.id, recordedAt: new Date().toISOString() };
   for (const runId of cancelledStartupIds) {
     await db.update(nativeRunFinalizations).set({

server/src/services/heartbeat.ts

@@ -7,7 +7,7 @@ import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLea
 import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
 import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
 import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
-import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
+import { admitExplicitNativeContinuation, undeliveredLegacyUserCommentIds } from "./explicit-native-continuation.js";
 import { connectionIntentService } from "./connection-intents.js";
 import { prepareManagedAiRuntime, assertManagedAiProjectAuth, stripAiAuthBindings, AI_AUTH_ENV_KEYS } from "./ai-connection-runtime.js";
 import { aiConnectionBindingSchema } from "@paperclipai/shared";
@@ -3538,6 +3538,8 @@ interface WakeupOptions {
   manualUserWake?: boolean;
   /** Internal resume of a queue with persisted board interruption intent. */
   queuedCommentInterruptId?: string;
+  /** Internal delivery of an existing undelivered user comment. */
+  queuedCommentRequestId?: string;
   /** Exact failed run selected by an authenticated board Retry request. */
   failedRunId?: string | null;
   durableChatRequest?: DurableChatWakeupRequest;
@@ -10221,16 +10223,34 @@ export function heartbeatService(
   }
 
   async function resumeQueuedCommentInterrupt(companyId: string, queueId: string, opts?: { retryCleanup?: boolean }) {
+    return resumeSavedLegacyComments(companyId, queueId, true, opts);
+  }
+
+  async function resumeSavedLegacyComments(companyId: string, queueId: string, interrupted = false, opts?: { retryCleanup?: boolean }) {
     const [wake] = await db.select().from(agentWakeupRequests).where(and(
       eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId),
       eq(agentWakeupRequests.status, "deferred_issue_execution"),
     ));
     if (!wake) return;
     const payload = parseObject(wake.payload);
-    const actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId);
-    const commentIds = queuedCommentIdsFromWakePayload(payload);
+    let actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId);
+    let commentIds = queuedCommentIdsFromWakePayload(payload);
     const issueId = readNonEmptyString(payload.issueId);
-    if (!actorId || !issueId || !commentIds.length) return;
+    if (!issueId || !commentIds.length || wake.idempotencyKey?.startsWith("chat-inbound:")) return;
+    if (!interrupted) {
+      commentIds = await undeliveredLegacyUserCommentIds(db, companyId, issueId, wake.agentId, commentIds);
+      if (!commentIds.length) return;
+      // The queue itself may have begun as a system wake. The saved human
+      // comment, not that wake's origin or mutable caller payload, is authority.
+      const [comment] = await db.select().from(issueComments).where(and(
+        eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
+        eq(issueComments.id, commentIds[commentIds.length - 1]!), eq(issueComments.authorType, "user"),
+        isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt),
+      )).orderBy(desc(issueComments.createdAt)).limit(1);
+      if (!comment?.body.trim() || !comment.authorUserId) return;
+      actorId = comment.authorUserId;
+    }
+    if (!actorId) return;
     const agent = await getAgent(wake.agentId);
     if (!agent || agent.companyId !== companyId || agent.adapterType === "paperclip_runner") return;
     const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
@@ -10239,7 +10259,7 @@ export function heartbeatService(
       inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
     )).limit(1);
     if (active) return;
-    if (opts?.retryCleanup) {
+    if (interrupted && opts?.retryCleanup) {
       // Only the HTTP click grants an extra cleanup attempt. Periodic retries
       // reuse the intent to deliver, never a fresh provider teardown budget.
       const sourceRun = await db.transaction(async tx => {
@@ -10278,6 +10298,9 @@ export function heartbeatService(
         )).for("update");
         for (const lease of historical) {
           if (!lease.provider || lease.provider === "local" || !lease.providerLeaseId || hasRemoteTerminationReceipt(lease)) continue;
+          // Provider resource IDs identify physical sandboxes. A lease in any
+          // company can still own this resource; never destroy it on behalf of
+          // this company. This existence-only guard exposes no foreign data.
           const [otherOwner] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
             ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider),
             eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
@@ -10293,7 +10316,7 @@ export function heartbeatService(
         companyId, runId: sourceRun.id, actorId, reason: "queued_comment_interrupt",
       } });
     }
-    const deliveryPayload = { ...payload };
+    const deliveryPayload = withQueuedCommentIdsInWakePayload(payload, commentIds);
     delete deliveryPayload.queuedCommentInterrupt;
     await enqueueWakeup(wake.agentId, {
       source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
@@ -10301,9 +10324,9 @@ export function heartbeatService(
         issueId, triggeredBy: "board", actorId, responsibleUserId: actorId,
       }, commentIds),
       requestedByActorType: "user", requestedByActorId: actorId,
-      queuedCommentInterruptId: queueId,
+      ...(interrupted ? { queuedCommentInterruptId: queueId } : { queuedCommentRequestId: queueId }),
       issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] },
-      idempotencyKey: `queued-comment-interrupt:${queueId}`,
+      idempotencyKey: `queued-comment-${interrupted ? "interrupt" : "delivery"}:${queueId}`,
     }, queueId);
   }
 
@@ -18960,6 +18983,13 @@ export function heartbeatService(
         eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"),
       ));
       if (!latest || latest.runtimeMode !== "legacy" || !isHeartbeatRunTerminalStatus(latest.status)) continue;
+      const cancelledAdmission = latest.status === "cancelled" && !latest.startedAt &&
+        latest.errorCode === "execution_reconciliation_required";
+      if ((latest.status !== "cancelled" || cancelledAdmission) && await getExecutionBlocker(db, wake.companyId, String(wake.payload?.issueId))) {
+        await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => {
+          logger.warn({ err, queueId: wake.id }, "failed to deliver saved legacy comment after recovery stopped");
+        });
+      }
       await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true }).catch(err => {
         logger.warn({ err, queueId: wake.id }, "failed to promote stranded legacy comments");
       });
@@ -26271,12 +26301,13 @@ export function heartbeatService(
               eq(agentWakeupRequests.id, executionWaitRequestId), eq(agentWakeupRequests.companyId, agent.companyId),
               eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
               // A user message can join a queue originally created by a
-              // system wake. The recorded board click supplies fresh authority.
-              opts.queuedCommentInterruptId === executionWaitRequestId
+              // system wake. Admission validates the saved user comment or board click.
+              (opts.queuedCommentInterruptId ?? opts.queuedCommentRequestId) === executionWaitRequestId
                 ? undefined : eq(agentWakeupRequests.requestedByActorType, "user"),
               opts.queuedCommentInterruptId === executionWaitRequestId
                 ? sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${opts.requestedByActorId ?? ""}`
-                : eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
+                : opts.queuedCommentRequestId === executionWaitRequestId ? undefined
+                  : eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
               sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
             ));
             // The issue lock serializes cleanup callbacks and periodic workers.
@@ -26293,7 +26324,16 @@ export function heartbeatService(
               sql`length(trim(${issueComments.body})) > 0`,
             ));
             if (!comment) return { kind: "deferred" as const };
-            if (!opts.queuedCommentInterruptId && pending.payload?.manualUserWake === true) {
+            if (opts.queuedCommentRequestId) {
+              const ids = await undeliveredLegacyUserCommentIds(tx as unknown as Db,
+                agent.companyId, issueId, agentId, queuedCommentIdsFromWakePayload(pending.payload));
+              if (!ids.includes(wakeCommentId)) return { kind: "deferred" as const };
+              pending.payload = withQueuedCommentIdsInWakePayload(parseObject(pending.payload), ids);
+              await tx.update(agentWakeupRequests).set({ payload: pending.payload }).where(and(
+                eq(agentWakeupRequests.id, pending.id), eq(agentWakeupRequests.companyId, agent.companyId),
+              ));
+            }
+            if (!opts.queuedCommentInterruptId && !opts.queuedCommentRequestId && pending.payload?.manualUserWake === true) {
               // A persisted manual wake keeps its actor when an execution wait
               // resumes. The locked receipt above has revalidated that actor.
               payload = { ...payload, manualUserWake: true };
@@ -26303,6 +26343,8 @@ export function heartbeatService(
               // The locked board receipt supplies execution authority even when
               // another user authored the messages. Dispatch revalidates the receipt.
               operatorResponsibleUserId = opts.requestedByActorId!;
+            }
+            if (opts.queuedCommentInterruptId || opts.queuedCommentRequestId) {
               // Edits/discards between the click and dispatch remain authoritative.
               Object.assign(enrichedContextSnapshot, withQueuedCommentIdsInRunContext(
                 enrichedContextSnapshot, queuedCommentIdsFromWakePayload(pending.payload),
@@ -26688,6 +26730,7 @@ export function heartbeatService(
             agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
             reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
             queuedCommentInterruptId: opts.queuedCommentInterruptId,
+            queuedCommentRequestId: opts.queuedCommentRequestId,
             dryRun: true,
             onBlocked: (reason, message) => { continuationWait = { reason, message }; },
           }))) return deferBlockedExecution(executionBlocker);
@@ -27457,6 +27500,7 @@ export function heartbeatService(
               continuationWait = { reason, message };
             },
             queuedCommentInterruptId: opts.queuedCommentInterruptId,
+            queuedCommentRequestId: opts.queuedCommentRequestId,
           });
           // Recovery can change while earlier admission gates await I/O. Use
           // the current blocker, not the snapshot from the start of admission.
@@ -27507,7 +27551,7 @@ export function heartbeatService(
                   .orderBy(asc(agentWakeupRequests.requestedAt))
               : [];
           const adoptedComments = pendingComments.filter((wake) => {
-            if (wake.id === opts.queuedCommentInterruptId) return true;
+            if (wake.id === opts.queuedCommentInterruptId || wake.id === opts.queuedCommentRequestId) return true;
             const deferredPayload = parseObject(wake.payload);
             const deferredContext = parseObject(
               deferredPayload[DEFERRED_WAKE_CONTEXT_KEY],
@@ -27524,7 +27568,7 @@ export function heartbeatService(
               queuedCommentIdsFromWakePayload(wake.payload).length > 0
             );
           });
-          const adoptedCommentIds = [
+          let adoptedCommentIds = [
             ...new Set([
               ...adoptedComments.flatMap((wake) =>
                 queuedCommentIdsFromWakePayload(wake.payload),
@@ -27532,6 +27576,10 @@ export function heartbeatService(
               ...queuedCommentIdsFromRunContext(enrichedContextSnapshot),
             ]),
           ];
+          if (opts.queuedCommentRequestId) {
+            adoptedCommentIds = await undeliveredLegacyUserCommentIds(tx as unknown as Db,
+              agent.companyId, issueId, agentId, adoptedCommentIds);
+          }
           const newRun = await tx
             .insert(heartbeatRuns)
             .values({