Merge master and preserve Boat execution through heartbeat extraction

Port remote branch validation and scoped startup cancellation into executeHeartbeatRuntime. Keep persistent workspace preparation at the heartbeat orchestration boundary.

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

doc/DEVELOPING.md

@@ -1358,8 +1358,16 @@ Native selection retains the row lock that serializes it with cancellation and
 controller lease renewal. The executor receives an explicit selection result;
 lifecycle and ownership callbacks update its teardown state immediately, including
 when preparation later throws. Public sandbox lifecycle and model fallback helpers
-remain re-exported by `heartbeat.ts`. Provider dispatch and final cleanup stay in
-the executor.
+remain re-exported by `heartbeat.ts`.
+
+Runtime execution is in `server/src/services/heartbeat/runtime-execution.ts`.
+`executeHeartbeatRuntime` dispatches a prepared native session or legacy adapter,
+captures usage, finalizes workspace operations, and releases gateway tokens and
+instruction copies. The executor supplies the atomic dispatch gate and retains
+admission, terminal run status, retry handling, and outer lease cleanup. An explicit
+dispatch result preserves early cancellation returns; ownership callbacks update
+teardown state before errors propagate. Instruction getters retain updates made
+during provider execution. Keep policy changes separate from this extraction.
 
 ## Wake Context Delivery
 

doc/composer-stop.md

@@ -159,7 +159,9 @@ pnpm exec playwright test --config tests/e2e/playwright-composer-stop.config.ts
 The suite boots a disposable local-trusted instance on port 3199 (override with
 `PAPERCLIP_E2E_PORT`). It never attaches to an existing server. Native coverage
 is explicitly skipped without the fixture; it must not use a logged-in provider
-as a fallback. Test companies are archived during cleanup.
+as a fallback. The acceptance server pins its Codex command resolver to the fixture
+and uses an empty credential home, because an installed Codex dependency takes
+precedence over `PATH`. Test companies are archived during cleanup.
 
 For each runner, the journey starts a parent, child, and unrelated task, plus a
 terminal child. It sends while running and verifies the durable queue, clicks

doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md

@@ -415,12 +415,23 @@ from event-driven normal paths and route only real deadlines through the interna
 scheduler. Preserve durable delivery, startup recovery, and domain-level claims.
 Longer polling intervals alone do not complete an item.
 
-- [ ] **JOB-01 — Chat reconciliation.** Current: every **1s**, with multiple
-  runtime/delivery/publication lanes and SQL even without connectors. Change:
-  disarm empty lanes and activate on committed events or deadlines. **Complete
-  when:** a tenant with no chat obligations issues no chat SQL and new inbound,
-  publication, and recovery work still runs. Sources:
+- [ ] **JOB-01 — Chat reconciliation.** Partially migrated on 2026-10-10.
+  Inbound deliveries, outbound publications, and Slack upload receipt recovery
+  now use transaction-aware commit wakes and durable retry/claim deadlines.
+  Empty queues disarm after startup. Publication deadlines use the dispatch
+  eligibility predicate so paused endpoints, unknown outcomes, blocked followers,
+  and receipts owned by a live sender do not cause empty scans. Completion wakes
+  refill free endpoint slots without waiting for slow peers. No new tables.
+  Remaining **1s** maintenance: provider runtimes, provider-specific action
+  outboxes, run milestone/wake-notice projection, GitHub operation recovery,
+  Teams protocol maintenance, and Slack session sync. GitHub pre-ingress recovery
+  must still inspect provider history: a lost callback has no local queue row.
+  This slice does **not** yet make the whole chat subsystem quiet. Independent
+  DB roots/direct SQL require startup recovery, as with the other migrated queues.
+  **Complete when:** all remaining lanes are demand/deadline-driven and a tenant
+  with no chat obligations issues no chat SQL. Sources:
   [app.ts](../../server/src/app.ts),
+  [chat-delivery-work.ts](../../server/src/services/chat-delivery-work.ts),
   [chat-channels.ts](../../server/src/services/chat-channels.ts).
 
 - [x] **JOB-02 — Email reconciliation.** Replaced the unconditional **1s**

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -90,17 +90,18 @@ describe("createChatReconciliationCoordinator", () => {
     expect(ordinary).not.toContain("processFailedGitHubWebhookDeliveries()");
   });
 
