Merge branch 'codex/sleep-run-task-recovery' into codex/sleep-issue-state-recovery

* codex/sleep-run-task-recovery:
  test: prove timer deadlines survive settled drain holds
  fix: back off expired deadlines during domain drain
  perf(heartbeat): let the dispatch worker own startup retry promotion
  fix(server): scope run retention hints to affected companies
  fix(heartbeat): scope and batch dispatch wakeups by owner
  fix(evals): support official TestOps job metadata (#18385)
  fix: release settled deadline work holds during drain
  fix(agents): revalidate error-clear authority before write intent
  test(skills): verify Cloud CLI reads through shared object storage

# Conflicts:
#	server/src/services/issues.ts
#	server/src/services/run-dispatch-notifications.ts

doc/plans/2026-10-30-run-dispatch-sleep.md

@@ -12,7 +12,7 @@ invalidate retries notify the retry worker in their transaction. Promotion uses
 the existing gate and compare-and-set, and its committed write wakes ordinary
 queued dispatch. A queued run behind occupied capacity waits for the owning
 run's lifecycle transition. Configuration and authority changes also wake it.
-An idle queue has no recurring read or timer-driven dispatch attempt.
+An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline.
 
 The supported writer contract is service writes through the shared database
 connection scope, including caller-owned and nested transactions. Existing

server/src/__tests__/agent-timer-deadlines.test.ts

@@ -597,8 +597,10 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => {
     await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } });
     await vi.advanceTimersByTimeAsync(2000);
 
-
-    expect(idleWorkSnapshot().active).toBe(1);
+    // The producer committed: retain its resume deadline without keeping the
+    // instance busy throughout the drain. Uncertain writes still hold admission.
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now());
     const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction");
     try {
       await vi.advanceTimersByTimeAsync(60_000);
@@ -608,6 +610,11 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => {
     stopTaskDrain();
     await vi.advanceTimersByTimeAsync(1000);
     await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 });
+    await vi.advanceTimersByTimeAsync(coordinator!.nextWakeAt()! - Date.now());
+    await vi.waitFor(() => expect(f.enqueueWakeup).toHaveBeenCalledOnce());
+    expect(f.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ source: "timer", reason: "heartbeat_timer" }));
+    await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: false } } });
+    await quiet();
   });
 
   it("does not inspect timer rows during startup drain, then reconstructs them on resume", async () => {

server/src/__tests__/deadline-work.test.ts

@@ -1,4 +1,5 @@
-import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { installDatabaseWorkSignals, signalDatabaseWork, databaseWorkPending } from "../../../packages/db/src/work-signals.js";
 import { afterEach, expect, it, vi } from "vitest";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { registerDeadlineWork } from "../services/deadline-work.js";
@@ -39,3 +40,132 @@ it.each([
   expect(run).toHaveBeenCalledTimes(2);
   expect(onError).not.toHaveBeenCalled();
 });
+
+it("releases a settled first-write hold during drain and retains a probe-free resume wake", async () => {
+  vi.useFakeTimers();
+  let draining = false;
+  const owner = installDatabaseWorkSignals({
+    execute: vi.fn(async () => [{ status: "committed" }]),
+    transaction: async (callback: (tx: object) => Promise<unknown>) => callback({ execute: async () => [{ xid: "1" }] }),
+  });
+  const run = vi.fn(async () => {}), next = vi.fn(async () => null);
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !draining, retryMs: 30_000 });
+  await worker.ready;
+  draining = true;
+  const before = idleWorkSnapshot().active;
+  await owner.transaction(async tx => { await signalDatabaseWork(tx, DELIVERY_QUEUES.agentTimers); });
+  await vi.advanceTimersByTimeAsync(0);
+  expect(idleWorkSnapshot().active).toBe(before);
+  expect(coordinator.nextWakeAt()).not.toBeNull();
+  await vi.advanceTimersByTimeAsync(3 * 30_000);
+  expect(run).toHaveBeenCalledOnce(); expect(next).toHaveBeenCalledOnce(); expect(owner.execute).not.toHaveBeenCalled();
+  draining = false;
+  await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("retains failed work through drain and preserves a future domain deadline", async () => {
+  vi.useFakeTimers();
+  let draining = false;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => { if (run.mock.calls.length === 1) { draining = true; throw new Error("retry debt"); } });
+  const deadline = Date.now() + 600_000;
+  const next = vi.fn(async () => deadline);
+  const onError = vi.fn();
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !draining, retryMs: 30_000 });
+  await worker.ready;
+  await vi.advanceTimersByTimeAsync(90_000);
+  expect(run).toHaveBeenCalledOnce(); expect(next).not.toHaveBeenCalled(); expect(owner.execute).not.toHaveBeenCalled();
+  draining = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBe(deadline);
+  draining = true; await vi.advanceTimersByTimeAsync(30_000);
+  expect(coordinator.nextWakeAt()).toBe(deadline); expect(onError).toHaveBeenCalledOnce();
+});
+
+it("does no domain probes in warm standby and reconstructs after admission opens", async () => {
+  vi.useFakeTimers();
+  let standby = true;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => {}), next = vi.fn(async () => null);
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !standby, canReconcile: () => !standby, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !standby, retryMs: 30_000 });
+  await worker.ready; await vi.advanceTimersByTimeAsync(90_000);
+  expect(run).not.toHaveBeenCalled(); expect(next).not.toHaveBeenCalled(); expect(owner.execute).not.toHaveBeenCalled();
+  standby = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("does not release an ambiguous database intent while dispatch is drained", async () => {
+  vi.useFakeTimers();
+  let draining = false, committed = false;
+  const owner = installDatabaseWorkSignals({
+    execute: vi.fn(async () => [{ status: committed ? "committed" : "in progress" }]),
+    transaction: async (callback: (tx: object) => Promise<unknown>) => {
+      await callback({ execute: async () => [{ xid: "2" }] });
+      throw new Error("lost acknowledgement");
+    },
+  });
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError: vi.fn() });
+  const run = vi.fn(async () => {});
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next: async () => null, canRun: () => !draining, retryMs: 30_000 });
+  await worker.ready; draining = true;
+  const before = idleWorkSnapshot().active;
+  await expect(owner.transaction(async tx => { await signalDatabaseWork(tx, DELIVERY_QUEUES.agentTimers); })).rejects.toThrow("lost acknowledgement");
+  await vi.advanceTimersByTimeAsync(0);
+  expect(idleWorkSnapshot().active).toBe(before + 1);
+  expect(databaseWorkPending(owner, DELIVERY_QUEUES.agentTimers)).toBe(true);
+  committed = true; await vi.advanceTimersByTimeAsync(30_000);
+  expect(databaseWorkPending(owner, DELIVERY_QUEUES.agentTimers)).toBe(false);
+  expect(idleWorkSnapshot().active).toBe(before);
+  draining = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("backs off an expired deadline when only the domain is drained and resumes promptly", async () => {
+  vi.useFakeTimers();
+  let paused = false;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => {});
+  const domainCanRun = vi.fn(() => !paused);
+  const deadline = Date.now() + 1_000;
+  const next = vi.fn(async () => run.mock.calls.length === 1 ? deadline : null);
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, {
+    run, next, canRun: domainCanRun, retryMs: 30_000,
+  });
+  await worker.ready;
+  paused = true;
+  const before = idleWorkSnapshot();
+  await vi.advanceTimersByTimeAsync(1_000);
+  expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
+  const attempts = domainCanRun.mock.calls.length;
+  await vi.advanceTimersByTimeAsync(90_000);
+  expect(domainCanRun.mock.calls.length - attempts).toBeLessThanOrEqual(12);
+  // Four attempts (expiry plus three retries), each begins and ends tracked work.
+  expect(idleWorkSnapshot().generation - before.generation).toBe(8);
+  expect(idleWorkSnapshot().active).toBe(before.active);
+  expect(run).toHaveBeenCalledOnce(); expect(next).toHaveBeenCalledOnce(); expect(owner.execute).not.toHaveBeenCalled();
+  paused = false;
+  await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("retains a prompt paused wake when a new request preempts a far-future deadline", async () => {
+  vi.useFakeTimers();
+  let paused = false;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => {});
+  const deadline = Date.now() + 86_400_000;
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, {
+    run, next: async () => run.mock.calls.length === 1 ? deadline : null, canRun: () => !paused, retryMs: 30_000,
+  });
+  await worker.ready; paused = true; worker.wake();
+  await vi.advanceTimersByTimeAsync(0);
+  expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
+  expect(owner.execute).not.toHaveBeenCalled();
+  paused = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});

