From 50cd007005c7971612d6543da30aee0b3bf94d2b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:24:46 -0500 Subject: [PATCH 1/6] perf(external-objects): schedule scoped refresh obligations Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 92 +++++ .../external-object-refresh-work.test.ts | 334 ++++++++++++++++++ .../external-objects-service.test.ts | 9 +- .../github-connection-events.test.ts | 20 +- server/src/index.ts | 17 +- server/src/services/companies.ts | 3 + server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../external-object-refresh-notifications.ts | 7 + server/src/services/external-objects.ts | 183 +++++++--- .../src/services/github-connection-events.ts | 44 ++- server/src/services/instance-settings.ts | 2 + server/src/services/issues.ts | 2 + server/src/services/plugin-registry.ts | 43 ++- 14 files changed, 659 insertions(+), 100 deletions(-) create mode 100644 doc/plans/2026-10-30-external-objects-sleep.md create mode 100644 server/src/__tests__/external-object-refresh-work.test.ts create mode 100644 server/src/services/external-object-refresh-notifications.ts diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md new file mode 100644 index 00000000000..639bbaea343 --- /dev/null +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -0,0 +1,92 @@ +# 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. + +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. 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 company intent before mention replacement and object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | +| Comment/document disappearance | Sync registers reconstruction before deleting obsolete mentions. | +| Last mention removal and reattachment | Mention replacement invalidates company deadlines. Retained objects stay available for reattachment. | +| Issue deletion | Issue service registers its company before cascading mention deletion. | +| Company status/archive/deletion | Company services register company reconstruction before their 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. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts new file mode 100644 index 00000000000..8d1f5cdc00c --- /dev/null +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -0,0 +1,334 @@ +import { randomUUID } from "node:crypto"; +import { setTimeout as delay } from "node:timers/promises"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { activityLog, companies, createDb, externalObjects, issues, 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 { 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 { 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; + let temp: Awaited> | undefined; + let coordinator: ReturnType | undefined; + let admitted = true; + let reconcile = true; + let enabled = true; + const provider = vi.fn(async (): Promise => terminal); + const errors = vi.fn(); + const companyIds: string[] = []; + beforeAll(async () => { + if (!url) temp = await startEmbeddedPostgresTestDatabase("external-refresh-work-"); + db = createDb(url ?? temp!.connectionString, { maxConnections: 6 }); + await db.execute(sql`truncate table ${companies} cascade`); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + for (const id of companyIds.splice(0)) { + 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) { + 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) { + await db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, companyId); + await tx.update(externalObjects).set(patch).where(eq(externalObjects.companyId, companyId)); + }); + } + + it("captures the old configured-empty aggregate SQL cost", async () => { + const svc = service(); await fixture(); + const selects = vi.spyOn(db, "select"); + for (let i = 0; i < 3; i++) await svc.refreshDueObjectsForActiveCompanies(); + expect(selects).toHaveBeenCalledTimes(6); + expect(provider).not.toHaveBeenCalled(); + }); + + 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(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(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("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 waits for in-flight 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 delay(10); expect(stopped).toBe(false); + finish(terminal); await stop; coordinator = undefined; + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(600_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(2); + }); + + 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(); + }); +}); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index e9022599779..e50e298b466 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -11,6 +11,7 @@ import { issues, plugins, } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -27,7 +28,8 @@ import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); -const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL"); +const describeEmbeddedPostgres = externalDatabaseUrl || embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { console.warn( @@ -709,8 +711,8 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); - db = createDb(tempDb.connectionString); + if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); + db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); }, 20_000); afterEach(async () => { @@ -725,6 +727,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 7c7b7025243..0417a434718 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -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(resolve => { markStarted = resolve; }); + const held = new Promise(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) }), }); diff --git a/server/src/index.ts b/server/src/index.ts index c127753889c..f6b59d277cf 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1323,19 +1323,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 @@ -1362,6 +1349,8 @@ async function startServerWithDatabaseTeardown( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const deliveryWork = app.locals.deliveryWork as ReturnType; + externalObjects.registerRefreshWork(deliveryWork, + () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby()); if (config.nativeHarnessBackupExpirationEnabled) { nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork); } @@ -2007,7 +1996,6 @@ async function startServerWithDatabaseTeardown( } if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduleExternalObjectRefreshSweep(new Date()); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleMergedPullRequestConfirmationSweep(); @@ -2133,7 +2121,6 @@ async function startServerWithDatabaseTeardown( // as the enabled path. await startEnvironmentCleanup(); startHeartbeatSchedulerInterval(() => { - scheduleExternalObjectRefreshSweep(new Date()); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); }); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 0f2ed547a79..f429b08badf 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,5 +1,6 @@ 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"; @@ -334,6 +335,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { 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 notifyCompanyRetentionWork(tx, id); @@ -490,6 +492,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { 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 notifyCompanyRetentionWork(tx, id); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 7ab88aadfa9..645e17f528b 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; @@ -71,6 +72,7 @@ export async function deleteCompany(db: Db, id: string) { const [existing] = await tx.select({ id: companies.id }).from(companies) .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; + await notifyExternalObjectRefresh(tx, id); await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..e9555d8ad53 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -6,6 +6,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", diff --git a/server/src/services/external-object-refresh-notifications.ts b/server/src/services/external-object-refresh-notifications.ts new file mode 100644 index 00000000000..95833421a8a --- /dev/null +++ b/server/src/services/external-object-refresh-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export async function notifyExternalObjectRefresh(tx: object, companyId?: string, objectId?: string) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.externalObjectRefresh, + companyId ? JSON.stringify({ companyId, ...(objectId ? { objectId } : {}) }) : "*"); +} + diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 56a8fb91834..c49df7ff076 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -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 } 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"; @@ -95,6 +99,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 }; +type RefreshDeadline = { id: string; companyId: string; dueAt: number }; + + +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), @@ -245,7 +261,7 @@ function objectSnapshot(object: ExternalObjectRecord): PluginExternalObjectRecor }; } -async function readyObjectReferencePlugins(db: Db) { +async function readyObjectReferencePlugins(db: Db, companyId?: string) { return db .select({ id: plugins.id, @@ -253,7 +269,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), @@ -267,7 +288,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) { @@ -326,7 +347,7 @@ async function resolveViaPluginProvider( object: ExternalObjectRecord, ): Promise { 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; @@ -374,10 +395,38 @@ export function externalObjectService( ...(opts.detectors ?? []), ...(githubProvider ? [githubProvider.detector] : []), ]); - const resolverRegistry = createExternalObjectResolverRegistry([ + const resolvers = [ ...(opts.resolvers ?? []), ...(githubProvider?.resolvers ?? []), - ]); + ]; + const resolverRegistry = createExternalObjectResolverRegistry(resolvers); + + function automaticRefreshWhere(database: Db = db) { + const supported = resolvers.map(resolver => and( + eq(externalObjects.providerKey, resolver.providerKey), + resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined, + )!); + if (opts.pluginWorkerManager) supported.push(exists(database.select({ id: plugins.id }).from(plugins).where(and( + eq(plugins.status, "ready"), + sql`${plugins.manifestJson}->'capabilities' ? 'external.objects.read'`, + sql`exists (select 1 from jsonb_array_elements(coalesce(${plugins.manifestJson}->'objectReferences', '[]'::jsonb)) as reference + where reference->>'providerKey' = ${externalObjects.providerKey} + and reference->'objectTypes' ? ${externalObjects.objectType})`, + not(exists(database.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings).where(and( + eq(pluginCompanySettings.pluginId, plugins.id), + eq(pluginCompanySettings.companyId, externalObjects.companyId), + eq(pluginCompanySettings.enabled, false), + )))), + )))); + return and( + refreshableObjectWhere(), isNotNull(externalObjects.nextRefreshAt), + supported.length ? or(...supported) : sql`false`, + exists(database.select({ id: externalObjectMentions.id }).from(externalObjectMentions).where(and( + eq(externalObjectMentions.companyId, externalObjects.companyId), + eq(externalObjectMentions.objectId, externalObjects.id), + ))), + )!; + } async function isEnabled() { if (typeof opts.enabled === "function") return await opts.enabled(); @@ -422,6 +471,7 @@ export function externalObjectService( nextRefreshAt: now, updatedAt: now, }; + await notifyExternalObjectRefresh(dbOrTx, companyId); const inserted = await dbOrTx .insert(externalObjects) .values(values) @@ -451,6 +501,7 @@ export function externalObjectService( dbOrTx: any = db, ) { const urls = extractExternalObjectCanonicalUrls(input.text ?? ""); + await notifyExternalObjectRefresh(dbOrTx, input.companyId); await dbOrTx.delete(externalObjectMentions).where(sourceWhere(input)); if (urls.length === 0) return; @@ -534,6 +585,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 notifyExternalObjectRefresh(tx); await tx .delete(externalObjectMentions) .where(and(eq(externalObjectMentions.sourceKind, "comment"), eq(externalObjectMentions.sourceRecordId, commentId))); @@ -568,6 +620,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 notifyExternalObjectRefresh(tx); await tx .delete(externalObjectMentions) .where(and(eq(externalObjectMentions.sourceKind, "document"), eq(externalObjectMentions.sourceRecordId, documentId))); @@ -780,6 +833,7 @@ export function externalObjectService( companyId: string; actor?: Pick; force?: boolean; + scheduled?: boolean; now?: Date; }; @@ -801,6 +855,18 @@ export function externalObjectService( return { object: toObjectPayload(latest, now), refreshed: false, reason: "refresh_superseded" as const }; } + async function persistRefresh(object: ExternalObjectRecord, token: string, + patch: Partial, scheduled = false) { + return db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, object.companyId, object.id); + const rows = await tx.update(externalObjects).set(patch) + .where(and(refreshOwnerWhere(object, token), scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined)).returning(); + if (scheduled && rows.length === 0) await tx.update(externalObjects) + .set({ refreshStartedAt: null, refreshToken: null }).where(refreshOwnerWhere(object, token)); + return rows; + }); + } + async function resolveObjectRefresh( object: ExternalObjectRecord, input: RefreshObjectInput, @@ -810,17 +876,13 @@ export function externalObjectService( 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({ + 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 }; } @@ -831,9 +893,7 @@ export function externalObjectService( // 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, @@ -843,9 +903,7 @@ export function externalObjectService( refreshStartedAt: null, refreshToken: null, updatedAt: now, - }) - .where(refreshOwnerWhere(object, refreshToken)) - .returning(); + }, input.scheduled); if (!updated) return refreshSupersededResult(object, now); publishLiveEvent({ companyId: object.companyId, @@ -879,14 +937,10 @@ export function externalObjectService( refreshToken: null, updatedAt: now, }; - const [updated] = await db - .update(externalObjects) - .set({ + const [updated] = await persistRefresh(object, refreshToken, { ...patch, lastChangedAt: objectChanged(object, { ...object, ...patch }) ? now : object.lastChangedAt, - }) - .where(refreshOwnerWhere(object, refreshToken)) - .returning(); + }, input.scheduled); if (!updated) return refreshSupersededResult(object, now); const next = updated ?? object; if (objectChanged(object, next) && input.actor) { @@ -947,6 +1001,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; @@ -1032,26 +1088,7 @@ export function externalObjectService( 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 refreshable = refreshableObjectWhere(); const due = await db .select({ id: externalObjects.id }) .from(externalObjects) @@ -1099,6 +1136,59 @@ export function externalObjectService( return { companies: activeCompanies.length, checked, refreshed }; } + function registerRefreshWork( + coordinator: ReturnType, + canRun: () => boolean, + ) { + const cursors = new Map(); + 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 }).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, + 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 }; + } + function deadline(object: Pick): RefreshDeadline { + return { id: object.id, companyId: object.companyId, + dueAt: Math.max(object.nextRefreshAt!.getTime(), object.refreshStartedAt + ? object.refreshStartedAt.getTime() + DEFAULT_REFRESH_LEASE_SECONDS * 1000 : 0) }; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.externalObjectRefresh, + batchSize: 50, + minimumPassIntervalMs: 0, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope, + matches: (entry, scope) => !scope || (entry.companyId === scope.companyId + && (!scope.objectId || entry.id === scope.objectId)), + 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, + 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 { syncIssue, syncComment, @@ -1114,5 +1204,6 @@ export function externalObjectService( refreshIssueObjects, refreshDueObjects, refreshDueObjectsForActiveCompanies, + registerRefreshWork, }; } diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 47bdac122a2..76755ce30c3 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -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,30 @@ 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); + 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([{ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 0208653a86f..a2f5eb5015c 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -13,6 +13,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"; @@ -628,6 +629,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + if (patch.enableExternalObjects !== undefined) await notifyExternalObjectRefresh(tx); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 42c65bb5d02..53ac68718a6 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -11953,6 +11954,7 @@ export function issueService(db: Db) { const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id)); if (owner) { await notifyExternalConversationSettlement(tx, { companyId: owner.companyId }); + await notifyExternalObjectRefresh(tx, owner.companyId); await notifyApprovalWork(tx, owner.companyId); } await notifyDecisionTargetIssueRemoval(tx, id); diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index 239c3c50490..62d9b7cde70 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { isDeepStrictEqual } from "node:util"; import { writePluginJobChange } from "./plugin-job-work-notifications.js"; import { asc, eq, isNull, ne, sql, and } from "drizzle-orm"; @@ -239,12 +240,15 @@ export function pluginRegistryService(db: Db) { setClause.categories = data.manifest.categories; } - return db - .update(plugins) - .set(setClause) - .where(eq(plugins.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + return db.transaction(async tx => { + await notifyExternalObjectRefresh(tx); + return tx + .update(plugins) + .set(setClause) + .where(eq(plugins.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + }); }, // ----- Status --------------------------------------------------------- @@ -254,16 +258,19 @@ export function pluginRegistryService(db: Db) { const plugin = await getById(id); if (!plugin) throw notFound("Plugin not found"); - return db - .update(plugins) - .set({ - status: input.status, - lastError: input.lastError ?? null, - updatedAt: new Date(), - }) - .where(eq(plugins.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + return db.transaction(async tx => { + await notifyExternalObjectRefresh(tx); + return tx + .update(plugins) + .set({ + status: input.status, + lastError: input.lastError ?? null, + updatedAt: new Date(), + }) + .where(eq(plugins.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + }); }, // ----- Uninstall / Remove -------------------------------------------- @@ -280,6 +287,7 @@ export function pluginRegistryService(db: Db) { if (!plugin) throw notFound("Plugin not found"); return writePluginJobChange(db, async db => { + await notifyExternalObjectRefresh(db); if (removeData) { // Hard delete – plugin_config cascades via FK onDelete return db @@ -343,6 +351,7 @@ export function pluginRegistryService(db: Db) { return db.transaction(async tx => { const db = tx as unknown as Db; await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + await notifyExternalObjectRefresh(db, companyId); const plugin = await getById(pluginId, db); if (!plugin) throw notFound("Plugin not found"); @@ -385,6 +394,7 @@ export function pluginRegistryService(db: Db) { return db.transaction(async tx => { const db = tx as unknown as Db; await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + await notifyExternalObjectRefresh(db, companyId); const plugin = await getById(pluginId, db); if (!plugin) throw notFound("Plugin not found"); @@ -467,6 +477,7 @@ export function pluginRegistryService(db: Db) { return db.transaction(async tx => { const db = tx as unknown as Db; await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + await notifyExternalObjectRefresh(db, companyId); const plugin = await getById(pluginId, db); if (!plugin) throw notFound("Plugin not found"); From 25a8a4392c807d4b5eb19cbf713b5eda99c017d3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:35:43 -0500 Subject: [PATCH 2/6] fix(external-objects): cover budget resume and scoped deletions Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 4 +- .../external-object-refresh-work.test.ts | 63 ++++++++++++++++++- .../external-objects-service.test.ts | 4 +- server/src/services/budgets.ts | 3 + server/src/services/external-objects.ts | 18 +++--- 5 files changed, 79 insertions(+), 13 deletions(-) diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index 639bbaea343..ab566999917 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -47,10 +47,10 @@ and fences automatic result persistence. It does not modify that draft PR. | Mutation | Owner and activation | | --- | --- | | Issue title/description, comment and document references | Existing sync methods register company intent before mention replacement and object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | -| Comment/document disappearance | Sync registers reconstruction before deleting obsolete mentions. | +| 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 | Issue service registers its company before cascading mention deletion. | -| Company status/archive/deletion | Company services register company reconstruction before their status or deletion writes. | +| 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. | diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index 8d1f5cdc00c..f429c31396b 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -2,10 +2,12 @@ import { randomUUID } from "node:crypto"; import { setTimeout as delay } from "node:timers/promises"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { activityLog, companies, createDb, externalObjects, issues, plugins } from "@paperclipai/db"; +import { activityLog, approvals, applyPendingMigrations, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, issues, issueComments, plugins } from "@paperclipai/db"; import { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { documentService } from "../services/documents.js"; +import { budgetService } from "../services/budgets.js"; import { pluginRegistryService } from "../services/plugin-registry.js"; import { companyService } from "../services/companies.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -34,14 +36,19 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { 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)); @@ -256,6 +263,58 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { expect(calls).toHaveBeenCalledOnce(); }); + it("reactivates refresh after ordinary budget pause and budget raise", async () => { + clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc); + const f = await fixture(); await svc.syncIssue(f.issueId); + await settle(async () => (await row(f.companyId))?.lastResolvedAt !== null); + await db.insert(costEvents).values({ companyId: f.companyId, usageKind: "decision", provider: "fixture", + biller: "fixture", billingType: "metered_api", model: "simulated", costCents: 200, occurredAt: new Date() }); + const budgets = budgetService(db); + await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, + amount: 100, hardStopEnabled: true }, null); + expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0]!.status).toBe("paused"); + await quiet(); + const reads = vi.spyOn(db, "select"); provider.mockClear(); + await vi.advanceTimersByTimeAsync(900_000); expect(reads).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled(); + provider.mockResolvedValue(terminal); + await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 300 }, null); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it.each(["comment", "document"] as const)("keeps hard-deleted %s activation scoped to its retained mention owner", async kind => { + clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc); + const a = await fixture(); const b = await fixture(); await svc.syncIssue(b.issueId); + let id: string; + if (kind === "comment") { + const [comment] = await db.insert(issueComments).values({ companyId: a.companyId, issueId: a.issueId, + body: `https://example.com/${a.companyId}` }).returning(); + id = comment!.id; await svc.syncComment(id); + } else { + const result = await documentService(db).upsertIssueDocument({ issueId: a.issueId, + key: "notes", format: "markdown", body: `https://example.com/${a.companyId}` }); + id = result.document.id; await svc.syncDocument(id); + } + await settle(async () => (await row(a.companyId))?.lastResolvedAt !== null && (await row(b.companyId))?.lastResolvedAt !== null); + await settle(() => coordinator!.nextWakeAt() !== null); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query, params) => { + if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"')) + reads.push({ query, params: [...params] }); + }; + try { + if (kind === "comment") { await db.delete(issueComments).where(eq(issueComments.id, id)); await svc.syncComment(id); } + else { await documentService(db).deleteIssueDocument(a.issueId, "notes"); await svc.syncDocument(id); } + await settle(() => reads.length > 0); + for (const read of reads) { + expect(read.query).toContain('"external_objects"."company_id" ='); + expect(read.params).toContain(a.companyId); expect(read.params).not.toContain(b.companyId); + } + } finally { db.$client.options.debug = debug; } + expect(provider).toHaveBeenCalledTimes(2); + }); + it("keeps warm standby startup SQL-free and resumes a first committed producer", async () => { clock(); admitted = false; reconcile = false; const svc = service(); const selects = vi.spyOn(db, "select"); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index e50e298b466..bc268d0ba6a 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -3,6 +3,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { eq } from "drizzle-orm"; import { companies, + applyPendingMigrations, createDb, activityLog, externalObjectMentions, @@ -711,9 +712,10 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { + if (externalDatabaseUrl) await applyPendingMigrations(externalDatabaseUrl); if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); - }, 20_000); + }, 90_000); afterEach(async () => { await db.delete(activityLog); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 6777124ee90..88c3255da4e 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { addCents, compareCents, normalizeCents } from "@paperclipai/shared"; @@ -301,6 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyExternalObjectRefresh(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); @@ -361,6 +363,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyExternalObjectRefresh(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index c49df7ff076..d524b473e1a 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -571,6 +571,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 }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId }) + .from(externalObjectMentions).where(source); + for (const owner of owners) await notifyExternalObjectRefresh(tx, owner.companyId); + await tx.delete(externalObjectMentions).where(source); + } + async function syncComment(commentId: string, dbOrTx: any = db) { const runSync = async (tx: any) => { if (!(await isEnabled())) return; @@ -585,10 +593,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 notifyExternalObjectRefresh(tx); - await tx - .delete(externalObjectMentions) - .where(and(eq(externalObjectMentions.sourceKind, "comment"), eq(externalObjectMentions.sourceRecordId, commentId))); + await deleteMissingSourceMentions("comment", commentId, tx); return; } await replaceSourceMentions({ @@ -620,10 +625,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 notifyExternalObjectRefresh(tx); - await tx - .delete(externalObjectMentions) - .where(and(eq(externalObjectMentions.sourceKind, "document"), eq(externalObjectMentions.sourceRecordId, documentId))); + await deleteMissingSourceMentions("document", documentId, tx); return; } await replaceSourceMentions({ From 30730dd61983ebcd1500170bef66df85c296871c Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:54:51 -0500 Subject: [PATCH 3/6] test(skills): verify Cloud CLI reads through shared object storage Use the real S3 provider with a controlled HTTP fixture and a stable test key. Preserve read-only/no-seeding behavior and verify corrupt, missing, and restored object outcomes under managed Cloud policy. Co-Authored-By: Paperclip --- .../__tests__/managed-skill-content.test.ts | 90 ++++++++++++++++--- 1 file changed, 77 insertions(+), 13 deletions(-) diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index cd4ddb8135f..59566d743e3 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; @@ -79,26 +81,88 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { const companyId = await company(); - await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const objects = new Map(); + const server = createServer((request, response) => { + void (async () => { + const key = new URL(request.url!, "http://fixture").pathname; + if (request.method === "PUT") { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + let body = Buffer.concat(chunks); + if (request.headers["content-encoding"]?.includes("aws-chunked")) { + const decoded: Buffer[] = []; + let offset = 0; + while (offset < body.length) { + const end = body.indexOf("\r\n", offset); + const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16); + if (size === 0) break; + decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2; + } + body = Buffer.concat(decoded); + } + objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return; + } + const body = objects.get(key); + if (!body) { response.writeHead(404).end(); return; } + response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' }); + response.end(request.method === "HEAD" ? undefined : body); + })().catch(() => { response.writeHead(500).end(); }); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address"); + const endpoint = `http://127.0.0.1:${address.port}`; + const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" }; + const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]])); + Object.assign(process.env, credentials); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({ + bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true, + }))); const key = "cloud-runtime-config:v1"; - const run = (phase: string) => promisify(execFile)(process.execPath, [ - "--import", fileURLToPath(import.meta.resolve("tsx")), - fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), - `--phase=${phase}`, `--company=${companyId}`, "--limit=50", - ], { env: { ...process.env, DATABASE_URL: database.connectionString, - PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", - PAPERCLIP_STORAGE_PROVIDER: "local_disk", PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects") }, - timeout: 60_000, maxBuffer: 1024 * 1024 }); - await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); - expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); - await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); try { + await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const run = (phase: string) => promisify(execFile)(process.execPath, [ + "--import", fileURLToPath(import.meta.resolve("tsx")), + fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), + `--phase=${phase}`, `--company=${companyId}`, "--limit=50", + ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString, + PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", + PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3", + PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", + PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" }, + timeout: 60_000, maxBuffer: 1024 * 1024 }); + await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); + await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); + const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const); const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(verification.results).toEqual([expect.objectContaining({ status: "local_storage" })]); + expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()]).toEqual(before); + const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!; + const archive = objects.get(archiveKey)!; + const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1; + objects.set(archiveKey, corrupted); + const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]); + objects.set(archiveKey, archive); + const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!; + const manifest = objects.get(manifestKey)!; objects.delete(manifestKey); + const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]); + objects.set(manifestKey, manifest); + const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()].sort()).toEqual([...before].sort()); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))) + .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]); } finally { await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")))); + for (const [key, value] of Object.entries(previousCredentials)) { + if (value === undefined) delete process.env[key]; else process.env[key] = value; + } + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); } }, 180_000); From b45b6a225d8a9983ace61cca426f0c0c949e2ab7 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:18:38 -0500 Subject: [PATCH 4/6] fix(external-objects): bound refresh lifecycle and activation scope Reuse the provider-owned request cancellation contract from merge-confirmation work. Keep host signals out of plugin RPC payloads, preserve refresh debt on shutdown, and target identities or known objects for source writes. Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 26 +++++- .../external-object-refresh-work.test.ts | 79 +++++++++++++++++-- .../server-startup-feedback-export.test.ts | 22 +++++- .../external-object-refresh-notifications.ts | 6 +- server/src/services/external-objects.ts | 60 ++++++++++---- .../src/services/github-connection-events.ts | 3 +- .../github-external-object-provider.ts | 18 +++-- 7 files changed, 179 insertions(+), 35 deletions(-) diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index ab566999917..61c7d8dded1 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -24,7 +24,9 @@ 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. A due callback reads only its +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 @@ -46,7 +48,7 @@ and fences automatic result persistence. It does not modify that draft PR. | Mutation | Owner and activation | | --- | --- | -| Issue title/description, comment and document references | Existing sync methods register company intent before mention replacement and object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | +| 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 | Issue service registers its company before cascading mention deletion. | @@ -90,3 +92,23 @@ 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 explicit legacy entry points + +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 two explicit `refreshDueObjects*` helpers remain manual diagnostic/test entry +points. They have no production scheduler caller. Their regression fixtures also +preserve the prior selection/lease semantics. 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. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index f429c31396b..754725c7d80 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -1,8 +1,9 @@ +import { createServer } from "node:http"; import { randomUUID } from "node:crypto"; import { setTimeout as delay } from "node:timers/promises"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { activityLog, approvals, applyPendingMigrations, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, issues, issueComments, plugins } from "@paperclipai/db"; +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"; @@ -12,6 +13,7 @@ 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"; @@ -295,8 +297,12 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { key: "notes", format: "markdown", body: `https://example.com/${a.companyId}` }); id = result.document.id; await svc.syncDocument(id); } - await settle(async () => (await row(a.companyId))?.lastResolvedAt !== null && (await row(b.companyId))?.lastResolvedAt !== null); - await settle(() => coordinator!.nextWakeAt() !== null); + 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) => { @@ -344,7 +350,7 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { finish(terminal); await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); }); - it("uses durable provider retry deadlines and waits for in-flight shutdown without overlap", async () => { + 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); @@ -360,13 +366,74 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { await svc.syncIssue(f.issueId); await vi.advanceTimersByTimeAsync(0); expect(provider).toHaveBeenCalledTimes(2); let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; }); - await delay(10); expect(stopped).toBe(false); - finish(terminal); await stop; coordinator = undefined; + 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(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(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]) { diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bbb020c865d..5002dc44a50 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -207,6 +207,8 @@ const { }; const executionWorkspaceServiceFactoryMock = vi.fn(() => executionWorkspaceServiceMock); const externalObjectsServiceMock = { + registerRefreshWork: vi.fn((coordinator: { register: typeof deliveryWorkRegisterMock }, _canRun: () => boolean) => + coordinator.register("external-object-refresh", { run: async () => undefined, hasPending: async () => false })), refreshDueObjectsForActiveCompanies: vi.fn(async () => ({ companies: 0, checked: 0, refreshed: 0 })), }; const externalObjectsServiceFactoryMock = vi.fn(() => externalObjectsServiceMock); @@ -1399,7 +1401,10 @@ 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(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); @@ -1422,6 +1427,16 @@ 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(() => {}), wake: vi.fn(), stop: vi.fn(async () => {}), + }); + await startServer(); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledOnce(); + expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); + }); + it("keeps external object refresh active when heartbeat scheduling is disabled", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: false, @@ -1449,7 +1464,10 @@ 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(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled(); diff --git a/server/src/services/external-object-refresh-notifications.ts b/server/src/services/external-object-refresh-notifications.ts index 95833421a8a..271d172e463 100644 --- a/server/src/services/external-object-refresh-notifications.ts +++ b/server/src/services/external-object-refresh-notifications.ts @@ -1,7 +1,9 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyExternalObjectRefresh(tx: object, companyId?: string, objectId?: string) { +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, ...(objectId ? { objectId } : {}) }) : "*"); + companyId ? JSON.stringify({ companyId, ...(typeof object === "string" ? { objectId: object } : object ? { identity: object } : {}) }) : "*"); } diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index d524b473e1a..0208cdab7b6 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -18,7 +18,7 @@ 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 } from "./external-object-refresh-notifications.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"; @@ -88,6 +88,7 @@ export interface ExternalObjectResolver { resolve(input: { companyId: string; object: ExternalObjectRecord; + signal?: AbortSignal; }): Promise; } @@ -99,8 +100,8 @@ 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 }; -type RefreshDeadline = { id: string; companyId: string; dueAt: number }; +type RefreshScope = { companyId: string; objectId?: string; identity?: ExternalObjectRefreshIdentity }; +type RefreshDeadline = { id: string; companyId: string; dueAt: number; providerKey: string; objectType: string; externalId: string }; function refreshableObjectWhere() { @@ -471,7 +472,7 @@ export function externalObjectService( nextRefreshAt: now, updatedAt: now, }; - await notifyExternalObjectRefresh(dbOrTx, companyId); + await notifyExternalObjectRefresh(dbOrTx, companyId, { providerKey: detection.providerKey, objectType: detection.objectType, externalId: detection.externalId }); const inserted = await dbOrTx .insert(externalObjects) .values(values) @@ -501,7 +502,9 @@ export function externalObjectService( dbOrTx: any = db, ) { const urls = extractExternalObjectCanonicalUrls(input.text ?? ""); - await notifyExternalObjectRefresh(dbOrTx, input.companyId); + 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; @@ -520,6 +523,7 @@ export function externalObjectService( if (seen.has(sourceKey)) continue; seen.add(sourceKey); const object = await upsertObjectFromDetection(input.companyId, detection, dbOrTx); + await notifyExternalObjectRefresh(dbOrTx, input.companyId, object.id); values.push({ companyId: input.companyId, sourceIssueId: input.sourceIssueId, @@ -573,9 +577,9 @@ export function externalObjectService( 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 }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId }) + 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) await notifyExternalObjectRefresh(tx, owner.companyId); + for (const owner of owners) if (owner.objectId) await notifyExternalObjectRefresh(tx, owner.companyId, owner.objectId); await tx.delete(externalObjectMentions).where(source); } @@ -836,6 +840,7 @@ export function externalObjectService( actor?: Pick; force?: boolean; scheduled?: boolean; + signal?: AbortSignal; now?: Date; }; @@ -875,7 +880,8 @@ export function externalObjectService( now: Date, refreshToken: string, ) { - const pluginResult = await resolveViaPluginProvider(db, opts.pluginWorkerManager, object); + input.signal?.throwIfAborted(); + const pluginResult = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal); const resolver = pluginResult ? null : resolverRegistry.find(object); if (!pluginResult && !resolver) { const [updated] = await persistRefresh(object, refreshToken, { @@ -889,7 +895,9 @@ export function externalObjectService( return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const }; } - const result = pluginResult ?? await resolver!.resolve({ companyId: object.companyId, object }); + const result = pluginResult ?? await observeRefreshCancellation( + resolver!.resolve({ companyId: object.companyId, object, signal: input.signal }), input.signal); + input.signal?.throwIfAborted(); if (!result.ok) { // A failed refresh means the object's status is currently unknown, not // terminal. Clear a previous terminal verdict so the object keeps being @@ -1027,6 +1035,16 @@ export function externalObjectService( }; } + function observeRefreshCancellation(work: Promise, signal?: AbortSignal): Promise { + 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>>>(); async function refreshObject( @@ -1046,7 +1064,7 @@ export function externalObjectService( const refreshKey = `${object.companyId}:${object.id}`; const existingRefresh = objectRefreshesInFlight.get(refreshKey); - if (existingRefresh) return existingRefresh; + if (existingRefresh) return observeRefreshCancellation(existingRefresh, input.signal); const claimed = await claimObjectRefresh(object, input, now); if (!claimed) { @@ -1063,7 +1081,10 @@ export function externalObjectService( } const stopRenewingRefreshLease = startRefreshLeaseRenewal(object, claimed.refreshToken); - const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).finally(() => { + const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).catch(async error => { + if (input.signal?.aborted) await persistRefresh(object, claimed.refreshToken!, { refreshStartedAt: null, refreshToken: null }); + throw error; + }).finally(() => { stopRenewingRefreshLease(); objectRefreshesInFlight.delete(refreshKey); }); @@ -1149,12 +1170,17 @@ export function externalObjectService( 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 }).from(externalObjects) + 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); @@ -1162,8 +1188,8 @@ export function externalObjectService( else cursors.delete(key); return { rows: entries, complete: rows.length < 250 }; } - function deadline(object: Pick): RefreshDeadline { - return { id: object.id, companyId: object.companyId, + function deadline(object: Pick): 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) }; } @@ -1173,7 +1199,10 @@ export function externalObjectService( minimumPassIntervalMs: 0, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope, matches: (entry, scope) => !scope || (entry.companyId === scope.companyId - && (!scope.objectId || entry.id === scope.objectId)), + && (!scope.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; @@ -1183,6 +1212,7 @@ export function externalObjectService( 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 }); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 76755ce30c3..643bc557b93 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -246,7 +246,8 @@ export function githubConnectionEventService( if (!snapshot) return; const appliedAt = now(); await db.transaction(async tx => { - await notifyExternalObjectRefresh(tx, companyId); + 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, diff --git a/server/src/services/github-external-object-provider.ts b/server/src/services/github-external-object-provider.ts index 6dedac97efb..83b47354035 100644 --- a/server/src/services/github-external-object-provider.ts +++ b/server/src/services/github-external-object-provider.ts @@ -15,6 +15,7 @@ type FetchLike = (url: string, init?: RequestInit) => Promise; export interface GitHubExternalObjectProviderOptions { fetch?: FetchLike; + requestTimeoutMs?: number; tokenProvider?: (companyId: string) => Promise | string | null; /** * Resolve the credential and its source. Prefer this over `tokenProvider` when the @@ -243,12 +244,12 @@ type RepoProbeOutcome = { async function probeRepoVisibility( identity: GitHubObjectIdentity, - opts: { fetch: FetchLike; headers?: Record | undefined }, + opts: { fetch: FetchLike; headers?: Record | undefined; signal?: AbortSignal }, ): Promise { try { const probe = await opts.fetch( `${gitHubApiBase(identity.host)}/repos/${encodeURIComponent(identity.owner)}/${encodeURIComponent(identity.repo)}`, - { headers: opts.headers }, + { headers: opts.headers, signal: opts.signal }, ); if (probe.ok) { const body = await safeJson(probe); @@ -427,12 +428,13 @@ export function createGitHubExternalObjectProvider( companyId: string, identity: GitHubObjectIdentity, headers: Record, + signal: AbortSignal, ): Promise { const credentialFingerprint = createHash("sha256").update(headers.authorization ?? "anonymous").digest("hex"); const key = `${companyId}\u0000${credentialFingerprint}\u0000${identity.owner.toLowerCase()}\u0000${identity.repo.toLowerCase()}`; const cached = probeCache.get(key); if (cached && cached.expiresAt > Date.now()) return cached.outcome; - const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (outcome.visibility === "visible" || outcome.visibility === "hidden") { probeCache.set(key, { outcome, expiresAt: Date.now() + GITHUB_OBJECT_TTL_SECONDS * 1000 }); } @@ -464,7 +466,9 @@ export function createGitHubExternalObjectProvider( return { providerKey: "github", objectType, - async resolve({ companyId, object }) { + async resolve({ companyId, object, signal: ownerSignal }) { + const deadline = AbortSignal.timeout(opts.requestTimeoutMs ?? 10_000); + const signal = ownerSignal ? AbortSignal.any([ownerSignal, deadline]) : deadline; const identity = parseGitHubObject(object); if (!identity || identity.objectType !== objectType) { return { @@ -509,7 +513,7 @@ export function createGitHubExternalObjectProvider( // "public" verdict must not authorize a later read of its private contents. // The cache is still used below to classify a 404, where a stale verdict can // only make the object look deleted — it can never expose private data. - const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (probe.visibility === "hidden") { return { ok: false, @@ -554,7 +558,7 @@ export function createGitHubExternalObjectProvider( let response: Response; try { - response = await fetchImpl(url, { headers }); + response = await fetchImpl(url, { headers, signal }); } catch { return { ok: false, @@ -573,7 +577,7 @@ export function createGitHubExternalObjectProvider( // Probe the repository metadata to classify: if the repo itself is visible // the object truly is gone; if it is not, surface credentials-needed // instead of permanently freezing a reference as "Not found". - const probe = await cachedProbeRepoVisibility(companyId, identity, headers); + const probe = await cachedProbeRepoVisibility(companyId, identity, headers, signal); if (probe.visibility === "visible") { return { ok: true, snapshot: notFoundSnapshot(identity, etag) }; } From 18b1637b6371a6a6adade857e2a73654ee626fef Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 30 Oct 2026 18:05:05 -0500 Subject: [PATCH 5/6] fix(runner): accept late Antigravity login URL during drain (#18254) ## Thinking Path > - Paperclip manages work performed by AI agents. > - The native Runner owns Antigravity subscription sign-in. > - Sign-in receives completion on stdout and the browser URL on stderr. > - The two pipes can deliver the completion result before the valid URL. > - The owner rejected that URL after it entered the completed phase. > - This pull request accepts one validated URL while the same child drains. > - A real child regression now proves the order and the failure controls. ## Linked Issues or Issue Description Refs #18188, #18189, and #18190. Related qualification work: #18345 and #18356. The duplicate search found this existing PR, so this update preserves its history and adds the real child proof here. **What happened?** A valid sign-in can fail when the authenticate response arrives before the authorization URL on the other pipe. The completed phase then rejects the valid URL. The same existing success case failed in [current follow-up CI](https://github.com/paperclipai/paperclip/actions/runs/41138618869/job/124347859445). That log has no stream-order trace, so it does not prove its inner failure branch. **Expected behavior** Both valid delivery orders must succeed. Invalid or duplicate URLs, unexpected protocol requests, failed retirement, and failed credential verification must still fail. **Steps to reproduce** 1. Run a synthetic native child that completes authenticate on stdout. 2. Wait for stdin EOF before it emits the authorization URL on stderr. EOF proves that the parent accepted the result and began retirement. 3. The valid case fails on frozen master `45a3a92e6dc52d20923b9aa34a22fee2ec5443f9`. The invalid and duplicate controls pass there. 4. Run the same regression with this PR's existing production fix. All three cases pass. **Paperclip version or commit** Existing reviewed PR parent: `646f5cdf37898ad54e1fa1cdfe39f59844da3e7c`. Frozen actual master for the real child RED: `45a3a92e6dc52d20923b9aa34a22fee2ec5443f9`. ## What Changed - Accept one fully validated authorization URL in the authentication or completed-drain phase. Keep duplicate rejection and all existing URL checks. - Retain the ten existing in-memory controls for both valid orders and eight failure paths. - Add three real child cases for valid, invalid, and duplicate late URLs. Use stdin EOF to order the streams without sleeps or a mocked retirement handler. - Check exact protocol requests, child exit, close-once behavior, Google-call counts, and persisted refresh rotation. This update changes only the existing test file; it adds no production delta to the reviewed parent. ## Verification - Frozen actual master: the new valid real child case fails, while both negative controls pass. The original eight-control parent calibration also remains recorded in this PR's history. - Existing fixed source plus the new tests: 24 passed. This includes 11 existing authentication tests, ten stream controls, and three real child cases. - Late invalid and duplicate URLs fail with zero Google calls. The valid case makes two mocked identity calls and saves the rotated synthetic refresh token. All three retire the real child and close the command lease once. - Scoped owning TypeScript `noEmit` passes for the auth module and both test files. The temporary configuration is not committed. Diff and credential-pattern checks pass. - The tests use real child pipes and private synthetic files. Google responses and the command installation are mocked. No provider account, network request, model run, app, or QA deployment was used. - Full monorepo typecheck, tests, and build were not run locally. Fresh [normal CI](https://github.com/paperclipai/paperclip/actions/runs/41140757050) passes on `1744c9f3374f78b10c2f15c42f3cf5d3af8c3119`: 56 successful checks, two skipped, and one neutral. Typecheck, build, Rust, static, Runner, server and browser lanes pass. Greptile gives that exact head a fresh 5/5 with zero unresolved threads. The PR is mergeable and clean. The earlier parent's passing CI remains historical evidence only. ## Risks The change accepts a URL while the same authenticated child drains. It still validates the exact OAuth URL and rejects a second URL. Success still requires the authenticate result, verified credentials, and proven process retirement. No provider pins, contracts, timeout bounds, retries, or public APIs change. Existing documentation already specifies these sign-in semantics; no documentation change is needed. ## Model Used OpenAI Codex, based on GPT-6, with tool use and code execution. The exact runtime model ID and context window were not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../antigravity-auth-stream-order.test.ts | 172 ++++++++++++++++++ .../src/drivers/acpx/antigravity-auth.ts | 2 +- 2 files changed, 173 insertions(+), 1 deletion(-) create mode 100644 packages/paperclip-runner/src/drivers/acpx/antigravity-auth-stream-order.test.ts diff --git a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth-stream-order.test.ts b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth-stream-order.test.ts new file mode 100644 index 00000000000..e44bb0ade13 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth-stream-order.test.ts @@ -0,0 +1,172 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PassThrough, Writable } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +const setup = vi.hoisted(() => ({ verify: vi.fn(), spawn: vi.fn(), close: vi.fn(), retire: vi.fn() })); +vi.mock("./antigravity-installation.js", () => ({ antigravityNativeDistributionSpec: () => ({ pinned: true }) })); +vi.mock("./installation-integrity.js", () => ({ verifyNativeAcpxInstallation: setup.verify })); +vi.mock("./antigravity-process.js", async importOriginal => ({ + ...await importOriginal(), retireAntigravityProcess: setup.retire, +})); +import { startAntigravityBrowserLogin } from "./antigravity-auth.js"; + +const clientId = "1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com"; +const scopes = ["https://www.googleapis.com/auth/cloud-platform", "https://www.googleapis.com/auth/userinfo.email", "https://www.googleapis.com/auth/aicode"]; +const authorizationUrl = "https://accounts.google.com/o/oauth2/v2/auth?" + new URLSearchParams({ client_id: clientId, response_type: "code", + redirect_uri: "http://127.0.0.1:18973/", scope: scopes.join(" "), state: "fixture_state_0123456789", code_challenge: "a".repeat(43), + code_challenge_method: "S256", access_type: "offline", prompt: "consent" }); +const roots: string[] = []; +const physicalChildren: ChildProcess[] = []; +afterEach(async () => { + for (const child of physicalChildren.splice(0)) { + if (child.exitCode === null && child.signalCode === null) { + try { process.kill(-child.pid!, "SIGKILL"); } catch {} + } + } + vi.unstubAllGlobals(); vi.clearAllMocks(); + await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true }))); +}); + +type Mode = "url-first" | "result-first" | "wrong-url" | "duplicate-url" | "unexpected-prompt" | "retirement-failure" | "credential-failure" | "auth-error" | "late-wrong-url" | "late-duplicate-url"; +async function fixture(mode: Mode) { + const home = await realpath(await mkdtemp(join(tmpdir(), "paperclip-antigravity-stream-test-"))); roots.push(home); + await mkdir(join(home, "antigravity-acp"), { mode: 0o700 }); + await writeFile(join(home, "antigravity-acp/acp_token.json"), JSON.stringify({ client_id: clientId, client_secret: "synthetic-client-secret", + refresh_token: "synthetic-refresh-token", token_uri: "https://oauth2.googleapis.com/token", scopes }), { mode: 0o600 }); + const stdout = new PassThrough(), stderr = new PassThrough(), requests: string[] = []; + const child = Object.assign(new EventEmitter(), { stdout, stderr, stdin: new Writable({ write(chunk, _encoding, done) { + const request = JSON.parse(String(chunk)); requests.push(request.method); + queueMicrotask(() => { + if (request.method === "initialize") { + stdout.write(JSON.stringify({ jsonrpc: "2.0", id: request.id, result: { protocolVersion: 1, + agentInfo: { name: "antigravity-acp", version: "1.4.3" }, agentCapabilities: { _meta: { antigravity: { paperclipAdmission: 1 } } }, + authMethods: [{ id: "oauth-personal" }] } }) + "\n"); + } else if (request.method === "authenticate") { + const url = mode === "wrong-url" || mode === "late-wrong-url" ? "https://example.invalid/synthetic-private-marker" : authorizationUrl; + const emitUrl = () => stderr.write("Open the following link to authenticate the ACP server: " + url + "\n"); + const resultFirst = ["result-first", "late-wrong-url", "late-duplicate-url"].includes(mode); + if (!resultFirst) emitUrl(); + if (mode === "duplicate-url") emitUrl(); + if (mode === "unexpected-prompt") stdout.write(JSON.stringify({ jsonrpc: "2.0", id: 9, method: "session/prompt", params: {} }) + "\n"); + stdout.write(JSON.stringify(mode === "auth-error" + ? { jsonrpc: "2.0", id: request.id, error: { code: -32000, message: "synthetic-private-marker" } } + : { jsonrpc: "2.0", id: request.id, result: {} }) + "\n"); + if (resultFirst) queueMicrotask(() => { emitUrl(); if (mode === "late-duplicate-url") emitUrl(); }); + } else throw new Error("Unexpected session/model request"); + }); + done(); + } }) }); + setup.close.mockResolvedValue(undefined); + setup.retire.mockImplementation(async (_child, closed: Promise) => { + await Promise.resolve(); child.emit("close", 0); await closed; + }); + if (mode === "retirement-failure") setup.retire.mockImplementationOnce(async () => { throw new Error("synthetic-private-marker"); }); + setup.spawn.mockReturnValue(child); + setup.verify.mockResolvedValue({ openCommand: async () => ({ spawn: setup.spawn, close: setup.close }) }); + const fetch = vi.fn(async (url: string) => { + if (mode === "credential-failure") return new Response(null, { status: 401 }); + if (url === "https://oauth2.googleapis.com/token") return Response.json({ access_token: "synthetic-access-token", token_type: "Bearer" }); + if (url === "https://www.googleapis.com/oauth2/v3/userinfo") return Response.json({ email: "person@example.test", email_verified: true }); + throw new Error("Unexpected network request"); + }); + vi.stubGlobal("fetch", fetch); + return { home, requests, fetch }; +} + +describe("Antigravity login cross-stream completion", () => { + it.each(["url-first", "result-first"] as const)("accepts the same authenticated exchange with %s delivery", async mode => { + const { home, requests, fetch } = await fixture(mode); + const login = startAntigravityBrowserLogin(home); + await vi.waitFor(() => expect(login.outcome).toBeDefined()); + expect(login.outcome).toBe("success"); + expect(login.authorizationUrl).toBe(authorizationUrl); + expect(requests).toEqual(["initialize", "authenticate"]); + expect(setup.retire).toHaveBeenCalledOnce(); expect(setup.close).toHaveBeenCalledOnce(); + expect(fetch).toHaveBeenCalledTimes(2); + await login.abort(); + }); + + it.each(["wrong-url", "duplicate-url", "unexpected-prompt", "retirement-failure", "credential-failure", "auth-error", "late-wrong-url", "late-duplicate-url"] as const)("keeps %s failures closed and private", async mode => { + const { home, requests, fetch } = await fixture(mode); + const login = startAntigravityBrowserLogin(home); + await vi.waitFor(() => expect(login.outcome).toBeDefined()); + expect(login.outcome).toBe("failure"); expect(login.error).toBe("Google sign-in did not complete. Start sign-in again."); + expect(login.error).not.toContain("synthetic-private-marker"); + await login.abort(); expect(setup.close).toHaveBeenCalledOnce(); + expect(requests).toEqual(["initialize", "authenticate"]); + expect(fetch).toHaveBeenCalledTimes(mode === "credential-failure" ? 1 : 0); + if (mode === "wrong-url" || mode === "late-wrong-url") expect(login.authorizationUrl).toBeUndefined(); + }); + + it.each(["valid", "invalid", "duplicate"] as const)("handles a %s URL from a real child after authenticate completion and stdin EOF", async mode => { + const home = await realpath(await mkdtemp(join(tmpdir(), "paperclip-antigravity-real-stream-"))); roots.push(home); + const peer = join(home, "peer.cjs"); + await writeFile(peer, ` +const fs = require('node:fs'); +const path = require('node:path'); +const readline = require('node:readline'); +const home = process.argv[2], mode = process.argv[3], requests = [], events = []; +const reader = readline.createInterface({input:process.stdin}); +reader.on('line', line => { + const request = JSON.parse(line); requests.push(request.method); + fs.writeFileSync(path.join(home,'requests.json'),JSON.stringify(requests)); + if(request.method==='initialize') { + process.stdout.write(JSON.stringify({jsonrpc:'2.0',id:request.id,result:{protocolVersion:1, + agentInfo:{name:'antigravity-acp',version:'1.4.3'},agentCapabilities:{_meta:{antigravity:{paperclipAdmission:1}}}, + authMethods:[{id:'oauth-personal'}]}})+'\\n'); + } else if(request.method==='authenticate') { + fs.mkdirSync(path.join(home,'antigravity-acp'),{mode:0o700}); + fs.writeFileSync(path.join(home,'antigravity-acp/acp_token.json'),${JSON.stringify(JSON.stringify({ client_id: clientId, + client_secret: "synthetic-client-secret", refresh_token: "synthetic-refresh-token", token_uri: "https://oauth2.googleapis.com/token", scopes }))},{mode:0o600}); + events.push('authenticate-result'); + process.stdout.write(JSON.stringify({jsonrpc:'2.0',id:request.id,result:{}})+'\\n'); + } else throw new Error('Unexpected session/model request'); +}); +reader.on('close',()=>{ + events.push('stdin-eof'); + const url = mode==='invalid' ? 'https://example.invalid/synthetic-private-marker' : ${JSON.stringify(authorizationUrl)}; + const message = 'Open the following link to authenticate the ACP server: '+url+'\\n'; + process.stderr.write(mode==='duplicate' ? message+message : message,()=>{ + events.push('url-flushed');fs.writeFileSync(path.join(home,'events.json'),JSON.stringify(events));process.exit(0); + }); +}); +`); + setup.spawn.mockImplementation((_args, options) => { + const child = spawn(process.execPath, [peer, home, mode], { ...options, detached: true }); + physicalChildren.push(child); return child; + }); + setup.close.mockResolvedValue(undefined); + setup.verify.mockResolvedValue({ openCommand: async () => ({ spawn: setup.spawn, close: setup.close }) }); + const realProcess = await vi.importActual("./antigravity-process.js"); + setup.retire.mockImplementation(realProcess.retireAntigravityProcess); + const fetch = vi.fn(async (url: string) => { + if (url === "https://oauth2.googleapis.com/token") return Response.json({ access_token: "synthetic-access-token", + refresh_token: "synthetic-rotated-refresh", token_type: "Bearer" }); + if (url === "https://www.googleapis.com/oauth2/v3/userinfo") return Response.json({ email: "person@example.test", email_verified: true }); + throw new Error("Unexpected network request"); + }); + vi.stubGlobal("fetch", fetch); + const login = startAntigravityBrowserLogin(home); + await vi.waitFor(() => expect(login.outcome).toBeDefined(), { timeout: 5_000 }); + expect(login.outcome).toBe(mode === "valid" ? "success" : "failure"); + expect(JSON.parse(await readFile(join(home, "events.json"), "utf8"))).toEqual(["authenticate-result", "stdin-eof", "url-flushed"]); + expect(JSON.parse(await readFile(join(home, "requests.json"), "utf8"))).toEqual(["initialize", "authenticate"]); + expect(physicalChildren.at(-1)!.exitCode).toBe(0); + expect(setup.retire).toHaveBeenCalledOnce(); expect(setup.close).toHaveBeenCalledOnce(); + expect(fetch).toHaveBeenCalledTimes(mode === "valid" ? 2 : 0); + if (mode === "valid") { + expect(login.authorizationUrl).toBe(authorizationUrl); + expect(JSON.parse(await readFile(join(home, "antigravity-acp", "acp_token.json"), "utf8")).refresh_token).toBe("synthetic-rotated-refresh"); + } else { + expect(login.error).toBe("Google sign-in did not complete. Start sign-in again."); + expect(login.error).not.toContain("synthetic-private-marker"); + if (mode === "invalid") expect(login.authorizationUrl).toBeUndefined(); + } + await login.abort(); + expect(setup.close).toHaveBeenCalledOnce(); + }, 15_000); +}); diff --git a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts index 2921fd141db..7f4b3bf54bd 100644 --- a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts +++ b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts @@ -203,7 +203,7 @@ export function startAntigravityBrowserLogin(home: string): AntigravityBrowserLo if (!line.startsWith(AUTH_PREFIX)) continue; try { const url = validateAntigravityAuthorizationUrl(line.slice(AUTH_PREFIX.length)); - if (phase !== 2 || state.authorizationUrl !== undefined) throw new Error(LOGIN_ERROR); + if ((phase !== 2 && phase !== 3) || state.authorizationUrl !== undefined) throw new Error(LOGIN_ERROR); state.authorizationUrl = url; } catch { fail(); } } From 4e97daafa06a54f02027335437c4f48e50b84236 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:09:41 -0500 Subject: [PATCH 6/6] refactor: remove obsolete external object batch refresh entrypoints Exercise lease, selection, and recovery behavior through the registered scheduler owner. Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 8 ++- .../external-object-refresh-work.test.ts | 8 --- .../external-objects-service.test.ts | 66 +++++++++++-------- .../server-startup-feedback-export.test.ts | 4 -- server/src/services/external-objects.ts | 52 --------------- 5 files changed, 43 insertions(+), 95 deletions(-) diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index 61c7d8dded1..41bf59440d1 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -106,9 +106,11 @@ 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 two explicit `refreshDueObjects*` helpers remain manual diagnostic/test entry -points. They have no production scheduler caller. Their regression fixtures also -preserve the prior selection/lease semantics. The scheduled check retains its +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. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index 754725c7d80..4c1817b9722 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -100,14 +100,6 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { }); } - it("captures the old configured-empty aggregate SQL cost", async () => { - const svc = service(); await fixture(); - const selects = vi.spyOn(db, "select"); - for (let i = 0; i < 3; i++) await svc.refreshDueObjectsForActiveCompanies(); - expect(selects).toHaveBeenCalledTimes(6); - expect(provider).not.toHaveBeenCalled(); - }); - it("activates saved work when the ordinary feature and company settings enable it", async () => { clock(); const settings = instanceSettingsService(db); await settings.updateExperimental({ enableExternalObjects: true }); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index bc268d0ba6a..e6de9780444 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -1,4 +1,5 @@ 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 { @@ -733,6 +734,18 @@ describeEmbeddedPostgres("externalObjectService", () => { await tempDb?.cleanup(); }); + async function runRefreshWork(service: ReturnType) { + 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({ @@ -882,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); }); @@ -921,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); @@ -969,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"); @@ -1032,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, @@ -1040,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(); @@ -1078,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); @@ -1139,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); }); @@ -1184,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({ @@ -1225,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"); @@ -1236,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({ @@ -1273,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({ @@ -1316,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(); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 5002dc44a50..f380ef478dd 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -209,7 +209,6 @@ const { const externalObjectsServiceMock = { registerRefreshWork: vi.fn((coordinator: { register: typeof deliveryWorkRegisterMock }, _canRun: () => boolean) => coordinator.register("external-object-refresh", { run: async () => undefined, hasPending: async () => false })), - refreshDueObjectsForActiveCompanies: vi.fn(async () => ({ companies: 0, checked: 0, refreshed: 0 })), }; const externalObjectsServiceFactoryMock = vi.fn(() => externalObjectsServiceMock); const routineServiceMock = { @@ -1404,7 +1403,6 @@ describe("startServer feedback export wiring", () => { 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(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); @@ -1434,7 +1432,6 @@ describe("startServer feedback export wiring", () => { }); await startServer(); expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledOnce(); - expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); }); it("keeps external object refresh active when heartbeat scheduling is disabled", async () => { @@ -1467,7 +1464,6 @@ describe("startServer feedback export wiring", () => { 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(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled(); diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 0208cdab7b6..a4d26014d26 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -1109,56 +1109,6 @@ 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); - const refreshable = refreshableObjectWhere(); - 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, - })); - } - 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; - } - return { companies: activeCompanies.length, checked, refreshed }; - } - function registerRefreshWork( coordinator: ReturnType, canRun: () => boolean, @@ -1234,8 +1184,6 @@ export function externalObjectService( getProjectSummary, refreshObject, refreshIssueObjects, - refreshDueObjects, - refreshDueObjectsForActiveCompanies, registerRefreshWork, }; }