perf(external-objects): schedule scoped refresh obligations (#18384)

Reviewed and verified at 3e58e12f87. Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-11-01 06:42:06 +01:00 · 2026-10-31 14:09:24 -05:00
1 parent 5bc029bf6c
commit 453f5fbb78
@@ -0,0 +1,145 @@
+# External object refresh deadlines
+
+External-object refresh uses the delivery coordinator. The server registers the
+service in both heartbeat-scheduler configurations. It no longer calls the
+company-wide aggregate sweep on each heartbeat tick.
+
+`external-objects.ts` owns automatic eligibility, resolver selection, refresh
+claims, provider results and the next deadline. `index.ts` supplies registration
+and admission gates. The existing scoped deadline helper owns wake coalescing,
+page reconciliation, fair company selection, retry backoff and shutdown.
+
+## Durable obligations and quiet states
+
+An automatic refresh needs an active company, a retained mention in that company,
+a non-null `next_refresh_at` and an available resolver. Injected resolvers,
+including custom `url/link` resolvers, remain supported. A plugin resolver needs a
+ready plugin with `external.objects.read`, a matching provider/object declaration,
+and no company setting that disables it. Built-in resolvers remain available
+independently of optional plugin support.
+
+A plugin result carries its selected plugin ID inside the host owner. Persistence
+rechecks that exact plugin's readiness, read capability, object declaration and
+same-company enabled setting in the result update. Another available resolver
+cannot authorize an obsolete plugin response. A rejected response releases only
+its own claim; the due row remains for a fresh allowed resolver. Plugin RPC
+payloads and public result shapes do not contain this host selection context.
+
+
+Terminal objects are passive. The existing exception remains: GitHub
+`not_found` snapshots without `data.probeVerified=true` need a repository-visibility
+probe before their terminal verdict is definitive. A null deadline retains its
+existing never-refresh meaning. Registration does not change stored timestamps.
+
+The worker reconstructs eligible deadlines in pages of at most 250 rows. A
+company notification reads that company only. Ordinary ingestion uses object IDs
+or stable provider/object/external identity hints, including concurrent upserts,
+so one mention write does not reconstruct all company history. A due callback reads only its
+object. Pages select scheduling columns and retain compact deadlines in memory.
+Dispatch uses fair batches of at most 50 objects. The deadline is the later of
+`next_refresh_at` and the current refresh lease expiry. Lease renewals and atomic
+refresh tokens continue to exclude another claimant.
+
+Unsupported generic links, orphan objects, disabled plugins, empty configuration
+and retained terminal history do not schedule fallback discovery. Manual refresh
+retains its existing `no_resolver` result. Actual provider failures retain their
+saved retry deadline. Unexpected failures retain scoped coordinator backoff;
+restart reconstructs the saved object and any unexpired claim.
+
+The supported-resolver and retained-reference approach follows the related draft
+[PR #17298](https://github.com/paperclipai/paperclip/pull/17298) by
+[@kaluli123123](https://github.com/kaluli123123). This change adapts that eligibility
+pattern to scoped scheduling, adds dynamic plugin/company permission predicates,
+and fences automatic result persistence. It does not modify that draft PR.
+
+## Producer and transaction audit
+
+| Mutation | Owner and activation |
+| --- | --- |
+| Issue title/description, comment and document references | Existing sync methods register object intent before mention replacement and stable identity intent before object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. |
+| Comment/document disappearance | Sync derives the retained mention owners and registers only those companies before deleting obsolete mentions. |
+| Last mention removal and reattachment | Mention replacement invalidates company deadlines. Retained objects stay available for reattachment. |
+| Issue deletion | The issue service selects distinct mentioned object IDs by company and `sourceIssueId`, then registers only those IDs before the cascade. The source owner includes title, description, comments and documents. |
+| Company status/archive/deletion, including budget pause/resume | Company and budget services register company reconstruction before status or deletion writes. |
+| Feature enable/disable | Instance settings registers global reconstruction before the experimental-settings write. |
+| Plugin manifest/status/uninstall | Plugin registry registers reconstruction before lifecycle writes. Installed plugins acquire support through the ordinary ready transition. |
+| Plugin company configuration and enabled setting | Registry writes register only the affected company. There is no plugin-readiness poll. |
+| Manual and automatic resolver result | Result persistence registers object intent in the same transaction. Automatic persistence rechecks reference and resolver eligibility. Lost support releases the worker's own claim without publishing the stale result. |
+| Normalized GitHub ingress | The event service registers company intent with the snapshot write. It clears the old refresh token, so an older provider response cannot overwrite that event. |
+| Import and restore | No independent object-table importer exists in the source baseline. Supported reference ingestion uses the ordinary sync boundary. Startup reconstructs restored durable objects. |
+
+Notifications are awaited before writes. Caller-owned and nested transactions
+retain the outer commit boundary through the existing database-work protocol.
+Rollback does not publish the write. Notifications received during a scan or an
+active resolver remain dirty until a successful reconciliation.
+
+The supported live writer model is one shared `createDb` signal owner, including
+its transactions and child pools. These process-local signals do not cover
+independent SQL clients or other replicas. Such writers must invoke the supported
+service boundary or arrange explicit reconstruction/restart. Existing normalized
+GitHub ingress uses the supported boundary. No permanent database listener or
+periodic global safety scan is added.
+
+## Acceptance evidence
+
+The focused suite uses real PostgreSQL with a controlled simulated provider.
+Ordinary sync and registry/settings writes create work; the fixture does not
+pre-create the tested resolved outcome. Assertions independently read saved
+objects. Tests measure owner reads and provider calls across former tick windows.
+
+Coverage includes empty and terminal quiet, unsupported links, orphan removal and
+reattachment, feature/company eligibility, plugin readiness and permission loss
+during a held provider call, exact and earlier deadlines, durable provider retry,
+rollback and outer commit after nested savepoint release, startup and paginated
+reconstruction, inactive companies, never-refresh rows, first work during drain,
+warm standby, failed refresh after restart, lease renewal, duplicate exclusion,
+and shutdown while a provider call remains active. The GitHub ingress suite also
+checks a real normalized event superseding a held resolver response.
+
+The conservative external-object host-sleep blocker remains. An in-process
+deadline cannot wake stopped compute. This change proves worker quiet and
+reactivation in the tested process; it does not establish Cloud alarm coverage,
+host/database suspension or deployed savings.
+
+
+## Host cancellation and registered ownership
+
+The resolver contract accepts an optional host `AbortSignal`. The worker passes
+its coordinator signal through refresh to the resolver. This reuses the narrow
+provider-owned capability introduced with merge-confirmation work: GitHub combines
+host cancellation with a ten-second request budget for real fetch, response-body
+reads and visibility probes. Plugin RPC payloads remain unchanged. The host
+observes cancellation while retaining handlers for any eventual plugin/custom
+result; it releases only its own claim and cannot publish that late result.
+Startup does not await provider work. Shutdown closes the active GitHub HTTP
+request, and its saved due row can resume immediately after restart.
+
+The old `refreshDueObjects*` batch helpers are removed. Existing selection,
+lease, concurrent manual refresh, terminal-history and not-found recovery tests
+now enter through the registered scheduler owner. The removed baseline control
+recorded six SQL reads over three empty aggregate ticks; the current quiet-work
+acceptance measures zero recurring SQL after settlement. The scheduled check retains its
+post-refresh scoped read to revalidate current company, reference and resolver
+eligibility and recover an authoritative lease after a competing writer; all such
+reads target one object rather than rebuilding a company.
+
+### Repeated eligibility changes
+
+Company status and company plugin settings affect every eligible object. Their
+company-wide reconstruction hints now coalesce for 30 seconds after a completed
+reconstruction. This is a deadline for a newly received dirty hint, not a periodic
+scan. Object and provider-identity hints remain immediate; already known object
+deadlines still execute on time and re-read current company/provider eligibility.
+Reactivation after a second broad change can wait up to 30 seconds to reconstruct
+previously removed future obligations. Startup and instance-wide feature changes
+retain their existing behavior.
+
+The shared scoped worker accepts an owner-selected reconstruction interval for
+its typed scope; existing numeric callers retain their policy. It does not know
+companies or provider identities. A real PostgreSQL regression with 1,100 future
+objects reproduces five immediate page reads after a status toggle before the
+fix. With the fix, five repeated status writes retain one owed reconstruction,
+then execute exactly one company-scoped five-page pass. Twenty quiet minutes add
+no reads or provider calls. A scheduler control proves an exact object hint and a
+250ms saved deadline still preempt the broad reconstruction. All 76 affected
+external-object, route, and scheduler tests and server types pass.
@@ -0,0 +1,645 @@
+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 { and, eq, inArray, 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 { issueService } from "../services/issues.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.each(["company setting", "status", "uninstall", "manifest", "other company"] as const)(
+    "revalidates the selected plugin after %s changes even while the built-in fallback remains available", async change => {
+      clock(); const registry = pluginRegistryService(db);
+      const manifest: PaperclipPluginManifestV1 = {
+        id: `paperclip.selected-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Selected provider",
+        description: "Selected resolver fixture", author: "Paperclip", categories: ["connector"],
+        capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" },
+        objectReferences: [{ providerKey: "github", displayName: "GitHub", objectTypes: ["pull_request"], urlPatterns: ["https://github.com/:owner/:repo/pull/:id"] }],
+      };
+      const plugin = await registry.install({ packageName: "selected-provider-fixture" }, manifest);
+      await registry.updateStatus(plugin!.id, { status: "ready" });
+      let finishPlugin!: (result: ExternalObjectResolveResult) => void;
+      const pluginResult = new Promise<ExternalObjectResolveResult>(resolve => { finishPlugin = resolve; });
+      let selected = false;
+      const calls = vi.fn(async (_id: string, method: string) => {
+        if (method === "detectExternalObjects") return { detections: [] };
+        selected = true; return pluginResult;
+      });
+      let finishFallback!: () => void;
+      const fallbackResult = new Promise<void>(resolve => { finishFallback = resolve; });
+      const fallback = vi.fn(async () => {
+        await fallbackResult;
+        return new Response(JSON.stringify({ state: "closed", merged: true, title: "Allowed fallback" }), { status: 200 });
+      });
+      const svc = externalObjectService(db, { pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager,
+        github: { fetch: fallback, 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(() => selected);
+        if (change === "company setting") await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} });
+        else if (change === "status") await registry.updateStatus(plugin!.id, { status: "error" });
+        else if (change === "uninstall") await registry.uninstall(plugin!.id, true);
+        else if (change === "manifest") await registry.update(plugin!.id, { manifest: { ...manifest, capabilities: [], objectReferences: [] } });
+        else {
+          const other = await fixture();
+          await registry.upsertCompanySettings(plugin!.id, other.companyId, { enabled: false, settingsJson: {} });
+        }
+        finishPlugin({ ok: true, snapshot: { statusKey: "plugin_result", statusCategory: "succeeded", statusTone: "success", isTerminal: true } });
+        if (change === "other company") {
+          await settle(async () => (await row(f.companyId))?.statusKey === "plugin_result"); await quiet();
+          expect(fallback).not.toHaveBeenCalled();
+        } else {
+          await settle(async () => fallback.mock.calls.length > 0 || Boolean((await row(f.companyId))?.lastResolvedAt));
+          const beforeFallback = await row(f.companyId);
+          expect(beforeFallback.isTerminal).toBe(false); expect(beforeFallback.statusKey).toBeNull(); expect(beforeFallback.lastResolvedAt).toBeNull();
+          expect(fallback).toHaveBeenCalledOnce();
+          finishFallback();
+          await settle(async () => (await row(f.companyId))?.statusKey === "merged"); await quiet();
+          expect((await row(f.companyId)).displayTitle).toBe("acme/app#42: Allowed fallback");
+        }
+      } finally { finishPlugin(terminal); finishFallback(); }
+    });
+
+  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("coalesces repeated company status writes without rebuilding every future object", async () => {
+    clock(); const svc = service(); const own = await fixture(), foreign = await fixture();
+    const future = new Date(Date.now() + 86_400_000);
+    for (let offset = 0; offset < 1100; offset += 250) {
+      const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID());
+      await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId,
+        providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future })));
+      await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId,
+        sourceIssueId: own.issueId, sourceKind: "description" as const, objectId })));
+    }
+    await svc.syncIssue(foreign.issueId);
+    await db.update(externalObjects).set({ nextRefreshAt: future });
+    await start(svc);
+    await settle(() => coordinator!.nextWakeAt() === future.getTime());
+    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 {
+      const companiesSvc = companyService(db);
+      await companiesSvc.update(own.companyId, { status: "paused" });
+      await settle(() => reads.length === 1 && coordinator!.nextWakeAt() === future.getTime()
+        && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+      for (const status of ["active", "paused", "active", "paused", "active"] as const) {
+        await companiesSvc.update(own.companyId, { status });
+        await settle(() => !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)
+          && (coordinator!.nextWakeAt() ?? 0) > Date.now());
+        expect(reads).toHaveLength(1);
+      }
+      const rebuildAt = coordinator!.nextWakeAt()!;
+      expect(rebuildAt - Date.now()).toBeGreaterThan(20_000);
+      expect(rebuildAt - Date.now()).toBeLessThanOrEqual(30_000);
+      await vi.advanceTimersByTimeAsync(rebuildAt - Date.now());
+      await settle(() => coordinator!.nextWakeAt() === future.getTime() && reads.length === 6);
+      for (const read of reads) {
+        expect(read.query).toContain('"external_objects"."company_id" =');
+        expect(read.params).toContain(own.companyId);
+        expect(read.params).not.toContain(foreign.companyId);
+        expect(read.params.at(-1)).toBe(250);
+      }
+      const count = reads.length;
+      await vi.advanceTimersByTimeAsync(20 * 60_000);
+      expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled();
+      expect(coordinator!.nextWakeAt()).toBe(future.getTime());
+      expect(errors).not.toHaveBeenCalled();
+    } finally { db.$client.options.debug = debug; }
+  });
+
+  it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => {
+    clock(); const svc = service();
+    const own = await fixture(), foreign = await fixture();
+    const future = new Date(Date.now() + 86_400_000), removedDeadline = new Date(Date.now() + 1_800_000);
+    const sharedUrl = "https://example.com/shared-issue-object";
+    const [retained] = await db.insert(issues).values({ companyId: own.companyId, title: "Retained",
+      description: sharedUrl }).returning();
+    await svc.syncIssue(retained!.id);
+    await db.update(issues).set({ description: sharedUrl }).where(eq(issues.id, foreign.issueId));
+    await svc.syncIssue(foreign.issueId);
+    const deletions: Array<{ id: string; affected: string[]; orphaned: string[] }> = [];
+    for (let index = 0; index < 3; index++) {
+      const [issue] = await db.insert(issues).values({ companyId: own.companyId,
+        title: sharedUrl, description: `https://example.com/deleted-${index}` }).returning();
+      await svc.syncIssue(issue!.id);
+      const [comment] = await db.insert(issueComments).values({ companyId: own.companyId, issueId: issue!.id,
+        body: `https://example.com/comment-${index}` }).returning();
+      await svc.syncComment(comment!.id);
+      const doc = await documentService(db).upsertIssueDocument({ issueId: issue!.id, key: "notes", format: "markdown",
+        body: `https://example.com/document-${index}` });
+      await svc.syncDocument(doc.document.id);
+      const mentions = await db.select().from(externalObjectMentions).where(and(
+        eq(externalObjectMentions.companyId, own.companyId), eq(externalObjectMentions.sourceIssueId, issue!.id)));
+      expect(new Set(mentions.map(mention => mention.sourceKind))).toEqual(new Set(["title", "description", "comment", "document"]));
+      deletions.push({ id: issue!.id, affected: [...new Set(mentions.map(mention => mention.objectId!))],
+        orphaned: mentions.filter(mention => mention.sourceKind !== "title").map(mention => mention.objectId!) });
+    }
+    const shared = (await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.sourceIssueId, retained!.id)))[0]!.objectId!;
+    for (let offset = 0; offset < 1100; offset += 250) {
+      const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID());
+      await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId,
+        providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future })));
+      await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId,
+        sourceIssueId: own.issueId, sourceKind: "description" as const, objectId })));
+    }
+    await db.update(externalObjects).set({ nextRefreshAt: future });
+    for (const entry of deletions) for (const id of entry.orphaned)
+      await db.update(externalObjects).set({ nextRefreshAt: removedDeadline }).where(eq(externalObjects.id, id));
+    await start(svc);
+    await settle(() => coordinator!.nextWakeAt() === removedDeadline.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+    expect(provider).not.toHaveBeenCalled();
+    const reads: Array<{ query: string; params: unknown[] }> = [];
+    const selections: 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] });
+      if (/^select distinct/i.test(query) && query.includes('from "external_object_mentions"'))
+        selections.push({ query, params: [...params] });
+    };
+    try {
+      for (let index = 0; index < deletions.length; index++) {
+        const entry = deletions[index], first = reads.length;
+        const result = await issueService(db).remove(entry.id);
+        expect(result?.id).toBe(entry.id);
+        const next = index === deletions.length - 1 ? future : removedDeadline;
+        await settle(() => reads.length > first && coordinator!.nextWakeAt() === next.getTime()
+          && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+        const deletionReads = reads.slice(first);
+        expect(deletionReads.length).toBeLessThanOrEqual(entry.affected.length);
+        for (const read of deletionReads) {
+          expect(read.query).toContain('"external_objects"."id" =');
+          expect(read.params).toContain(own.companyId); expect(read.params).not.toContain(foreign.companyId);
+          expect(read.params.some(value => entry.affected.includes(String(value)))).toBe(true);
+        }
+      }
+      expect(selections).toHaveLength(deletions.length);
+      for (let index = 0; index < selections.length; index++) {
+        expect(selections[index]!.query).toContain('"external_object_mentions"."company_id" =');
+        expect(selections[index]!.query).toContain('"external_object_mentions"."source_issue_id" =');
+        expect(selections[index]!.params).toContain(own.companyId); expect(selections[index]!.params).toContain(deletions[index]!.id);
+      }
+      const lastMentions = await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.objectId, shared));
+      expect(lastMentions).toHaveLength(1); expect(lastMentions[0]!.sourceIssueId).toBe(retained!.id);
+      const orphanIds = deletions.flatMap(entry => entry.orphaned);
+      const orphanRows = await db.select().from(externalObjects).where(inArray(externalObjects.id, orphanIds));
+      expect(orphanRows).toHaveLength(orphanIds.length);
+      for (const orphan of orphanRows) expect(orphan.nextRefreshAt).toEqual(removedDeadline);
+      expect(await db.select().from(externalObjectMentions).where(inArray(externalObjectMentions.objectId, orphanIds))).toEqual([]);
+      const foreignRows = await db.select().from(externalObjects).where(eq(externalObjects.companyId, foreign.companyId));
+      expect(foreignRows).toHaveLength(1); expect(foreignRows[0]!.nextRefreshAt).toEqual(future);
+      const count = reads.length;
+      await vi.advanceTimersByTimeAsync(removedDeadline.getTime() - Date.now() + 1);
+      expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled();
+      expect(coordinator!.nextWakeAt()).toBe(future.getTime());
+    } finally { db.$client.options.debug = debug; }
+  });
+
+  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,10 @@
-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 +13,7 @@ import {
  issues,
  plugins,
 } from "@paperclipai/db";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import {
  getEmbeddedPostgresTestSupport,
  startEmbeddedPostgresTestDatabase,
@@ -27,8 +29,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 +713,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 +734,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 +895,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 +934,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 +980,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 +1041,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 +1049,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 +1086,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 +1145,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 +1189,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 +1230,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 +1242,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 +1279,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 +1324,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) }),
    });