-  it("wires periodic publication reconciliation to bounded scheduled refill rather than awaiting provider sends", () => {
+  it("wires bounded publication refill to commit workers and isolates periodic maintenance", () => {
     const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
-    const flush = source.slice(
-      source.indexOf("const flushChatPublications ="),
-      source.indexOf("const flushChatPublications =") + 600,
+    expect(source).toContain("registerChatDeliveryWork(deliveryWork, chatChannels,");
+    const maintenance = source.slice(
+      source.indexOf("const reconcileChatPublicationMaintenance ="),
+      source.indexOf("const chatReconciliation ="),
     );
-    expect(flush).toContain("await chatChannels.schedulePendingPublications()");
-    expect(flush).not.toContain("chatChannels.processPendingPublications()");
-    expect(
-      flush.slice(0, flush.indexOf("const chatReconciliation")),
-    ).not.toContain("await enqueueChatRunMilestones");
+    expect(maintenance).toContain("await chatChannels.processPublicationMaintenance()");
+    expect(maintenance).not.toContain("chatChannels.schedulePendingPublications()");
+    expect(maintenance).not.toContain("chatChannels.processPendingPublications()");
+    const workers = readFileSync(new URL("../services/chat-delivery-work.ts", import.meta.url), "utf8");
+    expect(workers).toContain("run: service.scheduleQueuedPublications");
     // The service integration tests hold real publication workers while this
     // scheduled method returns; app shutdown must also join those workers.
     expect(source).toContain("await chatChannels.shutdown()");

server/src/__tests__/chat-channels.integration.test.ts

@@ -1,3 +1,7 @@
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
+import { notifyChatPublicationWork } from "../services/chat-work-notifications.js";
+import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
 import { chatSlackRegistrations, toolOauthStates } from "@paperclipai/db";
 import { buildSlackAppManifest } from "@paperclipai/shared";
 import { toolAccessService } from "../services/tool-access.js";
@@ -7345,7 +7349,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const countBefore = postFetch.mock.calls.length;
     await executeConnectorTool(db, binding, "slack_post_message", uncertain);
     expect(postFetch.mock.calls).toHaveLength(countBefore);
-    await expect(executeSlackTool(db, binding, "slack_delivery", { actionId: uncertainAction.id }, postFetch as typeof fetch)).resolves.toMatchObject({ state: "delivered", receipt: { reconciled: true } });
+    const receiptSettled = vi.fn();
+    const unsubscribeReceipt = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatPublications, receiptSettled);
+    try {
+      await expect(executeSlackTool(db, binding, "slack_delivery", { actionId: uncertainAction.id }, postFetch as typeof fetch)).resolves.toMatchObject({ state: "delivered", receipt: { reconciled: true } });
+      expect(receiptSettled).toHaveBeenCalledTimes(1);
+    } finally { unsubscribeReceipt(); }
     await expect(executeConnectorTool(db, binding, "slack_create_channel", { name: "approval-required", is_private: true, idempotencyKey: randomUUID() })).rejects.toMatchObject({ reasonCode: expect.stringMatching(/approval/) });
     expect(postFetch.mock.calls.some(call => String(call[0]).endsWith("/conversations.create"))).toBe(false);
     const [approval] = await db.select().from(toolActionRequests).where(eq(toolActionRequests.companyId, fixture.companyId));
@@ -14482,6 +14491,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
       .onDiscordGatewayEvent;
     if (!onGatewayEvent) throw new Error("Expected Discord Gateway callback");
 
+    const recovered = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatPublications, recovered);
     const disconnect = Promise.resolve(
       onGatewayEvent({
         endpointId: endpoint.id,
@@ -14505,7 +14516,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     await Promise.resolve();
     expect(recoveryFinished).toBe(false);
     releaseDisconnect();
-    await Promise.all([disconnect, recovery]);
+    try {
+      await Promise.all([disconnect, recovery]);
+      expect(recovered).toHaveBeenCalled();
+    } finally { unsubscribe(); }
     await expect(context.service.get(endpoint.id)).resolves.toMatchObject({
       status: "active",
       healthMessage: "Connected",
@@ -17360,10 +17374,22 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const fixture = await seedCompany();
     const { endpoint, service } = await configuredSlackEndpoint(fixture);
     const configuredEndpoint = await service.get(endpoint.id);
+    const originalTransaction = db.transaction.bind(db);
     const transaction = vi.spyOn(db, "transaction");
-    transaction.mockRejectedValueOnce(
-      new Error("injected lifecycle persistence failure"),
-    );
+    let injected = false;
+    transaction.mockImplementation(async (callback, options) => {
+      const [delivery] = await db.select({ state: chatDeliveries.state }).from(chatDeliveries).where(and(
+        eq(chatDeliveries.endpointId, endpoint.id),
+        eq(chatDeliveries.providerEventId, "lifecycle:Ev-lifecycle-retry"),
+      ));
+      // Fail the lifecycle effect after durable admission and claim, regardless
+      // of how many transaction boundaries those writes require.
+      if (!injected && delivery?.state === "processing") {
+        injected = true;
+        throw new Error("injected lifecycle persistence failure");
+      }
+      return originalTransaction(callback, options);
+    });
     await expect(
       service.handleWebhook(
         endpoint.publicId,
@@ -32754,6 +32780,158 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     return { ...fixture, storage, runtime, service, lanes, enqueue, cleanup };
   }
 
+  it("restores chat queue deadlines without polling blocked followers or receipt owners", async () => {
+    const f = await publicationLaneFixture(1);
+    const { service, lanes, enqueue } = f;
+    const lane = lanes[0]!;
+    try {
+      const head = await enqueue(0, "Head");
+      const tail = await enqueue(0, "Follower", 1);
+      const retryAt = new Date(Date.now() + 120_000);
+      await db.update(chatPublications).set({ state: "retry", nextAttemptAt: retryAt }).where(eq(chatPublications.id, head.id));
+      expect(await service.nextPublicationAt()).toBe(retryAt.getTime());
+      await db.update(chatPublications).set({ state: "delivery_unknown", nextAttemptAt: null }).where(eq(chatPublications.id, head.id));
+      expect(await service.nextPublicationAt()).toBeNull();
+      const leaseUntil = new Date(Date.now() + 180_000);
+      await db.update(chatPublications).set({ state: "streaming", updatedAt: new Date() }).where(eq(chatPublications.id, head.id));
+      await db.insert(chatEndpointLeases).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        leaseKey: `publication:${head.id}:0`, token: randomUUID(), expiresAt: leaseUntil });
+      expect(await service.nextPublicationAt()).toBe(leaseUntil.getTime());
+      await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        kind: "slack_file_upload_receipt", providerActionId: randomUUID(), status: "received",
+        payload: { publicationId: head.id, publicationAttempt: 0 } });
+      expect(await service.nextSlackReceiptAt()).toBeNull();
+      await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, lane.endpoint.id));
+      await db.update(chatPublications).set({ updatedAt: new Date(Date.now() - 70_000) }).where(eq(chatPublications.id, head.id));
+      const wake = vi.fn();
+      const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatReceipts, wake);
+      try {
+        await service.scheduleQueuedPublications();
+        expect(wake).toHaveBeenCalled();
+        expect(await service.nextSlackReceiptAt()).toBeLessThanOrEqual(Date.now());
+        await service.processPendingSlackFileUploadReceipts(); // Invalid retained payload is closed, not retried.
+        expect(await service.nextSlackReceiptAt()).toBeNull();
+      } finally { unsubscribe(); }
+      await db.update(chatPublications).set({ state: "cancelled" }).where(eq(chatPublications.id, tail.id));
+      expect(await service.nextPublicationAt()).toBeNull();
+    } finally { await f.cleanup(); }
+  });
+
+  it("restores inbound retry and abandoned wake deadlines and excludes paused queues", async () => {
+    const f = await publicationLaneFixture(1);
+    const { service, lanes } = f;
+    const lane = lanes[0]!;
+    try {
+      const retryAt = new Date(Date.now() + 120_000);
+      const [delivery] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        conversationId: lane.conversation.id, providerEventId: randomUUID(), deduplicationKey: randomUUID(),
+        eventKind: "message", normalizedEvent: { conversation: { externalThreadId: lane.conversation.externalThreadId } },
+        receivedAt: new Date(Date.now() - 1000), state: "retry", nextAttemptAt: retryAt }).returning();
+      const [follower] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        conversationId: lane.conversation.id, providerEventId: randomUUID(), deduplicationKey: randomUUID(),
+        eventKind: "message", normalizedEvent: delivery!.normalizedEvent, state: "received" }).returning();
+      expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
+      const [edit] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "message_updated", state: "received",
+        receivedAt: new Date(Date.now() - 2000), normalizedEvent: { ...delivery!.normalizedEvent,
+          message: { targetProviderEventId: delivery!.providerEventId } } }).returning();
+      expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
+      await db.update(chatDeliveries).set({ state: "filtered" }).where(eq(chatDeliveries.id, edit!.id));
+      expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
+      await db.update(chatDeliveries).set({ state: "processed" }).where(eq(chatDeliveries.id, delivery!.id));
+      const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        deliveryId: delivery!.id, kind: "inbound_wakeup", providerActionId: randomUUID(), status: "issued",
+        result: { retryAt: retryAt.toISOString() } }).returning();
+      expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
+      const claimedAt = new Date();
+      await db.update(chatActions).set({ status: "processing", updatedAt: claimedAt }).where(eq(chatActions.id, action!.id));
+      expect(await service.nextInboundDeliveryAt()).toBe(claimedAt.getTime() + 60_000);
+      await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+      expect(await service.nextInboundDeliveryAt()).toBeNull();
+      await db.update(chatActions).set({ status: "processed" }).where(eq(chatActions.id, action!.id));
+      await db.update(chatDeliveries).set({ state: "filtered" }).where(eq(chatDeliveries.id, follower!.id));
+      await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+      expect(await service.nextInboundDeliveryAt()).toBeNull();
+      await db.update(chatEndpoints).set({ status: "attention" }).where(eq(chatEndpoints.id, lane.endpoint.id));
+      await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "installation", state: "retry",
+        nextAttemptAt: retryAt, normalizedEvent: { lifecycle: { kind: "endpoint", provider: "slack",
+          providerEventId: randomUUID(), availability: "available", reason: "Recovered" } } });
+      expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
+    } finally { await f.cleanup(); }
+  });
+
+  it("sleeps behind an unresolved Slack tool send and wakes when its receipt settles", async () => {
+    const f = await publicationLaneFixture(1);
+    const lane = f.lanes[0]!;
+    const onError = vi.fn();
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    try {
+      const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, status: "succeeded" }).returning();
+      const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: lane.conversation.issueId,
+        authorAgentId: f.assignedAgentId, createdByRunId: run!.id, body: "Already sent" }).returning();
+      const publication = await f.enqueue(0, "Already sent");
+      await db.update(chatPublications).set({ commentId: comment!.id }).where(eq(chatPublications.id, publication.id));
+      const [send] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
+        conversationId: lane.conversation.id, kind: "slack_tool_write", providerActionId: randomUUID(), status: "processing",
+        payload: { name: "slack_post_message", binding: { runId: run!.id }, args: {
+          channel: lane.conversation.externalConversationId.replace(/^slack:/, ""),
+          thread_ts: lane.conversation.externalThreadId.split(":").at(-1), text: "  Already sent\n" } } }).returning();
+      await registerChatDeliveryWork(coordinator, f.service, () => true).ready;
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+      expect(await f.service.nextPublicationAt()).toBeNull();
+      for (const status of ["uncertain", "processed"]) {
+        await db.transaction(async tx => {
+          await notifyChatPublicationWork(tx);
+          await tx.update(chatActions).set({ status }).where(eq(chatActions.id, send!.id));
+        });
+        await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+      }
+      await vi.waitFor(async () => {
+        const [row] = await db.select().from(chatPublications).where(eq(chatPublications.id, publication.id));
+        expect(row!.state).toBe("cancelled");
+        expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(lane.providerRuntime.posts).toHaveLength(0);
+      expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); await f.cleanup(); }
+  });
+
+  it("wakes chat dispatch after commit and refills free endpoint slots without the maintenance loop", async () => {
+    const f = await publicationLaneFixture(6);
+    const onError = vi.fn();
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    f.lanes[0]!.providerRuntime.postHook = () => gate;
+    try {
+      await registerChatDeliveryWork(coordinator, f.service, () => true).ready;
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
+      let settle!: () => void;
+      const commitGate = new Promise<void>(resolve => { settle = resolve; });
+      let staged!: () => void;
+      const stagedGate = new Promise<void>(resolve => { staged = resolve; });
+      const writing = db.transaction(async tx => {
+        await notifyChatPublicationWork(tx);
+        await tx.insert(chatPublications).values(f.lanes.map((lane, index) => ({ companyId: f.companyId,
+          endpointId: lane.endpoint.id, conversationId: lane.conversation.id, issueId: lane.conversation.issueId,
+          idempotencyKey: `work-test:${randomUUID()}`, payload: { text: `Queued ${index}` }, state: "pending" as const })));
+        staged(); await commitGate;
+      });
+      await stagedGate;
+      expect(f.lanes.every(lane => lane.providerRuntime.posts.length === 0)).toBe(true);
+      settle(); await writing;
+      await vi.waitFor(() => expect(f.lanes.slice(1).every(lane => lane.providerRuntime.posts.length === 1)).toBe(true), { timeout: 10_000 });
+      release();
+      await vi.waitFor(async () => {
+        const rows = await db.select().from(chatPublications).where(eq(chatPublications.companyId, f.companyId));
+        expect(rows.every(row => row.state === "published")).toBe(true);
+        expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(onError).not.toHaveBeenCalled();
+    } finally { release(); await coordinator.stop(); await f.cleanup(); }
+  });
+
   it("refills four tracked publication endpoint lanes after a scheduled budget and joins shutdown", async () => {
     const fixture = await publicationLaneFixture(6);
     const { service, lanes, enqueue, storage } = fixture;

server/src/__tests__/chat-delivery-work.test.ts

@@ -0,0 +1,112 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
+import { notifyChatDeliveryWork, notifyChatPublicationWork } from "../services/chat-work-notifications.js";
+
+const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
+afterEach(async () => {
+  for (const coordinator of coordinators.splice(0)) await coordinator.stop();
+  vi.useRealTimers();
+});
+function setup() {
+  vi.useFakeTimers();
+  const execute = vi.fn(async () => [{ xid: "42", status: "committed" }]);
+  const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>) {
+    return callback({ execute });
+  } });
+  const enabled = vi.fn(() => true), primary = vi.fn(() => true), onError = vi.fn();
+  const coordinator = createDeliveryWorkCoordinator({ owner,
+    canRun: () => enabled() && primary(), canReconcile: primary, onError,
+  });
+  coordinators.push(coordinator);
+  const service = {
+    processQueuedDeliveries: vi.fn(async () => {}),
+    scheduleQueuedPublications: vi.fn(async () => {}),
+    processPendingSlackFileUploadReceipts: vi.fn(async () => {}),
+    nextInboundDeliveryAt: vi.fn(async (): Promise<number | null> => null),
+    nextPublicationAt: vi.fn(async (): Promise<number | null> => null),
+    nextSlackReceiptAt: vi.fn(async (): Promise<number | null> => null),
+    onPublicationsSettled: vi.fn(),
+  };
+  const start = () => registerChatDeliveryWork(coordinator, service, enabled);
+  return { owner, service, coordinator, start, enabled, primary, onError };
+}
+
+describe("chat delivery work", () => {
+  it("does one startup pass and no further queue scans when empty", async () => {
+    const s = setup();
+    await s.start().ready;
+    await vi.advanceTimersByTimeAsync(3_600_000);
+    for (const run of [s.service.processQueuedDeliveries, s.service.scheduleQueuedPublications, s.service.processPendingSlackFileUploadReceipts]) expect(run).toHaveBeenCalledTimes(1);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("coalesces commits, wakes receipt recovery after publication changes, and refills freed slots", async () => {
+    const s = setup();
+    await s.start().ready;
+    await s.owner.transaction(tx => notifyChatPublicationWork(tx));
+    await s.owner.transaction(tx => notifyChatPublicationWork(tx));
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(2);
+    expect(s.service.processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2);
+    expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
+    s.service.onPublicationsSettled.mock.calls[0]![0]();
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(3);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("keeps lanes independent and retains commits received during a blocked drain", async () => {
+    const s = setup();
+    let release!: () => void;
+    s.service.processQueuedDeliveries.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
+    const worker = s.start();
+    await vi.advanceTimersByTimeAsync(1);
+    await s.owner.transaction(tx => notifyChatDeliveryWork(tx));
+    await s.owner.transaction(tx => notifyChatPublicationWork(tx));
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
+    expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(2);
+    release();
+    await worker.ready;
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(2);
+  });
+  it("restores future retries and stops when the last retry finishes", async () => {
+    const s = setup();
+    const due = Date.now() + 60_000;
+    s.service.nextPublicationAt.mockResolvedValue(due);
+    await s.start().ready;
+    expect(s.coordinator.nextWakeAt()).toBe(due);
+    await vi.advanceTimersByTimeAsync(59_999);
+    expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(1);
+    s.service.nextPublicationAt.mockResolvedValue(null);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("does no queue SQL while standby or drain defers a first commit", async () => {
+    const s = setup();
+    s.primary.mockReturnValue(false);
+    await s.start().ready;
+    await vi.advanceTimersByTimeAsync(2000);
+    s.primary.mockReturnValue(true); s.enabled.mockReturnValue(false);
+    await s.owner.transaction(tx => notifyChatDeliveryWork(tx));
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(s.service.processQueuedDeliveries).not.toHaveBeenCalled();
+    expect(s.service.nextInboundDeliveryAt).not.toHaveBeenCalled();
+    s.enabled.mockReturnValue(true);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("retries failed selection without losing its startup recovery", async () => {
+    const s = setup();
+    s.service.nextSlackReceiptAt.mockRejectedValueOnce(new Error("connection lost"));
+    await s.start().ready;
+    expect(s.onError).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(s.service.processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+});

server/src/__tests__/chat-interaction-publications.test.ts

@@ -20,7 +20,7 @@ import {
   toolConnections,
 } from "@paperclipai/db";
 import type { ChatProvider } from "@paperclipai/shared";
-import { enqueueTerminalIssueInteractionChatPublications } from "../services/chat-interaction-publications.js";
+import { enqueueIssueInteractionChatPublications, enqueueTerminalIssueInteractionChatPublications } from "../services/chat-interaction-publications.js";
 import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
 import {
   getEmbeddedPostgresTestSupport,
@@ -226,6 +226,7 @@ describeEmbeddedPostgres(
           wrapQuery(Reflect.apply(db.select, db, args))) as typeof db.select,
         insert: db.insert.bind(db),
         update: db.update.bind(db),
+        transaction: db.transaction.bind(db),
       };
     }
 
@@ -312,6 +313,35 @@ describeEmbeddedPostgres(
       ).toBe(true);
     });
 
+    it("rolls back an interaction publication if its callback tokens cannot be stored", async () => {
+      const fixture = await seedBoundIssue(["telegram"]);
+      const interaction = await issueThreadInteractionService(db).create(
+        { id: fixture.issueId, companyId: fixture.companyId },
+        { kind: "request_confirmation", continuationPolicy: "wake_assignee", payload: { version: 1, prompt: "Proceed?" } },
+        { agentId: fixture.agentId },
+      );
+      expect(await publicationsForInteraction(fixture.companyId, interaction.id)).toHaveLength(1);
+      await db.delete(chatActions).where(eq(chatActions.companyId, fixture.companyId));
+      await db.delete(chatPublications).where(eq(chatPublications.companyId, fixture.companyId));
+      const failTokens = (insert: typeof db.insert): typeof db.insert =>
+        ((table: Parameters<typeof db.insert>[0]) => {
+          if (table === chatActions) throw new Error("Token write failed");
+          return insert(table);
+        }) as typeof db.insert;
+      const failingDb: Parameters<typeof enqueueIssueInteractionChatPublications>[0] = {
+        select: db.select.bind(db),
+        update: db.update.bind(db),
+        insert: failTokens(db.insert.bind(db)),
+        transaction: callback => db.transaction(async tx => {
+          tx.insert = failTokens(tx.insert.bind(tx));
+          return callback(tx);
+        }),
+      };
+      await expect(enqueueIssueInteractionChatPublications(failingDb, interaction))
+        .rejects.toThrow("Token write failed");
+      expect(await publicationsForInteraction(fixture.companyId, interaction.id)).toEqual([]);
+    });
+
     it("keeps unsupported governance interactions authoritative in Paperclip", async () => {
       const fixture = await seedBoundIssue();
       const interaction = await issueThreadInteractionService(db).create(

server/src/app.ts

@@ -9,6 +9,7 @@ import { customerSuccessRoutes } from "./routes/customer-success.js";
 import { cloudWarmStandbyMiddleware } from "./middleware/cloud-warm-standby.js";
 import type { CloudWarmStandby } from "./services/cloud-warm-standby.js";
 import { browserUseRoutes } from "./routes/browser-use.js";
+import { registerChatDeliveryWork } from "./services/chat-delivery-work.js";
 import { registerBrowserUseCleanup } from "./services/browser-use-work.js";
 import { computerRoutes } from "./routes/computers.js";
 import { browserUseService } from "./services/browser-use.js";
@@ -320,7 +321,7 @@ export function createChatReconciliationCoordinator(input: {
   processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
   projectRunMilestones: () => Promise<number>;
   flushPublications: () => Promise<unknown>;
-  processPendingSlackFileUploadReceipts: () => Promise<unknown>;
+  processPendingSlackFileUploadReceipts?: () => Promise<unknown>;
   processPendingSlackSessionSyncs: () => Promise<unknown>;
   onError: (lane: ChatReconciliationLane, error: unknown) => void;
 }) {
@@ -367,7 +368,7 @@ export function createChatReconciliationCoordinator(input: {
       }
       milestoneReconciliation.poll();
       publicationReconciliation.poll();
-      start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
+      if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
       start("Slack session status", input.processPendingSlackSessionSyncs);
     },
     notifyPublications() {
@@ -1234,25 +1235,24 @@ export async function createApp(
       hasPending: () => opts.feedbackExportService!.hasPendingFeedbackTraces(),
     });
   }
+  registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
   emailChannels.start();
-  const flushChatPublications = async () => {
-    await chatChannels.schedulePendingPublications();
+  const reconcileChatPublicationMaintenance = async () => {
+    await chatChannels.processPublicationMaintenance();
   };
   const chatReconciliation = createChatReconciliationCoordinator({
     reconcileProviderRuntimes: async () => {
       await chatChannels.reconcileProviderRuntimes();
       await chatChannels.voice.reconcile();
     },
-    processPendingDeliveries: () => chatChannels.processPendingDeliveries(),
+    processPendingDeliveries: () => chatChannels.processPendingChatMaintenance(),
     processFailedGitHubWebhookDeliveries: () =>
       chatChannels.processFailedGitHubWebhookDeliveries(),
     projectRunMilestones: () =>
       enqueueChatRunMilestones(db, {
         publicBaseUrl: opts.authPublicBaseUrl,
       }),
-    flushPublications: () => flushChatPublications(),
-    processPendingSlackFileUploadReceipts: () =>
-      chatChannels.processPendingSlackFileUploadReceipts(),
+    flushPublications: () => reconcileChatPublicationMaintenance(),
     processPendingSlackSessionSyncs: () =>
       chatChannels.processPendingSlackSessionSyncs(),
     onError: (lane, err) => {
@@ -1265,6 +1265,8 @@ export async function createApp(
         chatReconciliation.notifyPublications();
     },
   );
+  // Provider/action maintenance and milestone projection still use this cadence.
+  // Inbox, outbox, and file receipt queues have independent commit/deadline wakes.
   let chatPublicationTimer: ReturnType<typeof setInterval> | null = setInterval(
     () => {
       if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile();

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

@@ -0,0 +1,44 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+
+type ChatDeliveryService = {
+  processQueuedDeliveries(): Promise<unknown>;
+  scheduleQueuedPublications(): Promise<unknown>;
+  processPendingSlackFileUploadReceipts(): Promise<unknown>;
+  nextInboundDeliveryAt(): Promise<number | null>;
+  nextPublicationAt(): Promise<number | null>;
+  nextSlackReceiptAt(): Promise<number | null>;
+  onPublicationsSettled(wake: () => void): void;
+};
+
+/** Independent queue workers share the app's scheduler and DB transaction owner. */
+export function registerChatDeliveryWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ChatDeliveryService,
+  canRun: () => boolean,
+) {
+  const tasks = [
+    { queue: DELIVERY_QUEUES.chatInbound, run: service.processQueuedDeliveries, next: service.nextInboundDeliveryAt },
+    { queue: DELIVERY_QUEUES.chatPublications, run: service.scheduleQueuedPublications, next: service.nextPublicationAt },
+    { queue: DELIVERY_QUEUES.chatReceipts, run: service.processPendingSlackFileUploadReceipts, next: service.nextSlackReceiptAt },
+  ];
+  const workers = tasks.map(task => {
+    let next: number | null = null;
+    const worker = coordinator.register(task.queue, {
+      retryMs: 1000,
+      run: () => task.run(),
+      hasPending: async () => {
+        next = canRun() ? await task.next() : Date.now() + 1000;
+        return false;
+      },
+      nextRunAt: () => next === null ? null : Math.max(Date.now() + 1000, next),
+    });
+    if (task.queue === DELIVERY_QUEUES.chatPublications) {
+      // Dispatch returns after reserving endpoint slots. Refill them when a
+      // provider task ends, including failures before its final durable write.
+      service.onPublicationsSettled(worker.wake);
+    }
+    return worker;
+  });
+  return { ready: Promise.all(workers.map(worker => worker.ready)) };
+}

server/src/services/chat-interaction-publications.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import { nativePhotonInteraction } from "./photon/interactions.js";
 import { randomBytes } from "node:crypto";
 import { and, eq, inArray, sql } from "drizzle-orm";
@@ -30,7 +31,7 @@ const QUESTION_ACTION_TOKEN_BYTES = 16;
 export const CHAT_QUESTION_ACTION_TOKEN_TTL_MS = 7 * 24 * 60 * 60 * 1_000;
 export const TELEGRAM_CALLBACK_DATA_LIMIT_BYTES = 64;
 
-type ChatPublicationDb = Pick<Db, "select" | "insert" | "update">;
+type ChatPublicationDb = Pick<Db, "select" | "insert" | "update" | "transaction">;
 
 function terminalNativeInteractionCopy(
   interaction: IssueThreadInteraction,
@@ -408,83 +409,87 @@ export async function enqueueIssueInteractionChatPublications(
         },
       },
     });
-    const rows = await db
-      .insert(chatPublications)
-      .values({
-        companyId: interaction.companyId,
-        endpointId: endpoint.id,
-        conversationId: conversation.id,
-        issueId: interaction.issueId,
-        idempotencyKey: `interaction:${interaction.id}:${endpoint.id}`,
-        payload,
-        state: "pending",
-      })
-      .onConflictDoNothing()
-      .returning();
-    const publication = rows[0];
-    if (publication && endpoint.provider === "imessage-photon" && nativePhotonInteraction(interaction)) {
-      const reference = randomBytes(9).toString("base64url");
-      await db.insert(chatActions).values({ companyId: interaction.companyId, endpointId: endpoint.id, conversationId: conversation.id,
-        kind: "photon_interaction", providerActionId: `photon:${reference}`,
-        payload: { version: 1, reference, interactionId: interaction.id, publicationId: publication.id, sessionGeneration: conversation.sessionGeneration,
-          expiresAt: new Date(publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS).toISOString() }, status: "issued" });
-    } else if (publication && formDraft) {
-      await db.insert(chatActions).values(
-        chatQuestionFormActionRecords(formDraft, {
+    const rows = await db.transaction(async (tx) => {
+      await notifyChatPublicationWork(tx);
+      const rows = await tx
+        .insert(chatPublications)
+        .values({
           companyId: interaction.companyId,
           endpointId: endpoint.id,
           conversationId: conversation.id,
-          publicationId: publication.id,
-        }),
-      );
-    } else if (publication && question && questionActionTokens.length > 0) {
-      const expiresAt = new Date(
-        publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
-      ).toISOString();
-      await db.insert(chatActions).values(
-        questionActionTokens.map(({ actionId, option }) => ({
-          companyId: interaction.companyId,
-          endpointId: endpoint.id,
-          conversationId: conversation.id,
-          kind: "question_answer",
-          providerActionId: actionId,
-          payload: {
-            version: 1,
+          issueId: interaction.issueId,
+          idempotencyKey: `interaction:${interaction.id}:${endpoint.id}`,
+          payload,
+          state: "pending",
+        })
+        .onConflictDoNothing()
+        .returning();
+      const publication = rows[0];
+      if (publication && endpoint.provider === "imessage-photon" && nativePhotonInteraction(interaction)) {
+        const reference = randomBytes(9).toString("base64url");
+        await tx.insert(chatActions).values({ companyId: interaction.companyId, endpointId: endpoint.id, conversationId: conversation.id,
+          kind: "photon_interaction", providerActionId: `photon:${reference}`,
+          payload: { version: 1, reference, interactionId: interaction.id, publicationId: publication.id, sessionGeneration: conversation.sessionGeneration,
+            expiresAt: new Date(publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS).toISOString() }, status: "issued" });
+      } else if (publication && formDraft) {
+        await tx.insert(chatActions).values(
+          chatQuestionFormActionRecords(formDraft, {
+            companyId: interaction.companyId,
+            endpointId: endpoint.id,
+            conversationId: conversation.id,
             publicationId: publication.id,
-            interactionId: interaction.id,
-            questionId: question.id,
-            optionId: option.id,
-            expiresAt,
-          },
-          status: "issued",
-        })),
-      );
-    } else if (
-      publication &&
-      confirmation &&
-      confirmationActionTokens.length > 0
-    ) {
-      const expiresAt = new Date(
-        publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
-      ).toISOString();
-      await db.insert(chatActions).values(
-        confirmationActionTokens.map(({ actionId, decision }) => ({
-          companyId: interaction.companyId,
-          endpointId: endpoint.id,
-          conversationId: conversation.id,
-          kind: "confirmation_response",
-          providerActionId: actionId,
-          payload: {
-            version: 1,
-            publicationId: publication.id,
-            interactionId: interaction.id,
-            decision,
-            expiresAt,
-          },
-          status: "issued",
-        })),
-      );
-    }
+          }),
+        );
+      } else if (publication && question && questionActionTokens.length > 0) {
+        const expiresAt = new Date(
+          publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
+        ).toISOString();
+        await tx.insert(chatActions).values(
+          questionActionTokens.map(({ actionId, option }) => ({
+            companyId: interaction.companyId,
+            endpointId: endpoint.id,
+            conversationId: conversation.id,
+            kind: "question_answer",
+            providerActionId: actionId,
+            payload: {
+              version: 1,
+              publicationId: publication.id,
+              interactionId: interaction.id,
+              questionId: question.id,
+              optionId: option.id,
+              expiresAt,
+            },
+            status: "issued",
+          })),
+        );
+      } else if (
+        publication &&
+        confirmation &&
+        confirmationActionTokens.length > 0
+      ) {
+        const expiresAt = new Date(
+          publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
+        ).toISOString();
+        await tx.insert(chatActions).values(
+          confirmationActionTokens.map(({ actionId, decision }) => ({
+            companyId: interaction.companyId,
+            endpointId: endpoint.id,
+            conversationId: conversation.id,
+            kind: "confirmation_response",
+            providerActionId: actionId,
+            payload: {
+              version: 1,
+              publicationId: publication.id,
+              interactionId: interaction.id,
+              decision,
+              expiresAt,
+            },
+            status: "issued",
+          })),
+        );
+      }
+      return rows;
+    });
     inserted.push(...rows);
   }
   return inserted;
@@ -560,20 +565,23 @@ export async function enqueueTerminalIssueInteractionChatPublications(
     )
     .map((publication) => publication.id);
   if (unsentIds.length > 0) {
-    await db
-      .update(chatPublications)
-      .set({
-        state: "cancelled",
-        nextAttemptAt: null,
-        redactedError: "Interaction was resolved before provider publication",
-        updatedAt: new Date(),
-      })
-      .where(
-        and(
-          inArray(chatPublications.id, unsentIds),
-          inArray(chatPublications.state, ["pending", "retry"]),
-        ),
-      );
+    await db.transaction(async (tx) => {
+      await notifyChatPublicationWork(tx);
+      return tx
+        .update(chatPublications)
+        .set({
+          state: "cancelled",
+          nextAttemptAt: null,
+          redactedError: "Interaction was resolved before provider publication",
+          updatedAt: new Date(),
+        })
+        .where(
+          and(
+            inArray(chatPublications.id, unsentIds),
+            inArray(chatPublications.state, ["pending", "retry"]),
+          ),
+        );
+    });
   }
 
   const originalIds = originals.map((publication) => publication.id);
@@ -706,33 +714,37 @@ export async function enqueueTerminalIssueInteractionChatPublications(
   if (!copy) return [];
   const inserted: Array<typeof chatPublications.$inferSelect> = [];
   for (const original of providerVisibleOriginals) {
-    if (!original.payload.card) continue;
-    const rows = await db
-      .insert(chatPublications)
-      .values({
-        companyId: interaction.companyId,
-        endpointId: original.endpointId,
-        conversationId: original.conversationId,
-        issueId: interaction.issueId,
-        idempotencyKey: `interaction-resolution:${interaction.id}:${original.endpointId}`,
-        payload: projectSafeChatPublication({
-          classification: "external",
-          source: "issue_interaction",
-          text: copy.text,
-          interaction: {
-            id: interaction.id,
-            card: {
-              kind: original.payload.card.kind,
-              title: original.payload.card.title,
-              body: copy.body,
-              actions: [],
+    const card = original.payload.card;
+    if (!card) continue;
+    const rows = await db.transaction(async (tx) => {
+      await notifyChatPublicationWork(tx);
+      return tx
+        .insert(chatPublications)
+        .values({
+          companyId: interaction.companyId,
+          endpointId: original.endpointId,
+          conversationId: original.conversationId,
+          issueId: interaction.issueId,
+          idempotencyKey: `interaction-resolution:${interaction.id}:${original.endpointId}`,
+          payload: projectSafeChatPublication({
+            classification: "external",
+            source: "issue_interaction",
+            text: copy.text,
+            interaction: {
+              id: interaction.id,
+              card: {
+                kind: card.kind,
+                title: card.title,
+                body: copy.body,
+                actions: [],
+              },
             },
-          },
-        }),
-        state: "pending",
-      })
-      .onConflictDoNothing()
-      .returning();
+          }),
+          state: "pending",
+        })
+        .onConflictDoNothing()
+        .returning();
+    });
     inserted.push(...rows);
   }
   return inserted;
@@ -771,23 +783,26 @@ export async function cancelPendingIssueInteractionChatPublications(
         ]),
       ),
     );
-  return db
-    .update(chatPublications)
-    .set({
-      state: "cancelled",
-      nextAttemptAt: null,
-      redactedError: "Interaction was superseded before publication",
-      updatedAt: new Date(),
-    })
-    .where(
-      and(
-        eq(chatPublications.companyId, input.companyId),
-        eq(chatPublications.issueId, input.issueId),
-        inArray(chatPublications.state, ["pending", "retry"]),
-        inArray(sql<string>`${chatPublications.payload}->>'interactionId'`, [
-          ...input.interactionIds,
-        ]),
-      ),
-    )
-    .returning();
+  return db.transaction(async (tx) => {
+    await notifyChatPublicationWork(tx);
+    return tx
+      .update(chatPublications)
+      .set({
+        state: "cancelled",
+        nextAttemptAt: null,
+        redactedError: "Interaction was superseded before publication",
+        updatedAt: new Date(),
+      })
+      .where(
+        and(
+          eq(chatPublications.companyId, input.companyId),
+          eq(chatPublications.issueId, input.issueId),
+          inArray(chatPublications.state, ["pending", "retry"]),
+          inArray(sql<string>`${chatPublications.payload}->>'interactionId'`, [
+            ...input.interactionIds,
+          ]),
+        ),
+      )
+      .returning();
+  });
 }

server/src/services/chat-run-publications.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import {
   and,
   asc,
@@ -623,6 +624,7 @@ async function enqueueSafeNativeChatProgress(
           .limit(1);
         if (explicitlyAuthoredFinal.length > 0) return 0;
 
+        await notifyChatPublicationWork(tx);
         const insertedRows = await tx
           .insert(chatPublications)
           .values({
@@ -997,30 +999,33 @@ export async function enqueueChatRunMilestones(
           .limit(1);
         if (explicitlyAuthoredPublication.length > 0) continue;
       }
-      const result = await db
-        .insert(chatPublications)
-        .values({
-          companyId: row.companyId,
-          endpointId: row.endpointId,
-          conversationId: row.conversationId,
-          issueId: row.issueId,
-          idempotencyKey: `run:${row.runId}:${milestone}:${row.endpointId}`,
-          payload: projectSafeChatPublication({
-            classification: "external",
-            source: "safe_milestone",
-            text: safeMilestoneText({
-              agentName: row.agentName,
-              errorCode: row.runErrorCode,
-              milestone,
-              issueId: row.issueId,
-              publicBaseUrl: input.publicBaseUrl,
+      const result = await db.transaction(async (tx) => {
+        await notifyChatPublicationWork(tx);
+        return tx
+          .insert(chatPublications)
+          .values({
+            companyId: row.companyId,
+            endpointId: row.endpointId,
+            conversationId: row.conversationId,
+            issueId: row.issueId,
+            idempotencyKey: `run:${row.runId}:${milestone}:${row.endpointId}`,
+            payload: projectSafeChatPublication({
+              classification: "external",
+              source: "safe_milestone",
+              text: safeMilestoneText({
+                agentName: row.agentName,
+                errorCode: row.runErrorCode,
+                milestone,
+                issueId: row.issueId,
+                publicBaseUrl: input.publicBaseUrl,
+              }),
+              progressState: milestone,
             }),
-            progressState: milestone,
-          }),
-          state: "pending",
-        })
-        .onConflictDoNothing()
-        .returning({ id: chatPublications.id });
+            state: "pending",
+          })
+          .onConflictDoNothing()
+          .returning({ id: chatPublications.id });
+      });
       inserted += result.length;
     }
     if (rows.length < pageSize) break;

server/src/services/chat-teams-file-publication.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import { and, eq, sql } from "drizzle-orm";
 import { z } from "zod";
 import { CHAT_FILE_TRANSFER_PHASES } from "@paperclipai/shared";
@@ -375,6 +376,7 @@ export async function projectTeamsFilePublication(
     !sameDate(publication.nextAttemptAt, nextAttemptAt) ||
     !sameDate(publication.publishedAt, publishedAt)
   ) {
+    await notifyChatPublicationWork(tx);
     await tx
       .update(chatPublications)
       .set({

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

@@ -0,0 +1,16 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export async function notifyChatDeliveryWork(tx: object): Promise<void> {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInbound);
+}
+
+export async function notifyChatPublicationWork(tx: object): Promise<void> {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatPublications);
+  // Receipt recovery waits for the publication owner to release its claim.
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceipts);
+}
+
+export async function notifyChatEndpointWork(tx: object): Promise<void> {
+  await notifyChatDeliveryWork(tx);
+  await notifyChatPublicationWork(tx);
+}

server/src/services/connectors/slack-delivery.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "../chat-work-notifications.js";
 import { and, eq, sql, inArray } from "drizzle-orm";
 import { chatActions, type Db } from "@paperclipai/db";
 import { forbidden } from "../../errors.js";
@@ -40,17 +41,20 @@ export async function slackDelivery(
     action.status === "processing" &&
     Date.now() - action.updatedAt.getTime() > 5 * 60_000
   ) {
-    const [stale] = await db
-      .update(chatActions)
-      .set({ status: "uncertain", updatedAt: new Date() })
-      .where(
-        and(
-          eq(chatActions.id, action.id),
-          eq(chatActions.status, "processing"),
-          eq(chatActions.updatedAt, action.updatedAt),
-        ),
-      )
-      .returning();
+    const [stale] = await db.transaction(async (tx) => {
+      await notifyChatPublicationWork(tx);
+      return tx
+        .update(chatActions)
+        .set({ status: "uncertain", updatedAt: new Date() })
+        .where(
+          and(
+            eq(chatActions.id, action.id),
+            eq(chatActions.status, "processing"),
+            eq(chatActions.updatedAt, action.updatedAt),
+          ),
+        )
+        .returning();
+    });
     if (stale) action = stale;
   }
   const args = object(action.payload.args);
@@ -107,16 +111,19 @@ export async function slackDelivery(
         };
     }
     if (receipt) {
-      const [settled] = await db
-        .update(chatActions)
-        .set({ status: "processed", result: receipt, updatedAt: new Date() })
-        .where(
-          and(
-            eq(chatActions.id, action.id),
-            inArray(chatActions.status, ["uncertain", "processing"]),
-          ),
-        )
-        .returning();
+      const [settled] = await db.transaction(async (tx) => {
+        await notifyChatPublicationWork(tx);
+        return tx
+          .update(chatActions)
+          .set({ status: "processed", result: receipt, updatedAt: new Date() })
+          .where(
+            and(
+              eq(chatActions.id, action.id),
+              inArray(chatActions.status, ["uncertain", "processing"]),
+            ),
+          )
+          .returning();
+      });
       if (settled) action = settled;
     }
   }

server/src/services/connectors/slack-publication.ts

@@ -1,81 +1,54 @@
-import { and, eq, sql, inArray } from "drizzle-orm";
-import {
-  chatActions,
-  chatConversations,
-  issueComments,
-  type chatPublications,
-  type Db,
-} from "@paperclipai/db";
-import { object } from "./slack-client.js";
+import { sql, type SQL } from "drizzle-orm";
+import { chatPublications, type Db } from "@paperclipai/db";
 
-/** Explicit sends use a separate durable receipt. Suppress an identical automatic
- * final, but retain distinct summaries, questions, blockers and progress. */
+// ECMAScript String.trim whitespace, matching tool-send text comparison.
+const trimCharacters = "\u0009\u000a\u000b\u000c\u000d\u0020\u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000\ufeff";
+type PublicationMatch = {
+  companyId: SQL; endpointId: SQL; conversationId: SQL;
+  issueId: SQL; commentId: SQL; payload: SQL;
+};
+
+function matchingSlackSend(publication: PublicationMatch) {
+  const p = publication;
+  return sql`from chat_actions send
+    join issue_comments comment on comment.id = ${p.commentId}
+      and comment.company_id = ${p.companyId} and comment.issue_id = ${p.issueId}
+    join chat_conversations conversation on conversation.id = ${p.conversationId}
+      and conversation.company_id = ${p.companyId}
+    where send.company_id = ${p.companyId} and send.endpoint_id = ${p.endpointId}
+      and send.conversation_id = ${p.conversationId} and send.kind = 'slack_tool_write'
+      and send.payload->'binding'->>'runId' = coalesce(comment.created_by_run_id, comment.derived_created_by_run_id)::text
+      and send.payload->>'name' = 'slack_post_message'
+      and coalesce(${p.payload}->>'interactionId', '') = ''
+      and coalesce(${p.payload}->>'progressState', '') = ''
+      and (case when jsonb_typeof(${p.payload}->'attachmentIds') = 'array'
+        then jsonb_array_length(${p.payload}->'attachmentIds') else 0 end) = 0
+      and send.payload->'args'->>'channel' = regexp_replace(conversation.external_conversation_id, '^slack:', '')
+      and send.payload->'args'->>'thread_ts' = regexp_replace(conversation.external_thread_id, '^.*:', '')
+      and btrim(send.payload->'args'->>'text', ${trimCharacters}) = btrim(${p.payload}->>'text', ${trimCharacters})`;
+}
+
+/** Dispatch and deadlines must agree about pending explicit Slack sends. */
+export function unresolvedSlackPublicationCondition(): SQL {
+  return sql`exists (select 1 ${matchingSlackSend({
+    companyId: sql`${chatPublications.companyId}`, endpointId: sql`${chatPublications.endpointId}`,
+    conversationId: sql`${chatPublications.conversationId}`, issueId: sql`${chatPublications.issueId}`,
+    commentId: sql`${chatPublications.commentId}`, payload: sql`${chatPublications.payload}`,
+  })} and send.status in ('received', 'processing', 'uncertain'))`;
+}
+
+/** Explicit sends suppress identical automatic finals, but not distinct content. */
 export async function slackExplicitPublicationDuplicate(
   db: Db,
   publication: typeof chatPublications.$inferSelect,
 ) {
-  if (
-    !publication.commentId ||
-    publication.payload.interactionId ||
-    publication.payload.progressState ||
-    publication.payload.attachmentIds?.length
-  )
-    return null;
-  const [comment] = await db
-    .select({
-      runId: sql<string>`coalesce(${issueComments.createdByRunId}, ${issueComments.derivedCreatedByRunId})`,
-    })
-    .from(issueComments)
-    .where(
-      and(
-        eq(issueComments.id, publication.commentId),
-        eq(issueComments.companyId, publication.companyId),
-        eq(issueComments.issueId, publication.issueId),
-      ),
-    );
-  if (!comment?.runId) return null;
-  const [conversation] = await db
-    .select()
-    .from(chatConversations)
-    .where(
-      and(
-        eq(chatConversations.id, publication.conversationId),
-        eq(chatConversations.companyId, publication.companyId),
-      ),
-    );
-  if (!conversation) return null;
-  const sends = await db
-    .select({ payload: chatActions.payload, status: chatActions.status })
-    .from(chatActions)
-    .where(
-      and(
-        eq(chatActions.companyId, publication.companyId),
-        eq(chatActions.endpointId, publication.endpointId),
-        eq(chatActions.conversationId, publication.conversationId),
-        eq(chatActions.kind, "slack_tool_write"),
-        inArray(chatActions.status, [
-          "processed",
-          "received",
-          "processing",
-          "uncertain",
-        ]),
-        sql`${chatActions.payload}->'binding'->>'runId' = ${comment.runId}`,
-        sql`${chatActions.payload}->>'name' = 'slack_post_message'`,
-      ),
-    );
-  const match = sends.find(({ payload }) => {
-    const args = object(payload.args);
-    const thread = conversation.externalThreadId.split(":").at(-1);
-    return (
-      args.channel ===
-        conversation.externalConversationId.replace(/^slack:/, "") &&
-      args.thread_ts === thread &&
-      String(args.text).trim() === publication.payload.text.trim()
-    );
-  });
-  return match
-    ? match.status === "processed"
-      ? "delivered"
-      : "unresolved"
-    : null;
+  const p = publication;
+  if (!p.commentId || p.payload.interactionId || p.payload.progressState || p.payload.attachmentIds?.length) return null;
+  const rows = await db.execute<{ status: string }>(sql`select send.status ${matchingSlackSend({
+    companyId: sql`${p.companyId}::uuid`, endpointId: sql`${p.endpointId}::uuid`,
+    conversationId: sql`${p.conversationId}::uuid`, issueId: sql`${p.issueId}::uuid`,
+    commentId: sql`${p.commentId}::uuid`, payload: sql`${JSON.stringify(p.payload)}::jsonb`,
+  })} and send.status in ('processed', 'received', 'processing', 'uncertain')
+    order by (send.status = 'processed') asc limit 1`);
+  return rows[0] ? rows[0].status === "processed" ? "delivered" : "unresolved" : null;
 }

server/src/services/connectors/slack-retry.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "../chat-work-notifications.js";
 import { and, eq, sql, inArray } from "drizzle-orm";
 import { chatActions, toolInvocations, type Db } from "@paperclipai/db";
 import {
@@ -52,6 +53,7 @@ export async function claimSlackRateLimitRetry(
       !(Date.parse(action.result.retryAt) <= Date.now())
     )
       return false;
+    await notifyChatPublicationWork(tx);
     await tx
       .update(chatActions)
       .set({ status: "received", updatedAt: new Date() })

server/src/services/connectors/slack-writes.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "../chat-work-notifications.js";
 import { authorizeSlackDocument } from "./slack-document.js";
 import { slackMessage } from "./slack-message.js";
 import { createHash } from "node:crypto";
@@ -100,28 +101,31 @@ export async function executeSlackWrite(
     .update(JSON.stringify({ name, args }))
     .digest("hex");
   const key = `slack-tool:${binding.issueId}:${args.idempotencyKey}`;
-  const [created] = await db
-    .insert(chatActions)
-    .values({
-      companyId: binding.companyId,
-      endpointId: authority.endpoint.id,
-      conversationId: authority.conversation?.id ?? null,
-      principalId: authority.principalId,
-      kind: "slack_tool_write",
-      providerActionId: key,
-      status: "received",
-      payload: {
-        name,
-        args,
-        hash,
-        binding,
-        userId: authority.userId,
-        revision: authority.revision,
-        invocationId,
-      },
-    })
-    .onConflictDoNothing()
-    .returning();
+  const [created] = await db.transaction(async (tx) => {
+    await notifyChatPublicationWork(tx);
+    return tx
+      .insert(chatActions)
+      .values({
+        companyId: binding.companyId,
+        endpointId: authority.endpoint.id,
+        conversationId: authority.conversation?.id ?? null,
+        principalId: authority.principalId,
+        kind: "slack_tool_write",
+        providerActionId: key,
+        status: "received",
+        payload: {
+          name,
+          args,
+          hash,
+          binding,
+          userId: authority.userId,
+          revision: authority.revision,
+          invocationId,
+        },
+      })
+      .onConflictDoNothing()
+      .returning();
+  });
   const action =
     created ??
     (
@@ -151,13 +155,16 @@ export async function executeSlackWrite(
       instruction:
         "Do not retry with another key. Inspect delivery before reconciling an uncertain result.",
     };
-  const [claimed] = await db
-    .update(chatActions)
-    .set({ status: "processing", updatedAt: new Date() })
-    .where(
-      and(eq(chatActions.id, action.id), eq(chatActions.status, "received")),
-    )
-    .returning();
+  const [claimed] = await db.transaction(async (tx) => {
+    await notifyChatPublicationWork(tx);
+    return tx
+      .update(chatActions)
+      .set({ status: "processing", updatedAt: new Date() })
+      .where(
+        and(eq(chatActions.id, action.id), eq(chatActions.status, "received")),
+      )
+      .returning();
+  });
   if (!claimed) return { actionId: action.id, state: "processing" };
   try {
     authority = await resolveSlackTaskAuthority(db, binding);
@@ -252,13 +259,16 @@ export async function executeSlackWrite(
       )
         throw forbidden("Slack returned an unsupported upload destination");
       // Store the provider receipt before transport. An uncertain finish is never blindly retried.
-      await db
-        .update(chatActions)
-        .set({
-          result: { fileId: upload.file_id, phase: "upload_allocated" },
-          updatedAt: new Date(),
-        })
-        .where(eq(chatActions.id, action.id));
+      await db.transaction(async (tx) => {
+        await notifyChatPublicationWork(tx);
+        return tx
+          .update(chatActions)
+          .set({
+            result: { fileId: upload.file_id, phase: "upload_allocated" },
+            updatedAt: new Date(),
+          })
+          .where(eq(chatActions.id, action.id));
+      });
       const uploaded = await fetchImpl(url, {
         method: "POST",
         body: bytes,
@@ -436,10 +446,13 @@ export async function executeSlackWrite(
           }
         : {}),
     };
-    await db
-      .update(chatActions)
-      .set({ status: "processed", result: receipt, updatedAt: new Date() })
-      .where(eq(chatActions.id, action.id));
+    await db.transaction(async (tx) => {
+      await notifyChatPublicationWork(tx);
+      return tx
+        .update(chatActions)
+        .set({ status: "processed", result: receipt, updatedAt: new Date() })
+        .where(eq(chatActions.id, action.id));
+    });
     await logActivity(db, {
       companyId: binding.companyId,
       actorType: "agent",
@@ -468,19 +481,22 @@ export async function executeSlackWrite(
             "slack_request_timeout",
             "slack_service_unavailable",
           ].includes(details.code)));
-    await db
-      .update(chatActions)
-      .set({
-        status: definite ? "failed" : "uncertain",
-        result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify({ code: details.code ?? "slack_delivery_uncertain", ...(error instanceof HttpError && error.status === 429 ? { retryAt: new Date(Date.now() + Number(details.retryAfterSeconds ?? 60) * 1000).toISOString() } : {}) })}::jsonb`,
-        updatedAt: new Date(),
-      })
-      .where(
-        and(
-          eq(chatActions.id, action.id),
-          eq(chatActions.status, "processing"),
-        ),
-      );
+    await db.transaction(async (tx) => {
+      await notifyChatPublicationWork(tx);
+      return tx
+        .update(chatActions)
+        .set({
+          status: definite ? "failed" : "uncertain",
+          result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify({ code: details.code ?? "slack_delivery_uncertain", ...(error instanceof HttpError && error.status === 429 ? { retryAt: new Date(Date.now() + Number(details.retryAfterSeconds ?? 60) * 1000).toISOString() } : {}) })}::jsonb`,
+          updatedAt: new Date(),
+        })
+        .where(
+          and(
+            eq(chatActions.id, action.id),
+            eq(chatActions.status, "processing"),
+          ),
+        );
+    });
     if (error instanceof HttpError)
       error.details = {
         ...details,

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

@@ -2,6 +2,9 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  chatInbound: "chat-inbound",
+  chatPublications: "chat-publications",
+  chatReceipts: "chat-slack-receipts",
   browser: "browser-use",
   email: "email-channels",
   feedback: "feedback-exports",

server/src/services/heartbeat/runtime-execution.test.ts

@@ -0,0 +1,291 @@
+import { randomUUID } from "node:crypto";
+import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations, runUsageReceipts, workspaceOperations, type Db } from "@paperclipai/db";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import * as legacyRecovery from "../legacy-execution-recovery.js";
+import { createAgentIdentityRedactor } from "../agent-identity-redaction.js";
+import { createAdapterExecutionControl } from "../adapter-execution-control.js";
+import { executionWorkspaceService } from "../execution-workspaces.js";
+import { workspaceOperationService } from "../workspace-operations.js";
+import { instanceSettingsService } from "../instance-settings.js";
+import { executePaperclipNativeSession, NativeCancellationPendingRecoveryError, NativeControllerDetachedForRestartError } from "../native-runtime/native-session-executor.js";
+import { NativeRunnerOwnershipUnverifiedError } from "../native-runtime/native-runner-ownership.js";
+import { NativeWorkspaceFinalizationBusyError, NativeWorkspaceFinalizationOwnershipLostError } from "../native-runtime/native-workspace-finalization-ownership.js";
+import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mode.js";
+import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js";
+import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js";
+import { createHeartbeatRunState } from "./run-state.js";
+import { buildEffectiveRunSessionConfigMetadata } from "./workspaces.js";
+import { revokeHeartbeatRunGatewayTokens } from "./run-preparation.js";
+import { executeHeartbeatRuntime, type HeartbeatRuntimeExecutionInput } from "./runtime-execution.js";
+
+// Keep real accounting, workspace barriers, and task-session persistence. Providers
+// and gateway delivery are external boundaries; browser acceptance covers their wiring.
+vi.mock("../native-runtime/native-session-executor.js", async original => ({
+  ...await original<typeof import("../native-runtime/native-session-executor.js")>(),
+  executePaperclipNativeSession: vi.fn(),
+}));
+vi.mock("./run-preparation.js", async original => ({
+  ...await original<typeof import("./run-preparation.js")>(),
+  buildPaperclipRuntimeMcpServers: vi.fn(async () => []),
+  createAdapterRuntimeToolAccess: vi.fn(() => null),
+  createManagedMcpRunConfig: vi.fn(async () => null),
+  revokeHeartbeatRunGatewayTokens: vi.fn(async () => {}),
+}));
+
+const support = await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: Db;
+  let home: string;
+  beforeAll(async () => {
+    home = await mkdtemp(join(tmpdir(), "paperclip-runtime-execution-"));
+    vi.stubEnv("PAPERCLIP_HOME", home);
+    database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-execution");
+    db = createDb(database.connectionString);
+    await db.execute(sql`set client_min_messages = warning`);
+  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+  afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); });
+  beforeEach(async () => {
+    await db.execute(sql`truncate table companies restart identity cascade`);
+    vi.clearAllMocks();
+    vi.mocked(executePaperclipNativeSession).mockResolvedValue({ exitCode: 0, signal: null, timedOut: false });
+  });
+
+  async function fixture(native = false) {
+    const [company] = await db.insert(companies).values({ name: "Runtime execution", issuePrefix: "EXE" }).returning();
+    const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Executing agent", status: "running", adapterType: native ? "paperclip_runner" : "process", adapterConfig: { provider: "codex" } }).returning();
+    const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Complete this task", assigneeAgentId: agent.id }).returning();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: native ? "native" : "legacy" }).returning();
+    const state = createHeartbeatRunState(db);
+    const executionControl = createAdapterExecutionControl();
+    const adapter = { type: "process", label: "Test process", testEnvironment: vi.fn(), execute: vi.fn<HeartbeatRuntimeExecutionInput["runtime"]["adapter"]["execute"]>(async () => ({ exitCode: 0, signal: null, timedOut: false })) };
+    let instructionSave: Record<string, unknown> | null = null;
+    const nativeExecution = native ? buildNativeExecutionInput({
+      companyId: company.id, agentId: agent.id, runId: run.id, issue,
+      taskPrompt: issue.title, workspace: { id: run.id, cwd: home, repoUrl: null, repoRef: null, branchName: null },
+      normalizedSessionId: null, provider: "codex", runtimeContext: nativeRuntimeContextFixture(),
+      completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", contract: { revision: "1", objective: issue.title, criteria: [{ id: "output", requirement: issue.title }] } },
+    }) : null;
+    const input: HeartbeatRuntimeExecutionInput = {
+      run, agent, options: {},
+      task: { issueRef: issue, issueContext: null, context: {}, issueId: issue.id, taskKey: issue.id, executionContinuation: null },
+      runtime: { nativeExecution, nativeRunnerInstanceId: native ? randomUUID() : null, getNativeFreshSessionHandoff: undefined,
+        nativeRuntimeResolution: resolveHeartbeatNativeRuntimeMode({ persisted: run, enabled: native, runtimeConfig: {}, adapterConfig: agent.adapterConfig, agent, issue, target: null, workspaceId: null }),
+        adapter, runtimeForAdapter: { taskKey: issue.id, sessionId: null, sessionParams: null, sessionDisplayId: null }, getFreshSessionHandoff: undefined, runOptions: {},
+      },
+      workspace: { persistedExecutionWorkspace: null, executionWorkspace: { baseCwd: home, cwd: home, source: "agent_home", projectId: null, workspaceId: null, repoUrl: null, repoRef: null, strategy: "project_primary", branchName: null, worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false },
+        executionTarget: null, nativeWorkspaceSync: null, workspaceRestoreSource: undefined, remoteExecution: null,
+        workspaceOperationRecorder: workspaceOperationService(db).createRecorder({ companyId: company.id, heartbeatRunId: run.id, issueId: issue.id }),
+      },
+      config: { runtimeConfig: {}, resolvedConfig: {}, managedAiRuntime: undefined,
+        resolvedInstanceSettings: { experimental: { enableWorkspaceBranchReconcileForward: false, enableWorkspaceDirtyQuarantineRepair: false } },
+        configuredTaskEnvironment: {}, adapterEnv: {}, agentIdentity: undefined, runtimeEnv: {}, useHostGitHub: false, githubSelection: { configured: false },
+      },
+      session: { runtimeSessionParamsForAdapter: null, configuredModel: null,
+        sessionConfigMetadata: await buildEffectiveRunSessionConfigMetadata({ adapterType: agent.adapterType, effectiveAdapterConfig: {}, agentRuntimeConfig: {}, issueOverrides: null, workspaceConfig: null, environment: null, environmentEnv: null, projectEnv: null, routineEnv: null, runtimeSkills: [] }), goalCheckpointSession: { current: null },
+      },
+      instructions: { instructionCopy: null, getInstructionSave: () => instructionSave, nativeInstructionWorkingCopy: () => undefined,
+        collectStoppedInstructions: async () => { instructionSave = { state: "saved", revisionId: "during-execution" }; },
+        nativeInstructionReservation: null, releaseInstructionCopy: vi.fn(async () => {}),
+      },
+      trace: { providerTraceCapture: null, nativeRunnerPreparationSpans: [], attestedQuestionResponseAtMs: null, attemptStartedAtMs: Date.now(), environmentAcquireStartedAtMs: Date.now(), environmentRealizeEndedAtMs: Date.now() },
+      control: { executionControl, executionPhaseContext: { onExecutionPhase: executionControl.phases.enter },
+        dispatchResolvedInteractionContinuationWithAtomicGate: async dispatch => ({ dispatched: true, resultPromise: dispatch(vi.fn()) }),
+      },
+      output: { identityRedactor: createAgentIdentityRedactor(), appendIdentityRedactedLog: vi.fn(async () => {}), onLog: vi.fn(async () => {}), onAdapterEvent: vi.fn(async () => {}), onAdapterMeta: vi.fn(async () => {}) },
+      services: { ...state, executionWorkspacesSvc: executionWorkspaceService(db), instanceSettings: instanceSettingsService(db),
+        envOrchestrator: { releaseForRun: vi.fn() }, cancelRunInternal: vi.fn(), enqueueWakeup: vi.fn(),
+        persistRunProcessMetadata: vi.fn(), recordCurrentHeartbeatRunRuntimeProgress: vi.fn(), dispatchPendingNativeStatusWakeups: vi.fn(),
+      },
+      effects: { onUsageCaptureReady: vi.fn(), onNativeDispatchStarted: vi.fn(), onLegacyAdapterEntered: vi.fn(), onWorkspaceRestoreFailure: vi.fn(), onProviderResourceDisposition: vi.fn(), onNativeWorkspaceFinalizeScheduled: vi.fn(), onNativeSessionResumeScheduled: vi.fn(), onNativeOwnershipHeld: vi.fn() },
+    };
+    return { input, adapter, run };
+  }
+  const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!);
+  const finalizations = () => db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "workspace_finalize"));
+  function expectCleanup(input: HeartbeatRuntimeExecutionInput) {
+    expect(revokeHeartbeatRunGatewayTokens).toHaveBeenCalledWith({ db, companyId: input.agent.companyId, runId: input.run.id });
+    expect(input.instructions.releaseInstructionCopy).toHaveBeenCalledOnce();
+  }
+
+  it.each([false, true])("preserves the dispatch gate's early return and cleanup (native: %s)", async native => {
+    const { input, adapter } = await fixture(native);
+    input.control.dispatchResolvedInteractionContinuationWithAtomicGate = async () => ({ dispatched: false });
+    expect(await executeHeartbeatRuntime(db, input)).toEqual({ dispatched: false });
+    expect(adapter.execute).not.toHaveBeenCalled();
+    expect(executePaperclipNativeSession).not.toHaveBeenCalled();
+    expect(input.effects.onNativeDispatchStarted).not.toHaveBeenCalled();
+    expect(input.effects.onLegacyAdapterEntered).not.toHaveBeenCalled();
+    expect(await finalizations()).toHaveLength(0);
+    expectCleanup(input);
+  });
+
+  it("captures usage and live instruction saves before finalizing a legacy workspace", async () => {
+    const { input, adapter, run } = await fixture();
+    adapter.execute.mockImplementationOnce(async ctx => {
+      expect(input.effects.onLegacyAdapterEntered).toHaveBeenCalledOnce();
+      expect(input.effects.onUsageCaptureReady).toHaveBeenCalledOnce();
+      await ctx.onProviderStopped?.();
+      return { exitCode: 0, signal: null, timedOut: false, usage: { inputTokens: 10, outputTokens: 4 }, costUsd: 0.01, provider: "test" };
+    });
+    expect(await executeHeartbeatRuntime(db, input)).toMatchObject({ dispatched: true, adapterResult: { usageComplete: true, resultJson: { instructionSave: { state: "saved", revisionId: "during-execution" } } } });
+    expect(await read(run.id)).toMatchObject({ status: "running", costAccountingPending: true, usageJson: { inputTokens: 10, outputTokens: 4, accountingReceiptReady: true } });
+    expect(await db.select().from(runUsageReceipts)).toHaveLength(1);
+    expect(await finalizations()).toMatchObject([{ status: "succeeded" }]);
+    expectCleanup(input);
+  });
+
+  it("preserves Stop metadata and marks no-work accounting without manufacturing a receipt", async () => {
+    const { input, adapter, run } = await fixture();
+    await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: { stoppedByUser: true } }).where(eq(heartbeatRuns.id, run.id));
+    adapter.execute.mockResolvedValueOnce({ exitCode: 0, signal: null, timedOut: false, executionRecovery: { kind: "bootstrap", providerWorkStarted: false } });
+    await executeHeartbeatRuntime(db, input);
+    expect(await read(run.id)).toMatchObject({ status: "cancelled", resultJson: { stoppedByUser: true }, usageJson: { accountingProviderWorkStarted: false } });
+    expect(await db.select().from(runUsageReceipts)).toHaveLength(0);
+    expectCleanup(input);
+  });
+
+  it("records the failed workspace and keeps the original provider error", async () => {
+    const { input, adapter } = await fixture();
+    const error = new Error("provider failed");
+    adapter.execute.mockRejectedValueOnce(error);
+    await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error);
+    expect(await finalizations()).toMatchObject([{ status: "failed", metadata: { errorMessage: "provider failed" } }]);
+    expect(input.effects.onLegacyAdapterEntered).toHaveBeenCalledOnce();
+    expectCleanup(input);
+  });
+
+  it.each([
+    [new NativeCancellationPendingRecoveryError(), null],
+    [new NativeControllerDetachedForRestartError(), "onNativeSessionResumeScheduled"],
+    [new NativeRunnerOwnershipUnverifiedError(), "onNativeOwnershipHeld"],
+    [new NativeWorkspaceFinalizationBusyError(), "onNativeWorkspaceFinalizeScheduled"],
+    [new NativeWorkspaceFinalizationOwnershipLostError(), "onNativeWorkspaceFinalizeScheduled"],
+  ] as const)("transfers recovery ownership for %s before cleanup", async (error, effect) => {
+    const { input } = await fixture(true);
+    vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error);
+    input.instructions.releaseInstructionCopy = vi.fn(async () => {
+      if (effect) expect(input.effects[effect]).toHaveBeenCalledOnce();
+    });
+    await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error);
+    expect(input.effects.onNativeDispatchStarted).toHaveBeenCalledOnce();
+    expect(input.effects.onLegacyAdapterEntered).not.toHaveBeenCalled();
+    expect(await finalizations()).toHaveLength(0);
+    expectCleanup(input);
+  });
+
+  it("dispatches native execution with the selected session and persists its usage", async () => {
+    const { input, adapter, run } = await fixture(true);
+    vi.mocked(executePaperclipNativeSession).mockImplementationOnce(async native => {
+      expect(native.execution).toBe(input.runtime.nativeExecution);
+      expect(native.runnerInstanceId).toBe(input.runtime.nativeRunnerInstanceId);
+      await native.onUsage?.({ usage: { inputTokens: 7, outputTokens: 3 }, complete: true });
+      return { exitCode: 0, signal: null, timedOut: false, usageComplete: false };
+    });
+    expect(await executeHeartbeatRuntime(db, input)).toMatchObject({ dispatched: true, adapterResult: { usage: { inputTokens: 7, outputTokens: 3 }, usageComplete: true } });
+    expect(adapter.execute).not.toHaveBeenCalled();
+    expect(input.effects.onNativeDispatchStarted).toHaveBeenCalledOnce();
+    expect(await read(run.id)).toMatchObject({ status: "running", usageJson: { inputTokens: 7, outputTokens: 3, accountingReceiptReady: true } });
+    expect(await finalizations()).toMatchObject([{ status: "succeeded" }]);
+    expectCleanup(input);
+  });
+
+  it("retains the same-run retry signal without recording a failed workspace", async () => {
+    const { input, run } = await fixture(true);
+    await db.update(heartbeatRuns).set({ nativeIssueId: input.task.issueId }).where(eq(heartbeatRuns.id, run.id));
+    await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, issueId: input.task.issueId!, runId: run.id, phase: "retryable_failure" });
+    const error = new Error("session connection lost");
+    vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error);
+    await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({
+      name: "NativeSessionResumeScheduledError", original: error,
+    });
+    expect(input.effects.onNativeSessionResumeScheduled).toHaveBeenCalledOnce();
+    expect(await finalizations()).toHaveLength(0);
+    expectCleanup(input);
+  });
+
+  it("records a failed barrier if the successful finalization write fails", async () => {
+    const { input } = await fixture();
+    const error = new Error("finalization write failed");
+    vi.spyOn(input.workspace.workspaceOperationRecorder, "recordOperation").mockRejectedValueOnce(error);
+    await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error);
+    expect(await finalizations()).toMatchObject([{ status: "failed" }]);
+    expectCleanup(input);
+  });
+
+  it.each([true, false])("validates computer worktrees remotely before finalization (matching branch: %s)", async matches => {
+    const { input, adapter } = await fixture();
+    const remoteCwd = "/home/user/paperclip/company/projects/project/task";
+    const runner: CommandManagedRuntimeRunner = {
+      execute: vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false, pid: null, startedAt: new Date().toISOString(),
+        stdout: matches ? "paperclip/task\n" : "other-task\n", stderr: "" })),
+    };
+    input.workspace.executionTarget = {
+      kind: "remote", transport: "computer", remoteCwd, runner, processRunner: runner,
+      listenerPort: 43100,
+      resourceAuthority: { kind: "computer-owner", computerId: randomUUID(), ownerId: randomUUID(), generation: 1 },
+      fileAuthority: { kind: "remote-persistent", placementId: randomUUID(), root: remoteCwd, agentHome: "/home/user/paperclip/company/agents/agent" },
+      launch: vi.fn(), inspectProcess: vi.fn(async () => ({ running: false, claim: null })),
+      retainWarm: vi.fn(), retire: vi.fn(async () => true), computerTool: { command: "computer", args: [] },
+    };
+    Object.assign(input.workspace.executionWorkspace, {
+      cwd: remoteCwd, worktreePath: remoteCwd, strategy: "git_worktree", branchName: "paperclip/task",
+    });
+    adapter.execute.mockImplementationOnce(async ctx => {
+      expect(ctx.stopRemoteStartup).toBeTypeOf("function");
+      await expect(ctx.stopRemoteStartup!()).rejects.toThrow("Remote startup stop requires a cancelled run");
+      expect(input.services.envOrchestrator.releaseForRun).not.toHaveBeenCalled();
+      return { exitCode: 0, signal: null, timedOut: false };
+    });
+    if (matches) {
+      await expect(executeHeartbeatRuntime(db, input)).resolves.toMatchObject({ dispatched: true });
+      expect(await finalizations()).toMatchObject([{ status: "succeeded" }]);
+    } else {
+      await expect(executeHeartbeatRuntime(db, input)).rejects.toThrow("computer_workspace_branch_mismatch");
+      expect(await finalizations()).toMatchObject([{ status: "failed" }]);
+    }
+    expect(runner.execute).toHaveBeenCalledOnce();
+    expect(runner.execute).toHaveBeenCalledWith(expect.objectContaining({
+      command: "git", args: ["symbolic-ref", "--quiet", "--short", "HEAD"], cwd: remoteCwd, bypassSession: true,
+    }));
+    expectCleanup(input);
+  });
+
+  it("retains a failed sandbox restore even when persisting its receipt throws", async () => {
+    const { input, adapter, run } = await fixture();
+    input.workspace.executionTarget = { kind: "remote", transport: "sandbox", remoteCwd: home };
+    const evidence = { workspaceRestoreFailure: "restore_failed", workspaceRestoreDiagnostic: { message: "copy-back failed" } };
+    adapter.execute.mockResolvedValueOnce({ exitCode: 0, signal: null, timedOut: false, resultJson: evidence });
+    const order: string[] = [];
+    input.effects.onWorkspaceRestoreFailure = vi.fn(() => { order.push("restore evidence"); });
+    input.effects.onProviderResourceDisposition = vi.fn(() => { order.push("retain sandbox"); });
+    input.instructions.releaseInstructionCopy = vi.fn(async () => { order.push("cleanup"); });
+    const error = new Error("restore receipt write failed");
+    const record = vi.spyOn(legacyRecovery, "recordLegacyWorkspaceRestoreFailure").mockImplementationOnce(async () => {
+      order.push("receipt write");
+      throw error;
+    });
+    try {
+      await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error);
+      expect(record).toHaveBeenCalledWith(db, run, evidence, input.workspace.workspaceRestoreSource);
+      expect(input.effects.onWorkspaceRestoreFailure).toHaveBeenCalledWith(evidence);
+      expect(input.effects.onProviderResourceDisposition).toHaveBeenCalledWith("stop_and_retain");
+      expect(order).toEqual(["restore evidence", "retain sandbox", "receipt write", "cleanup"]);
+      expect(await finalizations()).toMatchObject([{ status: "failed" }]);
+      expectCleanup(input);
+    } finally { record.mockRestore(); }
+  });
+
+  it("runs instruction cleanup even when gateway revocation fails", async () => {
+    const { input } = await fixture();
+    vi.mocked(revokeHeartbeatRunGatewayTokens).mockRejectedValueOnce(new Error("gateway unavailable"));
+    expect(await executeHeartbeatRuntime(db, input)).toMatchObject({ dispatched: true });
+    expectCleanup(input);
+  });
+});

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import { recordChatHandoff, recordChatCompletion, existingChatCompletionReply, acknowledgeChatCompletionReply } from "./chat-completion-delivery.js";
 import { mirrorSlackBoardComment, slackBoardReplyBindings } from "./slack-board-messages.js";
 import { assertAgentRunWriteAllowed } from "../agent-run-cancellation.js";
