Integrate external refresh with existing deadline owners

Preserve all company mutation consumers and verify both merge-confirmation and external-refresh startup registrations. Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-11-11 05:11:49 +01:00 · 2026-10-30 18:19:58 -05:00
commit 2bca873c63
@@ -0,0 +1,452 @@
+import { createServer } from "node:http";
+import { randomUUID } from "node:crypto";
+import { setTimeout as delay } from "node:timers/promises";
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { activityLog, approvals, applyPendingMigrations, databaseWorkPending, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, externalObjectMentions, issues, issueComments, plugins } from "@paperclipai/db";
+import { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js";
+import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
+import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
+import { documentService } from "../services/documents.js";
+import { budgetService } from "../services/budgets.js";
+import { pluginRegistryService } from "../services/plugin-registry.js";
+import { companyService } from "../services/companies.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { notifyExternalObjectRefresh } from "../services/external-object-refresh-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+
+const url = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECT_REFRESH_TEST_DATABASE_URL");
+const support = url ? { supported: true } : await getEmbeddedPostgresTestSupport();
+const terminal: ExternalObjectResolveResult = { ok: true, snapshot: {
+  statusKey: "done", statusCategory: "succeeded", statusTone: "success", isTerminal: true,
+} };
+const active: ExternalObjectResolveResult = { ok: true, snapshot: {
+  statusKey: "open", statusCategory: "open", statusTone: "neutral", isTerminal: false, ttlSeconds: 60,
+} };
+
+(support.supported ? describe : describe.skip)("external object refresh work (real PostgreSQL, simulated provider)", () => {
+  let db: ReturnType<typeof createDb>;
+  let temp: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let admitted = true;
+  let reconcile = true;
+  let enabled = true;
+  const provider = vi.fn(async (): Promise<ExternalObjectResolveResult> => terminal);
+  const errors = vi.fn();
+  const companyIds: string[] = [];
+  beforeAll(async () => {
+    if (url) await applyPendingMigrations(url);
+    if (!url) temp = await startEmbeddedPostgresTestDatabase("external-refresh-work-");
+    db = createDb(url ?? temp!.connectionString, { maxConnections: 6 });
+    await db.execute(sql`truncate table ${companies} cascade`);
+  }, 90_000);
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    for (const id of companyIds.splice(0)) {
+      await db.delete(budgetIncidents).where(eq(budgetIncidents.companyId, id));
+      await db.delete(budgetPolicies).where(eq(budgetPolicies.companyId, id));
+      await db.delete(approvals).where(eq(approvals.companyId, id));
+      await db.delete(costEvents).where(eq(costEvents.companyId, id));
+      await db.delete(activityLog).where(eq(activityLog.companyId, id));
+      await db.delete(issues).where(eq(issues.companyId, id));
+      await db.delete(externalObjects).where(eq(externalObjects.companyId, id));
+      await db.delete(companies).where(eq(companies.id, id));
+    }
+    await db.delete(plugins);
+    admitted = true; reconcile = true; enabled = true; errors.mockClear();
+    provider.mockReset().mockResolvedValue(terminal);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temp?.cleanup(); });
+
+  function service() {
+    return externalObjectService(db, { github: false, enabled: () => enabled,
+      resolvers: [{ providerKey: "url", resolve: provider }] });
+  }
+  async function start(svc = service()) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted,
+      canReconcile: () => reconcile, onError: errors });
+    const worker = svc.registerRefreshWork(coordinator, () => admitted);
+    await worker.ready;
+    return worker;
+  }
+  async function fixture(status = "active") {
+    const id = randomUUID(); companyIds.push(id);
+    await db.insert(companies).values({ id, name: "Refresh test", status,
+      issuePrefix: `E${id.replaceAll("-", "").slice(0, 8)}` });
+    const [issue] = await db.insert(issues).values({ companyId: id, title: "Reference",
+      description: `https://example.com/${id}`, status: "todo" }).returning();
+    return { companyId: id, issueId: issue!.id };
+  }
+  async function row(companyId: string) {
+    return (await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)))[0]!;
+  }
+  async function settle(predicate: () => Promise<boolean> | boolean) {
+    for (let i = 0; i < 200; i++) {
+      await vi.advanceTimersByTimeAsync(0); await delay(5);
+      if (await predicate()) return;
+    }
+    throw new Error("refresh worker did not settle");
+  }
+  async function quiet() { await settle(() => coordinator!.nextWakeAt() === null); }
+  function clock() { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); }
+  async function change(companyId: string, patch: Partial<typeof externalObjects.$inferInsert>) {
+    await db.transaction(async tx => {
+      await notifyExternalObjectRefresh(tx, companyId);
+      await tx.update(externalObjects).set(patch).where(eq(externalObjects.companyId, companyId));
+    });
+  }
+
+  it("activates saved work when the ordinary feature and company settings enable it", async () => {
+    clock(); const settings = instanceSettingsService(db);
+    await settings.updateExperimental({ enableExternalObjects: true });
+    const svc = externalObjectService(db, { github: false,
+      enabled: async () => (await settings.getExperimental()).enableExternalObjects,
+      resolvers: [{ providerKey: "url", resolve: provider }] });
+    const f = await fixture("paused"); await svc.syncIssue(f.issueId);
+    await settings.updateExperimental({ enableExternalObjects: false });
+    await start(svc); await quiet();
+    await companyService(db).update(f.companyId, { status: "active" }); await quiet();
+    expect(provider).not.toHaveBeenCalled();
+    await settings.updateExperimental({ enableExternalObjects: true });
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledOnce();
+  });
+
+  it("settles empty, configured-idle and retained terminal history without recurring SQL or provider calls", async () => {
+    clock();
+    const svc = service(); await start(svc); await quiet();
+    const selects = vi.spyOn(db, "select");
+    await vi.advanceTimersByTimeAsync(15 * 60_000);
+    expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
+    const f = await fixture(); await svc.syncIssue(f.issueId);
+    await settle(async () => (await row(f.companyId))?.isTerminal === true); await quiet();
+    expect((await row(f.companyId)).statusKey).toBe("done");
+    selects.mockClear(); provider.mockClear();
+    await vi.advanceTimersByTimeAsync(15 * 60_000);
+    expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
+    expect(errors).not.toHaveBeenCalled();
+  });
+
+  it("registers exact future deadlines, preempts them on earlier writes and cancels terminal work", async () => {
+    clock(); provider.mockResolvedValue(active);
+    const svc = service(); await start(svc);
+    const f = await fixture(); await svc.syncIssue(f.issueId);
+    await settle(() => provider.mock.calls.length === 1);
+    await settle(async () => (await row(f.companyId)).lastResolvedAt !== null);
+    const saved = await row(f.companyId);
+    await settle(() => coordinator!.nextWakeAt() === saved.nextRefreshAt!.getTime());
+    const earlier = new Date(Date.now() + 1000); await change(f.companyId, { nextRefreshAt: earlier });
+    await settle(() => coordinator!.nextWakeAt() === earlier.getTime());
+    await vi.advanceTimersByTimeAsync(999); expect(provider).toHaveBeenCalledTimes(1);
+    provider.mockResolvedValue(terminal); await vi.advanceTimersByTimeAsync(1);
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    await vi.advanceTimersByTimeAsync(120_000); expect(provider).toHaveBeenCalledTimes(2);
+  });
+
+  it("keeps disabled and paused company work passive and reconstructs on eligibility mutation", async () => {
+    clock(); enabled = false;
+    const svc = service(); await start(svc); await quiet();
+    const f = await fixture("paused");
+    enabled = true; await svc.syncIssue(f.issueId); await quiet();
+    await vi.advanceTimersByTimeAsync(600_000); expect(provider).not.toHaveBeenCalled();
+    await db.transaction(async tx => {
+      await notifyExternalObjectRefresh(tx, f.companyId);
+      await tx.update(companies).set({ status: "active" }).where(eq(companies.id, f.companyId));
+    });
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledOnce();
+  });
+
+  it("waits for the outer commit and leaves rolled-back ingestion quiet", async () => {
+    clock(); const svc = service(); await start(svc); await quiet();
+    const f = await fixture();
+    await expect(db.transaction(async tx => {
+      await svc.syncIssue(f.issueId, tx); throw new Error("rollback");
+    })).rejects.toThrow("rollback");
+    await quiet(); expect(await row(f.companyId)).toBeUndefined(); expect(provider).not.toHaveBeenCalled();
+    let release!: () => void; const held = new Promise<void>(resolve => { release = resolve; });
+    let nestedDone = false;
+    const commit = db.transaction(async tx => {
+      await tx.transaction(async nested => { await svc.syncIssue(f.issueId, nested); });
+      nestedDone = true; await held;
+    });
+    await settle(() => nestedDone); await vi.advanceTimersByTimeAsync(60_000);
+    expect(provider).not.toHaveBeenCalled(); release(); await commit;
+    await settle(async () => (await row(f.companyId))?.isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledOnce();
+  });
+
+  it("retains first work during drain and recovers a provider failure after restart", async () => {
+    clock(); const svc = service(); await start(svc); await quiet();
+    admitted = false; const f = await fixture(); await svc.syncIssue(f.issueId);
+    const selects = vi.spyOn(db, "select");
+    await vi.advanceTimersByTimeAsync(90_000); expect(provider).not.toHaveBeenCalled();
+    selects.mockClear(); await vi.advanceTimersByTimeAsync(90_000); expect(selects).not.toHaveBeenCalled();
+    admitted = true; provider.mockRejectedValueOnce(new Error("simulated transport failure"));
+    await vi.advanceTimersByTimeAsync(30_000); await settle(() => errors.mock.calls.length === 1);
+    const claimed = await row(f.companyId); expect(claimed.refreshToken).not.toBeNull();
+    await coordinator!.stop(); coordinator = undefined; await start(svc);
+    const due = claimed.refreshStartedAt!.getTime() + 300_000;
+    expect(coordinator!.nextWakeAt()).toBe(due);
+    await vi.advanceTimersByTimeAsync(due - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(1);
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledTimes(2);
+  });
+
+  it("leaves unsupported URLs passive and cancels orphan work until ordinary reattachment", async () => {
+    clock(); const unsupported = externalObjectService(db, { github: false });
+    await start(unsupported); const f = await fixture(); await unsupported.syncIssue(f.issueId); await quiet();
+    const selects = vi.spyOn(db, "select");
+    await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled();
+    expect((await row(f.companyId)).lastResolvedAt).toBeNull();
+    await coordinator!.stop(); coordinator = undefined;
+    provider.mockResolvedValue(active); const svc = service(); await start(svc);
+    await settle(async () => (await row(f.companyId)).lastResolvedAt !== null);
+    await db.update(issues).set({ description: null }).where(eq(issues.id, f.issueId));
+    await svc.syncIssue(f.issueId); await quiet();
+    selects.mockClear(); provider.mockClear();
+    await vi.advanceTimersByTimeAsync(900_000);
+    expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
+    provider.mockResolvedValue(terminal);
+    await db.update(issues).set({ description: `https://example.com/${f.companyId}` }).where(eq(issues.id, f.issueId));
+    await svc.syncIssue(f.issueId);
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledOnce();
+  });
+
+  it("activates a newly ready plugin and fences a result after company permission is withdrawn", async () => {
+    clock(); const registry = pluginRegistryService(db);
+    const manifest: PaperclipPluginManifestV1 = {
+      id: `paperclip.refresh-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Refresh fixture",
+      description: "Simulated external object provider", author: "Paperclip", categories: ["connector"],
+      capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" },
+      objectReferences: [{ providerKey: "tracker", displayName: "Tracker", objectTypes: ["ticket"], urlPatterns: ["https://example.com/:id"] }],
+    };
+    const plugin = await registry.install({ packageName: "refresh-fixture" }, manifest);
+    let finish!: (result: ExternalObjectResolveResult) => void;
+    const calls = vi.fn(async (_id: string, method: string) => {
+      if (method === "detectExternalObjects") return { detections: [] };
+      if (calls.mock.calls.filter(call => call[1] === "resolveExternalObject").length === 1)
+        return new Promise<ExternalObjectResolveResult>(resolve => { finish = resolve; });
+      return terminal;
+    });
+    const svc = externalObjectService(db, { github: false,
+      pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager,
+      detectors: [{ key: "fixture", detect: ({ urls }) => urls.map(canonical => ({ canonical,
+        detectorKey: "fixture", providerKey: "tracker", objectType: "ticket", externalId: canonical.canonicalIdentityHash })) }],
+    });
+    await start(svc); const f = await fixture(); await svc.syncIssue(f.issueId); await quiet();
+    const selects = vi.spyOn(db, "select");
+    await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); expect(calls).not.toHaveBeenCalled();
+    await registry.updateStatus(plugin!.id, { status: "ready" });
+    await settle(() => typeof finish === "function");
+    await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} });
+    finish(terminal);
+    await settle(async () => (await row(f.companyId)).refreshToken === null); await quiet();
+    expect((await row(f.companyId)).isTerminal).toBe(false);
+    selects.mockClear(); calls.mockClear();
+    await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); expect(calls).not.toHaveBeenCalled();
+    calls.mockImplementation(async () => terminal);
+    await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: true, settingsJson: {} });
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    expect(calls).toHaveBeenCalledOnce();
+  });
+
+  it("reactivates refresh after ordinary budget pause and budget raise", async () => {
+    clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc);
+    const f = await fixture(); await svc.syncIssue(f.issueId);
+    await settle(async () => (await row(f.companyId))?.lastResolvedAt !== null);
+    await db.insert(costEvents).values({ companyId: f.companyId, usageKind: "decision", provider: "fixture",
+      biller: "fixture", billingType: "metered_api", model: "simulated", costCents: 200, occurredAt: new Date() });
+    const budgets = budgetService(db);
+    await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId,
+      amount: 100, hardStopEnabled: true }, null);
+    expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0]!.status).toBe("paused");
+    await quiet();
+    const reads = vi.spyOn(db, "select"); provider.mockClear();
+    await vi.advanceTimersByTimeAsync(900_000); expect(reads).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
+    provider.mockResolvedValue(terminal);
+    await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 300 }, null);
+    await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledOnce();
+  });
+
+  it.each(["comment", "document"] as const)("keeps hard-deleted %s activation scoped to its retained mention owner", async kind => {
+    clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc);
+    const a = await fixture(); const b = await fixture(); await svc.syncIssue(b.issueId);
+    let id: string;
+    if (kind === "comment") {
+      const [comment] = await db.insert(issueComments).values({ companyId: a.companyId, issueId: a.issueId,
+        body: `https://example.com/${a.companyId}` }).returning();
+      id = comment!.id; await svc.syncComment(id);
+    } else {
+      const result = await documentService(db).upsertIssueDocument({ issueId: a.issueId,
+        key: "notes", format: "markdown", body: `https://example.com/${a.companyId}` });
+      id = result.document.id; await svc.syncDocument(id);
+    }
+    await settle(async () => {
+      const left = await row(a.companyId), right = await row(b.companyId);
+      return Boolean(left?.lastResolvedAt && right?.lastResolvedAt);
+    });
+    const expectedDeadline = Math.min((await row(a.companyId)).nextRefreshAt!.getTime(), (await row(b.companyId)).nextRefreshAt!.getTime());
+    await settle(() => coordinator!.nextWakeAt() === expectedDeadline && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+    const reads: Array<{ query: string; params: unknown[] }> = [];
+    const debug = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query, params) => {
+      if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"'))
+        reads.push({ query, params: [...params] });
+    };
+    try {
+      if (kind === "comment") { await db.delete(issueComments).where(eq(issueComments.id, id)); await svc.syncComment(id); }
+      else { await documentService(db).deleteIssueDocument(a.issueId, "notes"); await svc.syncDocument(id); }
+      await settle(() => reads.length > 0);
+      for (const read of reads) {
+        expect(read.query).toContain('"external_objects"."company_id" =');
+        expect(read.params).toContain(a.companyId); expect(read.params).not.toContain(b.companyId);
+      }
+    } finally { db.$client.options.debug = debug; }
+    expect(provider).toHaveBeenCalledTimes(2);
+  });
+
+  it("keeps warm standby startup SQL-free and resumes a first committed producer", async () => {
+    clock(); admitted = false; reconcile = false;
+    const svc = service(); const selects = vi.spyOn(db, "select");
+    await start(svc); await vi.advanceTimersByTimeAsync(90_000);
+    expect(selects).not.toHaveBeenCalled();
+    const f = await fixture(); await svc.syncIssue(f.issueId);
+    selects.mockClear(); await vi.advanceTimersByTimeAsync(90_000);
+    expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled();
+    admitted = true; reconcile = true;
+    await vi.advanceTimersByTimeAsync(30_000);
+    await settle(async () => (await row(f.companyId))?.isTerminal); await quiet();
+    expect(provider).toHaveBeenCalledOnce();
+  });
+
+  it("renews an active provider lease so another service cannot duplicate its action", async () => {
+    clock(); const svc = service(); await start(svc);
+    let finish!: (result: ExternalObjectResolveResult) => void;
+    provider.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
+    const f = await fixture(); await svc.syncIssue(f.issueId);
+    await settle(() => provider.mock.calls.length === 1);
+    const claimed = await row(f.companyId);
+    await vi.advanceTimersByTimeAsync(60_000);
+    await settle(async () => (await row(f.companyId)).refreshStartedAt!.getTime() > claimed.refreshStartedAt!.getTime());
+    const competing = await service().refreshObject(claimed.id, { companyId: f.companyId,
+      force: true, now: new Date(claimed.refreshStartedAt!.getTime() + 301_000) });
+    expect(competing.reason).toBe("refresh_in_progress"); expect(provider).toHaveBeenCalledOnce();
+    finish(terminal); await settle(async () => (await row(f.companyId)).isTerminal); await quiet();
+  });
+
+  it("uses durable provider retry deadlines and cancels the observed result on shutdown without overlap", async () => {
+    clock(); const svc = service(); await start(svc);
+    provider.mockResolvedValueOnce({ ok: false, liveness: "rate_limited", errorCode: "slow_down", retryAfterSeconds: 5 });
+    const f = await fixture(); await svc.syncIssue(f.issueId);
+    await settle(async () => (await row(f.companyId))?.lastErrorCode === "slow_down");
+    const saved = await row(f.companyId);
+    await settle(() => coordinator!.nextWakeAt() === saved.nextRefreshAt!.getTime());
+    expect(saved.refreshToken).toBeNull();
+    await vi.advanceTimersByTimeAsync(saved.nextRefreshAt!.getTime() - Date.now() - 1);
+    expect(provider).toHaveBeenCalledOnce();
+    let finish!: (result: ExternalObjectResolveResult) => void;
+    provider.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
+    await vi.advanceTimersByTimeAsync(1); await settle(() => provider.mock.calls.length === 2);
+    await svc.syncIssue(f.issueId); await vi.advanceTimersByTimeAsync(0);
+    expect(provider).toHaveBeenCalledTimes(2);
+    let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; });
+    await stop; expect(stopped).toBe(true); coordinator = undefined;
+    const cancelled = await row(f.companyId); expect(cancelled.refreshToken).toBeNull(); expect(cancelled.isTerminal).toBe(false);
+    finish(terminal); await delay(10);
+    expect((await row(f.companyId)).isTerminal).toBe(false);
+    const selects = vi.spyOn(db, "select");
+    await vi.advanceTimersByTimeAsync(600_000);
+    expect(selects).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(2);
+  });
+
+  it("cancels an actual stalled GitHub HTTP body, retains due debt and resumes after restart", async () => {
+    clock(); let stall = true, started = false, closed = false;
+    const server = createServer((request, response) => {
+      if (request.url?.includes("/pulls/")) {
+        if (stall) {
+          started = true; response.on("close", () => { closed = true; });
+          response.writeHead(200, { "content-type": "application/json" }); response.write('{"state":');
+        } else response.end(JSON.stringify({ state: "closed", merged: true, title: "Done" }));
+      } else response.end(JSON.stringify({ private: false }));
+    });
+    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("Missing HTTP test port");
+    const svc = externalObjectService(db, { github: {
+      fetch: (url, init) => fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init),
+      credentialProvider: () => ({ token: "simulated-company-token", source: "company_secret" }),
+    } });
+    try {
+      await start(svc); const f = await fixture();
+      await db.update(issues).set({ description: "https://github.com/acme/app/pull/42" }).where(eq(issues.id, f.issueId));
+      await svc.syncIssue(f.issueId); await settle(() => started);
+      await coordinator!.stop(); coordinator = undefined;
+      await settle(() => closed);
+      const cancelled = await row(f.companyId);
+      expect(cancelled.refreshToken).toBeNull(); expect(cancelled.lastResolvedAt).toBeNull();
+      expect(cancelled.nextRefreshAt!.getTime()).toBeLessThanOrEqual(Date.now());
+      stall = false; await start(svc);
+      await settle(async () => Boolean((await row(f.companyId))?.isTerminal)); await quiet();
+    } finally { await coordinator?.stop(); coordinator = undefined; server.closeAllConnections(); await new Promise<void>(resolve => server.close(() => resolve())); }
+  });
+
+  it("loads only the changed identity and objects after ingestion into ten thousand future objects", async () => {
+    clock(); const svc = service(); const f = await fixture();
+    const future = new Date(Date.now() + 86_400_000);
+    const ids = Array.from({ length: 10_000 }, () => randomUUID());
+    for (let offset = 0; offset < ids.length; offset += 500) {
+      const batch = ids.slice(offset, offset + 500);
+      await db.insert(externalObjects).values(batch.map(id => ({ id, companyId: f.companyId,
+        providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future })));
+      await db.insert(externalObjectMentions).values(batch.map(objectId => ({ companyId: f.companyId,
+        sourceIssueId: f.issueId, sourceKind: "description" as const, objectId })));
+    }
+    await start(svc); await settle(() => coordinator!.nextWakeAt() === future.getTime());
+    const [incoming] = await db.insert(issues).values({ companyId: f.companyId, title: "Incoming",
+      description: "https://example.com/new-object" }).returning();
+    const reads: Array<{ query: string; params: unknown[] }> = [];
+    const debug = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query, params) => {
+      if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"'))
+        reads.push({ query, params: [...params] });
+    };
+    try {
+      await svc.syncIssue(incoming!.id);
+      await settle(() => provider.mock.calls.length === 1);
+      await settle(() => coordinator!.nextWakeAt() === future.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+      expect(reads.length).toBeGreaterThan(0); expect(reads.length).toBeLessThanOrEqual(6);
+      for (const read of reads) expect(read.query).toMatch(/"external_objects"\."(?:id|external_id)" =/);
+    } finally { db.$client.options.debug = debug; }
+  });
+
+  it("reconstructs paginated multi-company work and never invokes inactive or never-refresh objects", async () => {
+    clock(); const svc = service(); const a = await fixture(); const b = await fixture(); const inactive = await fixture("paused");
+    for (const companyId of [b.companyId, inactive.companyId]) {
+      await db.insert(externalObjects).values(Array.from({ length: 255 }, (_, index) => ({
+        companyId, providerKey: "url", objectType: "link", externalId: String(index),
+        sanitizedCanonicalUrl: `https://example.com/${companyId}/${index}`, canonicalIdentityHash: `${companyId}-${index}`,
+        nextRefreshAt: null,
+      })));
+    }
+    await db.update(issues).set({ description: Array.from({ length: 255 }, (_, index) =>
+      `https://example.com/${a.companyId}/${index}`).join(" ") }).where(eq(issues.id, a.issueId));
+    await svc.syncIssue(a.issueId);
+    const future = new Date(Date.now() + 60_000);
+    await db.update(externalObjects).set({ nextRefreshAt: future }).where(eq(externalObjects.companyId, a.companyId));
+    await svc.syncIssue(b.issueId);
+    await start(svc); await settle(() => provider.mock.calls.length === 1);
+    await settle(() => coordinator!.nextWakeAt() === future.getTime());
+    expect((await row(a.companyId)).nextRefreshAt).toEqual(future);
+    const companiesSeen = provider.mock.calls.map(call => (call as unknown as [{ companyId: string }])[0].companyId);
+    expect(companiesSeen).toEqual([b.companyId]);
+    await change(a.companyId, { isTerminal: true }); await quiet();
+    await vi.advanceTimersByTimeAsync(600_000); expect(provider).toHaveBeenCalledOnce();
+  });
+});
@@ -1,9 +1,11 @@
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { eq } from "drizzle-orm";
 import {
  companies,
+  applyPendingMigrations,
  createDb,
  activityLog,
  externalObjectMentions,
@@ -12,6 +14,7 @@ import {
  issues,
  plugins,
 } from "@paperclipai/db";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import {
  getEmbeddedPostgresTestSupport,
  startEmbeddedPostgresTestDatabase,
@@ -27,8 +30,8 @@ import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
 import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js";

-const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL");
-const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL");
+const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;

 if (!embeddedPostgresSupport.supported) {
@@ -711,9 +714,10 @@ describeEmbeddedPostgres("externalObjectService", () => {
  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;

  beforeAll(async () => {
-    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-");
-    db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 });
-  }, 20_000);
+    if (externalDatabaseUrl) await applyPendingMigrations(externalDatabaseUrl);
+    if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-");
+    db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
+  }, 90_000);

  afterEach(async () => {
    await db.delete(activityLog);
@@ -731,6 +735,18 @@ describeEmbeddedPostgres("externalObjectService", () => {
    await tempDb?.cleanup();
  });

+  async function runRefreshWork(service: ReturnType<typeof externalObjectService>) {
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      onError: error => errors.push(error) });
+    try {
+      await service.registerRefreshWork(coordinator, () => true).ready;
+      expect(errors).toEqual([]);
+    } finally {
+      await coordinator.stop();
+    }
+  }
+
  async function createIssue(companyId = randomUUID()) {
    const issueId = randomUUID();
    await db.insert(companies).values({
@@ -880,9 +896,9 @@ describeEmbeddedPostgres("externalObjectService", () => {

    expect(object.nextRefreshAt).toBeInstanceOf(Date);

-    const refreshed = await svc.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000));
+    await runRefreshWork(svc);

-    expect(refreshed).toHaveLength(1);
+    expect((await db.select().from(externalObjects))[0]?.lastResolvedAt).toBeInstanceOf(Date);
    expect(resolve).toHaveBeenCalledTimes(1);
  });

@@ -919,15 +935,13 @@ describeEmbeddedPostgres("externalObjectService", () => {
    await svc.syncIssue(issueId);
    const object = await db.select().from(externalObjects).then((rows) => rows[0]!);

-    const dueRefresh = svc.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000));
+    const dueRefresh = runRefreshWork(svc);
    await resolverStarted;
    const manualRefresh = svc.refreshObject(object.id, { companyId, force: true });
    finishResolver();

-    const [dueResults, manualResult] = await Promise.all([dueRefresh, manualRefresh]);
+    const [, manualResult] = await Promise.all([dueRefresh, manualRefresh]);

-    expect(dueResults).toHaveLength(1);
-    expect(dueResults[0]?.refreshed).toBe(true);
    expect(manualResult.refreshed).toBe(true);
    expect(manualResult.object.statusLabel).toBe("Open");
    expect(resolve).toHaveBeenCalledTimes(1);
@@ -967,14 +981,12 @@ describeEmbeddedPostgres("externalObjectService", () => {
    await scheduledService.syncIssue(issueId);
    const object = await db.select().from(externalObjects).then((rows) => rows[0]!);

-    const dueRefresh = scheduledService.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000));
+    const dueRefresh = runRefreshWork(scheduledService);
    await resolverStarted;
    const manualRefresh = await manualService.refreshObject(object.id, { companyId, force: true });
    finishResolver();
-    const dueResults = await dueRefresh;
+    await dueRefresh;

-    expect(dueResults).toHaveLength(1);
-    expect(dueResults[0]?.refreshed).toBe(true);
    expect(manualRefresh.refreshed).toBe(false);
    expect(manualRefresh.reason).toBe("refresh_in_progress");
    expect(manualRefresh.object).not.toHaveProperty("refreshToken");
@@ -1030,7 +1042,7 @@ describeEmbeddedPostgres("externalObjectService", () => {
    const object = await db.select().from(externalObjects).then((rows) => rows[0]!);
    const claimedAt = new Date(Date.now() + 1_000);

-    const dueRefresh = scheduledService.refreshDueObjects(companyId, 50, claimedAt);
+    const dueRefresh = runRefreshWork(scheduledService);
    await resolverStarted;
    const replacementResult = await replacementService.refreshObject(object.id, {
      companyId,
@@ -1038,11 +1050,10 @@ describeEmbeddedPostgres("externalObjectService", () => {
      now: new Date(claimedAt.getTime() + 301_000),
    });
    finishResolver();
-    const dueResults = await dueRefresh;
+    await dueRefresh;
    const finalObject = await db.select().from(externalObjects).then((rows) => rows[0]!);

    expect(replacementResult.refreshed).toBe(true);
-    expect(dueResults[0]).toMatchObject({ refreshed: false, reason: "refresh_superseded" });
    expect(finalObject.statusLabel).toBe("Closed");
    expect(finalObject.refreshStartedAt).toBeNull();
    expect(finalObject.refreshToken).toBeNull();
@@ -1076,9 +1087,7 @@ describeEmbeddedPostgres("externalObjectService", () => {
    await svc.syncIssue(active.issueId);
    await svc.syncIssue(paused.issueId);

-    const result = await svc.refreshDueObjectsForActiveCompanies(50, new Date(Date.now() + 1_000));
-
-    expect(result).toEqual({ companies: 1, checked: 1, refreshed: 1 });
+    await runRefreshWork(svc);
    expect(resolve).toHaveBeenCalledTimes(1);
    const rows = await db.select().from(externalObjects);
    const activeObject = rows.find((row) => row.companyId === active.companyId);
@@ -1137,9 +1146,8 @@ describeEmbeddedPostgres("externalObjectService", () => {
      .set({ nextRefreshAt: new Date(0) })
      .where(eq(externalObjects.id, object.id));

-    const refreshed = await svc.refreshDueObjects(companyId);
+    await runRefreshWork(svc);

-    expect(refreshed).toEqual([]);
    expect(resolve).toHaveBeenCalledTimes(1);
  });

@@ -1182,7 +1190,7 @@ describeEmbeddedPostgres("externalObjectService", () => {
  });

  it("re-checks previously frozen GitHub not-found objects so they can self-heal", async () => {
-    const { companyId } = await createIssue();
+    const { companyId, issueId } = await createIssue();
    const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42");
    if (!canonical) throw new Error("expected canonical url");
    const inserted = await db.insert(externalObjects).values({
@@ -1223,9 +1231,10 @@ describeEmbeddedPostgres("externalObjectService", () => {
    const svc = externalObjectService(db, { resolvers: [resolver], github: false });
    expect(object.isTerminal).toBe(true);

-    const refreshed = await svc.refreshDueObjects(companyId);
-    expect(refreshed).toHaveLength(1);
-    expect(refreshed[0]?.refreshed).toBe(true);
+    const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId));
+    await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId,
+      sourceKind: "description", objectId: linked!.id });
+    await runRefreshWork(svc);

    const updated = await db.select().from(externalObjects).then((rows) => rows[0]!);
    expect(updated.statusLabel).toBe("Open");
@@ -1234,7 +1243,7 @@ describeEmbeddedPostgres("externalObjectService", () => {
  });

  it("does not re-check probe-verified GitHub not-found objects (deleted references are terminal)", async () => {
-    const { companyId } = await createIssue();
+    const { companyId, issueId } = await createIssue();
    const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42");
    if (!canonical) throw new Error("expected canonical url");
    await db.insert(externalObjects).values({
@@ -1271,14 +1280,16 @@ describeEmbeddedPostgres("externalObjectService", () => {
    };
    const svc = externalObjectService(db, { resolvers: [resolver], github: false });

-    const refreshed = await svc.refreshDueObjects(companyId);
+    const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId));
+    await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId,
+      sourceKind: "description", objectId: linked!.id });
+    await runRefreshWork(svc);

-    expect(refreshed).toEqual([]);
    expect(resolver.resolve).not.toHaveBeenCalled();
  });

  it("does not re-check terminal objects that are not GitHub not-found verdicts", async () => {
-    const { companyId } = await createIssue();
+    const { companyId, issueId } = await createIssue();
    const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42");
    if (!canonical) throw new Error("expected canonical url");
    await db.insert(externalObjects).values({
@@ -1314,9 +1325,11 @@ describeEmbeddedPostgres("externalObjectService", () => {
    };
    const svc = externalObjectService(db, { resolvers: [resolver], github: false });

-    const refreshed = await svc.refreshDueObjects(companyId);
+    const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId));
+    await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId,
+      sourceKind: "description", objectId: linked!.id });
+    await runRefreshWork(svc);

-    expect(refreshed).toEqual([]);
    expect(resolver.resolve).not.toHaveBeenCalled();
  });

@@ -1,3 +1,4 @@
+import { externalObjectService } from "../services/external-objects.js";
 import { startToolHealthWorker } from "./helpers/tool-health-worker.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
@@ -35,7 +36,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
  beforeAll(async () => {
    if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-");
    if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl);
-    db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString);
+    db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);

  afterEach(async () => {
@@ -171,12 +172,27 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
      expect(connector.leaseEvents).toHaveBeenCalledTimes(5);
    }

+    let releaseRefresh!: () => void;
+    let markStarted!: () => void;
+    const started = new Promise<void>(resolve => { markStarted = resolve; });
+    const held = new Promise<void>(resolve => { releaseRefresh = resolve; });
+    const resolver = externalObjectService(db, { github: false, resolvers: [{
+      providerKey: "github", resolve: async () => {
+        markStarted(); await held;
+        return { ok: true, snapshot: { statusKey: "open", statusCategory: "open", statusTone: "info", isTerminal: false } };
+      },
+    }] });
+    const oldRefresh = resolver.refreshObject(externalObjectId, { companyId, force: true });
+    await started;
+
    const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
    await expect(service.pollOnce({ forSetup: activeSetup })).resolves.toMatchObject({ leased: 2, processed: 1, duplicate: 0, failed: 1 });
    expect(warning.mock.calls).toEqual([[{
      deliveryId: "undecryptable-delivery", reason: "inner_event_authentication_failed",
    }, "GitHub connector event retained for retry"]]);
    warning.mockRestore();
+    releaseRefresh();
+    await expect(oldRefresh).resolves.toMatchObject({ refreshed: false, reason: "refresh_superseded" });
    const [snapshot] = await db.select().from(externalObjects).where(eq(externalObjects.id, externalObjectId));
    expect(snapshot).toMatchObject({
      statusKey: "merged",
@@ -184,6 +200,8 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => {
      statusCategory: "succeeded",
      statusTone: "success",
      isTerminal: true,
+      refreshToken: null,
+      refreshStartedAt: null,
      remoteVersion: "2026-09-04T11:59:01.000Z",
      data: expect.objectContaining({ marker: "preserved", merged: true, headRef: "feature", baseSha: "b".repeat(40) }),
    });
@@ -213,7 +213,8 @@ const {
  };
  const executionWorkspaceServiceFactoryMock = vi.fn(() => executionWorkspaceServiceMock);
  const externalObjectsServiceMock = {
-    refreshDueObjectsForActiveCompanies: vi.fn(async () => ({ companies: 0, checked: 0, refreshed: 0 })),
+    registerRefreshWork: vi.fn((coordinator: { register: typeof deliveryWorkRegisterMock }, _canRun: () => boolean) =>
+      coordinator.register("external-object-refresh", { run: async () => undefined, hasPending: async () => false })),
  };
  const externalObjectsServiceFactoryMock = vi.fn(() => externalObjectsServiceMock);
  const routineServiceMock = {
@@ -1484,10 +1485,12 @@ describe("startServer feedback export wiring", () => {
      await Promise.resolve();

      expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1);
-      expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1);
      expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalledTimes(1);
      expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled();
      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true);
+      expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledTimes(1);
+      expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledWith(expect.any(Object), expect.any(Function));
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.externalObjectRefresh)).toBe(true);
      expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1);
      expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1);
      const routineWorkerIndex = deliveryWorkRegisterMock.mock.calls.findIndex(([queue]) => queue === DELIVERY_QUEUES.routineSchedules);
@@ -1509,6 +1512,15 @@ describe("startServer feedback export wiring", () => {
    }
  });

+  it.each([true, false])("does not wait for external-object provider work before readiness (scheduler=%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    externalObjectsServiceMock.registerRefreshWork.mockReturnValueOnce({
+      ready: new Promise<void>(() => {}), wake: vi.fn(), stop: vi.fn(async () => {}),
+    });
+    await startServer();
+    expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledOnce();
+  });
+
  it("keeps external object refresh active when heartbeat scheduling is disabled", async () => {
    loadConfigMock.mockReturnValue(buildTestConfig({
      heartbeatSchedulerEnabled: false,
@@ -1536,7 +1548,9 @@ describe("startServer feedback export wiring", () => {
      await Promise.resolve();
      await Promise.resolve();

-      expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1);
+      expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledTimes(1);
+      expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledWith(expect.any(Object), expect.any(Function));
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.externalObjectRefresh)).toBe(true);
      expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled();
      expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
      expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled();
@@ -1325,19 +1325,6 @@ async function startServerWithDatabaseTeardown(
    pluginWorkerManager,
    enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true,
  });
-  const scheduleExternalObjectRefreshSweep = (now = new Date()) => {
-    if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
-    trackHeartbeatSchedulerWork(externalObjects
-      .refreshDueObjectsForActiveCompanies(50, now)
-      .then((result) => {
-        if (result.checked > 0 || result.refreshed > 0) {
-          logger.info({ ...result }, "external-object scheduler tick refreshed due objects");
-        }
-      })
-      .catch((err) => {
-        logger.error({ err }, "external-object scheduler tick failed");
-      }));
-  };

  // The retry backstop for orphan sandboxes. An acquire that rejects a
  // foreign-company insert tears the provisioned sandbox down. If that teardown
@@ -1364,6 +1351,8 @@ async function startServerWithDatabaseTeardown(
    resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction),
  });
  const deliveryWork = app.locals.deliveryWork as ReturnType<typeof createDeliveryWorkCoordinator>;
+  externalObjects.registerRefreshWork(deliveryWork,
+    () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby());
  if (config.nativeHarnessBackupExpirationEnabled) {
    nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork);
  }
@@ -1971,7 +1960,6 @@ async function startServerWithDatabaseTeardown(
        }

        if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
-        scheduleExternalObjectRefreshSweep(new Date());

        if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
        scheduleGitHubConnectionEventPoll();
@@ -2073,7 +2061,6 @@ async function startServerWithDatabaseTeardown(
    // as the enabled path.
    await startEnvironmentCleanup();
    startHeartbeatSchedulerInterval(() => {
-      scheduleExternalObjectRefreshSweep(new Date());
      scheduleGitHubConnectionEventPoll();
      scheduleGitHubConnectionContinuitySweep();
    });
@@ -1,5 +1,6 @@
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyApprovalWork } from "./approval-work-notifications.js";
 import { notifyAttentionRetentionWork, notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { addCents, compareCents, normalizeCents } from "@paperclipai/shared";
@@ -306,6 +307,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
    await notifyRunDispatch(db, { companyId: policy.companyId });
    await notifyCompanyRetentionWork(db, policy.companyId);
    await notifyStatusCardIssueChange(db, policy.companyId);
+    await notifyExternalObjectRefresh(db, policy.companyId);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors);
@@ -369,6 +371,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
    await notifyRunDispatch(db, { companyId: policy.companyId });
    await notifyCompanyRetentionWork(db, policy.companyId);
    await notifyStatusCardIssueChange(db, policy.companyId);
+    await notifyExternalObjectRefresh(db, policy.companyId);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors);
@@ -2,6 +2,7 @@ import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { isRunnerApiResponseAsset } from "./asset-content-access.js";
 import { deleteCompany } from "./company-deletion.js";