@@ -295,3 +295,40 @@ it("keeps an exact deadline and late reconstruction debt with concurrency enable
  expect(load).toHaveBeenCalledTimes(3); expect(check).toHaveBeenCalledOnce();
  expect(onError).not.toHaveBeenCalled();
 });
+
+it("coalesces broad reconstruction without delaying an exact object hint or saved deadline", async () => {
+  vi.useFakeTimers();
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
+  const now = Date.now();
+  let dueAt = now + 60_000;
+  const load = vi.fn(async () => [{ id: "object", companyId: "company", dueAt }]);
+  const check = vi.fn(async () => null);
+  const worker = registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.externalObjectRefresh,
+    minimumPassIntervalMs: 0,
+    reconstructionIntervalMs: (scope: string | undefined) => scope === "company" ? 30_000 : 0,
+    decodeScope: hint => hint === "*" ? undefined : hint,
+    matches: () => true,
+    load,
+    check,
+  }, () => true);
+  await worker.ready;
+  worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0);
+  expect(load).toHaveBeenCalledTimes(2);
+  worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0);
+  expect(load).toHaveBeenCalledTimes(2);
+  expect(coordinator.nextWakeAt()).toBe(now + 30_000);
+  dueAt = now + 250;
+  worker.invalidate("object"); await vi.advanceTimersByTimeAsync(0);
+  expect(load).toHaveBeenCalledTimes(3);
+  expect(coordinator.nextWakeAt()).toBe(dueAt);
+  await vi.advanceTimersByTimeAsync(250);
+  expect(check).toHaveBeenCalledOnce();
+  load.mockResolvedValue([]);
+  await vi.advanceTimersByTimeAsync(29_750);
+  expect(load).toHaveBeenCalledTimes(4);
+  expect(coordinator.nextWakeAt()).toBeNull();
+  await vi.advanceTimersByTimeAsync(120_000);
+  expect(load).toHaveBeenCalledTimes(4); expect(check).toHaveBeenCalledOnce();
+});
@@ -215,7 +215,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 = {
@@ -1489,7 +1490,9 @@ describe("startServer feedback export wiring", () => {
      await Promise.resolve();

      expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1);
-      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(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalledTimes(1);
      expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled();
      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true);
@@ -1514,6 +1517,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,
@@ -1541,7 +1553,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();
@@ -1321,19 +1321,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
@@ -1360,6 +1347,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);
  }