@@ -13047,49 +13048,55 @@ export function issueService(db: Db) {
               );
         const publicationCreatedAt = new Date();
         for (const binding of bindings) {
-          await dbOrTx
-            .insert(chatPublications)
-            .values({
-              companyId: binding.companyId,
-              endpointId: binding.endpointId,
-              conversationId: binding.conversationId,
-              issueId,
-              commentId: comment.id,
-              idempotencyKey: `comment:${comment.id}:${binding.endpointId}`,
-              payload: projectSafeChatPublication({
-                classification: "external",
-                source: "agent_comment",
-                text: redactedBody,
-              }),
-              state: "pending",
-              createdAt: publicationCreatedAt,
-              updatedAt: publicationCreatedAt,
-            })
-            .onConflictDoNothing();
-          for (const [index, attachment] of boundAttachments.entries()) {
-            const attachmentCreatedAt = new Date(
-              publicationCreatedAt.getTime() + index + 1,
-            );
-            await dbOrTx
+          await dbOrTx.transaction(async (tx: DbTransaction) => {
+            await notifyChatPublicationWork(tx);
+            return tx
               .insert(chatPublications)
               .values({
                 companyId: binding.companyId,
                 endpointId: binding.endpointId,
                 conversationId: binding.conversationId,
                 issueId,
-                commentId: attachment.commentId,
-                idempotencyKey: `attachment:${attachment.id}:${binding.endpointId}`,
+                commentId: comment.id,
+                idempotencyKey: `comment:${comment.id}:${binding.endpointId}`,
                 payload: projectSafeChatPublication({
                   classification: "external",
                   source: "agent_comment",
-                  text: `Shared ${attachment.originalFilename ?? "a file"}.`,
-                  attachmentIds: [attachment.id],
+                  text: redactedBody,
                 }),
                 state: "pending",
-                createdAt: attachmentCreatedAt,
-                updatedAt: attachmentCreatedAt,
+                createdAt: publicationCreatedAt,
+                updatedAt: publicationCreatedAt,
               })
               .onConflictDoNothing();
+          });
+          for (const [index, attachment] of boundAttachments.entries()) {
+            const attachmentCreatedAt = new Date(
+              publicationCreatedAt.getTime() + index + 1,
+            );
+            await dbOrTx.transaction(async (tx: DbTransaction) => {
+              await notifyChatPublicationWork(tx);
+              return tx
+                .insert(chatPublications)
+                .values({
+                  companyId: binding.companyId,
+                  endpointId: binding.endpointId,
+                  conversationId: binding.conversationId,
+                  issueId,
+                  commentId: attachment.commentId,
+                  idempotencyKey: `attachment:${attachment.id}:${binding.endpointId}`,
+                  payload: projectSafeChatPublication({
+                    classification: "external",
+                    source: "agent_comment",
+                    text: `Shared ${attachment.originalFilename ?? "a file"}.`,
+                    attachmentIds: [attachment.id],
+                  }),
+                  state: "pending",
+                  createdAt: attachmentCreatedAt,
+                  updatedAt: attachmentCreatedAt,
+                })
+                .onConflictDoNothing();
+            });
           }
         }
       }