server/src/__tests__/execution-control-work.test.ts

@@ -383,11 +383,11 @@ describe.skipIf(!support.supported)("execution-control durable work", () => {
     const due = new Date(Date.now() + 120_000), earlier = new Date(Date.now() + 60_000);
     for (const at of [due, earlier]) {
       await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns)
-        .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)));
+        .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)), { companyId });
       await settle(); expect(c.nextWakeAt()).toBe(at.getTime());
     }
     await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns)
-      .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)));
+      .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)), { companyId });
     await quiet(c);
   });
   it("delivers a reconciled continuation when marking and resolving commit together", async () => {

server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts

@@ -1,3 +1,4 @@
+import { idleWorkSnapshot } from "../services/task-admission.js";
 import { PgDialect } from "drizzle-orm/pg-core";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
@@ -28,7 +29,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean
     if (state.due !== null && state.due <= Date.now()) state.due = null;
   });
   const next = vi.fn(async () => state.due);
-  const dispatch = vi.fn(async () => {});
+  const dispatch = vi.fn(async (_scope?: { companyId: string }) => {});
   const coordinator = createDeliveryWorkCoordinator({ owner: db,
     canRun: () => !state.draining && !state.standby, canReconcile: () => !state.standby, onError: vi.fn() });
   const worker = registerRunDispatchWork(coordinator, {
@@ -37,8 +38,8 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean
   }, () => !state.draining && !state.standby);
   cleanup.push(() => coordinator.stop());
   return { db, state, promote, next, dispatch, coordinator, worker,
-    notify: () => db.transaction(tx => notifyRunDispatch(tx)),
-    settle: () => vi.advanceTimersByTimeAsync(0) };
+    notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)),
+    settle: () => vi.advanceTimersByTimeAsync(1_000) };
 }
 
 describe("run dispatch obligations", () => {
@@ -67,20 +68,39 @@ describe("run dispatch obligations", () => {
     const before = f.next.mock.calls.length;
     await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before);
   });
+  it("batches repeated tenant writes and passes only their company to queued dispatch", async () => {
+    const f = fixture(); await f.worker.ready;
+    for (let index = 0; index < 100; index++) {
+      await f.notify({ companyId: "company-a" });
+      await vi.advanceTimersByTimeAsync(10);
+    }
+    expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2);
+    expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2);
+    await f.settle();
+    expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" });
+    expect(f.dispatch.mock.calls.slice(1).every(([scope]) => scope?.companyId === "company-a")).toBe(true);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+    const count = f.dispatch.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.dispatch).toHaveBeenCalledTimes(count);
+  });
   it("remembers the first producer arriving during drain without owner SQL", async () => {
     const f = fixture(); await f.worker.ready; f.state.draining = true;
     await f.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
     expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce();
+    expect(idleWorkSnapshot().active).toBe(0);
     f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
     expect(f.promote).toHaveBeenCalledTimes(2); expect(f.dispatch).toHaveBeenCalledTimes(2);
     expect(f.coordinator.nextWakeAt()).toBeNull();
   });
-  it("retains failed work across drain and settles after recovery", async () => {
-    const f = fixture(); f.state.fail = true; await f.worker.ready;
+  it("releases a failed producer hold during drain and retains recovery for resume", async () => {
+    const f = fixture(); await f.worker.ready;
+    f.state.fail = true; await f.notify(); await f.settle();
+    expect(f.promote).toHaveBeenCalledTimes(2);
     f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000);
-    expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).not.toHaveBeenCalled();
+    expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce();
+    expect(idleWorkSnapshot().active).toBe(0);
     f.state.fail = false; f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
-    expect(f.promote).toHaveBeenCalledTimes(2); expect(f.coordinator.nextWakeAt()).toBeNull();
+    expect(f.promote).toHaveBeenCalledTimes(3); expect(f.coordinator.nextWakeAt()).toBeNull();
   });
   it("preserves a notification during a nonoverlapping pass", async () => {
     const f = fixture(); await f.worker.ready;

server/src/__tests__/heartbeat-run-dispatch-work.test.ts

@@ -1,3 +1,4 @@
+import { idleWorkSnapshot } from "../services/task-admission.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -76,9 +77,11 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
     return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]);
   }
   async function completed(id: string) {
-    await vi.waitFor(async () => expect((await status(id))?.status).toBe("succeeded"), { timeout: 10_000 });
+    await vi.waitFor(async () => expect(["running", "succeeded"]).toContain((await status(id))?.status), { timeout: 10_000 });
     await heartbeat.drainActiveRunExecutions();
-    expect((await status(id)).stdoutExcerpt).toContain("dispatch-proof");
+    const completed = await status(id);
+    expect(completed.status).toBe("succeeded");
+    expect(completed.stdoutExcerpt).toContain("dispatch-proof");
   }
   it("keeps fresh empty, configured idle and retained terminal history quiet", async () => {
     await start(); await quiet();
@@ -87,10 +90,10 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
   });
   it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => {
     const f = await fixture(); await start(); await quiet();
-    const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 2_000, random: () => 0 });
+    const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 });
     expect(result.outcome).toBe("scheduled");
     if (result.outcome !== "scheduled") throw new Error("Expected retry creation");
-    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(result.dueAt.getTime()));
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(result.dueAt.getTime()), { timeout: 5_000 });
     expect((await status(result.run.id)).status).toBe("scheduled_retry");
     await new Promise(resolve => setTimeout(resolve, 100));
     expect(Date.now()).toBeLessThan(result.dueAt.getTime());
@@ -110,7 +113,7 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
     const f = await fixture(); await start();
     const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 600_000 });
     if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
-    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()));
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()), { timeout: 5_000 });
     await heartbeat.cancelRun(scheduled.run.id, "operator stop", { suppressImmediateRecovery: true });
     await quiet(); expect((await status(scheduled.run.id)).status).toBe("cancelled");
   });
@@ -118,7 +121,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
     const f = await fixture(); await start(); await quiet(); draining = true;
     const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 200 });
     if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
-    await new Promise(resolve => setTimeout(resolve, 400));
+    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    const count = reads.mock.calls.length;
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      await vi.advanceTimersByTimeAsync(90_000);
+      expect(reads).toHaveBeenCalledTimes(count);
+      expect(idleWorkSnapshot().active).toBe(0);
+    } finally { vi.useRealTimers(); }
     expect((await status(scheduled.run.id)).status).toBe("scheduled_retry");
     draining = false; worker!.wake();
     await completed(scheduled.run.id); await quiet();
@@ -151,13 +161,39 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
         await blocker.select().from(agents).where(eq(agents.id, f.agentId)).for("update");
         writing = agentService(db).clearError(f.agentId);
         void writing.catch(() => {});
-        await vi.waitFor(() => expect(databaseWorkPending(db, DELIVERY_QUEUES.queuedRunDispatch)).toBe(true));
+        await new Promise(resolve => setTimeout(resolve, 100));
+        expect(databaseWorkPending(db, DELIVERY_QUEUES.queuedRunDispatch)).toBe(false);
         expect((await status(queued.id)).status).toBe("queued");
         expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("error");
       });
     } finally { await writing; }
     await completed(queued.id); await quiet();
   });