@@ -1933,7 +1922,6 @@ async function startServerWithDatabaseTeardown(
        }

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

        if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
        scheduleGitHubConnectionEventPoll();
@@ -1999,7 +1987,6 @@ async function startServerWithDatabaseTeardown(
    // as the enabled path.
    await startEnvironmentCleanup();
    startHeartbeatSchedulerInterval(() => {
-      scheduleExternalObjectRefreshSweep(new Date());
      scheduleGitHubConnectionEventPoll();
      scheduleGitHubConnectionContinuitySweep();
    });
@@ -1,3 +1,4 @@
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
@@ -326,6 +327,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
    await notifyRunDispatch(db, { companyId: policy.companyId });
    await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId });
    await notifyCompanyRetentionWork(db, policy.companyId);
+    await notifyExternalObjectRefresh(db, policy.companyId);
    await notifyStatusCardIssueChange(db, policy.companyId);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
@@ -391,6 +393,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
    await notifyRunDispatch(db, { companyId: policy.companyId });
    await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId });
    await notifyCompanyRetentionWork(db, policy.companyId);
+    await notifyExternalObjectRefresh(db, policy.companyId);
    await notifyStatusCardIssueChange(db, policy.companyId);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
    await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
@@ -3,6 +3,7 @@ import { notifyDeferredExecutionEligibility } from "./deferred-execution-notific
 import { notifyRunDispatch } from "./run-dispatch-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";