@@ -13333,6 +13340,7 @@ export function issueService(db: Db) {
             registeredRunId,
           );
           for (const binding of bindings) {
+            await notifyChatPublicationWork(tx);
             await tx
               .insert(chatPublications)
               .values({

server/src/services/native-runtime/native-session-resume.test.ts

@@ -24,7 +24,6 @@ import {
   nativeRunResults,
 } from "@paperclipai/db";
 import { describe, expect, it, vi } from "vitest";
-import * as codexCommandResolver from "../../../../packages/paperclip-runner/src/drivers/codex/codex-command.js";
 import {
   CodexAppServerDriver,
   HarnessDriverBackend,
@@ -746,6 +745,9 @@ const recoveryFakeCodex = resolve(
     };
     // The production resolver prefers installed Codex over PATH. Explicitly
     // bind this disposable fixture so it can never invoke a real provider.
+    const codexCommandResolver = await vi.importActual<{ resolveCodexCommand: () => string }>(
+      "../../../../packages/paperclip-runner/src/drivers/codex/codex-command.js",
+    );
     const resolveFixtureCodex = vi.spyOn(codexCommandResolver, "resolveCodexCommand")
       .mockReturnValue(join(bin, "codex"));
     process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase;

server/src/services/slack-board-messages.ts

@@ -1,3 +1,4 @@
+import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
 import {
   assets,
@@ -121,6 +122,7 @@ export async function mirrorSlackBoardComment(
       .onConflictDoNothing()
       .returning();
     if (!receipt) continue;
+    await notifyChatPublicationWork(db);
     const [publication] = await db
       .insert(chatPublications)
       .values({
@@ -161,7 +163,7 @@ export async function mirrorSlackBoardComment(
           byId.has(id) ? [byId.get(id)!] : [],
         )
       : files;
-    for (const [index, file] of orderedFiles.entries())
+    for (const [index, file] of orderedFiles.entries()) {
       await db
         .insert(chatPublications)
         .values({
@@ -181,6 +183,7 @@ export async function mirrorSlackBoardComment(
           createdAt: new Date(Date.now() + index + 1),
         })
         .onConflictDoNothing();
+    }
     await logActivity(db as Db, {
       companyId: comment.companyId,
       actorType: "user",

server/src/services/voice/voice-session-store.ts

@@ -1,3 +1,4 @@
+import { notifyChatDeliveryWork } from "../chat-work-notifications.js";
 import { environmentService } from "../environments.js";
 import { logger } from "../../middleware/logger.js";
 import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
@@ -210,6 +211,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
     const threadId = conversation.externalThreadId;
     const providerEventId = `${threadId}:${session.id}:${toolId}`;
     const trigger = prior.length ? "subscribed_message" : "direct_message";
+    await notifyChatDeliveryWork(tx);
     const [delivery] = await tx.insert(chatDeliveries).values({
       companyId: session.companyId, endpointId: session.endpointId, conversationId: session.conversationId,
       providerEventId, deduplicationKey: createHash("sha256").update(providerEventId).digest("hex"),

tests/e2e/fixtures/composer-stop-codex-loader.mjs

@@ -0,0 +1,23 @@
+// The installed Codex dependency takes precedence over PATH. Pin only this
+// acceptance server's command resolver to its deterministic protocol fixture.
+import { accessSync, constants } from "node:fs";
+import { registerHooks } from "node:module";
+import { isAbsolute } from "node:path";
+
+const command = process.env.PAPERCLIP_STOP_CODEX_COMMAND;
+if (process.env.NODE_ENV !== "test" || !command || !isAbsolute(command)) {
+  throw new Error("Composer Stop requires an absolute fixture command in NODE_ENV=test");
+}
+accessSync(command, constants.X_OK);
+registerHooks({
+  load(url, context, nextLoad) {
+    if (/\/packages\/paperclip-runner\/(src|dist)\/drivers\/codex\/codex-command\.(js|ts)$/.test(url)) {
+      return {
+        format: "module",
+        shortCircuit: true,
+        source: `export const resolveCodexCommand = () => ${JSON.stringify(command)}; export const resolvePinnedCodexCommand = resolveCodexCommand;`,
+      };
+    }
+    return nextLoad(url, context);
+  },
+});

tests/e2e/fixtures/composer-stop-codex-loader.test.mjs

@@ -0,0 +1,41 @@
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { test } from "node:test";
+
+const loader = fileURLToPath(new URL("./composer-stop-codex-loader.mjs", import.meta.url));
+const resolver = new URL("../../../packages/paperclip-runner/src/drivers/codex/codex-command.ts", import.meta.url).href;
+function launch(env, source = "") {
+  return spawnSync(process.execPath, ["--import", loader, "--input-type=module", "-e", source], {
+    encoding: "utf8", env: { ...process.env, NODE_ENV: "test", PAPERCLIP_STOP_CODEX_COMMAND: "", ...env },
+  });
+}
+
+test("pins the fixture even when a bundled Codex dependency is installed", () => {
+  const dir = mkdtempSync(join(tmpdir(), "composer-stop-loader-"));
+  const command = join(dir, "codex");
+  try {
+    writeFileSync(command, "#!/bin/sh\nprintf 'composer-stop-fixture'\n", { mode: 0o755 });
+    const result = launch({ PAPERCLIP_STOP_CODEX_COMMAND: command }, `
+      import { resolveCodexCommand, resolvePinnedCodexCommand } from ${JSON.stringify(resolver)};
+      import { execFileSync } from "node:child_process";
+      import assert from "node:assert/strict";
+      assert.equal(resolveCodexCommand(), ${JSON.stringify(command)});
+      assert.equal(resolvePinnedCodexCommand(), ${JSON.stringify(command)});
+      process.stdout.write(execFileSync(resolveCodexCommand(), ["--version"]));
+    `);
+    assert.equal(result.status, 0, result.stderr);
+    assert.equal(result.stdout, "composer-stop-fixture");
+  } finally { rmSync(dir, { recursive: true, force: true }); }
+});
+
+for (const env of [{}, { PAPERCLIP_STOP_CODEX_COMMAND: "codex" }, { NODE_ENV: "production", PAPERCLIP_STOP_CODEX_COMMAND: process.execPath }]) {
+  test(`rejects a missing or unsafe fixture configuration: ${JSON.stringify(env)}`, () => {
+    const result = launch(env);
+    assert.notEqual(result.status, 0);
+    assert.match(result.stderr, /Composer Stop requires an absolute fixture command/);
+  });
+}

tests/e2e/playwright-composer-stop.config.ts

@@ -19,6 +19,7 @@ if (fixture) {
     throw new Error(
       "PAPERCLIP_STOP_FAKE_CODEX must name the built fake-codex-app-server binary",
     );
+  fs.mkdirSync(path.join(fixtureDir, "codex-home"));
   const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
   fs.writeFileSync(
     path.join(fixtureDir, "codex"),
@@ -37,6 +38,14 @@ export default defineConfig({
       ...server?.env,
       HEARTBEAT_SCHEDULER_INTERVAL_MS: "10000",
       PATH: `${fixtureDir}${path.delimiter}${process.env.PATH ?? ""}`,
+      ...(fixture ? {
+        // Provider selection prefers the installed dependency over PATH.
+        // Scope this override and its empty credential home to the test server.
+        PAPERCLIP_STOP_CODEX_COMMAND: path.join(fixtureDir, "codex"),
+        CODEX_HOME: path.join(fixtureDir, "codex-home"),
+        OPENAI_API_KEY: "composer-stop-fixture-key",
+        NODE_OPTIONS: `${server?.env?.NODE_OPTIONS ?? ""} --import=${path.resolve(import.meta.dirname, "fixtures/composer-stop-codex-loader.mjs")}`,
+      } : {}),
     },
   },
 });