Merge commit '554341a79a6779e194eec2243f2e4a05b2061d31' into codex/sleep-09-native-lock-followup

* commit '554341a79a6779e194eec2243f2e4a05b2061d31':
  Preserve native receipt grouping within each bounded worker pass
  fix(github): lock company before binding handoff children

server/src/__tests__/deferred-execution-receipt-work.test.ts

@@ -1,3 +1,4 @@
+import { createServer } from "node:http";
 import { conflict, HttpError } from "../errors.js";
 import { setImmediate as immediate } from "node:timers/promises";
 import { setTimeout as realTimeout, clearTimeout as clearRealTimeout } from "node:timers";
@@ -55,6 +56,7 @@ describe("deferred receipt durable dispatch", () => {
   const errorTimes: number[] = [];
   const statements: string[] = [];
   const passes = new Set<Promise<unknown>>();
+  const extraStops: Array<() => Promise<void>> = [];
   const lifecycleWorkers = new Set<ReturnType<typeof configureAgentLifecycle>>();
   beforeAll(async () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL");
@@ -80,6 +82,7 @@ describe("deferred receipt durable dispatch", () => {
   });
   afterEach(async () => {
     await coordinator.stop(); await heartbeat.drainActiveRunExecutions();
+    for (const stop of extraStops.splice(0)) await stop();
     await Promise.all([...lifecycleWorkers].map(worker => worker.stop())); lifecycleWorkers.clear();
     vi.useRealTimers(); vi.restoreAllMocks(); expect(errors).toEqual([]);
   });
@@ -333,7 +336,7 @@ describe("deferred receipt durable dispatch", () => {
     const resume = vi.fn(async (scope: { onError?: (error: unknown) => void }) => {
       if (fail) { failedAt = Date.now(); scope.onError?.(failure); }
     });
-    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {},
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, createNativeBatch: () => async () => {},
       schedulingSuppressed: async () => false }, () => !drain).ready;
     expect(errors).toHaveLength(1);
     expect((errors[0] as AggregateError).errors[0].errors).toEqual([failure]);
@@ -345,6 +348,154 @@ describe("deferred receipt durable dispatch", () => {
     await quiet();
   });
 
+  async function nativeCohort(f: Awaited<ReturnType<typeof fixture>>, count = 2, completeAfter = 1) {
+    const failures: unknown[] = [];
+    let delivered = 0;
+    const listener = createServer(async (request, response) => {
+      try {
+        let body = "";
+        for await (const chunk of request) body += chunk;
+        const { runId } = JSON.parse(body) as { runId: string };
+        const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId),
+          eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, f.agent.id)));
+        const issueId = run?.contextSnapshot?.issueId;
+        const [target] = typeof issueId === "string" ? await db.select().from(issues).where(and(eq(issues.id, issueId),
+          eq(issues.companyId, f.companyId), eq(issues.assigneeAgentId, f.agent.id))) : [];
+        if (!run || !target) throw new Error("Unknown native cohort run");
+        delivered++;
+        if (delivered >= completeAfter) await issueService(db).update(target.id, { status: "done" });
+        else await issueService(db).addComment(target.id, "This native batch was handled.", { agentId: f.agent.id, runId: run.id });
+        response.end("ok");
+      } catch (error) { failures.push(error); response.statusCode = 500; response.end("completion failed"); }
+    });
+    await new Promise<void>(resolve => listener.listen(0, "127.0.0.1", resolve));
+    extraStops.push(async () => { listener.closeAllConnections(); await new Promise<void>(resolve => listener.close(() => resolve())); });
+    const port = (listener.address() as { port: number }).port;
+    const script = `fetch("http://127.0.0.1:${port}/complete", {method:"POST",body:JSON.stringify({runId:process.env.PAPERCLIP_RUN_ID})}).then(async response=>{if(!response.ok)throw new Error(await response.text());console.log("receipt-proof")}).catch(error=>{console.error(error);process.exitCode=1})`;
+    await db.update(agents).set({ adapterConfig: { command: process.execPath, args: ["-e", script], cwd: process.cwd() } })
+      .where(eq(agents.id, f.agent.id));
+    const at = Date.now();
+    const intents = await db.insert(agentWakeupRequests).values(Array.from({ length: count }, (_, index) => ({
+      reason: index % 2 ? "issue_blockers_resolved" : "issue_children_completed",
+      companyId: f.companyId, agentId: f.agent.id, source: "automation", triggerDetail: "system",
+      status: "queued", requestedByActorType: "system", requestedByActorId: "native-status-committer",
+      requestedAt: new Date(at - (count - index) * 1000), idempotencyKey: `native-cohort:${index}:${f.issue.id}`,
+      payload: { issueId: f.issue.id, taskId: f.issue.id,
+        _paperclipWakeContext: { issueId: f.issue.id, taskId: f.issue.id, source: "native_status_decision" } },
+    }))).returning();
+    return { intents, failures };
+  }
+  async function nativeCohortResult(f: Awaited<ReturnType<typeof fixture>>, cohort: Awaited<ReturnType<typeof nativeCohort>>,
+    owners = [heartbeat]) {
+    await waitForReal(async () => {
+      const rows = await db.select().from(agentWakeupRequests).where(sql`${agentWakeupRequests.id} in ${cohort.intents.map(row => row.id)}`);
+      expect(rows.map(row => row.status)).toEqual(cohort.intents.map(() => "coalesced"));
+    }, { timeout: 15_000 });
+    const dispatched = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId),
+      sql`${agentWakeupRequests.requestedByActorId} like 'native-status-wake-dispatch:%'`,
+      sql`${agentWakeupRequests.payload}->>'issueId' = ${f.issue.id}`));
+    expect(dispatched.map(row => ({ id: row.id, runId: row.runId, status: row.status }))).toHaveLength(1);
+    const [first, second] = await db.select().from(agentWakeupRequests)
+      .where(sql`${agentWakeupRequests.id} in ${cohort.intents.map(row => row.id)}`);
+    expect(first.runId).toBeTruthy(); expect(second.runId).toBe(first.runId);
+    expect((await db.select().from(agentWakeupRequests).where(sql`${agentWakeupRequests.id} in ${cohort.intents.map(row => row.id)}`))
+      .every(row => row.runId === first.runId)).toBe(true);
+    await waitForReal(async () => {
+      await Promise.all(owners.map(owner => owner.drainActiveRunExecutions()));
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, first.runId!));
+      expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
+      expect(run.stdoutExcerpt).toContain("receipt-proof");
+    }, { timeout: 15_000 });
+    expect(cohort.failures).toEqual([]);
+    return first.runId!;
+  }
+  it("coalesces two due native status intents through the registered receipt worker", async () => {
+    const f = await fixture(), cohort = await nativeCohort(f);
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await nativeCohortResult(f, cohort); await quiet();
+  });
+  it("preserves one dispatch for the canonical hundred-intent native batch", async () => {
+    const f = await fixture(), cohort = await nativeCohort(f, 100);
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await nativeCohortResult(f, cohort); await quiet();
+  });
+  it("keeps other issues and companies separate within one native worker batch", async () => {
+    const f = await fixture(), cohort = await nativeCohort(f), foreign = await fixture(), foreignCohort = await nativeCohort(foreign);
+    const [otherIssue] = await db.insert(issues).values({ companyId: f.companyId, assigneeAgentId: f.agent.id,
+      status: "todo", title: "Separate native issue" }).returning();
+    const otherIntents = await db.insert(agentWakeupRequests).values(cohort.intents.map((_, index) => ({
+      companyId: f.companyId, agentId: f.agent.id, source: "automation", triggerDetail: "system",
+      reason: index ? "issue_blockers_resolved" : "issue_children_completed", status: "queued",
+      requestedByActorType: "system", requestedByActorId: "native-status-committer",
+      idempotencyKey: `other-native-issue:${index}:${otherIssue.id}`,
+      payload: { issueId: otherIssue.id, taskId: otherIssue.id,
+        _paperclipWakeContext: { issueId: otherIssue.id, taskId: otherIssue.id, source: "native_status_decision" } },
+    }))).returning();
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    const first = await nativeCohortResult(f, cohort);
+    const other = await nativeCohortResult({ ...f, issue: otherIssue }, { intents: otherIntents, failures: cohort.failures });
+    const third = await nativeCohortResult(foreign, foreignCohort);
+    expect(new Set([first, other, third]).size).toBe(3); await quiet();
+  });
+  it("dispatches a notice committed during held admission in a fresh native pass even with old requestedAt", async () => {
+    const f = await fixture(), cohort = await nativeCohort(f, 2, 2);
+    let allowLate!: () => void, began!: () => void, releaseIssue!: () => void, locked!: () => void;
+    const lateGate = new Promise<void>(resolve => { allowLate = resolve; });
+    const lateBegan = new Promise<void>(resolve => { began = resolve; });
+    const issueGate = new Promise<void>(resolve => { releaseIssue = resolve; });
+    const issueLocked = new Promise<void>(resolve => { locked = resolve; });
+    let late: typeof agentWakeupRequests.$inferSelect | undefined;
+    const lateRequestedAt = new Date(Date.now() - 60_000);
+    const lateTransaction = db.transaction(async tx => {
+      await tx.execute(sql`select 1`); began(); await lateGate;
+      const id = randomUUID();
+      await notifyNativeStatusWakeWork(tx, { companyId: f.companyId, requestId: id });
+      [late] = await tx.insert(agentWakeupRequests).values({ id, companyId: f.companyId, agentId: f.agent.id,
+        source: "automation", triggerDetail: "system", reason: "issue_children_completed", status: "queued",
+        requestedByActorType: "system", requestedByActorId: "native-status-committer", requestedAt: lateRequestedAt,
+        idempotencyKey: `late-native:${id}`,
+        payload: { issueId: f.issue.id, taskId: f.issue.id,
+          _paperclipWakeContext: { issueId: f.issue.id, taskId: f.issue.id, source: "native_status_decision" } },
+      }).returning();
+    });
+    await lateBegan;
+    const issueLock = db.transaction(async tx => {
+      await tx.select().from(issues).where(eq(issues.id, f.issue.id)).for("update"); locked(); await issueGate;
+    });
+    await issueLocked;
+    const registration = heartbeat.registerDeferredReceiptWork(coordinator, () => !drain);
+    try {
+      await waitForReal(async () => expect((await db.select().from(agentWakeupRequests)
+        .where(sql`${agentWakeupRequests.id} in ${cohort.intents.map(row => row.id)}`)).some(row => row.status === "claimed")).toBe(true), { timeout: 5000 });
+      const [claimed] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId),
+        eq(agentWakeupRequests.status, "claimed")));
+      allowLate(); await lateTransaction;
+      expect(late!.requestedAt.getTime()).toBeLessThanOrEqual(claimed.claimedAt!.getTime());
+    } finally { allowLate(); releaseIssue(); await Promise.all([lateTransaction, issueLock]); }
+    await registration.ready;
+    let lateRunId: string | null = null;
+    await waitForReal(async () => {
+      const [saved] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId),
+        eq(agentWakeupRequests.requestedByActorType, "system"),
+        eq(agentWakeupRequests.requestedByActorId, `native-status-wake-dispatch:${late!.id}`)));
+      expect(saved?.runId, JSON.stringify(saved)).toBeTruthy(); lateRunId = saved.runId;
+    }, { timeout: 45_000 });
+    const originals = await db.select().from(agentWakeupRequests).where(sql`${agentWakeupRequests.id} in ${cohort.intents.map(row => row.id)}`);
+    expect(originals.every(row => row.runId === originals[0].runId)).toBe(true);
+    expect(lateRunId).not.toBe(originals[0].runId);
+    await waitForReal(async () => {
+      await heartbeat.drainActiveRunExecutions();
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, lateRunId!));
+      expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
+      expect(run.stdoutExcerpt).toContain("receipt-proof");
+      expect(run.contextSnapshot?.nativeStatusWakeIntentId).toBe(late!.id);
+    }, { timeout: 15_000 });
+    expect(cohort.failures).toEqual([]);
+    expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, late!.id)))[0].status).toBe("coalesced");
+    expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId),
+      sql`${agentWakeupRequests.requestedByActorId} like 'native-status-wake-dispatch:%'`))).toHaveLength(2);
+    await quiet();
+  });
   it.each(["native claim", "execution wait"])("reconstructs an exact fresh %s deadline without probing early", async kind => {
     const f = await fixture(), updatedAt = new Date(), native = kind === "native claim";
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
@@ -354,7 +505,7 @@ describe("deferred receipt durable dispatch", () => {
     const check = vi.fn(async (scope: { companyId?: string; requestId?: string }) => {
       await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!));
     });
-    await registerDeferredReceiptWork(db, coordinator, { resume: check, settleTaskChat: vi.fn(), dispatchNative: check,
+    await registerDeferredReceiptWork(db, coordinator, { resume: check, settleTaskChat: vi.fn(), createNativeBatch: () => check,
       schedulingSuppressed: async () => false }, () => !drain).ready;
     const due = updatedAt.getTime() + (native ? 60_000 : 30_000);
     expect(coordinator.nextWakeAt()).toBe(due); expect(check).not.toHaveBeenCalled();
@@ -366,7 +517,7 @@ describe("deferred receipt durable dispatch", () => {
   it("reconciles the first drained producer and waits for resume without domain SQL", async () => {
     const f = await fixture(); drain = true;
     const resume = vi.fn(async () => {}), dispatchNative = vi.fn(async () => {});
-    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), dispatchNative,
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => dispatchNative,
       schedulingSuppressed: async () => false }, () => !drain).ready;
     await db.transaction(async tx => {
       await notifyDeferredExecutionWork(tx, { companyId: f.companyId });
@@ -387,7 +538,7 @@ describe("deferred receipt durable dispatch", () => {
   ])("activates only the committed company's $name receipt after an empty start", async ({ payload }) => {
     const f = await fixture(), foreign = await fixture();
     const resume = vi.fn(async () => {});
-    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), dispatchNative: vi.fn(),
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => vi.fn(),
       schedulingSuppressed: async () => false }, () => !drain).ready;
     const [other] = await db.insert(agentWakeupRequests).values({ companyId: foreign.companyId, agentId: foreign.agent.id,
       source: "on_demand", status: "deferred_issue_execution", payload }).returning();
@@ -412,7 +563,7 @@ describe("deferred receipt durable dispatch", () => {
     await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: { queuedCommentInterruptQueueId: wake.id } })
       .where(eq(heartbeatRuns.id, f.source.id));
     const resume = vi.fn(async () => {});
-    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), dispatchNative: vi.fn(),
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => vi.fn(),
       schedulingSuppressed: async () => false }, () => !drain).ready;
     expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id }));
     await quiet();