@@ -339,6 +340,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
          await notifyDeferredExecutionEligibility(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);
@@ -498,6 +500,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
          await notifyDeferredExecutionEligibility(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,3 +1,4 @@
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -73,6 +74,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 notifyIssueRecovery(tx, { companyId: id });
        await notifyChatCompanyConfiguration(tx, id);
        await chatCompanyDeletion.deleteCompanyData(tx, id);
@@ -18,6 +18,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) {
@@ -325,31 +346,35 @@ async function resolveViaPluginProvider(
  db: Db,
  pluginWorkerManager: PluginWorkerManager | undefined,
  object: ExternalObjectRecord,
-): Promise<PluginExternalObjectResolveResult | null> {
+): Promise<{ pluginId: string; result: 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;
    if (!manifestProvidesObject(manifest, object)) continue;
    try {
-      return await pluginWorkerManager.call(provider.id, "resolveExternalObject", {
+      const result = await pluginWorkerManager.call(provider.id, "resolveExternalObject", {
        companyId: object.companyId,
        providerKey: object.providerKey,
        objectType: object.objectType,
        externalId: object.externalId,
        object: objectSnapshot(object),
      });
+      return { pluginId: provider.id, result };
    } catch (err) {
      logger.warn(
        { err, pluginId: provider.id, pluginKey: provider.pluginKey, objectId: object.id },
        "plugin external object resolver failed",
      );
      return {
-        ok: false,
-        liveness: "unreachable",
-        errorCode: "plugin_resolver_failed",
-        errorMessage: err instanceof Error ? err.message : String(err),
+        pluginId: provider.id,
+        result: {
+          ok: false,
+          liveness: "unreachable",
+          errorCode: "plugin_resolver_failed",
+          errorMessage: err instanceof Error ? err.message : String(err),
+        },
      };
    }
  }
@@ -375,10 +400,43 @@ export function externalObjectService(
    ...(opts.detectors ?? []),
    ...(githubProvider ? [githubProvider.detector] : []),
  ]);
-  const resolverRegistry = createExternalObjectResolverRegistry([
+  const resolvers = [
    ...(opts.resolvers ?? []),
    ...(githubProvider?.resolvers ?? []),
-  ]);
+  ];
+  const resolverRegistry = createExternalObjectResolverRegistry(resolvers);
+
+  function pluginRefreshWhere(database: Db, selectedPluginId?: string) {
+    return exists(database.select({ id: plugins.id }).from(plugins).where(and(
+      selectedPluginId ? eq(plugins.id, selectedPluginId) : undefined,
+      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),
+      )))),
+    )));
+  }
+
+  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(pluginRefreshWhere(database));
+    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 +481,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 +511,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 +532,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 +584,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 +606,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 +638,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 +848,8 @@ export function externalObjectService(
    companyId: string;
    actor?: Pick<LogActivityInput, "actorType" | "actorId" | "agentId" | "runId">;
    force?: boolean;
+    scheduled?: boolean;
+    signal?: AbortSignal;
    now?: Date;
  };