@@ -337,6 +338,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
          await notifyRunDispatch(tx, { companyId: id });
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
+          await notifyExternalObjectRefresh(tx, id);
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
          await notifyStatusCardIssueChange(tx, id);
@@ -495,6 +497,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
          await notifyRunDispatch(tx, { companyId: id });
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
+          await notifyExternalObjectRefresh(tx, id);
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
          await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
          await notifyStatusCardIssueChange(tx, id);
@@ -1,4 +1,5 @@
 import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
 import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
@@ -72,6 +73,7 @@ export async function deleteCompany(db: Db, id: string) {
        const [existing] = await tx.select({ id: companies.id }).from(companies)
          .where(eq(companies.id, id)).for("no key update");
        if (!existing) return null;
+        await notifyExternalObjectRefresh(tx, id);
        await notifyChatCompanyConfiguration(tx, id);
        await chatCompanyDeletion.deleteCompanyData(tx, id);
        await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
@@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = {
  providerTraceRetention: "provider-trace-retention",
  nativeRunnerEvidenceRetention: "native-runner-evidence-retention",
  skillVersionRetention: "skill-version-retention",
+  externalObjectRefresh: "external-object-refresh",
  databaseBackups: "database-backups",
  importSpoolCleanup: "import-spool-cleanup",
  externalConversationSettlement: "external-conversation-settlement",
@@ -0,0 +1,8 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type ExternalObjectRefreshIdentity = { providerKey: string; objectType: string; externalId: string; ignoreExternalIdCase?: boolean };
+
+export async function notifyExternalObjectRefresh(tx: object, companyId?: string, object?: string | ExternalObjectRefreshIdentity) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.externalObjectRefresh,
+    companyId ? JSON.stringify({ companyId, ...(typeof object === "string" ? { objectId: object } : object ? { identity: object } : {}) }) : "*");
+}
@@ -1,7 +1,7 @@
 import { randomUUID } from "node:crypto";