@@ -560,7 +711,7 @@ describe("deferred receipt durable dispatch", () => {
       source: "on_demand", status: "deferred_issue_execution", payload, updatedAt: initialAt }).returning();
     const deps = createHeartbeatQueueTestDependencies(db), queue = createHeartbeatQueue(db, deps);
     await registerDeferredReceiptWork(db, coordinator, { resume: scope => queue.resumeDeferredExecutionReceipts(scope),
-      settleTaskChat: async () => {}, dispatchNative: async () => {}, schedulingSuppressed: async () => false }, () => !drain).ready;
+      settleTaskChat: async () => {}, createNativeBatch: () => async () => {}, schedulingSuppressed: async () => false }, () => !drain).ready;
     await vi.advanceTimersByTimeAsync(initialAt.getTime() + 10_000 - Date.now());
     const movedAt = new Date();
     const writer = createQueuedCommentIssueLockWriter(db, { syncCommentReferences: async () => {},
@@ -615,7 +766,7 @@ describe("deferred receipt durable dispatch", () => {
       requestedByActorType: "system", requestedByActorId: "native-status-committer" }).returning();
     const dispatchNative = vi.fn(async () => {});
     await registerDeferredReceiptWork(db, coordinator, { resume: async () => {}, settleTaskChat: async () => {},
-      dispatchNative, schedulingSuppressed: async () => false }, () => !drain).ready;
+      createNativeBatch: () => dispatchNative, schedulingSuppressed: async () => false }, () => !drain).ready;
     await vi.advanceTimersByTimeAsync(initialAt.getTime() + 10_000 - Date.now());
     const movedAt = new Date();
     await db.update(agentWakeupRequests).set({ updatedAt: movedAt }).where(eq(agentWakeupRequests.id, wake.id));
