Preserve accepted messages for paused recipients (#16679)

Keep accepted deferred requests pending while their recipient is paused and deliver through normal gates after resume. Preserve company, task hold, and cleanup checks.

Verification: 247 focused checks, recursive typecheck, the full CI test and build matrix, and Apex 5/5. The local Git fixture initial-branch limitation is documented in the PR.

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

server/src/modules/wake-queue/adapters/postgres.test.ts

@@ -177,6 +177,41 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
     return id;
   }
 
+  it.each(["assignee", "mentioned"])("preserves a paused recipient's accepted message until resume (%s)", async kind => {
+    const companyId = await seedCompany();
+    const ownerId = await seedAgent({ companyId });
+    const recipientId = kind === "assignee" ? ownerId : await seedAgent({ companyId });
+    await db.update(agents).set({ status: "paused" }).where(eq(agents.id, recipientId));
+    const issueId = await seedIssue({ companyId, status: "blocked", assigneeAgentId: ownerId });
+    const runId = await seedRun({ companyId, agentId: ownerId, status: "succeeded", contextSnapshot: { issueId } });
+    await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId));
+    const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user",
+      authorUserId: "responsible-user", body: "Accepted input must survive an agent pause" }).returning();
+    const wakeId = await seedDeferredWake({ companyId, agentId: recipientId, issueId, requestedByActorId: "responsible-user",
+      payload: { _paperclipWakeContext: { issueId, wakeReason: kind === "mentioned" ? "issue_comment_mentioned" : "issue_commented",
+        wakeCommentIds: [comment.id] } },
+    });
+    const [accepted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId));
+    const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps),
+      recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined },
+    });
+    const input = { companyId, runId, now: new Date(), suppressImmediateRecovery: true, resumeStrandedQueue: true };
+    for (let pass = 0; pass < 2; pass++) {
+      expect((await release(input)).outcome.kind).toBe("released");
+      expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]).toEqual(accepted);
+      expect(await db.select().from(heartbeatRuns)).toHaveLength(1);
+    }
+    await db.update(agents).set({ status: "active" }).where(eq(agents.id, recipientId));
+    expect((await release(input)).outcome.kind).toBe("promoted");
+    const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId));
+    expect(delivered).toMatchObject({ status: "queued", runId: expect.any(String), payload: accepted.payload });
+    const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!));
+    expect(successor).toMatchObject({ agentId: recipientId, contextSnapshot: { wakeCommentIds: [comment.id] } });
+    await release(input);
+    expect(await db.select().from(heartbeatRuns)).toHaveLength(2);
+    expect((await db.select().from(issueComments).where(eq(issueComments.id, comment.id)))[0].body).toBe(comment.body);
+  });
+
   it.each(["queued", "running", "scheduled_retry", "cancelled"])("preserves a stranded queue when another agent has a newer %s turn", async status => {
     const companyId = await seedCompany();
     const agentId = await seedAgent({ companyId });

server/src/modules/wake-queue/adapters/postgres.ts

@@ -213,7 +213,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .then((rows) => rows[0] ?? null);
       if (!agent) return null;
       const invokability = await evaluateAgentInvokabilityFromDb(tx, agent);
-      return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable };
+      return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable,
+        paused: !invokability.invokable && invokability.reason === "paused" };
     },
 
     async findNextDeferredWake({ companyId, issueId, excludedWakeIds }) {

server/src/modules/wake-queue/application/types.ts

@@ -58,6 +58,7 @@ export type InvokableAgentSnapshot = {
   companyId: string;
   name: string | null;
   invokable: boolean;
+  paused?: boolean;
 };
 
 /** A new heartbeat run reached the queued state and should be published and dispatched. */

server/src/modules/wake-queue/application/use-cases.test.ts

@@ -146,6 +146,28 @@ function createFakeRecovery(): RecoveryEscalationPort {
 }
 
 describe("releaseIssueExecution", () => {
+  it.each([false, true])("preserves a paused head without blocking an eligible follower (%s)", async withFollower => {
+    const paused = wakeCandidate({ id: "paused-wake", wakeReason: "issue_comment_mentioned", queuedCommentIds: ["paused-comment"] });
+    const ready = wakeCandidate({ id: "ready-wake", agentId: "ready-agent", queuedCommentIds: ["ready-comment"] });
+    const pending = withFollower ? [paused, ready] : [paused];
+    const transaction = createFakeTransaction({
+      findNextDeferredWake: vi.fn(async input => pending.find(wake => !input.excludedWakeIds?.includes(wake.id)) ?? null),
+      findInvokableAgent: vi.fn(async input => input.agentId === paused.agentId
+        ? { ...AGENT, invokable: false, paused: true } : { ...AGENT, id: ready.agentId }),
+      getQueuedCommentLiveness: vi.fn(async input => ({ liveNonSelfCommentIds: input.queuedCommentIds, containedSelfAuthoredComment: false })),
+    });
+    const release = createReleaseIssueExecution({ issueLock: createFakeIssueLock(createFakeHost(), transaction,
+      { ...ISSUE, status: "blocked", assigneeAgentId: ready.agentId }, { ...RUN, status: "succeeded" }), recovery: createFakeRecovery() });
+    const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date(), resumeStrandedQueue: true });
+    expect(result.outcome.kind).toBe(withFollower ? "promoted" : "released");
+    expect(transaction.failDeferredWake).not.toHaveBeenCalled();
+    expect(transaction.cancelDeferredWake).not.toHaveBeenCalled();
+    expect(transaction.normalizeDeferredWakeCommentIds).not.toHaveBeenCalled();
+    expect(transaction.findNextDeferredWake).toHaveBeenNthCalledWith(2, expect.objectContaining({ excludedWakeIds: [paused.id] }));
+    if (withFollower) expect(transaction.finalizePromotedWake).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ wakeId: ready.id }));
+    else expect(transaction.finalizePromotedWake).not.toHaveBeenCalled();
+  });
+
   it.each(["closing", "mixed_comments", "explicit_resume", "interaction", "other_source_task", "parent_open"])(
     "checks late completed-delegation mentions without dropping independent input (%s)", async (scenario) => {
       const originalCommentIds = ["closing-comment", "second-comment"];

server/src/modules/wake-queue/application/use-cases.ts

@@ -152,12 +152,12 @@ async function runReleaseDrain(
   // The `processedWakeIds` guard below makes a break of the invariant
   // fail loudly, instead of holding this transaction open forever.
   const processedWakeIds = new Set<string>();
-  const handoffWakeIds: string[] = [];
+  const preservedWakeIds: string[] = [];
 
   while (true) {
     const candidate = await ports.transaction.findNextDeferredWake({
       companyId: run.companyId, issueId: issue.id,
-      ...(handoffWakeIds.length ? { excludedWakeIds: handoffWakeIds } : {}),
+      ...(preservedWakeIds.length ? { excludedWakeIds: preservedWakeIds } : {}),
     });
     if (!candidate) break;
     if (processedWakeIds.has(candidate.id)) {
@@ -205,7 +205,7 @@ async function runReleaseDrain(
         // The old owner can release before assignment admission adopts these
         // exact IDs. Leave its receipt intact, skip it for this drain, and let
         // a current-assignee wake behind it proceed.
-        handoffWakeIds.push(candidate.id);
+        preservedWakeIds.push(candidate.id);
       } else {
         // The current owner has finished. An obsolete assignment cannot
         // launch another former-owner run or reopen its completed task.
@@ -283,7 +283,7 @@ async function runReleaseDrain(
 
     // A comment-id rewrite cannot change the agent or pause-hold facts already fetched above, so one decision covers both the rewritten and un-rewritten cases.
     const wakeOutcome = decideWakeOutcome({
-      agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false },
+      agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false, paused: deferredAgent?.paused === true },
       pauseHold: { activePauseHold: pauseHold.activePauseHold, treeHoldInteractionWake: pauseHold.treeHoldInteractionWake },
     });
 
@@ -292,6 +292,11 @@ async function runReleaseDrain(
       continue;
     }
 
+    if (wakeOutcome.kind === "defer_agent_pause") {
+      preservedWakeIds.push(workingCandidate.id);
+      continue;
+    }
+
     if (wakeOutcome.kind === "cancel_pause_hold") {
       await ports.transaction.cancelDeferredWake({
         companyId: run.companyId,

server/src/modules/wake-queue/domain/policy.test.ts

@@ -247,6 +247,28 @@ describe("decideWakeOutcome", () => {
       facts: { ...baseWakeOutcomeFacts, agent: { agentFound: false, invokable: false } },
       expected: { kind: "fail_not_invokable" },
     },
+    {
+      name: "defer_agent_pause: a paused recipient keeps its accepted request",
+      facts: { ...baseWakeOutcomeFacts, agent: { agentFound: true, invokable: false, paused: true } },
+      expected: { kind: "defer_agent_pause" },
+    },
+    {
+      name: "fail_not_invokable: a pause cannot authorize a missing recipient",
+      facts: { ...baseWakeOutcomeFacts, agent: { agentFound: false, invokable: false, paused: true } },
+      expected: { kind: "fail_not_invokable" },
+    },
+    {
+      name: "cancel_pause_hold: an explicit task hold still suppresses a paused recipient",
+      facts: { agent: { agentFound: true, invokable: false, paused: true },
+        pauseHold: { activePauseHold: true, treeHoldInteractionWake: false } },
+      expected: { kind: "cancel_pause_hold" },
+    },
+    {
+      name: "defer_agent_pause: a permitted tree interaction cannot invoke a paused agent",
+      facts: { agent: { agentFound: true, invokable: false, paused: true },
+        pauseHold: { activePauseHold: true, treeHoldInteractionWake: true } },
+      expected: { kind: "defer_agent_pause" },
+    },
     {
       name: "fail_not_invokable: the agent is found but not invokable",
       facts: { ...baseWakeOutcomeFacts, agent: { agentFound: true, invokable: false } },

server/src/modules/wake-queue/domain/policy.ts

@@ -142,6 +142,7 @@ export type DeferredWakeAgentFacts = {
   agentFound: boolean;
   /** True when the agent is invokable (status, org chain). Meaningless when agentFound is false. */
   invokable: boolean;
+  paused?: boolean;
 };
 
 export type DeferredWakePauseHoldFacts = {
@@ -158,6 +159,7 @@ export type DeferredWakeOutcomeFacts = {
 
 export type DeferredWakeOutcomeDecision =
   | { kind: "fail_not_invokable" }
+  | { kind: "defer_agent_pause" }
   | { kind: "cancel_pause_hold" }
   | { kind: "promote" };
 
@@ -180,11 +182,10 @@ export function decideQueuedCommentAction(
   return { kind: "proceed" };
 }
 
-/** Decides the outcome for a deferred wake whose queued-comment action resolved to "proceed" (or finished its rewrite): fail, cancel, or promote. */
 export function decideWakeOutcome(facts: DeferredWakeOutcomeFacts): DeferredWakeOutcomeDecision {
   const { agent, pauseHold } = facts;
 
-  if (!agent.agentFound || !agent.invokable) {
+  if (!agent.agentFound || (!agent.invokable && !agent.paused)) {
     return { kind: "fail_not_invokable" };
   }
 
@@ -192,6 +193,8 @@ export function decideWakeOutcome(facts: DeferredWakeOutcomeFacts): DeferredWake
     return { kind: "cancel_pause_hold" };
   }
 
+  if (agent.paused) return { kind: "defer_agent_pause" };
+
   return { kind: "promote" };
 }
 

server/src/services/heartbeat/queue.test.ts

@@ -6,6 +6,8 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vites
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
 import { createRunDispatch } from "../../modules/run-dispatch/index.js";
 import type { createWakeQueue } from "../../modules/wake-queue/index.js";
+import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js";
+import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js";
 import { createHeartbeatRunState } from "./run-state.js";
 import { createHeartbeatRunPreparation } from "./run-preparation.js";
 import { createHeartbeatQueue, type HeartbeatQueueDependencies } from "./queue.js";
@@ -51,7 +53,7 @@ function callbacks(db: Db) {
     canStartNewExternalBotWork: vi.fn(async () => true),
     executeRun: vi.fn(async () => undefined),
     applyWakeQueuePostCommitEffects: vi.fn(async () => undefined),
-    releaseIssueExecutionAndPromote: vi.fn(async () => undefined),
+    releaseIssueExecutionAndPromote: vi.fn<HeartbeatQueueDependencies["releaseIssueExecutionAndPromote"]>(async () => undefined),
     publishRunLifecyclePluginEvent: vi.fn(),
     cancelRunInternal: vi.fn(async () => undefined),
     cancelActiveForAgentInternal: vi.fn(async () => undefined),
@@ -165,6 +167,29 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => {
     return { ...seeded, recipient, pending };
   }
 
+  it.each(["legacy", "native"])("periodic recovery preserves a paused recipient through the real %s release adapter", async runtimeMode => {
+    const { company, issue, recipient, pending } = await strandedMention(runtimeMode);
+    await db.update(agents).set({ status: "paused" }).where(eq(agents.id, recipient.id));
+    await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issue.id));
+    const deps = callbacks(db);
+    const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, {
+      resolveResponsibleUserId: async () => "responsible-user",
+      getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }),
+      resolveSessionBeforeForWakeup: async () => null,
+    }), recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined } });
+    deps.releaseIssueExecutionAndPromote.mockImplementation(async (run, options) => {
+      await release({ companyId: run.companyId, runId: run.id, now: new Date(), ...options });
+    });
+    for (let pass = 0; pass < 2; pass++) {
+      await createHeartbeatQueue(db, deps).resumeQueuedRuns();
+      expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, pending.id)))[0])
+        .toMatchObject({ status: "deferred_issue_execution", runId: null, finishedAt: null, error: null, payload: pending.payload });
+      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, company.id))).toHaveLength(1);
+    }
+    expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(2);
+    expect(deps.executeRun).not.toHaveBeenCalled();
+  });
+
   it.each(["legacy", "native"])("resumes an accepted non-assignee queue through its %s owner", async runtimeMode => {
     const { run, pending } = await strandedMention(runtimeMode);
     const deps = callbacks(db);