-import { and, asc, eq, inArray, isNull, lte, not, or, sql } from "drizzle-orm";
+import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, lte, not, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
-import { companies, documents, externalObjectMentions, externalObjects, issueComments, issueDocuments, issues, plugins } from "@paperclipai/db";
+import { companies, documents, externalObjectMentions, externalObjects, issueComments, issueDocuments, issues, plugins, pluginCompanySettings } from "@paperclipai/db";
 import {
  formatExternalObjectMentionSourceLabel,
  type ExternalObjectCanonicalUrl,
@@ -18,6 +18,10 @@ import { notFound } from "../errors.js";
 import { logger } from "../middleware/logger.js";
 import { logActivity, type LogActivityInput } from "./activity-log.js";
 import { createGitHubExternalObjectProvider, type GitHubExternalObjectProviderOptions } from "./github-external-object-provider.js";
+import { notifyExternalObjectRefresh, type ExternalObjectRefreshIdentity } from "./external-object-refresh-notifications.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
 import { publishLiveEvent } from "./live-events.js";
 import type { PluginWorkerManager } from "./plugin-worker-manager.js";

@@ -96,6 +100,18 @@ const DEFAULT_RETRY_AFTER_SECONDS = 300;
 const DEFAULT_REFRESH_LEASE_SECONDS = 300;
 const REFRESH_LEASE_RENEW_INTERVAL_MS = 60_000;

+type RefreshScope = { companyId: string; objectId?: string; identity?: ExternalObjectRefreshIdentity };
+type RefreshDeadline = { id: string; companyId: string; dueAt: number; providerKey: string; objectType: string; externalId: string };
+
+
+function refreshableObjectWhere() {
+  return or(eq(externalObjects.isTerminal, false), and(
+    eq(externalObjects.providerKey, "github"),
+    eq(externalObjects.statusKey, "not_found"),
+    not(sql`coalesce(${externalObjects.data}->>'probeVerified', 'false') = 'true'`),
+  ))!;
+}
+
 function sourceWhere(input: ExternalObjectSourceContext) {
  const conditions = [
    eq(externalObjectMentions.companyId, input.companyId),
@@ -246,7 +262,7 @@ function objectSnapshot(object: ExternalObjectRecord): PluginExternalObjectRecor
  };
 }

-async function readyObjectReferencePlugins(db: Db) {
+async function readyObjectReferencePlugins(db: Db, companyId?: string) {
  return db
    .select({
      id: plugins.id,
@@ -254,7 +270,12 @@ async function readyObjectReferencePlugins(db: Db) {
      manifestJson: plugins.manifestJson,
    })
    .from(plugins)
-    .where(eq(plugins.status, "ready"))
+    .where(and(eq(plugins.status, "ready"), companyId ? not(exists(db.select({ id: pluginCompanySettings.id })
+      .from(pluginCompanySettings).where(and(
+        eq(pluginCompanySettings.pluginId, plugins.id),
+        eq(pluginCompanySettings.companyId, companyId),
+        eq(pluginCompanySettings.enabled, false),
+      )))) : undefined))
    .orderBy(asc(plugins.installOrder))
    .then((rows) =>
      rows.filter((row) => (row.manifestJson.objectReferences?.length ?? 0) > 0),
@@ -268,7 +289,7 @@ function createPluginProviderDetector(
  return {
    key: "plugin-object-reference-providers",
    async detect(input) {
-      const providers = await readyObjectReferencePlugins(db);
+      const providers = await readyObjectReferencePlugins(db, input.companyId);
      const detections: ExternalObjectDetection[] = [];

      for (const provider of providers) {
@@ -327,7 +348,7 @@ async function resolveViaPluginProvider(
  object: ExternalObjectRecord,
 ): Promise<PluginExternalObjectResolveResult | null> {
  if (!pluginWorkerManager) return null;
-  const providers = await readyObjectReferencePlugins(db);
+  const providers = await readyObjectReferencePlugins(db, object.companyId);
  for (const provider of providers) {
    const manifest = provider.manifestJson;
    if (!manifest.capabilities.includes("external.objects.read")) continue;
@@ -375,10 +396,38 @@ export function externalObjectService(
    ...(opts.detectors ?? []),
    ...(githubProvider ? [githubProvider.detector] : []),
  ]);
-  const resolverRegistry = createExternalObjectResolverRegistry([
+  const resolvers = [
    ...(opts.resolvers ?? []),
    ...(githubProvider?.resolvers ?? []),
-  ]);
+  ];
+  const resolverRegistry = createExternalObjectResolverRegistry(resolvers);
+
+  function automaticRefreshWhere(database: Db = db) {
+    const supported = resolvers.map(resolver => and(
+      eq(externalObjects.providerKey, resolver.providerKey),
+      resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined,
+    )!);
+    if (opts.pluginWorkerManager) supported.push(exists(database.select({ id: plugins.id }).from(plugins).where(and(
+      eq(plugins.status, "ready"),
+      sql`${plugins.manifestJson}->'capabilities' ? 'external.objects.read'`,
+      sql`exists (select 1 from jsonb_array_elements(coalesce(${plugins.manifestJson}->'objectReferences', '[]'::jsonb)) as reference
+        where reference->>'providerKey' = ${externalObjects.providerKey}
+          and reference->'objectTypes' ? ${externalObjects.objectType})`,
+      not(exists(database.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings).where(and(
+        eq(pluginCompanySettings.pluginId, plugins.id),
+        eq(pluginCompanySettings.companyId, externalObjects.companyId),
+        eq(pluginCompanySettings.enabled, false),
+      )))),
+    ))));
+    return and(
+      refreshableObjectWhere(), isNotNull(externalObjects.nextRefreshAt),
+      supported.length ? or(...supported) : sql`false`,
+      exists(database.select({ id: externalObjectMentions.id }).from(externalObjectMentions).where(and(
+        eq(externalObjectMentions.companyId, externalObjects.companyId),
+        eq(externalObjectMentions.objectId, externalObjects.id),
+      ))),
+    )!;
+  }

  async function isEnabled() {
    if (typeof opts.enabled === "function") return await opts.enabled();
@@ -423,6 +472,7 @@ export function externalObjectService(
      nextRefreshAt: now,
      updatedAt: now,
    };
+    await notifyExternalObjectRefresh(dbOrTx, companyId, { providerKey: detection.providerKey, objectType: detection.objectType, externalId: detection.externalId });
    const inserted = await dbOrTx
      .insert(externalObjects)
      .values(values)
@@ -452,6 +502,9 @@ export function externalObjectService(
    dbOrTx: any = db,
  ) {
    const urls = extractExternalObjectCanonicalUrls(input.text ?? "");
+    const previous: Array<{ objectId: string | null }> = await dbOrTx.selectDistinct({ objectId: externalObjectMentions.objectId })
+      .from(externalObjectMentions).where(sourceWhere(input));
+    for (const row of previous) if (row.objectId) await notifyExternalObjectRefresh(dbOrTx, input.companyId, row.objectId);
    await dbOrTx.delete(externalObjectMentions).where(sourceWhere(input));
    if (urls.length === 0) return;

@@ -470,6 +523,7 @@ export function externalObjectService(
      if (seen.has(sourceKey)) continue;
      seen.add(sourceKey);
      const object = await upsertObjectFromDetection(input.companyId, detection, dbOrTx);
+      await notifyExternalObjectRefresh(dbOrTx, input.companyId, object.id);
      values.push({
        companyId: input.companyId,
        sourceIssueId: input.sourceIssueId,
@@ -521,6 +575,14 @@ export function externalObjectService(
    return dbOrTx === db ? db.transaction(runSync) : runSync(dbOrTx);
  }

+  async function deleteMissingSourceMentions(kind: "comment" | "document", recordId: string, tx: any) {
+    const source = and(eq(externalObjectMentions.sourceKind, kind), eq(externalObjectMentions.sourceRecordId, recordId));
+    const owners: Array<{ companyId: string; objectId: string | null }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId, objectId: externalObjectMentions.objectId })
+      .from(externalObjectMentions).where(source);
+    for (const owner of owners) if (owner.objectId) await notifyExternalObjectRefresh(tx, owner.companyId, owner.objectId);
+    await tx.delete(externalObjectMentions).where(source);
+  }
+
  async function syncComment(commentId: string, dbOrTx: any = db) {
    const runSync = async (tx: any) => {
      if (!(await isEnabled())) return;
@@ -535,9 +597,7 @@ export function externalObjectService(
        .where(eq(issueComments.id, commentId))
        .then((rows: Array<{ id: string; companyId: string; issueId: string; body: string }>) => rows[0] ?? null);
      if (!comment) {
-        await tx
-          .delete(externalObjectMentions)
-          .where(and(eq(externalObjectMentions.sourceKind, "comment"), eq(externalObjectMentions.sourceRecordId, commentId)));
+        await deleteMissingSourceMentions("comment", commentId, tx);
        return;
      }
      await replaceSourceMentions({
@@ -569,9 +629,7 @@ export function externalObjectService(
        .where(eq(documents.id, documentId))
        .then((rows: Array<{ documentId: string; companyId: string; issueId: string; key: string; body: string }>) => rows[0] ?? null);
      if (!document) {
-        await tx
-          .delete(externalObjectMentions)
-          .where(and(eq(externalObjectMentions.sourceKind, "document"), eq(externalObjectMentions.sourceRecordId, documentId)));
+        await deleteMissingSourceMentions("document", documentId, tx);
        return;
      }
      await replaceSourceMentions({
@@ -781,6 +839,8 @@ export function externalObjectService(
    companyId: string;
    actor?: Pick<LogActivityInput, "actorType" | "actorId" | "agentId" | "runId">;
    force?: boolean;
+    scheduled?: boolean;
+    signal?: AbortSignal;
    now?: Date;
  };

@@ -802,39 +862,48 @@ export function externalObjectService(
    return { object: toObjectPayload(latest, now), refreshed: false, reason: "refresh_superseded" as const };
  }

+  async function persistRefresh(object: ExternalObjectRecord, token: string,
+    patch: Partial<typeof externalObjects.$inferInsert>, scheduled = false) {
+    return db.transaction(async tx => {
+      await notifyExternalObjectRefresh(tx, object.companyId, object.id);
+      const rows = await tx.update(externalObjects).set(patch)
+        .where(and(refreshOwnerWhere(object, token), scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined)).returning();
+      if (scheduled && rows.length === 0) await tx.update(externalObjects)
+        .set({ refreshStartedAt: null, refreshToken: null }).where(refreshOwnerWhere(object, token));
+      return rows;
+    });
+  }
+
  async function resolveObjectRefresh(
    object: ExternalObjectRecord,
    input: RefreshObjectInput,
    now: Date,
    refreshToken: string,
  ) {
-    const pluginResult = await resolveViaPluginProvider(db, opts.pluginWorkerManager, object);
+    input.signal?.throwIfAborted();
+    const pluginResult = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal);
    const resolver = pluginResult ? null : resolverRegistry.find(object);
    if (!pluginResult && !resolver) {
-      const [updated] = await db
-        .update(externalObjects)
-        .set({
+      const [updated] = await persistRefresh(object, refreshToken, {
          liveness: object.isTerminal ? object.liveness : visibleLiveness(object, now) === "fresh" ? "stale" : object.liveness,
          nextRefreshAt: addSeconds(now, DEFAULT_RETRY_AFTER_SECONDS),
          refreshStartedAt: null,
          refreshToken: null,
          updatedAt: now,
-        })
-        .where(refreshOwnerWhere(object, refreshToken))
-        .returning();
+        }, input.scheduled);
      if (!updated) return refreshSupersededResult(object, now);
      return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const };
    }

-    const result = pluginResult ?? await resolver!.resolve({ companyId: object.companyId, object });
+    const result = pluginResult ?? await observeRefreshCancellation(
+      resolver!.resolve({ companyId: object.companyId, object, signal: input.signal }), input.signal);
+    input.signal?.throwIfAborted();
    if (!result.ok) {
      // A failed refresh means the object's status is currently unknown, not
      // terminal. Clear a previous terminal verdict so the object keeps being
      // scheduled for automatic refresh and self-heals (e.g. a GitHub "not
      // found" that later resolves once repo access improves).
-      const [updated] = await db
-        .update(externalObjects)
-        .set({
+      const [updated] = await persistRefresh(object, refreshToken, {
          liveness: result.liveness,
          isTerminal: false,
          lastErrorAt: now,
@@ -844,9 +913,7 @@ export function externalObjectService(
          refreshStartedAt: null,
          refreshToken: null,
          updatedAt: now,
-        })
-        .where(refreshOwnerWhere(object, refreshToken))
-        .returning();
+        }, input.scheduled);
      if (!updated) return refreshSupersededResult(object, now);
      publishLiveEvent({
        companyId: object.companyId,
@@ -880,14 +947,10 @@ export function externalObjectService(
      refreshToken: null,
      updatedAt: now,
    };
-    const [updated] = await db
-      .update(externalObjects)
-      .set({
+    const [updated] = await persistRefresh(object, refreshToken, {
          ...patch,
          lastChangedAt: objectChanged(object, { ...object, ...patch }) ? now : object.lastChangedAt,
-        })
-      .where(refreshOwnerWhere(object, refreshToken))
-      .returning();
+        }, input.scheduled);
    if (!updated) return refreshSupersededResult(object, now);
    const next = updated ?? object;
    if (objectChanged(object, next) && input.actor) {
@@ -948,6 +1011,8 @@ export function externalObjectService(
        eq(externalObjects.id, object.id),
        eq(externalObjects.companyId, object.companyId),
        dueOrForced,
+        input.scheduled ? automaticRefreshWhere() : undefined,
+        input.scheduled ? sql`exists (select 1 from ${companies} where ${companies.id} = ${externalObjects.companyId} and ${companies.status} = 'active')` : undefined,
      ))
      .returning();
    return claimed ?? null;
@@ -970,6 +1035,16 @@ export function externalObjectService(
    };
  }

+  function observeRefreshCancellation<T>(work: Promise<T>, signal?: AbortSignal): Promise<T> {
+    if (!signal) return work;
+    return new Promise((resolve, reject) => {
+      const abort = () => reject(signal.reason);
+      work.then(value => { signal.removeEventListener("abort", abort); resolve(value); },
+        error => { signal.removeEventListener("abort", abort); reject(error); });
+      if (signal.aborted) abort(); else signal.addEventListener("abort", abort, { once: true });
+    });
+  }
+
  const objectRefreshesInFlight = new Map<string, Promise<Awaited<ReturnType<typeof resolveObjectRefresh>>>>();

  async function refreshObject(
@@ -989,7 +1064,7 @@ export function externalObjectService(

    const refreshKey = `${object.companyId}:${object.id}`;
    const existingRefresh = objectRefreshesInFlight.get(refreshKey);
-    if (existingRefresh) return existingRefresh;
+    if (existingRefresh) return observeRefreshCancellation(existingRefresh, input.signal);

    const claimed = await claimObjectRefresh(object, input, now);
    if (!claimed) {
@@ -1006,7 +1081,10 @@ export function externalObjectService(
    }

    const stopRenewingRefreshLease = startRefreshLeaseRenewal(object, claimed.refreshToken);
-    const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).finally(() => {
+    const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).catch(async error => {
+      if (input.signal?.aborted) await persistRefresh(object, claimed.refreshToken!, { refreshStartedAt: null, refreshToken: null });
+      throw error;
+    }).finally(() => {
      stopRenewingRefreshLease();
      objectRefreshesInFlight.delete(refreshKey);
    });
@@ -1031,73 +1109,66 @@ export function externalObjectService(
    return results;
  }

-  async function refreshDueObjectsUnchecked(companyId: string, limit = 50, now = new Date()) {
-    const staleRefreshStartedBefore = new Date(now.getTime() - DEFAULT_REFRESH_LEASE_SECONDS * 1000);
-    // Terminal objects are usually skipped forever, but a GitHub "not found"
-    // verdict is only trustworthy while the repository is actually visible to
-    // the configured credentials. GitHub hides inaccessible (often private)
-    // repos behind a 404, so a not-found object can become resolvable later
-    // when credentials or repo visibility improve. Re-check objects that were
-    // frozen as "not found" before the repository probe existed so they
-    // self-heal instead of freezing. Once the provider has probed the repo and
-    // confirmed it is visible (data.probeVerified = "true"), the not-found
-    // verdict is definitive, so stop re-polling that object to avoid burning
-    // the company's GitHub API budget on deleted references forever.
-    const probeVerifiedNotFound = sql`coalesce(${externalObjects.data}->>'probeVerified', 'false') = 'true'`;
-    const refreshable = or(
-      eq(externalObjects.isTerminal, false),
-      and(
-        eq(externalObjects.isTerminal, true),
-        eq(externalObjects.providerKey, "github"),
-        eq(externalObjects.statusKey, "not_found"),
-        not(probeVerifiedNotFound),
-      ),
-    )!;
-    const due = await db
-      .select({ id: externalObjects.id })
-      .from(externalObjects)
-      .where(
-        and(
-          eq(externalObjects.companyId, companyId),
-          refreshable,
-          lte(externalObjects.nextRefreshAt, now),
-          or(
-            isNull(externalObjects.refreshStartedAt),
-            lte(externalObjects.refreshStartedAt, staleRefreshStartedBefore),
-          ),
-        ),
-      )
-      .limit(limit);
-    const results = [];
-    for (const row of due) {
-      results.push(await refreshObject(row.id, {
-        companyId,
-        actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null },
-        now,
-      }));
+  function registerRefreshWork(
+    coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+    canRun: () => boolean,
+  ) {
+    const cursors = new Map<string, string>();
+    const scopeKey = (scope?: RefreshScope) => JSON.stringify(scope ?? null);
+    async function load(scope?: RefreshScope) {
+      if (!(await isEnabled())) { cursors.clear(); return []; }
+      const key = scopeKey(scope);
+      const cursor = cursors.get(key);
+      const rows = await db.select({ id: externalObjects.id, companyId: externalObjects.companyId,
+        nextRefreshAt: externalObjects.nextRefreshAt, refreshStartedAt: externalObjects.refreshStartedAt,
+        providerKey: externalObjects.providerKey, objectType: externalObjects.objectType, externalId: externalObjects.externalId }).from(externalObjects)
+        .innerJoin(companies, eq(companies.id, externalObjects.companyId))
+        .where(and(
+          eq(companies.status, "active"), automaticRefreshWhere(),
+          scope ? eq(externalObjects.companyId, scope.companyId) : undefined,
+          scope?.objectId ? eq(externalObjects.id, scope.objectId) : undefined,
+          scope?.identity ? and(eq(externalObjects.providerKey, scope.identity.providerKey),
+            eq(externalObjects.objectType, scope.identity.objectType), scope.identity.ignoreExternalIdCase
+              ? sql`lower(${externalObjects.externalId}) = lower(${scope.identity.externalId})`
+              : eq(externalObjects.externalId, scope.identity.externalId)) : undefined,
+          cursor ? gt(externalObjects.id, cursor) : undefined,
+        )).orderBy(asc(externalObjects.id)).limit(250);
+      const entries = rows.map(deadline);
+      if (rows.length === 250) cursors.set(key, rows[rows.length - 1]!.id);
+      else cursors.delete(key);
+      return { rows: entries, complete: rows.length < 250 };
    }
-    return results;
-  }
-
-  async function refreshDueObjects(companyId: string, limit = 50, now = new Date()) {
-    if (!(await isEnabled())) return [];
-    return refreshDueObjectsUnchecked(companyId, limit, now);
-  }
-
-  async function refreshDueObjectsForActiveCompanies(limitPerCompany = 50, now = new Date()) {
-    if (!(await isEnabled())) return { companies: 0, checked: 0, refreshed: 0 };
-    const activeCompanies = await db
-      .select({ id: companies.id })
-      .from(companies)
-      .where(eq(companies.status, "active"));
-    let checked = 0;
-    let refreshed = 0;
-    for (const company of activeCompanies) {
-      const results = await refreshDueObjectsUnchecked(company.id, limitPerCompany, now);
-      checked += results.length;
-      refreshed += results.filter((result) => result.refreshed).length;
+    function deadline(object: Pick<ExternalObjectRecord, "id" | "companyId" | "nextRefreshAt" | "refreshStartedAt" | "providerKey" | "objectType" | "externalId">): RefreshDeadline {
+      return { id: object.id, companyId: object.companyId, providerKey: object.providerKey, objectType: object.objectType, externalId: object.externalId,
+        dueAt: Math.max(object.nextRefreshAt!.getTime(), object.refreshStartedAt
+          ? object.refreshStartedAt.getTime() + DEFAULT_REFRESH_LEASE_SECONDS * 1000 : 0) };
    }
-    return { companies: activeCompanies.length, checked, refreshed };
+    return registerScopedDeadlineWork<RefreshDeadline, RefreshScope>(coordinator, {
+      queue: DELIVERY_QUEUES.externalObjectRefresh,
+      batchSize: 50,
+      minimumPassIntervalMs: 0,
+      decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope,
+      matches: (entry, scope) => !scope || (entry.companyId === scope.companyId
+        && (!scope.objectId || entry.id === scope.objectId)
+        && (!scope.identity || (entry.providerKey === scope.identity.providerKey && entry.objectType === scope.identity.objectType
+          && (scope.identity.ignoreExternalIdCase ? entry.externalId.toLowerCase() === scope.identity.externalId.toLowerCase()
+            : entry.externalId === scope.identity.externalId)))),
+      load,
+      check: async (entry, signal) => {
+        if (signal?.aborted || !canRun()) return entry;
+        const current = await load({ companyId: entry.companyId, objectId: entry.id });
+        const scheduled = Array.isArray(current) ? current[0] : current.rows[0];
+        if (!scheduled || scheduled.dueAt > Date.now()) return scheduled ?? null;
+        await refreshObject(entry.id, {
+          companyId: entry.companyId,
+          scheduled: true,
+          signal,
+          actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null },
+        });
+        const updated = await load({ companyId: entry.companyId, objectId: entry.id });
+        return (Array.isArray(updated) ? updated[0] : updated.rows[0]) ?? null;
+      },
+    }, canRun);
  }

  return {
@@ -1113,7 +1184,6 @@ export function externalObjectService(
    getProjectSummary,
    refreshObject,
    refreshIssueObjects,
-    refreshDueObjects,
-    refreshDueObjectsForActiveCompanies,
+    registerRefreshWork,
  };
 }
@@ -1,3 +1,4 @@
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
 import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js";
 import { chatEndpoints } from "@paperclipai/db";
@@ -244,25 +245,31 @@ export function githubConnectionEventService(
    const snapshot = githubSnapshotUpdate(event.payload);
    if (!snapshot) return;
    const appliedAt = now();
-    await db.update(externalObjects).set({
-      statusKey: snapshot.statusKey,
-      statusLabel: snapshot.statusLabel,
-      statusCategory: snapshot.statusCategory,
-      statusTone: snapshot.statusTone,
-      statusIconKey: snapshot.statusIconKey,
-      isTerminal: snapshot.merged || snapshot.state === "closed",
-      data: sql`${externalObjects.data} || ${JSON.stringify(snapshot.data)}::jsonb`,
-      remoteVersion: snapshot.remoteVersion,
-      lastResolvedAt: appliedAt,
-      lastChangedAt: appliedAt,
-      nextRefreshAt: appliedAt,
-      updatedAt: appliedAt,
-    }).where(and(
-      eq(externalObjects.companyId, companyId),
-      eq(externalObjects.providerKey, "github"),
-      eq(externalObjects.objectType, "pull_request"),
-      sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`,
-    ));
+    await db.transaction(async tx => {
+      await notifyExternalObjectRefresh(tx, companyId, { providerKey: "github", objectType: "pull_request",
+        externalId: snapshot.externalId, ignoreExternalIdCase: true });
+      await tx.update(externalObjects).set({
+        statusKey: snapshot.statusKey,
+        statusLabel: snapshot.statusLabel,
+        statusCategory: snapshot.statusCategory,
+        statusTone: snapshot.statusTone,
+        statusIconKey: snapshot.statusIconKey,
+        isTerminal: snapshot.merged || snapshot.state === "closed",
+        data: sql`${externalObjects.data} || ${JSON.stringify(snapshot.data)}::jsonb`,
+        remoteVersion: snapshot.remoteVersion,
+        lastResolvedAt: appliedAt,
+        lastChangedAt: appliedAt,
+        nextRefreshAt: appliedAt,
+        refreshStartedAt: null,
+        refreshToken: null,
+        updatedAt: appliedAt,
+      }).where(and(
+        eq(externalObjects.companyId, companyId),
+        eq(externalObjects.providerKey, "github"),
+        eq(externalObjects.objectType, "pull_request"),
+        sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`,
+      ));
+    });
    if (snapshot.merged && event.action === "closed") {
      await issueThreadInteractionService(db, { wakeup: options.wakeup })
        .sweepMergedPullRequestConfirmations([{
@@ -15,6 +15,7 @@ import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
 import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
 import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { companies, instanceSettings } from "@paperclipai/db";
@@ -631,6 +632,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
        await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
        await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
        await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
+        if (patch.enableExternalObjects !== undefined) await notifyExternalObjectRefresh(tx);
        if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
        if (patch.enableWorktreeRunExecution !== undefined) {
          await notifyRunDispatch(tx);
@@ -3,6 +3,7 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js";
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
@@ -11986,6 +11987,7 @@ export function issueService(db: Db) {
          if (owner) {
            await notifyAttentionRetentionWork(tx, { companyId: owner.companyId });
            await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
+            await notifyExternalObjectRefresh(tx, owner.companyId);
            await notifyApprovalWork(tx, owner.companyId);
            await notifyStatusCardIssueChange(tx, owner.companyId);
          }
@@ -1,3 +1,4 @@
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { isDeepStrictEqual } from "node:util";
 import { writePluginJobChange } from "./plugin-job-work-notifications.js";
 import { asc, eq, isNull, ne, sql, and } from "drizzle-orm";
@@ -239,12 +240,15 @@ export function pluginRegistryService(db: Db) {
        setClause.categories = data.manifest.categories;
      }

-      return db
-        .update(plugins)
-        .set(setClause)
-        .where(eq(plugins.id, id))
-        .returning()
-        .then((rows) => rows[0] ?? null);
+      return db.transaction(async tx => {
+        await notifyExternalObjectRefresh(tx);
+        return tx
+          .update(plugins)
+          .set(setClause)
+          .where(eq(plugins.id, id))
+          .returning()
+          .then((rows) => rows[0] ?? null);
+      });
    },

    // ----- Status ---------------------------------------------------------
@@ -254,16 +258,19 @@ export function pluginRegistryService(db: Db) {
      const plugin = await getById(id);
      if (!plugin) throw notFound("Plugin not found");

-      return db
-        .update(plugins)
-        .set({
-          status: input.status,
-          lastError: input.lastError ?? null,
-          updatedAt: new Date(),
-        })
-        .where(eq(plugins.id, id))
-        .returning()
-        .then((rows) => rows[0] ?? null);
+      return db.transaction(async tx => {
+        await notifyExternalObjectRefresh(tx);
+        return tx
+          .update(plugins)
+          .set({
+            status: input.status,
+            lastError: input.lastError ?? null,
+            updatedAt: new Date(),
+          })
+          .where(eq(plugins.id, id))
+          .returning()
+          .then((rows) => rows[0] ?? null);
+      });
    },

    // ----- Uninstall / Remove --------------------------------------------
@@ -280,6 +287,7 @@ export function pluginRegistryService(db: Db) {
      if (!plugin) throw notFound("Plugin not found");

      return writePluginJobChange(db, async db => {
+        await notifyExternalObjectRefresh(db);
        if (removeData) {
          // Hard delete – plugin_config cascades via FK onDelete
          return db
@@ -343,6 +351,7 @@ export function pluginRegistryService(db: Db) {
      return db.transaction(async tx => {
        const db = tx as unknown as Db;
        await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle);
+        await notifyExternalObjectRefresh(db, companyId);
        const plugin = await getById(pluginId, db);
        if (!plugin) throw notFound("Plugin not found");

@@ -385,6 +394,7 @@ export function pluginRegistryService(db: Db) {
      return db.transaction(async tx => {
        const db = tx as unknown as Db;
        await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle);
+        await notifyExternalObjectRefresh(db, companyId);
        const plugin = await getById(pluginId, db);
        if (!plugin) throw notFound("Plugin not found");

@@ -467,6 +477,7 @@ export function pluginRegistryService(db: Db) {
      return db.transaction(async tx => {
        const db = tx as unknown as Db;
        await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle);
+        await notifyExternalObjectRefresh(db, companyId);
        const plugin = await getById(pluginId, db);
        if (!plugin) throw notFound("Plugin not found");