@@ -630,7 +781,7 @@ describe("deferred receipt durable dispatch", () => {
     const resume = vi.fn(async (scope: { companyId?: string; requestId?: string }) => {
       await writeDeferredExecutionWait(db, { companyId: scope.companyId!, requestId: scope.requestId! }, { reason });
     });
-    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {},
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, createNativeBatch: () => async () => {},
       schedulingSuppressed: async () => false }, () => !drain).ready;
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
       source: "on_demand", status: "deferred_issue_execution", reason: "task_retry" }).returning();
@@ -720,7 +871,7 @@ describe("deferred receipt durable dispatch", () => {
   });
   it("outer rollback and nested savepoint rollback do not dispatch receipt intent", async () => {
     const f = await fixture(); const dispatchNative = vi.fn(async () => {});
-    await registerDeferredReceiptWork(db, coordinator, { resume: vi.fn(), settleTaskChat: vi.fn(), dispatchNative,
+    await registerDeferredReceiptWork(db, coordinator, { resume: vi.fn(), settleTaskChat: vi.fn(), createNativeBatch: () => dispatchNative,
       schedulingSuppressed: async () => false }, () => !drain).ready;
     await expect(db.transaction(async tx => {
       await notifyNativeStatusWakeWork(tx, { companyId: f.companyId });

server/src/__tests__/github-connection-work.test.ts

@@ -1,4 +1,5 @@
 import { notifyChatEndpointConfiguration } from "../services/chat-channels/configuration-work-notifications.js";
+import { deleteCompany } from "../services/company-deletion.js";
 import { toolAccessService } from "../services/tool-access.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { randomUUID, generateKeyPairSync, verify, createHash, diffieHellman, hkdfSync, randomBytes, createCipheriv } from "node:crypto";
@@ -376,6 +377,146 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
     } finally { heldBindingCompany = undefined; release(); await outcome; bindingGate = undefined; bindingArrived = undefined; bindingClosed = undefined; }
   });
 
+  it("serializes binding handoff behind canonical company deletion without a child lock cycle", async () => {
+    const f = await fixture([]);
+    const [pending] = await db.update(connectionGrants).set({ providerTenant: {
+      ...f.grant.providerTenant, github: { ...f.grant.providerTenant!.github!,
+        bindingSync: { id: randomUUID(), removeInstallationIds: ["101"] } },
+    } }).where(eq(connectionGrants.id, f.grant.id)).returning();
+    let release!: () => void, entered!: (pid: number) => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const companyLocked = new Promise<number>(resolve => { entered = resolve; });
+    const transaction = db.transaction.bind(db);
+    // Pause after the canonical deletion's actual company lock, before its child writes.
+    const spy = vi.spyOn(db, "transaction").mockImplementation(((callback: (tx: unknown) => Promise<unknown>) =>
+      transaction(async tx => {
+        const select = tx.select.bind(tx);
+        tx.select = ((...args: Parameters<typeof select>) => {
+          const query = select(...args), from = query.from.bind(query);
+          query.from = ((table: unknown) => {
+            const builder = from(table as typeof companies);
+            if (table === companies) {
+              const lock = builder.for.bind(builder);
+              builder.for = ((strength: Parameters<typeof lock>[0]) => {
+                const locked = lock(strength);
+                if (strength === "no key update") {
+                  const then = locked.then.bind(locked);
+                  locked.then = ((resolve: (rows: unknown) => unknown, reject: (error: unknown) => unknown) => then(async rows => {
+                    const [owner] = await tx.execute(sql`select pg_backend_pid() as pid`);
+                    entered(Number(owner!.pid)); await gate; return resolve(rows);
+                  }, reject)) as typeof locked.then;
+                }
+                return locked;
+              }) as typeof builder.for;
+            }
+            return builder;
+          }) as typeof query.from;
+          return query;
+        }) as typeof tx.select;
+        return callback(tx);
+      })) as typeof db.transaction);
+    const deletion = deleteCompany(db, f.company.id);
+    const deleted = Promise.allSettled([deletion]);
+    let handoff: Promise<unknown> | undefined;
+    try {
+      const deletionPid = await companyLocked;
+      handoff = f.service.syncGitHubGrantBindings(f.connection, pending!);
+      const completed = Promise.allSettled([deletion, handoff]);
+      await vi.waitFor(async () => {
+        const waiting = await db.execute(sql`select pid from pg_stat_activity
+          where datname = current_database() and query like '%"companies"%' and query like '%for share%'
+            and ${deletionPid} = any(pg_blocking_pids(pid))`);
+        expect(waiting).toHaveLength(1);
+      }, { timeout: 5_000, interval: 20 });
+      release();
+      const results = await completed;
+      expect(results.map(row => row.status), JSON.stringify(results)).toEqual(["fulfilled", "fulfilled"]);
+      expect(await db.select().from(companies).where(eq(companies.id, f.company.id))).toHaveLength(0);
+      expect(await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id))).toHaveLength(0);
+      expect(bindings).toHaveLength(0);
+    } finally { release(); await deleted; await handoff?.catch(() => {}); spy.mockRestore(); }
+  });
+
+  it("lets an admitted binding finish before canonical company deletion", async () => {
+    const f = await fixture([]);
+    const [pending] = await db.update(connectionGrants).set({ providerTenant: {
+      ...f.grant.providerTenant, github: { ...f.grant.providerTenant!.github!,
+        bindingSync: { id: randomUUID(), removeInstallationIds: ["101"] } },
+    } }).where(eq(connectionGrants.id, f.grant.id)).returning();
+    heldBindingCompany = f.company.id;
+    let release!: () => void;
+    bindingGate = new Promise<void>(resolve => { release = resolve; });
+    const arrived = new Promise<void>(resolve => { bindingArrived = resolve; });
+    const handoff = f.service.syncGitHubGrantBindings(f.connection, pending!);
+    const handedOff = Promise.allSettled([handoff]);
+    let deletion: Promise<unknown> | undefined;
+    try {
+      await arrived;
+      deletion = deleteCompany(db, f.company.id);
+      const deleted = Promise.allSettled([deletion]);
+      await vi.waitFor(async () => {
+        const waiting = await db.execute(sql`select pid from pg_stat_activity
+          where datname = current_database() and query like '%"companies"%' and query like '%for no key update%'
+            and cardinality(pg_blocking_pids(pid)) > 0`);
+        expect(waiting).toHaveLength(1);
+      }, { timeout: 5_000, interval: 20 });
+      release();
+      expect((await handedOff)[0]!.status).toBe("fulfilled");
+      expect((await deleted)[0]!.status).toBe("fulfilled");
+      expect(bindings).toEqual([expect.objectContaining({ grantId: f.grant.id, connectionId: f.connection.id, installationId: "101", active: false })]);
+      expect(await db.select().from(companies).where(eq(companies.id, f.company.id))).toHaveLength(0);
+      expect(await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id))).toHaveLength(0);
+    } finally { release(); await handedOff; await deletion?.catch(() => {}); heldBindingCompany = undefined;
+      bindingGate = undefined; bindingArrived = undefined; }
+  });
+
+  it.each(["revocation", "revision", "rollback"] as const)("revalidates a late %s after waiting for company admission", async change => {
+    const f = await fixture([]), revision = randomUUID();
+    const [pending] = await db.update(connectionGrants).set({ providerTenant: {
+      ...f.grant.providerTenant, github: { ...f.grant.providerTenant!.github!,
+        bindingSync: { id: randomUUID(), removeInstallationIds: ["101"] } },
+    } }).where(eq(connectionGrants.id, f.grant.id)).returning();
+    let entered!: (pid: number) => void, release!: () => void;
+    const locked = new Promise<number>(resolve => { entered = resolve; });
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const mutation = db.transaction(async tx => {
+      await tx.select().from(companies).where(eq(companies.id, f.company.id)).for("no key update");
+      const [owner] = await tx.execute(sql`select pg_backend_pid() as pid`);
+      entered(Number(owner!.pid)); await gate;
+      if (change === "revision") await tx.update(connectionGrants).set({ providerTenant: {
+        ...pending!.providerTenant, github: { ...pending!.providerTenant!.github!,
+          bindingSync: { id: revision, removeInstallationIds: ["202"] } },
+      } }).where(eq(connectionGrants.id, f.grant.id));
+      else await tx.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, f.grant.id));
+      if (change === "rollback") throw new Error("rollback late revocation");
+    });
+    const mutated = Promise.allSettled([mutation]);
+    let handoff: Promise<unknown> | undefined;
+    try {
+      const ownerPid = await locked;
+      handoff = f.service.syncGitHubGrantBindings(f.connection, pending!);
+      const handedOff = Promise.allSettled([handoff]);
+      await vi.waitFor(async () => {
+        const waiting = await db.execute(sql`select pid from pg_stat_activity
+          where datname = current_database() and query like '%"companies"%' and query like '%for share%'
+            and ${ownerPid} = any(pg_blocking_pids(pid))`);
+        expect(waiting).toHaveLength(1);
+      }, { timeout: 5_000, interval: 20 });
+      release();
+      expect((await mutated)[0]!.status).toBe(change === "rollback" ? "rejected" : "fulfilled");
+      expect((await handedOff)[0]!.status).toBe("fulfilled");
+      const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id));
+      if (change === "rollback") {
+        expect(bindings).toEqual([expect.objectContaining({ grantId: f.grant.id, installationId: "101", active: false })]);
+        expect(saved!.status).toBe("active"); expect(saved!.providerTenant!.github!.bindingSync).toBeUndefined();
+      } else {
+        expect(bindings).toHaveLength(0);
+        if (change === "revocation") expect(saved!.status).toBe("revoked");
+        else expect(saved!.providerTenant!.github!.bindingSync).toEqual({ id: revision, removeInstallationIds: ["202"] });
+      }
+    } finally { release(); await mutated; await handoff?.catch(() => {}); }
+  });
+
   it("renews an installation created without metadata using the saved connection expiry", async () => {
     const f = await fixture();
     vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "01234567890123456789012345678901");

server/src/services/heartbeat.ts

@@ -4220,7 +4220,7 @@ export function heartbeatService(
       registerDeferredReceiptWork(db, coordinator, {
         resume: scope => heartbeatQueue.resumeDeferredExecutionReceipts(scope),
         settleTaskChat: scope => reconcileTaskChatRetryReceipts(db, { ...scope, settleOnly: true }),
-        dispatchNative: dispatchPendingNativeStatusWakeups,
+        createNativeBatch: heartbeatQueue.createNativeStatusWakeBatch,
         schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed,
       }, () => canRun() && readTaskDrain(new Date()) === null),
     registerIssueStateRecoveryWork: (coordinator: Parameters<typeof registerIssueRecoveryWork>[1], canRun: () => boolean) =>

server/src/services/heartbeat/deferred-receipt-work.ts

@@ -18,7 +18,7 @@ export function registerDeferredReceiptWork(
   owner: {
     resume(scope: DeferredReceiptRecoveryOptions): Promise<unknown>;
     settleTaskChat(scope: DeferredExecutionScope): Promise<unknown>;
-    dispatchNative(scope: NativeIntentOptions): Promise<unknown>;
+    createNativeBatch(): (scope: NativeIntentOptions) => Promise<unknown>;
     schedulingSuppressed(): Promise<boolean>;
   },
   canRun: () => boolean,
@@ -26,12 +26,17 @@ export function registerDeferredReceiptWork(
   function register(kind: "deferred" | "task_chat" | "native") {
     const cursors = new Map<string, string>();
     let enabled = false;
+    let dispatchNative: ReturnType<typeof owner.createNativeBatch>;
     const queue = kind === "deferred" ? DELIVERY_QUEUES.deferredExecutionReceipts
       : kind === "task_chat" ? DELIVERY_QUEUES.taskRetryReceiptSettlement : DELIVERY_QUEUES.nativeStatusWakeIntents;
     return registerScopedDeadlineWork<Entry, DeferredExecutionScope>(coordinator, {
       queue,
+      batchSize: kind === "native" ? 100 : undefined,
       reconstructionIntervalMs: 30_000,
-      beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); },
+      beforePass: async () => {
+        enabled = !(await owner.schedulingSuppressed());
+        if (kind === "native") dispatchNative = owner.createNativeBatch();
+      },
       decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint),
       matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.requestId || entry.id === scope.requestId),
       load: async scope => {
@@ -101,7 +106,7 @@ export function registerDeferredReceiptWork(
         const errors: unknown[] = [];
         let next: number | null = null;
         const scope = { companyId: entry.companyId, requestId: entry.id };
-        if (kind === "native") await owner.dispatchNative({ ...scope,
+        if (kind === "native") await dispatchNative({ ...scope,
           onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); },
           onError: error => { errors.push(error); },
         });

server/src/services/heartbeat/queue.ts

@@ -4784,15 +4784,27 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
    * wake was inserted but before the intent was linked to it. Dispatch failure
    * only requeues the intent; it never changes the provider run outcome.
    */
-  async function dispatchPendingNativeStatusWakeups(
-    input: {
-      companyId?: string;
-      limit?: number;
-      staleClaimMs?: number;
-      requestId?: string;
-      onDeferredDeadline?: (at: number) => void;
-      onError?: (error: unknown) => void;
-    } = {},
+  type NativeStatusWakeDispatchInput = {
+    companyId?: string;
+    limit?: number;
+    staleClaimMs?: number;
+    requestId?: string;
+    onDeferredDeadline?: (at: number) => void;
+    onError?: (error: unknown) => void;
+  };
+  type NativeStatusWakeDelivery = { runId: string | null; status: string };
+
+  function createNativeStatusWakeBatch() {
+    const deliveredByIssueScope = new Map<string, NativeStatusWakeDelivery>();
+    return (input: NativeStatusWakeDispatchInput = {}) => dispatchNativeStatusWakeBatch(input, deliveredByIssueScope);
+  }
+
+  function dispatchPendingNativeStatusWakeups(input: NativeStatusWakeDispatchInput = {}) {
+    return createNativeStatusWakeBatch()(input);
+  }
+
+  async function dispatchNativeStatusWakeBatch(input: NativeStatusWakeDispatchInput,
+    deliveredByIssueScope: Map<string, NativeStatusWakeDelivery>,
   ) {
     const now = new Date();
     const staleClaimMs = Math.max(1_000, input.staleClaimMs ?? 60_000);
@@ -4817,10 +4829,6 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
     let dispatched = 0;
     let recovered = 0;
     let deferred = 0;
-    const deliveredByIssueScope = new Map<
-      string,
-      { runId: string | null; status: string }
-    >();
 
     for (const candidate of candidates) {
       const payload = parseObject(candidate.payload);
@@ -5093,6 +5101,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
   return {
     enqueueWakeup,
     dispatchPendingNativeStatusWakeups,
+    createNativeStatusWakeBatch,
     startNextQueuedRunForAgent,
     countRunningRunsForAgent,
     releaseRunClaimedJustBeforeSuppression,

server/src/services/tool-access/github-grant-continuity.ts

@@ -234,6 +234,11 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
     // Serialize external binding changes with grant/connection lifecycle writes.
     // Requests occur only for saved debt; no idle lock or connection is retained.
     return db.transaction(async tx => {
+      // Company deletion locks its parent before cascading to grants/connections.
+      // Keep that order while retaining admission through the external handoff.
+      const [company] = await tx.select({ status: companies.status }).from(companies)
+        .where(eq(companies.id, connection.companyId)).for("share").limit(1);
+      if (!company) return grant;
       const [current] = await tx.select().from(connectionGrants).where(and(
         eq(connectionGrants.id, grant.id), eq(connectionGrants.companyId, connection.companyId),
         eq(connectionGrants.connectionId, connection.id),
@@ -243,9 +248,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
         eq(toolConnections.id, connection.id), eq(toolConnections.companyId, connection.companyId),
       )).for("update").limit(1);
       if (!liveConnection || liveConnection.status !== "active" || !liveConnection.enabled) return current;
-      const [company] = await tx.select({ status: companies.status }).from(companies)
-        .where(eq(companies.id, connection.companyId)).for("share").limit(1);
-      if (company?.status !== "active") return current;
+      if (company.status !== "active") return current;
       const github = current.providerTenant?.github;
       if (!github?.bindingSync || github.bindingSync.id !== grant.providerTenant?.github?.bindingSync?.id) return current;
       const subject = current.kind === "agent" && current.subjectAgentId ? `agent:${current.subjectAgentId}`