+  it("leaves cancellation-owned retries quiet even when their old promotion time is overdue", async () => {
+    const f = await fixture();
+    const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { now: new Date(Date.now() - 60_000), delayMs: 1_000 });
+    if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await db.update(heartbeatRuns).set({ resultJson: { startupCancellation: { requestedAt: new Date().toISOString() },
+      cancellation: { expected: true, reason: "operator stop", source: "operator" } } }).where(eq(heartbeatRuns.id, scheduled.run.id));
+    await start(); await quiet();
+    expect((await status(scheduled.run.id)).status).toBe("scheduled_retry");
+    expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] });
+    await quiet();
+  });
+  it("rejects an error clear without publishing work when the lifecycle is no longer ready", async () => {
+    const f = await fixture(); await start(); await quiet();
+    await db.update(agents).set({ status: "error", lifecycleState: "paused" }).where(eq(agents.id, f.agentId));
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      await expect(agentService(db).clearError(f.agentId)).rejects.toThrow("Only agents in error status");
+      expect(coordinator!.nextWakeAt()).toBeNull();
+      const count = reads.mock.calls.length;
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(reads).toHaveBeenCalledTimes(count);
+    } finally { vi.useRealTimers(); }
+    expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("error");
+    await quiet();
+  });
   it("keeps a foreign company's future retry intact while concurrent workers claim one due retry", async () => {
     const selected = await fixture(), foreign = await fixture(); await start();
     const later = await heartbeat.scheduleBoundedRetry(foreign.runId, { delayMs: 600_000 });

server/src/__tests__/managed-skill-content.test.ts

@@ -1,3 +1,5 @@
+import { createServer } from "node:http";
+import { createS3StorageProvider } from "../storage/s3-provider.js";
 import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js";
 import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js";
 import { eq } from "drizzle-orm";
@@ -79,7 +81,96 @@ describe.skipIf(!support.supported)("managed skills use committed versions", ()
 
   it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => {
     const companyId = await company();
-    await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" });
+    const objects = new Map<string, Buffer>();
+    const server = createServer((request, response) => {
+      void (async () => {
+        const key = new URL(request.url!, "http://fixture").pathname;
+        if (request.method === "PUT") {
+          const chunks: Buffer[] = [];
+          for await (const chunk of request) chunks.push(Buffer.from(chunk));
+          let body = Buffer.concat(chunks);
+          if (request.headers["content-encoding"]?.includes("aws-chunked")) {
+            const decoded: Buffer[] = [];
+            let offset = 0;
+            while (offset < body.length) {
+              const end = body.indexOf("\r\n", offset);
+              const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16);
+              if (size === 0) break;
+              decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2;
+            }
+            body = Buffer.concat(decoded);
+          }
+          objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return;
+        }
+        const body = objects.get(key);
+        if (!body) { response.writeHead(404).end(); return; }
+        response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' });
+        response.end(request.method === "HEAD" ? undefined : body);
+      })().catch(() => { response.writeHead(500).end(); });
+    });
+    await new Promise<void>(resolve => { server.listen(0, "127.0.0.1", resolve); });
+    const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address");
+    const endpoint = `http://127.0.0.1:${address.port}`;
+    const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" };
+    const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]]));
+    Object.assign(process.env, credentials);
+    vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({
+      bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true,
+    })));
+    const key = "cloud-runtime-config:v1";
+    try {
+      await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" });
+      const run = (phase: string) => promisify(execFile)(process.execPath, [
+        "--import", fileURLToPath(import.meta.resolve("tsx")),
+        fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)),
+        `--phase=${phase}`, `--company=${companyId}`, "--limit=50",
+      ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString,
+        PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture",
+        PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3",
+        PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1",
+        PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" },
+        timeout: 60_000, maxBuffer: 1024 * 1024 });
+      await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration");
+      expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0);
+      await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } });
+      const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!);
+      expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]);
+      const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const);
+      const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]);
+      expect([...objects.entries()]).toEqual(before);
+      const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!;
+      const archive = objects.get(archiveKey)!;
+      const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1;
+      objects.set(archiveKey, corrupted);
+      const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]);
+      objects.set(archiveKey, archive);
+      const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!;
+      const manifest = objects.get(manifestKey)!; objects.delete(manifestKey);
+      const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]);
+      objects.set(manifestKey, manifest);
+      const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]);
+      expect([...objects.entries()].sort()).toEqual([...before].sort());
+      expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key)))
+        .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]);
+    } finally {
+      await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key));
+      vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects"))));
+      for (const [key, value] of Object.entries(previousCredentials)) {
+        if (value === undefined) delete process.env[key]; else process.env[key] = value;
+      }
+      server.closeAllConnections(); await new Promise<void>(resolve => { server.close(() => resolve()); });
+    }
+  }, 180_000);
+
+  it("reports a provider mismatch for local versions under persisted Cloud configuration", async () => {
+    const companyId = await company();
+    const skill = await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" });
+    const local = await skillInstallationMaintenance(db).verifyVersions({ companyId, limit: 50 });
+    expect(local.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "local_storage" })]);
     const key = "cloud-runtime-config:v1";
     const run = (phase: string) => promisify(execFile)(process.execPath, [
       "--import", fileURLToPath(import.meta.resolve("tsx")),
@@ -87,16 +178,18 @@ describe.skipIf(!support.supported)("managed skills use committed versions", ()
       `--phase=${phase}`, `--company=${companyId}`, "--limit=50",
     ], { env: { ...process.env, DATABASE_URL: database.connectionString,
       PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture",
-      PAPERCLIP_STORAGE_PROVIDER: "local_disk", PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects") },
+      PAPERCLIP_STORAGE_PROVIDER: "s3", PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-cli-fixture",
+      PAPERCLIP_STORAGE_S3_REGION: "us-east-1", PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64) },
       timeout: 60_000, maxBuffer: 1024 * 1024 });
     await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration");
     expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0);
-    await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } });
+    const [persisted] = await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }).returning();
     try {
       const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!);
-      expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]);
+      expect(eligibility.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "version_backed" })]);
       const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
-      expect(verification.results).toEqual([expect.objectContaining({ status: "local_storage" })]);
+      expect(verification.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "provider_mismatch" })]);
+      expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toEqual([persisted]);
     } finally {
       await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key));
     }

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -1341,7 +1341,7 @@ describe("startServer feedback export wiring", () => {
       expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
       expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true });
       expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
-      expect(heartbeatServiceMock.promoteDueScheduledRetries).toHaveBeenCalledTimes(1);
+      expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled();
       expect(retiredDetector).not.toHaveBeenCalled();
     } finally {
       delete (runtime as Partial<typeof runtime>).reconcileProductivityReviews;

server/src/index.ts

@@ -1778,7 +1778,6 @@ async function startServerWithDatabaseTeardown(
           }
         }
 