@@ -802,39 +871,50 @@ 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, selectedPluginId?: string) {
+    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,
+          selectedPluginId ? pluginRefreshWhere(tx as unknown as Db, selectedPluginId) : undefined)).returning();
+      if ((scheduled || selectedPluginId) && 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);
-    const resolver = pluginResult ? null : resolverRegistry.find(object);
-    if (!pluginResult && !resolver) {
-      const [updated] = await db
-        .update(externalObjects)
-        .set({
+    input.signal?.throwIfAborted();
+    const pluginResolution = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal);
+    const resolver = pluginResolution ? null : resolverRegistry.find(object);
+    if (!pluginResolution && !resolver) {
+      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 = pluginResolution?.result ?? 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 +924,7 @@ export function externalObjectService(
          refreshStartedAt: null,
          refreshToken: null,
          updatedAt: now,
-        })
-        .where(refreshOwnerWhere(object, refreshToken))
-        .returning();
+        }, input.scheduled, pluginResolution?.pluginId);
      if (!updated) return refreshSupersededResult(object, now);
      publishLiveEvent({
        companyId: object.companyId,
@@ -880,14 +958,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, pluginResolution?.pluginId);
    if (!updated) return refreshSupersededResult(object, now);
    const next = updated ?? object;
    if (objectChanged(object, next) && input.actor) {
@@ -948,6 +1022,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 +1046,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 +1075,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 +1092,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 +1120,69 @@ 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,
+      // Eligibility changes affect every object in the company. Coalesce repeated
+      // rebuilds while object/identity hints and saved deadlines remain immediate.
+      reconstructionIntervalMs: scope => scope && !scope.objectId && !scope.identity ? 30_000 : 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 +1198,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([{
@@ -16,6 +16,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";
@@ -635,6 +636,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 (chatBecameEnabled) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAssignments, "*");
        if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
        if (patch.enableWorktreeRunExecution !== undefined) {
@@ -1,3 +1,4 @@
+import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
 import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js";
 import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js";
@@ -96,6 +97,7 @@ import {
  heartbeatRuns,
  routineRuns,
  executionWorkspaces,
+  externalObjectMentions,
  issueApprovals,
  issueAccessGrants,
  issueAttachments,
@@ -12022,6 +12024,14 @@ export function issueService(db: Db) {
          if (owner) {
            await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id });
            await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
+            const mentionedObjects = await tx.selectDistinct({ objectId: externalObjectMentions.objectId })
+              .from(externalObjectMentions)
+              .where(and(
+                eq(externalObjectMentions.companyId, owner.companyId),
+                eq(externalObjectMentions.sourceIssueId, id),
+                isNotNull(externalObjectMentions.objectId),
+              ));
+            for (const mention of mentionedObjects) await notifyExternalObjectRefresh(tx, owner.companyId, mention.objectId!);
            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";
@@ -89,9 +90,9 @@ export function pluginRegistryService(db: Db) {
  }

  async function writeWorkspaceProviderChange<T>(reconsiderWorkspaces: boolean, write: (database: Db) => Promise<T>): Promise<T> {
-    if (!reconsiderWorkspaces) return write(db);
    return db.transaction(async tx => {
-      await notifyDeliveryWork(tx, DELIVERY_QUEUES.terminalWorkspaces);
+      await notifyExternalObjectRefresh(tx);
+      if (reconsiderWorkspaces) await notifyDeliveryWork(tx, DELIVERY_QUEUES.terminalWorkspaces);
      return write(tx as unknown as Db);
    });
  }
@@ -292,6 +293,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
@@ -355,6 +357,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");

@@ -397,6 +400,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");

@@ -479,6 +483,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");

@@ -13,7 +13,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
    batchSize?: number;
    minimumPassIntervalMs?: number;
    concurrentChecks?: { limit: number; key(entry: Entry): string };
-    reconstructionIntervalMs?: number;
+    reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number);
    beforePass?(): Promise<void>;
    decodeScope(hint: string): Scope | undefined;
    matches(entry: Entry, scope?: Scope): boolean;
@@ -62,7 +62,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
    } },
    run: async signal => {
      if (!canRun() || Date.now() < earliest) return;
-      if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) {
+      for (const [hint, until] of reconstructAfter) {
        if (until <= Date.now()) reconstructAfter.delete(hint);
      }
      const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
@@ -98,8 +98,10 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
            entries.delete(entry.id); failures.delete(entry.id);
          }
          scans.delete(hint);
-          if ((options.reconstructionIntervalMs ?? 0) > 0) {
-            reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!);
+          const reconstructionInterval = typeof options.reconstructionIntervalMs === "function"
+            ? options.reconstructionIntervalMs(scope) : (options.reconstructionIntervalMs ?? 0);
+          if (reconstructionInterval > 0) {
+            reconstructAfter.set(hint, Date.now() + reconstructionInterval);
          }
          if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint);
          if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);