fix: resume queued turns after maintenance (#17967)

Keep an accepted turn's starting input in task history, show its admission wait, and resume existing queued runs promptly when maintenance ends or expires.

Verified with behavioral red/green regressions, real PostgreSQL and process execution, browser/native Runner acceptance with a deterministic provider, workspace typecheck, production build, and current-head CI/review.

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

DESIGN.md

@@ -87,13 +87,19 @@ that relayed a message from another channel.
 
 ## Contextual feedback
 
-Keep the runner turn's avatar, name, Working header and current-activity slot
+Keep the runner turn's avatar, name, status header and current-activity slot
 in place from message submission through queued admission and native startup.
 Saving the human comment does not end startup feedback. Keep the identity mounted
 until its run appears or a newer execution outcome ends admission.
 Runtime discovery must not replace the turn with a generic spinner or a
 "run started" receipt. Queued and running headers use the same spacing.
 
+A queued turn shows `Queued`, or `Waiting for maintenance` while the server holds
+admission. Reserve `Working` and `Thinking` for execution. A saved message that
+starts this queued turn stays in history, including when it reopens an older
+task. This placement is not a consumption receipt. Later follow-ups remain in
+the composer queue with their existing edit, order and discard controls.
+
 Stream the provider's final reply as the conversational answer. Completion-tool
 reports remain inspectable run data while work continues. Use their summary as
 an answer fallback only after the run ends and its final transcript fetch

doc/execution-semantics.md

@@ -326,6 +326,16 @@ continues automatically when the workspace is available. Internal scheduling
 attempts remain in the run log without conversation cancellation markers,
 cancellation toasts, or manual Retry controls. Users can keep sending instructions.
 
+An instance admission drain keeps accepted new turns queued and displays
+**Waiting for maintenance**. Ending the audited drain requests admission for
+these existing runs immediately. The shared delivery coordinator owns the drain's
+actual expiry deadline and cancels it on shutdown; expiry also wakes admission.
+Neither path waits for a general recovery scan or creates a new wake
+request. The ordinary claim and dispatch checks still enforce company status,
+identity, budgets, pauses, ownership, and concurrency. A restarted server restores
+queued work through startup recovery. A failed admission request leaves its
+durable rows available for existing recovery.
+
 The legacy remote ACP process-session relay runs on the control-plane host. Its
 launch command uses the host's absolute Node executable even when the adapter's
 launch environment is sanitized for a remote sandbox; the sandbox PATH remains

maps/agent-access-map/steering.md

@@ -33,6 +33,10 @@ Compact task run reads retain `wakeupRequestId` and `queuedCommentSourceWakeIds`
 to identify consumed input queues, including coalesced requests. These receipt
 fields follow the existing task/run read checks and grant no queue mutation authority.
 
+The UI can place a queued turn's starting input in history without recording
+consumption. Drain release and expiry only reconsider existing queued runs under
+normal admission authority. They grant no new agent control or replay authority.
+
 Accepted deferred requests remain pending while their recipient agent is paused.
 Resume uses the normal admission checks. This grants no new pause, resume, queue
 mutation, or execution authority to an agent.

maps/feature-map/steering.md

@@ -77,6 +77,10 @@ consumed message whose send response and queue snapshot arrive late. A started
 run's input receipt removes that message from pending, while a separate unsent
 message stays queued. Admission to a run that has not started is not delivery.
 A later retry of the same message remains visible in its new authoritative queue.
+A queued turn's own starting input stays in history before execution, including
+board messages that reopen a task. Later inputs keep their queue controls and
+full canonical order. A maintenance hold displays its current admission wait;
+drain release and expiry prompt normal admission without a manual retry.
 Receipts retain the primary wake and coalesced source queue IDs. They clear a late
 snapshot even when its target names the preceding run or is null.
 

server/src/__tests__/heartbeat-task-drain-admission-release.test.ts

@@ -1,4 +1,9 @@
 import { randomUUID } from "node:crypto";
+import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { registerTaskDrainExpiryWork } from "../services/task-admission.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
 import {
@@ -22,6 +27,10 @@ import {
 } from "./helpers/embedded-postgres.js";
 import { heartbeatService, getTaskDrainStatus, startTaskDrain, stopTaskDrain } from "../services/heartbeat.ts";
 import { subscribeCompanyLiveEvents } from "../services/live-events.ts";
+import express from "express";
+import request from "supertest";
+import { instanceSettingsRoutes } from "../routes/instance-settings.js";
+import { errorHandler } from "../middleware/error-handler.js";
 
 const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
@@ -234,6 +243,88 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => {
     expect(finished?.status).toBe("succeeded");
   }, 20_000);
 
+  it.each(["release", "paused agent", "new drain"] as const)("reconsiders queued work after maintenance with current gates: %s", async gate => {
+    const { companyId, agentId, issueId } = await seedAgentAndIssue();
+    const heartbeat = heartbeatService(db);
+    const counterDir = mkdtempSync(join(tmpdir(), "drain-executor-"));
+    const counter = join(counterDir, "executions");
+    await db.update(agents).set({ adapterConfig: { command: process.execPath,
+      args: ["-e", `require("node:fs").appendFileSync(${JSON.stringify(counter)}, process.env.PAPERCLIP_RUN_ID + "\\n")`] } })
+      .where(eq(agents.id, agentId));
+    startTaskDrain({});
+    await heartbeat.wakeup(agentId, {
+      source: "assignment", triggerDetail: "system", reason: "issue_assigned",
+      payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" },
+      requestedByActorType: "system", requestedByActorId: "issue_assignment",
+    });
+    await heartbeat.drainActiveRunExecutions();
+    const [queued] = await db.select({ id: heartbeatRuns.id, wakeupRequestId: heartbeatRuns.wakeupRequestId })
+      .from(heartbeatRuns);
+    expect(queued).toBeDefined();
+    if (gate === "paused agent") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId));
+    const unsubscribe = subscribeCompanyLiveEvents(companyId, event => {
+      if (gate === "new drain" && event.type === "activity.logged" && event.payload.action === "instance.task_drain.stopped") startTaskDrain({});
+    });
+    try {
+      const app = express();
+      app.use(express.json());
+      app.use((req, _res, next) => {
+        req.actor = { type: "board", userId: "local-board", source: "local_implicit", isInstanceAdmin: true };
+        next();
+      });
+      app.use("/api", instanceSettingsRoutes(db));
+      app.use(errorHandler);
+      const response = await request(app).delete("/api/instance/task-drain");
+      expect(response.status).toBe(200);
+      await heartbeat.drainActiveRunExecutions();
+      const runs = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, startedAt: heartbeatRuns.startedAt })
+        .from(heartbeatRuns).where(eq(heartbeatRuns.wakeupRequestId, queued!.wakeupRequestId!));
+      expect(runs).toHaveLength(1);
+      const executions = existsSync(counter) ? readFileSync(counter, "utf8").trim().split("\n") : [];
+      if (gate === "release") {
+        expect(runs[0]).toMatchObject({ id: queued!.id, status: "succeeded" });
+        expect(executions.filter(id => id === queued!.id)).toHaveLength(1);
+        await request(app).delete("/api/instance/task-drain").expect(200);
+        await heartbeat.drainActiveRunExecutions();
+        expect(readFileSync(counter, "utf8").trim().split("\n").filter(id => id === queued!.id)).toHaveLength(1);
+      } else {
+        expect(runs[0]?.startedAt).toBeNull();
+        expect(executions).toEqual([]);
+      }
+    } finally {
+      unsubscribe();
+      await heartbeat.drainActiveRunExecutions();
+      rmSync(counterDir, { recursive: true, force: true });
+    }
+  }, 20_000);
+
+  it("dispatches existing queued work at the drain deadline", async () => {
+    const { agentId, issueId } = await seedAgentAndIssue();
+    const heartbeat = heartbeatService(db);
+    startTaskDrain({});
+    await heartbeat.wakeup(agentId, {
+      source: "assignment", triggerDetail: "system", reason: "issue_assigned",
+      payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" },
+      requestedByActorType: "system", requestedByActorId: "issue_assignment",
+    });
+    await heartbeat.drainActiveRunExecutions();
+    const [queued] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns);
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      onError: error => { throw error; } });
+    let worker: ReturnType<typeof registerTaskDrainExpiryWork> | undefined;
+    try {
+      await new Promise<void>((resolve, reject) => {
+        worker = registerTaskDrainExpiryWork(coordinator, () => {
+          void heartbeat.resumeQueuedRunsAfterDrain().then(resolve, reject);
+        });
+        startTaskDrain({ ttlMs: 5 });
+      });
+      await heartbeat.drainActiveRunExecutions();
+      const [run] = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, queued!.id));
+      expect(run?.status).toBe("succeeded");
+    } finally { await worker?.stop(); await coordinator.stop(); }
+  }, 20_000);
+
   it("keeps a wake that arrives during a task drain queued and runs it once the drain lifts", async () => {
     const { agentId, issueId } = await seedAgentAndIssue();
     const heartbeat = heartbeatService(db);

server/src/__tests__/instance-settings-routes.test.ts

@@ -18,6 +18,7 @@ const mockHeartbeatService = vi.hoisted(() => ({
   applyTaskDrain: vi.fn(),
   stopTaskDrain: vi.fn(),
   getTaskDrainStatus: vi.fn(),
+  resumeQueuedRunsAfterDrain: vi.fn(),
 }));
 const mockEnvironmentService = vi.hoisted(() => ({
   getById: vi.fn(),
@@ -119,6 +120,8 @@ describe("instance settings routes", () => {
     mockHeartbeatService.applyTaskDrain.mockReset();
     mockHeartbeatService.stopTaskDrain.mockReset();
     mockHeartbeatService.getTaskDrainStatus.mockReset();
+    mockHeartbeatService.resumeQueuedRunsAfterDrain.mockReset();
+    mockHeartbeatService.resumeQueuedRunsAfterDrain.mockResolvedValue(undefined);
     mockEnvironmentService.getById.mockReset();
     mockEnvironmentService.findManagedSandboxEnvironment.mockReset();
     mockEnvironmentService.findManagedSandboxEnvironment.mockResolvedValue(null);
@@ -1345,6 +1348,23 @@ describe("instance settings routes", () => {
       expect(mockHeartbeatService.stopTaskDrain).not.toHaveBeenCalled();
     });
 
+    it("does not dispatch work when the drain-stop audit transaction rejects", async () => {
+      mockHeartbeatService.getTaskDrainStatus.mockReturnValue({ ...idleStatus, draining: true });
+      mockLogActivity.mockRejectedValue(new Error("activity insert failed"));
+      const res = await request(createApp(adminActor)).delete("/api/instance/task-drain");
+      expect(res.status).toBeGreaterThanOrEqual(500);
+      expect(mockHeartbeatService.resumeQueuedRunsAfterDrain).not.toHaveBeenCalled();
+    });
+
+    it("reports the committed drain release even when queued dispatch fails", async () => {
+      mockHeartbeatService.getTaskDrainStatus.mockReturnValue({ ...idleStatus, draining: true });
+      mockHeartbeatService.resumeQueuedRunsAfterDrain.mockRejectedValue(new Error("database unavailable"));
+      const res = await request(createApp(adminActor)).delete("/api/instance/task-drain");
+      expect(res.status).toBe(200);
+      expect(res.body).toEqual({ wasActive: true });
+      expect(mockHeartbeatService.resumeQueuedRunsAfterDrain).toHaveBeenCalledTimes(1);
+    });
+
     it("does not stop the drain when the audit transaction rejects", async () => {
       mockHeartbeatService.getTaskDrainStatus.mockReturnValue({
         draining: true,

server/src/__tests__/task-admission-deadline.test.ts

@@ -0,0 +1,74 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { registerTaskDrainExpiryWork, readTaskDrain, startTaskDrain, stopTaskDrain } from "../services/task-admission.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+
+describe("task drain deadline", () => {
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
+  let worker: ReturnType<typeof registerTaskDrainExpiryWork>;
+  let resume = vi.fn<() => void>();
+  const execute = vi.fn(async () => []);
+  const transaction = async () => { throw new Error("No transaction is expected for a process-local deadline"); };
+  beforeEach(async () => {
+    vi.useFakeTimers();
+    resume = vi.fn<() => void>();
+    execute.mockClear();
+    coordinator = createDeliveryWorkCoordinator({ owner: installDatabaseWorkSignals({ execute, transaction }), canRun: () => false,
+      canReconcile: () => true, onError: error => { throw error; } });
+    worker = registerTaskDrainExpiryWork(coordinator, resume);
+    await worker.ready;
+  });
+  afterEach(async () => {
+    await worker.stop(); await coordinator.stop(); stopTaskDrain(); vi.useRealTimers();
+  });
+
+  it("releases admission once at its deadline without a status read or periodic sweep", async () => {
+    startTaskDrain({ ttlMs: 1000 });
+    await vi.advanceTimersByTimeAsync(0);
+    expect(coordinator.nextWakeAt()).toBe(Date.now() + 1000);
+    await vi.advanceTimersByTimeAsync(999);
+    expect(resume).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(1);
+    expect(resume).toHaveBeenCalledTimes(1);
+    expect(readTaskDrain(new Date())).toBeNull();
+    await vi.advanceTimersByTimeAsync(3_600_000);
+    expect(resume).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+    expect(execute).not.toHaveBeenCalled();
+  });
+
+  it.each(["replace", "stop", "shutdown"] as const)("does not resume from an obsolete expiry after %s", async action => {
+    startTaskDrain({ ttlMs: 1000 });
+    await vi.advanceTimersByTimeAsync(0);
+    if (action === "replace") startTaskDrain({ ttlMs: null });
+    if (action === "stop") stopTaskDrain();
+    if (action === "shutdown") {
+      await worker.stop();
+      await coordinator.stop();
+      expect(vi.getTimerCount()).toBe(0);
+    }
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(resume).not.toHaveBeenCalled();
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("removes its listener on close and rearms an existing drain on a new lifecycle", async () => {
+    startTaskDrain({ ttlMs: 1000 });
+    await vi.advanceTimersByTimeAsync(0);
+    await worker.stop();
+    expect(coordinator.nextWakeAt()).toBeNull();
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(readTaskDrain(new Date())).toBeNull();
+    expect(resume).not.toHaveBeenCalled();
+    startTaskDrain({ ttlMs: 1000 });
+    const next = createDeliveryWorkCoordinator({ owner: installDatabaseWorkSignals({ execute, transaction }), canRun: () => true, onError: vi.fn() });
+    const resumed = vi.fn();
+    const nextWorker = registerTaskDrainExpiryWork(next, resumed);
+    try {
+      await nextWorker.ready;
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(resumed).toHaveBeenCalledOnce();
+      expect(resume).not.toHaveBeenCalled();
+    } finally { await nextWorker.stop(); await next.stop(); }
+  });
+});

server/src/index.ts

@@ -46,7 +46,7 @@ import { caseHandoffDeliveryService } from "./services/case-handoff-delivery.js"
 import { registerScheduledWork } from "./services/scheduled-work.js";
 import type { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js";
 import { registerTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./services/terminal-workspace-work.js";
-import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js";
+import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork, registerTaskDrainExpiryWork } from "./services/task-admission.js";
 import { markIdleStartupComplete, registerIdleLocalWorkProbe } from "./services/idle-local-work.js";
 import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./services/idle-database-backup.js";
 import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js";
@@ -1368,6 +1368,13 @@ async function startServerWithDatabaseTeardown(
   if (config.nativeHarnessBackupExpirationEnabled) {
     nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork);
   }
+  const taskDrainExpiryWork = registerTaskDrainExpiryWork(deliveryWork, () => {
+    if (heartbeatSchedulerStopped || isWarmStandby()) return;
+    trackHeartbeatSchedulerWork(accountingHeartbeat.resumeQueuedRunsAfterDrain().catch(err => {
+      logger.error({ err }, "failed to resume queued work after task drain expiry");
+    }));
+  });
+  server.on("close", () => { void taskDrainExpiryWork.stop(); });
   const executionSchedule = executionControlSchedule(db);
   const workspaceRecoveryRuntime = environmentRuntimeService(db, { pluginWorkerManager });
   const workspaceRecoveryWork = registerWorkspaceRecoveryWork(db, deliveryWork, workspaceRecoveryRuntime,
@@ -2122,6 +2129,7 @@ async function startServerWithDatabaseTeardown(
   ) => {
     await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]);
     heartbeatSchedulerStopped = true;
+    await taskDrainExpiryWork.stop();
     clearInterval(skillVersionRetentionInterval);
     clearTimeout(skillVersionRetentionStartup);
     clearTimeout(skillVersionRetentionContinuation);

server/src/routes/instance-settings.ts

@@ -452,6 +452,11 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
       publishActivitiesBestEffort(postCommitActivityPublications, "instance.task_drain.stopped");
       return wasActive;
     });
+    if (wasActive) {
+      void heartbeat.resumeQueuedRunsAfterDrain().catch(err => {
+        logger.error({ err }, "failed to resume queued work after task drain; durable work remains queued");
+      });
+    }
     res.json({ wasActive });
   });
 

server/src/services/delivery-work-coordinator.ts

@@ -125,7 +125,7 @@ export function createDeliveryWorkCoordinator(input: {
       } };
       workers.set(queue, worker);
       const ready = Promise.resolve().then(() => { start(); return running; });
-      return { ready, wake };
+      return { ready, ...worker };
     },
     async stop() {
       stopped = true;

server/src/services/delivery-work-notifications.ts

@@ -32,6 +32,7 @@ export const DELIVERY_QUEUES = {
   localLoginCleanup: "local-login-cleanup",
   budgetEnforcement: "budget-enforcement",
   accounting: "cost-accounting",
+  taskDrainExpiry: "task-drain-expiry",
   agentLifecycle: "agent-lifecycle",
   chatBoardMessages: "chat-board-messages",
   chatTaskStarts: "chat-task-starts",

server/src/services/execution-projection.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from "vitest";
 import { projectExecution } from "./execution-projection.js";
+import { startTaskDrain, stopTaskDrain } from "./task-admission.js";
 
 type Run = Parameters<typeof projectExecution>[0];
 type Coordinator = NonNullable<Parameters<typeof projectExecution>[1]>;
@@ -37,6 +38,16 @@ const project = (
 ) => projectExecution(r, c, pending, undefined, now);
 
 describe("execution truth projection", () => {
+  it("shows maintenance only for admission that has not started", () => {
+    startTaskDrain();
+    try {
+      expect(projectExecution(run({ status: "queued", startedAt: null }), undefined, [], undefined, now))
+        .toMatchObject({ phase: "queued", label: "Waiting for maintenance", cause: "task_drain" });
+      expect(project().phase).toBe("working");
+    } finally { stopTaskDrain(); }
+    expect(projectExecution(run({ status: "queued", startedAt: null }), undefined, [], undefined, now))
+      .toMatchObject({ phase: "queued", label: "Queued", cause: null });
+  });
   it.each(["queued", "recovering", "blocked", "recovered"] as const)("shows the exact saved workspace recovery state instead of stale repair instructions: %s", status => {
     const sourceRunId = "ec036e06-90a3-4b69-81db-a8eab7dbfbac";
     const source = run({ id: sourceRunId, status: "failed", errorCode: "native_continuation_requires_reconciliation" });

server/src/services/execution-projection.ts

@@ -15,6 +15,7 @@ import { EXECUTION_CONTROL_DEADLINE_MS } from "./execution-control-deadline.js";
 import { executionFailureRetryCount } from "./execution-recovery-attempt.js";
 import { retryIdempotentDatabaseOperation } from "../database-retry.js";
 import { workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js";
+import { readTaskDrain } from "./task-admission.js";
 const text = (v: unknown) => (typeof v === "string" ? v : null);
 const executionRunColumns = {
   id: heartbeatRuns.id,
@@ -382,5 +383,10 @@ export function projectExecution(
       ? set("working", "Working")
       : set("reconnecting", "Confirming execution");
   }
+  if (run.status === "queued" && !run.startedAt && readTaskDrain(now)) {
+    projection.cause = "task_drain";
+    projection.nextAction = "This turn will start automatically when maintenance finishes.";
+    return set("queued", "Waiting for maintenance");
+  }
   return projection;
 }

server/src/services/heartbeat.ts

@@ -1311,6 +1311,16 @@ export function heartbeatService(
     return heartbeatQueue.dispatchPendingNativeStatusWakeups(...args);
   }
 
+  function resumeQueuedRunsAfterDrain() {
+    const promise = heartbeatQueue.resumeQueuedRunsAfterDrain();
+    activeWakeupPromises.add(promise);
+    void promise.then(
+      () => { activeWakeupPromises.delete(promise); },
+      () => { activeWakeupPromises.delete(promise); },
+    );
+    return promise;
+  }
+
   const runLifecycle = createHeartbeatLifecycle(db, {
     issuesSvc,
     recovery,
@@ -4095,6 +4105,8 @@ export function heartbeatService(
 
     resumeQueuedRuns,
 
+    resumeQueuedRunsAfterDrain,
+
     resumeQueuedRunsForAgent: async (companyId: string, agentId: string) => {
       const agent = await getAgent(agentId);
       if (!agent || agent.companyId !== companyId) return [];

server/src/services/heartbeat/queue.ts

@@ -1908,6 +1908,14 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       });
     }
 
+    await resumeQueuedRunsAfterDrain();
+  }
+
+  // Release admission promptly without first scanning unrelated recovery work.
+  // Every run still passes the ordinary per-agent claim and dispatch gates.
+  async function resumeQueuedRunsAfterDrain() {
+    if ((await getSchedulingSuppression()).suppressed) return;
+    const cutoff = await getWorktreeExecutionCutoff();
     const queuedRuns = await db
       .select({ agentId: heartbeatRuns.agentId })
       .from(heartbeatRuns)
@@ -4941,6 +4949,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
     withChatControlRecoveryGate,
     trackWakeup,
     resumeQueuedRuns,
+    resumeQueuedRunsAfterDrain,
     parseHeartbeatPolicy,
     claimDueTimerHeartbeat,
   };

server/src/services/task-admission.ts

@@ -1,10 +1,49 @@
 import { randomUUID } from "node:crypto";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 
 export type TaskDrainState = { startedAt: Date; expiresAt: Date | null; ownerId?: string };
 let taskDrainState: TaskDrainState | null = null;
+const expiryListeners = new Set<() => void>();
+const changeListeners = new Set<() => void>();
+
+function notifyDrainChanged() {
+  for (const listener of changeListeners) listener();
+}
+
+// The shared coordinator owns this process-local deadline and its shutdown.
+// Reading admission can also establish expiry, so both paths notify the same
+// listener. No database scan or recurring timer is needed while idle.
+export function registerTaskDrainExpiryWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  onExpired: () => void,
+) {
+  const worker = coordinator.register(DELIVERY_QUEUES.taskDrainExpiry, {
+    retryMs: 1000,
+    runDuringDrain: true,
+    run: async () => { readTaskDrain(new Date()); },
+    hasPending: async () => false,
+    nextRunAt: () => taskDrainState?.expiresAt?.getTime() ?? null,
+  });
+  expiryListeners.add(onExpired);
+  changeListeners.add(worker.wake);
+  return {
+    ready: worker.ready,
+    async stop() {
+      changeListeners.delete(worker.wake);
+      expiryListeners.delete(onExpired);
+      await worker.stop();
+    },
+  };
+}
 
 export function readTaskDrain(now: Date): TaskDrainState | null {
-  if (taskDrainState?.expiresAt && taskDrainState.expiresAt.getTime() <= now.getTime()) taskDrainState = null;
+  if (taskDrainState?.expiresAt && taskDrainState.expiresAt.getTime() <= now.getTime()) {
+    taskDrainState = null;
+    for (const listener of expiryListeners) {
+      try { listener(); } catch (error) { process.emitWarning(`Task drain expiry notification failed: ${String(error)}`); }
+    }
+  }
   return taskDrainState;
 }
 
@@ -14,7 +53,10 @@ export function computeTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle"
     ...(opts.purpose === "idle" ? { ownerId: randomUUID() } : {}) };
 }
 
-export function applyTaskDrain(drain: TaskDrainState): void { taskDrainState = drain; }
+export function applyTaskDrain(drain: TaskDrainState): void {
+  taskDrainState = drain;
+  notifyDrainChanged();
+}
 export function startTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle" } = {}): TaskDrainState {
   const drain = computeTaskDrain(opts);
   applyTaskDrain(drain);
@@ -23,6 +65,7 @@ export function startTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle" }
 export function stopTaskDrain(): { wasActive: boolean } {
   const wasActive = readTaskDrain(new Date()) !== null;
   taskDrainState = null;
+  notifyDrainChanged();
   return { wasActive };
 }
 export function isIdleTaskDrainActive(): boolean { return Boolean(readTaskDrain(new Date())?.ownerId); }

ui/src/components/TaskChatThread.test.tsx

@@ -439,7 +439,7 @@ it.each([true, false])("keeps an upload attached when a conversation draft becom
   expect(onAdd.mock.calls[0]?.[3]).toEqual([id]);
 });
 
-it("shows Thinking during admission and clears it when the saved follow-up enters backoff", async () => {
+it("shows Queued during admission and clears it when the saved follow-up enters backoff", async () => {
   const pending = { id: "optimistic-recovery", companyId: "company", issueId: "issue", authorType: "user" as const,
     authorAgentId: null, authorUserId: "board", body: "Confirm the existing result.", presentation: null, metadata: null,
     createdAt: new Date(), updatedAt: new Date(), clientStatus: "pending" as const };
@@ -452,7 +452,7 @@ it("shows Thinking during admission and clears it when the saved follow-up enter
     agentMap={new Map([["agent", { id: "agent", name: "Worker", adapterType: "paperclip_runner" } as Agent]])} />);
   show(false);
   expect(container.textContent).toContain("Confirm the existing result.");
-  expect(container.textContent).toContain("Thinking");
+  expect(container.textContent).toContain("Queued");
   show(true);
   expect(container.textContent).not.toContain("Thinking");
   expect(container.textContent).not.toContain("Working");
@@ -482,7 +482,7 @@ it.each((["task", "legacy"] as const).flatMap(surface =>
   try {
     await act(async () => { send!.click(); });
     expect(onAdd).toHaveBeenCalledTimes(1);
-    expect(container.textContent).toContain("Thinking");
+    expect(container.textContent).toContain("Queued");
     const identity = container.querySelector('[data-testid="task-chat-agent-identity"]');
     render(<Component {...props} composerDisabledReason="Could not load task pause state." />);
     expect(container.querySelector('[data-testid="task-chat-agent-identity"]')).toBe(identity);
@@ -1208,7 +1208,7 @@ describe("TaskChatThread runtime transcript selection", () => {
     const activity = container.querySelector('[data-testid="task-chat-current-activity-label"]');
     expect(identity?.textContent).toBe("Runner");
     expect(identity?.querySelector("svg, img")).not.toBeNull();
-    expect(activity?.textContent).toBe("Thinking");
+    expect(activity?.textContent).toBe("Queued");
 
     renderThread(props);
     expect(container.querySelector('[data-testid="task-chat-agent-identity"]')).toBe(identity);
@@ -1224,7 +1224,7 @@ describe("TaskChatThread runtime transcript selection", () => {
       expect(container.querySelector('[data-testid="task-chat-agent-identity"]')).toBe(identity);
       expect(container.querySelector('[data-testid="task-chat-runner-identity-row"]')).toBe(header);
       expect(container.querySelector('[data-testid="task-chat-current-activity-label"]')).toBe(activity);
-      expect(activity?.textContent).toBe("Thinking");
+      expect(activity?.textContent).toBe(startup.status === "queued" ? "Queued" : "Thinking");
       expect(container.querySelector('[data-testid="task-chat-live-run-pill"]')).toBeNull();
       expect(container.textContent).not.toContain("run started");
     }

ui/src/components/task-chat/TaskChatRunnerTurn.test.tsx

@@ -35,7 +35,7 @@ describe("TaskChatRunnerTurn", () => {
   const render = (
     items: TaskChatItem[],
     status = "running",
-    runId = "run-1",
+    runId: string | null = "run-1",
     onRuntimeRequestDecision?: (
       item: TaskChatRuntimeRequestItem,
       decision: TaskChatRuntimeRequestDecision,
@@ -43,6 +43,7 @@ describe("TaskChatRunnerTurn", () => {
     suppressFinal = false,
     continuedAfterSteering = false,
     execution?: ExecutionProjection,
+    preparationLabel?: string,
   ) =>
     act(() =>
       root.render(
@@ -55,6 +56,7 @@ describe("TaskChatRunnerTurn", () => {
               items={items}
               status={status}
               execution={execution}
+              preparationLabel={preparationLabel}
               startedAtMs={Date.now() - 2_000}
               suppressFinal={suppressFinal}
               continuedAfterSteering={continuedAfterSteering}
@@ -66,6 +68,38 @@ describe("TaskChatRunnerTurn", () => {
       ),
     );
 
+  it("shows admission waiting instead of Thinking for a queued turn", () => {
+    render([], "queued");
+    expect(container.querySelector('[data-testid="task-chat-turn-status-header"]')?.textContent).toContain("Queued");
+    expect(container.querySelector('[data-testid="task-chat-current-activity-label"]')?.textContent).toBe("Queued");
+    expect(container.textContent).not.toContain("Working");
+    expect(container.textContent).not.toContain("Thinking");
+  });
+
+  it("shows queued startup before a run ID or preparation label exists", () => {
+    render([], "queued", null);
+    expect(container.querySelector('[data-testid="task-chat-turn-status-header"]')?.textContent).toContain("Queued");
+    expect(container.querySelector('[data-testid="task-chat-current-activity-label"]')?.textContent).toBe("Queued");
+    expect(container.textContent).not.toContain("Thinking");
+  });
+
+  it("preserves a specific preparation label before a run ID exists", () => {
+    render([], "queued", null, undefined, false, false, undefined, "Preparing workspace");
+    expect(container.querySelector('[data-testid="task-chat-current-activity-label"]')?.textContent).toBe("Preparing workspace");
+  });
+
+  it("shows the server's maintenance wait while admission is held", () => {
+    render([], "queued", "run-1", undefined, false, false, {
+      phase: "queued", label: "Waiting for maintenance", cause: "task_drain",
+      lastConfirmedActivityAt: null, retryAt: null, attempt: 1, maxAttempts: 3,
+      recoveryOwner: null, nextAction: "Starts automatically when maintenance finishes.",
+      permittedActions: ["inspect_run"], predecessorRunId: null, successorRunId: null,
+    });
+    expect(container.querySelector('[data-testid="task-chat-turn-status-header"]')?.textContent).toContain("Waiting for maintenance");
+    expect(container.querySelector('[data-testid="task-chat-current-activity-label"]')?.textContent).toBe("Waiting for maintenance");
+    expect(container.textContent).not.toContain("Thinking");
+  });
+
   const resolvedQuestion = (
     id: string,
     prompt: string,
@@ -95,7 +129,7 @@ describe("TaskChatRunnerTurn", () => {
   });
 
   it("shows a stable Working header with Thinking at the turn tail", () => {
-    render([], "queued");
+    render([], "running");
     expect(container.textContent).toContain("Thinking");
     expect(container.textContent).toContain("Runner");
     expect(

ui/src/components/task-chat/TaskChatRunnerTurn.tsx

@@ -156,7 +156,8 @@ export function TaskChatRunnerTurn({
 }) {
   const terminal = isTerminalRunStatus(status);
   const waitingLabel = !terminal && execution?.phase === "waiting_for_provider" ? execution.label
-    : !runId && !terminal ? preparationLabel ?? null : null;
+    : !runId && !terminal ? preparationLabel ?? (status === "queued" ? execution?.label ?? "Queued" : null)
+    : status === "queued" ? execution?.label ?? "Queued" : null;
   const yielded = items.some(
     (item) =>
       item.kind === "protocol" &&

ui/src/lib/issue-queued-comment-queue.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
 import {
   mergePendingIssueQueuedComments,
   normalizeIssueQueuedCommentQueue,
+  startingQueuedCommentIds,
 } from "./issue-queued-comment-queue";
 
 function comment(id: string, body: string) {
@@ -22,6 +23,22 @@ function comment(id: string, body: string) {
 }
 
 describe("normalizeIssueQueuedCommentQueue", () => {
+  it("keeps later inputs and uncertain or retry queues editable", () => {
+    const opener = comment("opener", "Start this turn");
+    const later = { ...comment("later", "Follow up"), createdAt: new Date("2026-09-04T12:00:02Z") };
+    const queue = normalizeIssueQueuedCommentQueue({ queueId: "wake", state: "queued",
+      entries: [opener, later].map(comment => ({ comment })) }, "issue-1");
+    const run = { runId: "run", contextIssueId: "issue-1", wakeupRequestId: "wake", status: "queued",
+      startedAt: null, createdAt: "2026-09-04T12:00:01Z", wakeCommentIds: [opener.id, later.id] };
+    const ids = (runs: unknown[], value = queue) => [...startingQueuedCommentIds(value, runs as import("../api/activity").RunForIssue[])];
+    expect(ids([run])).toEqual(["opener"]);
+    expect(ids([{ ...run, retryOfRunId: "prior" }])).toEqual([]);
+    expect(ids([{ ...run, wakeupRequestId: "new-wake" }])).toEqual([]);
+    expect(ids([{ ...run, contextIssueId: "other-issue" }])).toEqual([]);
+    expect(ids([])).toEqual([]);
+    expect(ids([run], { ...queue, targetRunId: "previous", state: "deferred" })).toEqual([]);
+    expect(ids([run], { ...queue, entries: [{ ...queue.entries[0]!, source: { kind: "interaction", interactionId: "decision", interactionKind: "request_confirmation", requiresFreshSession: false } }] })).toEqual([]);
+  });
   it("retains the immutable response and fresh-turn requirement", () => {
     const source = { kind: "interaction", interactionId: "approval-1", interactionKind: "request_confirmation", requiresFreshSession: true };
     const normalized = normalizeIssueQueuedCommentQueue({ entries: [{ comment: comment("approval-1", "Accepted: Plan"), source,

ui/src/lib/issue-queued-comment-queue.ts

@@ -5,6 +5,28 @@ import type {
   IssueQueuedCommentQueueState,
   IssueQueuedCommentSteeringDisposition,
 } from "@paperclipai/shared";
+import type { RunForIssue } from "../api/activity";
+
+/** Presentation only. Admission has accepted these inputs, but has not consumed them. */
+export function startingQueuedCommentIds(
+  queue: IssueQueuedCommentQueue | null | undefined,
+  runs: readonly RunForIssue[],
+): ReadonlySet<string> {
+  if (!queue?.queueId || queue.state !== "queued" || queue.targetRunId) return new Set();
+  const owners = runs.filter(run => run.contextIssueId === queue.issueId &&
+    run.wakeupRequestId === queue.queueId && run.status === "queued" &&
+    !run.startedAt && !run.retryOfRunId);
+  if (owners.length !== 1 || runs.some(run => run.contextIssueId === queue.issueId &&
+    run.runId !== owners[0]!.runId && run.status === "running")) return new Set();
+  const run = owners[0]!;
+  const inputs = new Set(run.wakeCommentIds?.length ? run.wakeCommentIds
+    : [run.wakeCommentId ?? run.contextCommentId]);
+  const createdAt = new Date(run.createdAt).getTime();
+  return new Set(queue.entries.flatMap(entry =>
+    !entry.source && !entry.comment.authorAgentId && !entry.comment.deletedAt &&
+    inputs.has(entry.comment.id) && new Date(entry.comment.createdAt).getTime() <= createdAt
+      ? [entry.comment.id] : []));
+}
 
 function record(value: unknown): Record<string, unknown> | null {
   return typeof value === "object" && value !== null && !Array.isArray(value)

ui/src/pages/IssueDetail.test.tsx

@@ -3468,6 +3468,45 @@ describe("IssueDetail", () => {
     });
   });
 
+  it("keeps a queued turn's starting input in history after reopen", async () => {
+    const opener = createIssueComment({ id: "reopen-input" });
+    const followUp = createIssueComment({ id: "later-input", createdAt: new Date("2026-04-21T00:00:07.000Z") });
+    mockIssuesApi.get.mockResolvedValue(createIssue({ status: "todo", assigneeAgentId: "agent-1", executionRunId: null }));
+    mockIssuesApi.listComments.mockResolvedValue([opener, followUp]);
+    mockIssuesApi.recentComments.mockResolvedValue({ comments: [opener, followUp], latestCommentId: followUp.id, hasMore: false });
+    const queuedRuns = [{
+      runId: "reopened-run", status: "queued", contextIssueId: "issue-1", agentId: "agent-1",
+      adapterType: "paperclip_runner", startedAt: null, finishedAt: null,
+      createdAt: "2026-04-21T00:00:06.000Z", invocationSource: "automation",
+      wakeupRequestId: "reopen-wake", wakeCommentIds: [opener.id, followUp.id],
+    }];
+    mockActivityApi.recentRunsForIssue.mockResolvedValue(queuedRuns);
+    mockActivityApi.runsForIssue.mockResolvedValue(queuedRuns);
+    const queue = createQueuedCommentQueue({
+      queueId: "reopen-wake", state: "queued", targetRunId: null,
+      steeringDisposition: "temporarily_unavailable",
+      entries: [opener, followUp].map((comment, position) => ({ comment, position, canEdit: true, canDiscard: true })),
+    });
+    mockIssuesApi.getQueuedComments.mockResolvedValue(queue);
+    mockIssuesApi.reorderQueuedComments.mockResolvedValue(queue);
+    await act(async () => renderIssueView());
+    await waitForAssertion(() => {
+      const props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0];
+      expect(mockIssuesApi.getQueuedComments).toHaveBeenCalled();
+      expect(props?.comments.find((comment: IssueComment) => comment.id === opener.id))
+        .toMatchObject({ body: opener.body });
+      expect(props?.comments.find((comment: IssueComment) => comment.id === opener.id)?.consumedByRunId).toBeFalsy();
+      expect(props?.comments.find((comment: IssueComment) => comment.id === opener.id)?.queueState).toBeUndefined();
+      expect(props?.queuedCommentQueue?.entries.map((entry: { comment: IssueComment }) => entry.comment.id))
+        .toEqual([followUp.id]);
+    });
+    const props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0];
+    await act(async () => props.onReorderQueuedComments([followUp.id], queue.revision));
+    expect(mockIssuesApi.reorderQueuedComments).toHaveBeenCalledWith("issue-1", {
+      orderedCommentIds: [opener.id, followUp.id], queueId: queue.queueId, revision: queue.revision,
+    });
+  });
+
   it("does not mark the wake comment for the current live run as queued when active-run cache is stale", async () => {
     mockIssuesApi.get.mockResolvedValue(
       createIssue({

ui/src/pages/IssueDetail.tsx

@@ -111,6 +111,7 @@ import { useIssueWorkProducts } from "../hooks/useIssueWorkProducts";
 import {
   mergePendingIssueQueuedComments,
   normalizeIssueQueuedCommentQueue,
+  startingQueuedCommentIds,
 } from "../lib/issue-queued-comment-queue";
 import { collectLiveIssueIds } from "../lib/liveIssueIds";
 import {
@@ -2170,6 +2171,9 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({
     resolvedLiveRuns,
     interruptibleIssueRun,
   ]);
+  const startingCommentIds = useMemo(() =>
+    startingQueuedCommentIds(authoritativeQueuedCommentQueue, resolvedLinkedRuns),
+    [authoritativeQueuedCommentQueue, resolvedLinkedRuns]);
   const effectiveQueuedCommentQueue = useMemo(() => {
     if (!queuedCommentQueueEnabled) return null;
     const deliveredCommentRunIds = new Map<string, string>();
@@ -2235,11 +2239,21 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({
     locallyQueuedCommentRunIds,
     queuedCommentQueueEnabled,
   ]);
+  const visibleQueuedCommentQueue = useMemo(() => {
+    if (!effectiveQueuedCommentQueue) return null;
+    const entries = effectiveQueuedCommentQueue.entries
+      .filter(entry => !startingCommentIds.has(entry.comment.id))
+      .map((entry, position) => ({ ...entry, position }));
+    return entries.length ? { ...effectiveQueuedCommentQueue, entries } : null;
+  }, [effectiveQueuedCommentQueue, startingCommentIds]);
   const commentsForThread = useMemo(
     () =>
-      commentsWithRunMeta.flatMap((comment) => {
+      [...commentsWithRunMeta, ...(authoritativeQueuedCommentQueue?.entries ?? [])
+        .filter(entry => startingCommentIds.has(entry.comment.id) &&
+          !commentsWithRunMeta.some(comment => comment.id === entry.comment.id))
+        .map((entry): IssueDetailComment => entry.comment)].flatMap((comment) => {
         if (discardedQueuedCommentIds.has(comment.id)) return [];
-        if (!consumedQueuedCommentIds.has(comment.id)) return [comment];
+        if (!consumedQueuedCommentIds.has(comment.id) && !startingCommentIds.has(comment.id)) return [comment];
         return [
           {
             ...comment,
@@ -2261,6 +2275,8 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({
       consumedQueuedCommentIds,
       discardedQueuedCommentIds,
       localInterruptPlacements,
+      authoritativeQueuedCommentQueue,
+      startingCommentIds,
     ],
   );
 
@@ -2324,9 +2340,12 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({
           "The queued messages are awaiting server acknowledgement.",
         );
       try {
+        let nextPosition = 0;
+        const fullOrder = effectiveQueuedCommentQueue.entries.map(entry =>
+          startingCommentIds.has(entry.comment.id) ? entry.comment.id : orderedCommentIds[nextPosition++]!);
         storeQueuedCommentQueue(
           await issuesApi.reorderQueuedComments(issueId, {
-            orderedCommentIds,
+            orderedCommentIds: fullOrder,
             queueId,
             revision,
           }),
@@ -2336,7 +2355,8 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({
       }
     },
     [
-      effectiveQueuedCommentQueue?.queueId,
+      effectiveQueuedCommentQueue,
+      startingCommentIds,
       issueId,
       refreshQueueAfterConflict,
       storeQueuedCommentQueue,
@@ -2702,7 +2722,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({
             imageUploadHandler={onImageUpload}
             onAttachImage={onAttachImage}
             onInterruptQueued={interruptQueuedComments}
-            queuedCommentQueue={effectiveQueuedCommentQueue}
+            queuedCommentQueue={visibleQueuedCommentQueue}
             onEditQueuedComment={editQueuedComment}
             onReorderQueuedComments={reorderQueuedComments}
             onSteerQueuedComment={steerQueuedComment}