-        const promotion = await timeStartupRecovery("scheduled_retries", () => heartbeat.promoteDueScheduledRetries());
         await timeStartupRecovery("queued_runs", () => heartbeat.resumeQueuedRuns());
         const recoveredGoalActions = await timeStartupRecovery("goal_actions", () => heartbeat.recoverPendingSessionGoalActions());
         if (
@@ -1799,7 +1798,6 @@ async function startServerWithDatabaseTeardown(
         }
         const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
         if (
-          promotion.promoted > 0 ||
           reconciled.assignmentDispatched > 0 ||
           reconciled.dispatchRequeued > 0 ||
           reconciled.continuationRequeued > 0 ||
@@ -1808,7 +1806,7 @@ async function startServerWithDatabaseTeardown(
           reconciled.escalated > 0
         ) {
           logger.warn(
-            { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled },
+            { ...reconciled },
             "startup heartbeat recovery changed assigned issue state",
           );
         }

server/src/modules/agent-lifecycle/adapters/records.ts

@@ -81,22 +81,19 @@ export function agentRecords(db: Db, effects: LifecycleEffects) {
       }
 
       const updated = await effects.transaction(db, existing.companyId, async tx => {
+        const [current] = await tx.select({ status: agents.status, lifecycleState: agents.lifecycleState })
+          .from(agents).where(and(eq(agents.id, id), eq(agents.companyId, existing.companyId))).for("no key update");
+        if (current?.status !== "error" || current.lifecycleState !== "ready") {
+          throw conflict("Only agents in error status can have their error cleared");
+        }
         await effects.beforeStatusChange(tx, existing.companyId, id, "error", "idle");
-        return tx.update(agents).set({
-          status: "idle",
-          pauseReason: null,
-          pausedAt: null,
-          errorReason: null,
-          updatedAt: new Date(),
-        })
-          .where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready")))
-          .returning()
-          .then((rows) => rows[0] ?? null);
+        const [saved] = await tx.update(agents).set({
+          status: "idle", pauseReason: null, pausedAt: null, errorReason: null, updatedAt: new Date(),
+        }).where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready"))).returning();
+        if (!saved) throw conflict("Only agents in error status can have their error cleared");
+        return saved;
       });
 
-      if (!updated) {
-        throw conflict("Only agents in error status can have their error cleared");
-      }
       return getById(updated.id);
     },
 

server/src/modules/run-dispatch/adapters/postgres.ts

@@ -514,6 +514,7 @@ export function createPostgresRunDispatchAdapter(
   async function nextScheduledRetryAt(input: { cutoff: Date | null }): Promise<number | null> {
     const [row] = await db.select({ at: heartbeatRuns.scheduledRetryAt }).from(heartbeatRuns)
       .where(and(eq(heartbeatRuns.status, "scheduled_retry"),
+        sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') is distinct from 'object'`,
         input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined))
       .orderBy(asc(heartbeatRuns.scheduledRetryAt)).limit(1);
     return row?.at?.getTime() ?? null;
@@ -526,6 +527,7 @@ export function createPostgresRunDispatchAdapter(
       .where(
         and(
           eq(heartbeatRuns.status, "scheduled_retry"),
+          sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') is distinct from 'object'`,
           lte(heartbeatRuns.scheduledRetryAt, input.now),
           input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined,
         ),

server/src/services/deadline-work.ts

@@ -26,7 +26,7 @@ export function registerDeadlineWork(
     pausePassiveMaintenance: true,
     onSettled: hints => { requested++; task.onSettled?.(hints); },
     run: async signal => {
-      if (!task.canRun()) { requested++; next = null; return; }
+      if (!task.canRun()) { requested++; return; }
       if (Date.now() < earliest) { next = earliest; return; }
       if (Date.now() < coalesceUntil && (next === null || next > Date.now())) {
         next = next === null ? coalesceUntil : Math.min(next, coalesceUntil);
@@ -39,8 +39,6 @@ export function registerDeadlineWork(
       if (signal.aborted || !task.canRun()) { next = null; return; }
       const deadline = await task.next();
 
-
-
       if (!task.canRun()) { next = null; return; }
       earliest = deadline === null || deadline > Date.now()
         ? (task.minimumSpacingMs ? Date.now() + task.minimumSpacingMs : 0)
@@ -49,8 +47,13 @@ export function registerDeadlineWork(
       coalesceUntil = Date.now() + (task.notificationCoalesceMs ?? 0);
       reconciled = generation;
     },
-    hasPending: async () => requested !== reconciled,
-    nextRunAt: () => next,
+    // A settled producer must release its admission hold during drain. Keep
+    // the unreconciled generation and an in-memory resume wake, without asking
+    // the domain to probe its database while dispatch is closed.
+    hasPending: async () => task.canRun() && requested !== reconciled,
+    nextRunAt: () => !task.canRun() && requested !== reconciled
+      ? Date.now() + task.retryMs
+      : next,
   });
   return { ...worker, wake: () => { requested++; worker.wake(); } };
 }

server/src/services/execution-control-notifications.ts

@@ -2,8 +2,10 @@ import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
 
-export async function notifyExecutionRunWork(transaction: object) {
-  await notifyRunDispatch(transaction);
+export type ExecutionWorkScope = { companyId: string; agentId?: string; runId?: string };
+
+export async function notifyExecutionRunWork(transaction: object, scope?: ExecutionWorkScope) {
+  if (scope) await notifyRunDispatch(transaction, scope);
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
     await notifyDeliveryWork(transaction, queue);
@@ -14,9 +16,10 @@ export function writeExecutionControl<T>(
   db: Db | Parameters<Parameters<Db["transaction"]>[0]>[0],
   queue: DeliveryQueue | readonly DeliveryQueue[],
   write: (tx: Db) => PromiseLike<T>,
+  scope: ExecutionWorkScope | undefined,
 ): Promise<T> {
   return db.transaction(async transaction => {
-    await notifyRunDispatch(transaction);
+    await notifyRunDispatch(transaction, scope);
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

server/src/services/execution-control-reconciliation.ts

@@ -196,7 +196,7 @@ export async function reconcileAbandonedExecutionControl(
           await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id });
           await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
           await notifyUnsafeWorkspaceExportRunChange(tx, run);
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           const [updatedRun] = await tx
             .update(heartbeatRuns)
             .set({

server/src/services/execution-recovery-resolution.ts

@@ -546,7 +546,7 @@ export async function settleUnrecoverableExecutions(
             companyId: run.companyId, issueId: task.id, runId: run.id,
           }).limit(1).for("update");
           if (!hold) return;
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           const note = "Exact session recovery stopped. The recorded workspace recovery holds this execution without replaying its actions.";
           await tx.update(issueRecoveryActions).set({ status: "cancelled", outcome: "cancelled",
             resolvedAt: now, updatedAt: now, wakePolicy: null, monitorPolicy: null,
@@ -654,7 +654,7 @@ export async function settleUnrecoverableExecutions(
             replay: "not_authorized",
           },
         });
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await tx
           .update(heartbeatRuns)
           .set({ executionStatusDeliveryId: randomUUID() })

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

@@ -475,7 +475,7 @@ export async function admitExplicitNativeContinuation(input: {
   }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.queuedCommentInterruptId),
     eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution")));
   if (cancelledStartupIds.size) await db.transaction(async tx => {
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId, agentId });
     for (const runId of cancelledStartupIds) {
       await tx.update(nativeRunFinalizations).set({
         phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null,

server/src/services/heartbeat.ts

@@ -1336,8 +1336,8 @@ export function heartbeatService(
     return heartbeatQueue.dispatchPendingNativeStatusWakeups(...args);
   }
 
-  function resumeQueuedRunsAfterDrain() {
-    const promise = heartbeatQueue.resumeQueuedRunsAfterDrain();
+  function resumeQueuedRunsAfterDrain(...args: Parameters<typeof heartbeatQueue.resumeQueuedRunsAfterDrain>) {
+    const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(...args);
     activeWakeupPromises.add(promise);
     void promise.then(
       () => { activeWakeupPromises.delete(promise); },
@@ -3721,7 +3721,7 @@ export function heartbeatService(
           await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx
             .update(heartbeatRuns)
             .set({ executionControlDeadlineAt: null })
-            .where(eq(heartbeatRuns.id, run.id)));
+            .where(eq(heartbeatRuns.id, run.id)), { companyId: run.companyId });
         }
         nativeOwnershipHeld =
           nativeOwnershipHeld ||

server/src/services/heartbeat/dispatch-work.ts

@@ -1,3 +1,4 @@
+import type { RunDispatchScope } from "../run-dispatch-notifications.js";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { registerDeadlineWork } from "../deadline-work.js";
 import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
@@ -7,7 +8,7 @@ export function registerRunDispatchWork(
   owner: {
     promoteDueScheduledRetries(): Promise<unknown>;
     nextScheduledRetryAt(): Promise<number | null>;
-    resumeQueuedRunsAfterDrain(): Promise<unknown>;
+    resumeQueuedRunsAfterDrain(scope?: RunDispatchScope): Promise<unknown>;
     schedulingSuppressed(): Promise<boolean>;
   },
   canRun: () => boolean,
@@ -16,6 +17,7 @@ export function registerRunDispatchWork(
   const retries = registerDeadlineWork(coordinator, DELIVERY_QUEUES.scheduledRetryDispatch, {
     retryMs: 30_000,
     continuationMs: 1_000,
+    notificationCoalesceMs: 1_000,
     canRun,
     run: async signal => {
       suppressed = await owner.schedulingSuppressed();
@@ -23,17 +25,34 @@ export function registerRunDispatchWork(
     },
     next: () => suppressed ? Promise.resolve(null) : owner.nextScheduledRetryAt(),
   });
+  const scopes = new Set<string>(["*"]);
   const queue = registerDeadlineWork(coordinator, DELIVERY_QUEUES.queuedRunDispatch, {
     retryMs: 30_000,
+    notificationCoalesceMs: 1_000,
     canRun,
+    onSettled: hints => { for (const hint of hints ?? ["*"]) scopes.add(hint); },
     run: async signal => {
-      if (!signal.aborted && canRun()) await owner.resumeQueuedRunsAfterDrain();
+      const requested = scopes.has("*") ? ["*"] : [...scopes];
+      if (requested[0] === "*") scopes.clear();
+      else for (const hint of requested) scopes.delete(hint);
+      for (let index = 0; index < requested.length; index++) {
+        if (signal.aborted || !canRun()) {
+          for (const hint of requested.slice(index)) scopes.add(hint);
+          return;
+        }
+        try {
+          await owner.resumeQueuedRunsAfterDrain(requested[index] === "*" ? undefined : JSON.parse(requested[index]) as RunDispatchScope);
+        } catch (error) {
+          for (const hint of requested.slice(index)) scopes.add(hint);
+          throw error;
+        }
+      }
     },
     next: async () => null,
   });
   return {
     ready: Promise.all([retries.ready, queue.ready]),
-    wake() { retries.wake(); queue.wake(); },
+    wake() { scopes.add("*"); retries.wake(); queue.wake(); },
     async stop() { await Promise.all([retries.stop(), queue.stop()]); },
   };
 }

server/src/services/heartbeat/queue.ts

@@ -1973,7 +1973,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
 
   // Release admission promptly without first scanning unrelated recovery work.
   // Every run still passes the ordinary per-agent claim and dispatch gates.
-  async function resumeQueuedRunsAfterDrain() {
+  async function resumeQueuedRunsAfterDrain(scope?: { companyId: string }) {
     if ((await getSchedulingSuppression()).suppressed) return;
     const cutoff = await getWorktreeExecutionCutoff();
     const queuedRuns = await db
@@ -1983,6 +1983,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       .where(
         and(
           eq(heartbeatRuns.status, "queued"),
+          scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
           eq(companies.status, "active"),
           cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined,
         ),

server/src/services/heartbeat/run-lifecycle.ts

@@ -1045,7 +1045,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
       if (previous && isHeartbeatRunTerminalStatus(status))
         await notifyRetainedNativeSessionCleanup(tx, { companyId: previous.companyId, runId: previous.id });
       if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
-      await notifyExecutionRunWork(tx);
+      await notifyExecutionRunWork(tx, companyId ? { companyId } : undefined);
       if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id,
         issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined });
       return write(tx);

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

@@ -966,7 +966,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
               eq(heartbeatRuns.id, run.id),
               eq(heartbeatRuns.status, "running"),
             ),
-          ));
+          ), { companyId: run.companyId });
         const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded";
         await recordWorkspaceFinalize(workspaceFinalizeStatus);
         if (adapterResult.nativeFinalization) {
@@ -1042,7 +1042,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
           eq(heartbeatRuns.id, run.id),
           eq(heartbeatRuns.status, "running"),
         ),
-      ));
+      ), { companyId: run.companyId });
     if (
       issueRef &&
       context.resumeSessionGoalHeartbeat === true &&

server/src/services/issue-recovery-actions.ts

@@ -392,7 +392,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now })
             .where(and(eq(issueRecoveryActions.id, existing.id),
               inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
-              sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning());
+              sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning(), { companyId: input.companyId });
           if (updated) return toReadModel(updated);
         }
         return existing;
@@ -467,7 +467,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
               ),
             )
             .returning();
-        });
+        }, { companyId: input.companyId });
         if (!exhausted) {
           return retryUpsertSourceScoped(input, retryCount);
         }
@@ -539,7 +539,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
             ),
           )
           .returning();
-      });
+      }, { companyId: input.companyId });
       if (!updated) {
         return retryUpsertSourceScoped(input, retryCount);
       }
@@ -554,7 +554,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           .insert(issueRecoveryActions)
           .values(buildInsertValues(input, ownerType, now))
           .returning();
-      });
+      }, { companyId: input.companyId });
       return toReadModel(created!);
     } catch (error) {
       if (!isUniqueRecoveryActionConflict(error)) throw error;
@@ -608,7 +608,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
         })
         .where(and(...predicates))
         .returning();
-    });
+    }, { companyId: input.companyId });
 
     return updated ? toReadModel(updated) : null;
   }

server/src/services/legacy-execution-recovery.ts

@@ -135,7 +135,7 @@ export async function terminalizeLegacyExecution(input: {
     };
     if (issueId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: current.companyId, issueId });
     await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id });
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: current.companyId });
     let [updated] = await tx
       .update(heartbeatRuns)
       .set({

server/src/services/native-runtime/native-failed-workspace-recovery.ts

@@ -65,7 +65,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
     let retained = !retainFiles || alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source);
     if (!retained || retainFiles && !hasRequiredWorkspaceRecovery(retained.resultJson)) return;
     if (options.allowPreterminal && coordinator.phase === "observed") {
-      await notifyExecutionRunWork(tx);
+      await notifyExecutionRunWork(tx, { companyId: run.companyId });
       const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure",
         leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null,
         recoveryState: "blocked", failureCode: coordinator.failureCode ?? "native_continuation_requires_reconciliation", updatedAt: new Date(),

server/src/services/native-runtime/native-restart-recovery.ts

@@ -899,7 +899,7 @@ export async function claimNativeRestartRecoveries(input: {
           recycledProviderPids: providerProcesses.recycledPids,
           stderrTail: row.run.stderrExcerpt,
         });
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: row.run.companyId });
         await tx
           .update(nativeRunFinalizations)
           .set({

server/src/services/native-runtime/native-run-finalizer.ts

@@ -401,7 +401,7 @@ async function recordRetryableFailure(input: {
     if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") {
       await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     }
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
     await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     await tx
       .update(nativeRunFinalizations)
@@ -1659,7 +1659,7 @@ export async function finalizeNativeRun(input: {
         if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
         if (input.projectRunStatus) await notifyUnsafeWorkspaceExportRunChange(tx, run);
         if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
         const [after] = await projectRun(tx);
         return after ? { before, after } : null;

server/src/services/native-runtime/native-run-projection.ts

@@ -138,7 +138,7 @@ export async function projectCommittedRun(input: {
     await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id });
     await notifyUnsafeWorkspaceExportRunChange(tx, input.run);
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
     await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     const [after] = await tx
       .update(heartbeatRuns)

server/src/services/native-runtime/native-safe-replacement.ts

@@ -267,7 +267,7 @@ export async function reconcileSafeNativeReplacements(
             current.failureDetail?.replacementDenied
           )
             return;
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           await tx
             .update(nativeRunFinalizations)
             .set({
@@ -409,7 +409,7 @@ export async function reconcileSafeNativeReplacements(
             idempotencyKey: `native-safe-replacement:${run.id}`,
           })
           .returning();
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await tx.insert(heartbeatRuns).values({
           id: successorRunId,
           companyId: run.companyId,

server/src/services/native-runtime/native-session-executor.ts

@@ -9691,7 +9691,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.runId, input.execution.binding.runId),
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
           ),
-        ));
+        ), { companyId: input.execution.binding.companyId });
 
       const failedAtMs = Date.now();
       const executionFailureMessage = redactIdentityText(
@@ -9743,7 +9743,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
             eq(nativeRunFinalizations.attempt, attempt),
             isNull(nativeRunFinalizations.resultId),
-          )).returning({ runId: nativeRunFinalizations.runId }));
+          )).returning({ runId: nativeRunFinalizations.runId }), { companyId: input.execution.binding.companyId });
           if (settled) {
             await stoppedLeaseRenewal;
             if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
@@ -9830,7 +9830,7 @@ async function executePaperclipNativeSessionWithinScope(
             ),
           )
           .for("update");
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: input.execution.binding.companyId });
         await notifyNativeFinalizationWork(tx, input.execution.binding);
         const updated = await tx
           .update(nativeRunFinalizations)

server/src/services/native-runtime/native-superseded-retry.ts

@@ -39,7 +39,7 @@ export async function retireSupersededNativeRetry(db: Db, input: {
   )).limit(1);
   if (!source) return false;
 
-  await notifyExecutionRunWork(db);
+  await notifyExecutionRunWork(db, { companyId: run.companyId });
   await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null,
     leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: "blocked",
     failureCode: "native_retry_superseded", failureDetail: { ...coordinator.failureDetail,

server/src/services/run-dispatch-notifications.ts

@@ -1,8 +1,11 @@
-import { notifyIssueRecovery, type IssueRecoveryScope } from "./issue-recovery-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export async function notifyRunDispatch(transaction: object, scope?: IssueRecoveryScope) {
+export type RunDispatchScope = { companyId: string };
+
+export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
   await notifyIssueRecovery(transaction, scope);
-  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch);
-  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch);
+  const hint = scope ? JSON.stringify(scope) : "*";
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);
 }

server/src/services/system-recovery.ts

@@ -71,7 +71,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour
       sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(previous.revision)}`))
       .returning({ id: issueRecoveryActions.id });
     return Boolean(updated);
-  });
+  }, { companyId: source.companyId });
 }
 
 export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) {

server/src/services/task-retry.ts

@@ -334,7 +334,7 @@ export async function requestTaskRetry(input: {
       details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy) } });
     return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id,
       chatActionId: intent.chatActionId };
-  });
+  }, { companyId: input.companyId });
 }
 
 /** Chat owns admission and authorization. The task receipt follows that

tests/runner-e2e/CHAT-FEATURES.md

@@ -223,8 +223,12 @@ actors before any TestOps mutation. A later dispatch or replay cannot reuse that
 approved number. Missing or uncertain metadata stops; an uncertain request needs
 reconciliation rather than a new send. The nonce and launch search provide
 additional evidence. Normal TestOps **Run** uses `evaluate` and is independent of
-this one-time administrator fence. The operation parameter identifier and launch
-nonce are also protected configuration, not dispatch inputs.
+this one-time administrator fence. The launch nonce is protected configuration,
+not a dispatch input. `CHAT_TESTOPS_OPERATION_PARAMETER_ID` is required only when
+the registered job defines `operation`; the launcher sets that parameter to
+`evaluate`. An official job with only `Branch` keeps its exact
+`refs/heads/master` value and uses the fixed workflow's `evaluate` default.
+Unknown parameters and other branch values are rejected.
 
 `CHAT_TESTOPS_INTEGRATION_ID` names the reviewed existing project GitHub
 integration. The launcher verifies its project, type and enabled state, then binds

tests/runner-e2e/chat-feature-testops-api.mjs

@@ -17,7 +17,8 @@ const diagnosticCodes = new Set(['untrusted_chat_testops_context', 'testops_toke
   'testops_created_case_identity_invalid', 'testops_created_plan_identity_invalid', 'testops_plan_scope_mismatch',
   'testops_operation_invalid', 'testops_output_invalid', 'untrusted_public_testops_context', 'chat_testops_identity_missing',
   'chat_testops_case_mismatch', 'chat_testops_plan_mismatch', 'chat_testops_plan_job_mismatch', 'chat_testops_job_mismatch',
-  'chat_testops_jobrun_mismatch', 'chat_testops_launch_mismatch', 'chat_testops_selection_mismatch', 'chat_testops_parameter_mismatch', 'chat_testops_nonce_invalid', 'chat_testops_nonce_conflict', 'chat_testops_catalog_incomplete', 'chat_testops_launch_admission', 'chat_testops_github_http', 'chat_testops_integration_mismatch', 'chat_testops_upload_mismatch', 'chat_testops_upload_unverified']);
+  'chat_testops_job_id_mismatch', 'chat_testops_job_type_mismatch', 'chat_testops_job_parameters_shape_mismatch',
+  'chat_testops_job_parameters_limit', 'chat_testops_job_url_mismatch', 'chat_testops_jobrun_mismatch', 'chat_testops_launch_mismatch', 'chat_testops_selection_mismatch', 'chat_testops_parameter_mismatch', 'chat_testops_nonce_invalid', 'chat_testops_nonce_conflict', 'chat_testops_catalog_incomplete', 'chat_testops_launch_admission', 'chat_testops_github_http', 'chat_testops_integration_mismatch', 'chat_testops_upload_mismatch', 'chat_testops_upload_unverified']);
 export function failureProjection(error, stage) {
   const stages = ['context', 'output_admission', 'authentication', 'metadata_inspection', 'plan_bootstrap', 'selection_validation', 'launch_admission', 'launch_preflight', 'plan_launch', 'bootstrap_job_metadata', 'result_export', 'upload_verification', 'bootstrap_context', 'receipt'];
   return { stage: stages.includes(stage) ? stage : 'unknown',
@@ -146,9 +147,7 @@ export async function validateChatSelection(api, env) {
   const assignments = await api(`/api/testplan/${planId}/job`);
   if (!Array.isArray(assignments) || assignments.length !== 1 || assignments[0].job?.id !== jobId || assignments[0].testCasesCount !== 1) fail('chat_testops_plan_job_mismatch');
   const job = await api(`/api/job/${jobId}`);
-  const jobUrls = ['https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml'];
-  if (positive(env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID)) jobUrls.push(`https://github.com/paperclipai/paperclip/actions/workflows/${env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID}`);
-  if (job.id !== jobId || job.type !== 'github' || !jobUrls.includes(job.url)) fail('chat_testops_job_mismatch');
+  if (job.id !== jobId || job.type !== 'github' || !publicJobUrl(job.url, env)) fail('chat_testops_job_mismatch');
   const run = await api(`/api/jobrun/${jobRunId}`);
   if (run.id !== jobRunId || run.job?.id !== jobId || run.job?.type !== 'github'
     || !['scheduled', 'in_progress'].includes(run.stage) || !id(run.launchId)
@@ -162,12 +161,16 @@ export async function validateChatSelection(api, env) {
 }
 
 function configuredIds(env) {
-  const names = ['CHAT_TESTOPS_CASE_ID', 'CHAT_TESTOPS_PLAN_ID', 'CHAT_TESTOPS_JOB_ID', 'CHAT_TESTOPS_OPERATION_PARAMETER_ID', 'CHAT_TESTOPS_INTEGRATION_ID'];
+  const names = ['CHAT_TESTOPS_CASE_ID', 'CHAT_TESTOPS_PLAN_ID', 'CHAT_TESTOPS_JOB_ID', 'CHAT_TESTOPS_INTEGRATION_ID'];
   if (names.some(name => !positive(env[name]))) fail('chat_testops_identity_missing');
-  return Object.fromEntries(names.map((name, index) => [['caseId', 'planId', 'jobId', 'parameterId', 'integrationId'][index], Number(env[name])]));
+  const operation = env.CHAT_TESTOPS_OPERATION_PARAMETER_ID;
+  if (operation !== undefined && operation !== '' && !positive(operation)) fail('chat_testops_identity_missing');
+  return { ...Object.fromEntries(names.map((name, index) => [['caseId', 'planId', 'jobId', 'integrationId'][index], Number(env[name])])),
+    parameterId: positive(operation) ? Number(operation) : null };
 }
 function publicJobUrl(url, env) {
   return url === 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml'
+    || url === 'https://github.com/paperclipai/paperclip/actions?query=workflow%3A%22Chat+feature+TestOps%22'
     || positive(env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID) && url === `https://github.com/paperclipai/paperclip/actions/workflows/${env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID}`;
 }
 async function registeredPlan(api, env) {
@@ -182,10 +185,24 @@ async function registeredPlan(api, env) {
   if (job.id !== ids.jobId || job.type !== 'github' || !publicJobUrl(job.url, env)) fail('chat_testops_job_mismatch');
   const integration = await api(`/api/integration/${ids.integrationId}/project/1`);
   if (integration.id !== ids.integrationId || integration.projectId !== 1 || integration.info?.type !== 'github' || integration.disabled !== false) fail('chat_testops_integration_mismatch');
-  const definition = await api(`/api/environment/${ids.parameterId}`);
-  if (definition.id !== ids.parameterId || definition.name !== 'operation' || !Array.isArray(job.parameters) || job.parameters.length !== 1
-    || job.parameters[0].id !== ids.parameterId || !['bootstrap', 'evaluate'].includes(job.parameters[0].value)) fail('chat_testops_parameter_mismatch');
-  return { ...ids, selected, plan, job };
+  if (!Array.isArray(job.parameters) || job.parameters.length < 1 || job.parameters.length > 2) fail('chat_testops_parameter_mismatch');
+  const seenIds = new Set(), seenNames = new Set(), parameters = [];
+  for (const parameter of job.parameters) {
+    if (!id(parameter.id) || seenIds.has(parameter.id)) fail('chat_testops_parameter_mismatch');
+    seenIds.add(parameter.id);
+    const definition = await api(`/api/environment/${parameter.id}`);
+    if (definition.id !== parameter.id || !['Branch', 'operation'].includes(definition.name) || seenNames.has(definition.name)) fail('chat_testops_parameter_mismatch');
+    seenNames.add(definition.name);
+    if (definition.name === 'Branch') {
+      if (parameter.value !== 'refs/heads/master') fail('chat_testops_parameter_mismatch');
+      parameters.push({ id: parameter.id, value: parameter.value });
+    } else {
+      if (ids.parameterId !== parameter.id || !['bootstrap', 'evaluate'].includes(parameter.value)) fail('chat_testops_parameter_mismatch');
+      parameters.push({ id: parameter.id, value: 'evaluate' });
+    }
+  }
+  if (ids.parameterId !== null && !seenNames.has('operation')) fail('chat_testops_parameter_mismatch');
+  return { ...ids, selected, plan, job, parameters };
 }
 
 export async function bootstrapJobMetadata(api, env, expectedLaunchId) {
@@ -205,8 +222,11 @@ export async function bootstrapJobMetadata(api, env, expectedLaunchId) {
     if (launch.id !== expectedLaunchId || launch.projectId !== 1) fail('chat_testops_launch_mismatch');
   }
   const job = await api(`/api/job/${run.job.id}`);
-  if (job.id !== run.job.id || job.type !== 'github' || !Array.isArray(job.parameters) || job.parameters.length > 4) fail('chat_testops_job_mismatch');
-  if (strict && !publicJobUrl(job.url, env)) fail('chat_testops_job_mismatch');
+  if (job.id !== run.job.id) fail('chat_testops_job_id_mismatch');
+  if (job.type !== 'github') fail('chat_testops_job_type_mismatch');
+  if (!Array.isArray(job.parameters)) fail('chat_testops_job_parameters_shape_mismatch');
+  if (job.parameters.length > 4) fail('chat_testops_job_parameters_limit');
+  if (strict && !publicJobUrl(job.url, env)) fail('chat_testops_job_url_mismatch');
   const parameters = [];
   const seen = new Set();
   for (const parameter of job.parameters) {
@@ -214,9 +234,10 @@ export async function bootstrapJobMetadata(api, env, expectedLaunchId) {
     seen.add(parameter.id);
     const definition = await api(`/api/environment/${parameter.id}`);
     if (strict && definition.id !== parameter.id) fail('chat_testops_parameter_mismatch');
-    const known = ['operation', 'ALLURE_JOB_RUN_ID', 'ALLURE_USERNAME'].includes(definition.name);
+    const known = ['operation', 'ALLURE_JOB_RUN_ID', 'ALLURE_USERNAME', 'Branch'].includes(definition.name);
     parameters.push({ id: parameter.id, name: known ? definition.name : 'UNKNOWN',
-      value: definition.name === 'operation' && ['bootstrap', 'evaluate', 'launch-plan'].includes(parameter.value) ? parameter.value : 'NOT_PROJECTED' });
+      value: definition.name === 'operation' && ['bootstrap', 'evaluate', 'launch-plan'].includes(parameter.value)
+        || definition.name === 'Branch' && parameter.value === 'refs/heads/master' ? parameter.value : 'NOT_PROJECTED' });
   }
   return { ...(strict ? { schema: 'paperclip.chat-feature-testops-bootstrap-registration/v1', ...context, capturedAfterWatch: true, runnableBindingQualified: false } : {}),
     status: 'observed', projectId: 1, jobId: job.id, allureJobRunId: env.ALLURE_JOB_RUN_ID,
@@ -274,7 +295,7 @@ export async function launchChatPlan(api, env, markOnce, admission) {
       jobId: known.jobId, caseId: known.caseId, launchId: existing[0].id, status: 'EXISTING_NO_RESEND', actualResult: null };
   }
   await markOnce({ ...source, launchAdmission: admission, nonce, planId: known.planId, jobId: known.jobId, caseId: known.caseId });
-  const parameters = [{ id: known.parameterId, value: 'evaluate' }];
+  const parameters = known.parameters;
   await api(`/api/job/${known.jobId}`, 'PATCH', { integrationId: known.integrationId, canRun: true, parameters });
   const patched = await api(`/api/job/${known.jobId}`);
   if (patched.id !== known.jobId || patched.type !== 'github' || !publicJobUrl(patched.url, env) || !same(patched.parameters, parameters)) fail('chat_testops_parameter_mismatch');

tests/runner-e2e/chat-feature-testops-api.test.mjs

@@ -168,21 +168,24 @@ async function fence(env = launchEnv, changes = {}, workflowChanges = {}, repoCh
     : url.includes('/actions/runs/') ? { ...ownRun, ...changes }
     : { id: 99, path: '.github/workflows/chat-feature-testops.yml', state: 'active', ...workflowChanges }) }));
 }
-function launchServer({ wrongParameter = false, wrongAssignment = false, uncertainRun = false, integrationChange = {} } = {}) {
-  const calls = [], parameters = [{ id: 17, value: 'bootstrap' }]; let assigned = false, launched = false, requests = 0;
+function launchServer({ wrongParameter = false, wrongAssignment = false, uncertainRun = false, integrationChange = {}, initialParameters = [{ id: 17, value: 'bootstrap' }], expectedParameters = [{ id: 17, value: 'evaluate' }], definitions = {}, jobUrl = 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml' } = {}) {
+  const calls = [], parameters = structuredClone(initialParameters); let assigned = false, launched = false, requests = 0;
   const api = async (route, method, body) => {
     calls.push({ route, method, body });
     if (route.startsWith('/api/launch?')) return { totalPages: 1, content: launched ? [{ id: 10, name: `Chat feature evals · ${'1'.repeat(32)}` }] : [] };
     if (route === '/api/integration/2/project/1') return { id: 2, projectId: 1, info: { type: 'github' }, disabled: false, ...integrationChange };
-    if (route === '/api/environment/17') return { id: 17, name: wrongParameter ? 'OTHER_PRIVATE_PARAMETER' : 'operation' };
+    if (route.startsWith('/api/environment/')) {
+      const id = Number(route.split('/').at(-1));
+      return definitions[id] ?? { id, name: id === 17 ? wrongParameter ? 'OTHER_PRIVATE_PARAMETER' : 'operation' : 'Branch' };
+    }
     if (route === '/api/job/7') {
-      if (method === 'PATCH') { assert.deepEqual(body, { integrationId: 2, canRun: true, parameters: [{ id: 17, value: 'evaluate' }] }); parameters[0].value = 'evaluate'; }
-      return { id: 7, type: 'github', url: 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml', parameters };
+      if (method === 'PATCH') { assert.deepEqual(body, { integrationId: 2, canRun: true, parameters: expectedParameters }); parameters.splice(0, parameters.length, ...structuredClone(body.parameters)); }
+      return { id: 7, type: 'github', url: jobUrl, parameters };
     }
     if (route === '/api/testplan/8/job') return wrongAssignment ? [{ job: { id: 70 }, testCasesCount: 1 }]
       : assigned ? [{ job: { id: 7, parameters }, testCasesCount: 1 }] : [];
     if (route === '/api/testplan/8/assign') { assert.equal(method, 'POST'); assert.deepEqual(body, { jobId: 7, selection: { inverted: false, manual: false, leafsInclude: [91] } }); assigned = true; return null; }
-    if (route === '/api/testplan/8/jobparameter') { assert.deepEqual(body, { jobId: 7, parameters: [{ id: 17, value: 'evaluate' }] }); return null; }
+    if (route === '/api/testplan/8/jobparameter') { assert.deepEqual(body, { jobId: 7, parameters: expectedParameters }); return null; }
     if (route === '/api/testplan/8/run') { requests++; assert.equal(method, 'POST'); if (uncertainRun) throw new Error('synthetic uncertain transport'); launched = true; return { id: 10, projectId: 1, name: body.launchName }; }
     return selectedApi()(route);
   };
@@ -432,3 +435,106 @@ test('post-watch reader caps parameter definitions and refuses duplicate IDs', a
     await assert.rejects(bootstrapJobMetadata(strictBootstrapApi({ job: { parameters } }), publicEnv, 10));
   }
 });
+
+
+const observedLegacyJobUrl = 'https://github.com/paperclipai/paperclip/actions?query=workflow%3A%22Chat+feature+TestOps%22';
+
+test('official observed legacy and existing workflow URLs pass all three actual job consumers', async () => {
+  for (const url of [observedLegacyJobUrl, 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml',
+    'https://github.com/paperclipai/paperclip/actions/workflows/99']) {
+    const env = { ...publicEnv, CHAT_TESTOPS_GITHUB_WORKFLOW_ID: '99' };
+    const selected = await validateChatSelection(selectedApi({ '/api/job/7': { id: 7, type: 'github', url } }), env);
+    assert.equal(selected.caseId, 91);
+    const registered = await bootstrapJobMetadata(strictBootstrapApi({ job: { url } }), env, 10);
+    assert.equal(registered.publicWorkflowMatches, true);
+    const server = launchServer({ jobUrl: url });
+    const launched = await launchChatPlan(server.api, launchEnv, async () => {}, await fence());
+    assert.equal(launched.status, 'REQUESTED');
+    assert.equal(server.requests(), 1);
+  }
+});
+
+test('official legacy URL admission rejects foreign workflow repo query credentials and fragment in all consumers', async () => {
+  for (const url of [observedLegacyJobUrl.replace('/paperclip/actions', '/paperclip-cloud/actions'),
+    observedLegacyJobUrl.replace('Chat+feature+TestOps', 'Other+workflow'), `${observedLegacyJobUrl}&extra=1`,
+    observedLegacyJobUrl.replace('https://github.com', 'https://user:password@github.com'), `${observedLegacyJobUrl}#extra`]) {
+    await assert.rejects(validateChatSelection(selectedApi({ '/api/job/7': { id: 7, type: 'github', url } }), publicEnv), /chat_testops_job_mismatch/);
+    await assert.rejects(bootstrapJobMetadata(strictBootstrapApi({ job: { url } }), publicEnv, 10), error => {
+      assert.deepEqual(failureProjection(error, 'bootstrap_context'), { stage: 'bootstrap_context', code: 'chat_testops_job_url_mismatch', httpStatus: null });
+      return true;
+    });
+    const server = launchServer({ jobUrl: url });
+    await assert.rejects(launchChatPlan(server.api, launchEnv, async () => assert.fail('no mutation marker'), await fence()), /chat_testops_job_mismatch/);
+    assert.equal(server.calls.some(call => call.method), false);
+  }
+});
+
+test('bootstrap job predicate failures retain only distinct fixed codes and no observed values', async () => {
+  for (const [job, code] of [
+    [{ id: 99 }, 'chat_testops_job_id_mismatch'],
+    [{ type: 'SYNTHETIC_PRIVATE_TYPE' }, 'chat_testops_job_type_mismatch'],
+    [{ parameters: undefined }, 'chat_testops_job_parameters_shape_mismatch'],
+    [{ parameters: null }, 'chat_testops_job_parameters_shape_mismatch'],
+    [{ parameters: 'SYNTHETIC_PRIVATE_PARAMETERS' }, 'chat_testops_job_parameters_shape_mismatch'],
+    [{ parameters: Array.from({ length: 5 }, (_, i) => ({ id: i + 1 })) }, 'chat_testops_job_parameters_limit'],
+    [{ url: 'https://SYNTHETIC_PRIVATE_URL.invalid/' }, 'chat_testops_job_url_mismatch'],
+  ]) {
+    await assert.rejects(bootstrapJobMetadata(strictBootstrapApi({ job }), publicEnv, 10), error => {
+      const projected = failureProjection(error, 'bootstrap_context');
+      assert.deepEqual(projected, { stage: 'bootstrap_context', code, httpStatus: null });
+      assert.ok(!JSON.stringify(projected).includes('SYNTHETIC_PRIVATE'));
+      return true;
+    });
+  }
+});
+
+
+test('launch preserves exact Branch and defaults evaluate only when operation is genuinely absent', async () => {
+  for (const [parameters, opId, expectedParameters] of [
+    [[{ id: 19, value: 'refs/heads/master' }], undefined, [{ id: 19, value: 'refs/heads/master' }]],
+    [[{ id: 19, value: 'refs/heads/master' }], '', [{ id: 19, value: 'refs/heads/master' }]],
+    [[{ id: 17, value: 'bootstrap' }], '17', [{ id: 17, value: 'evaluate' }]],
+    [[{ id: 19, value: 'refs/heads/master' }, { id: 17, value: 'bootstrap' }], '17', [{ id: 19, value: 'refs/heads/master' }, { id: 17, value: 'evaluate' }]],
+  ]) {
+    const env = { ...launchEnv, CHAT_TESTOPS_OPERATION_PARAMETER_ID: opId };
+    const server = launchServer({ initialParameters: parameters, expectedParameters });
+    let marked = 0;
+    const result = await launchChatPlan(server.api, env, async () => { marked++; assert.equal(server.calls.some(call => call.method), false); }, await fence(env));
+    assert.equal(result.status, 'REQUESTED'); assert.equal(server.requests(), 1); assert.equal(marked, 1);
+    assert.deepEqual(server.calls.filter(call => call.route === '/api/job/7' && call.method === 'PATCH')[0].body.parameters, expectedParameters);
+    assert.deepEqual(server.calls.filter(call => call.route === '/api/testplan/8/jobparameter')[0].body.parameters, expectedParameters);
+  }
+});
+
+test('launch refuses empty unknown duplicate wrong Branch or conflicting operation before any mutation', async () => {
+  const branch = { id: 19, value: 'refs/heads/master' };
+  for (const [parameters, opId, definitions] of [
+    [[], undefined, {}],
+    [[{ id: 19, value: 'refs/heads/other' }], undefined, {}],
+    [[branch, branch], undefined, {}],
+    [[branch, { id: 20, value: 'refs/heads/master' }], undefined, {}],
+    [[{ id: 19, value: 'refs/heads/master' }], undefined, { 19: { id: 99, name: 'Branch' } }],
+    [[{ id: 19, value: 'PRIVATE_PARAMETER_VALUE' }], undefined, { 19: { id: 19, name: 'PRIVATE_PARAMETER_NAME' } }],
+    [[branch], '17', {}],
+    [[branch], 'bad-id', {}],
+    [[{ id: 17, value: 'bootstrap' }], undefined, {}],
+    [[{ id: 17, value: 'bootstrap' }], '18', {}],
+    [[{ id: 17, value: 'launch-plan' }], '17', {}],
+    [[branch, { id: 17, value: 'bootstrap' }, { id: 20, value: 'refs/heads/master' }], '17', {}],
+  ]) {
+    const env = { ...launchEnv, CHAT_TESTOPS_OPERATION_PARAMETER_ID: opId };
+    const server = launchServer({ initialParameters: parameters, definitions });
+    await assert.rejects(launchChatPlan(server.api, env, async () => assert.fail('no mutation marker'), await fence(env)), /chat_testops_(?:identity_missing|parameter_mismatch)/);
+    assert.equal(server.calls.some(call => call.method), false); assert.equal(server.requests(), 0);
+  }
+});
+
+test('bootstrap projects only fixed Branch name and exact master value alongside genuine numeric ID', async () => {
+  const base = strictBootstrapApi({ job: { parameters: [{ id: 19, value: 'refs/heads/master' }] } });
+  const result = await bootstrapJobMetadata(route => route === '/api/environment/19' ? { id: 19, name: 'Branch' } : base(route), publicEnv, 10);
+  assert.deepEqual(result.parameters, [{ id: 19, name: 'Branch', value: 'refs/heads/master' }]);
+  const hidden = strictBootstrapApi({ job: { parameters: [{ id: 19, value: 'PRIVATE_BRANCH_VALUE' }] } });
+  const unknown = await bootstrapJobMetadata(route => route === '/api/environment/19' ? { id: 19, name: 'Branch' } : hidden(route), publicEnv, 10);
+  assert.deepEqual(unknown.parameters, [{ id: 19, name: 'Branch', value: 'NOT_PROJECTED' }]);
+  assert.ok(!JSON.stringify(unknown).includes('PRIVATE_BRANCH'));
+});