diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index 41bf59440d1..e5401596a0e 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -18,6 +18,14 @@ ready plugin with `external.objects.read`, a matching provider/object declaratio and no company setting that disables it. Built-in resolvers remain available independently of optional plugin support. +A plugin result carries its selected plugin ID inside the host owner. Persistence +rechecks that exact plugin's readiness, read capability, object declaration and +same-company enabled setting in the result update. Another available resolver +cannot authorize an obsolete plugin response. A rejected response releases only +its own claim; the due row remains for a fresh allowed resolver. Plugin RPC +payloads and public result shapes do not contain this host selection context. + + Terminal objects are passive. The existing exception remains: GitHub `not_found` snapshots without `data.probeVerified=true` need a repository-visibility probe before their terminal verdict is definitive. A null deadline retains its @@ -94,7 +102,7 @@ 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 +## Host cancellation and registered ownership The resolver contract accepts an optional host `AbortSignal`. The worker passes its coordinator signal through refresh to the resolver. This reuses the narrow diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index 4c1817b9722..8aab42b35b1 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -257,6 +257,60 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { expect(calls).toHaveBeenCalledOnce(); }); + it.each(["company setting", "status", "uninstall", "manifest", "other company"] as const)( + "revalidates the selected plugin after %s changes even while the built-in fallback remains available", async change => { + clock(); const registry = pluginRegistryService(db); + const manifest: PaperclipPluginManifestV1 = { + id: `paperclip.selected-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Selected provider", + description: "Selected resolver fixture", author: "Paperclip", categories: ["connector"], + capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" }, + objectReferences: [{ providerKey: "github", displayName: "GitHub", objectTypes: ["pull_request"], urlPatterns: ["https://github.com/:owner/:repo/pull/:id"] }], + }; + const plugin = await registry.install({ packageName: "selected-provider-fixture" }, manifest); + await registry.updateStatus(plugin!.id, { status: "ready" }); + let finishPlugin!: (result: ExternalObjectResolveResult) => void; + const pluginResult = new Promise(resolve => { finishPlugin = resolve; }); + let selected = false; + const calls = vi.fn(async (_id: string, method: string) => { + if (method === "detectExternalObjects") return { detections: [] }; + selected = true; return pluginResult; + }); + let finishFallback!: () => void; + const fallbackResult = new Promise(resolve => { finishFallback = resolve; }); + const fallback = vi.fn(async () => { + await fallbackResult; + return new Response(JSON.stringify({ state: "closed", merged: true, title: "Allowed fallback" }), { status: 200 }); + }); + const svc = externalObjectService(db, { pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager, + github: { fetch: fallback, credentialProvider: () => ({ token: "simulated-company-token", source: "company_secret" }) } }); + try { + await start(svc); const f = await fixture(); + await db.update(issues).set({ description: "https://github.com/acme/app/pull/42" }).where(eq(issues.id, f.issueId)); + await svc.syncIssue(f.issueId); await settle(() => selected); + if (change === "company setting") await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} }); + else if (change === "status") await registry.updateStatus(plugin!.id, { status: "error" }); + else if (change === "uninstall") await registry.uninstall(plugin!.id, true); + else if (change === "manifest") await registry.update(plugin!.id, { manifest: { ...manifest, capabilities: [], objectReferences: [] } }); + else { + const other = await fixture(); + await registry.upsertCompanySettings(plugin!.id, other.companyId, { enabled: false, settingsJson: {} }); + } + finishPlugin({ ok: true, snapshot: { statusKey: "plugin_result", statusCategory: "succeeded", statusTone: "success", isTerminal: true } }); + if (change === "other company") { + await settle(async () => (await row(f.companyId))?.statusKey === "plugin_result"); await quiet(); + expect(fallback).not.toHaveBeenCalled(); + } else { + await settle(async () => fallback.mock.calls.length > 0 || Boolean((await row(f.companyId))?.lastResolvedAt)); + const beforeFallback = await row(f.companyId); + expect(beforeFallback.isTerminal).toBe(false); expect(beforeFallback.statusKey).toBeNull(); expect(beforeFallback.lastResolvedAt).toBeNull(); + expect(fallback).toHaveBeenCalledOnce(); + finishFallback(); + await settle(async () => (await row(f.companyId))?.statusKey === "merged"); await quiet(); + expect((await row(f.companyId)).displayTitle).toBe("acme/app#42: Allowed fallback"); + } + } finally { finishPlugin(terminal); finishFallback(); } + }); + it("reactivates refresh after ordinary budget pause and budget raise", async () => { clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc); const f = await fixture(); await svc.syncIssue(f.issueId); diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index a4d26014d26..81e10a208f2 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -346,7 +346,7 @@ async function resolveViaPluginProvider( db: Db, pluginWorkerManager: PluginWorkerManager | undefined, object: ExternalObjectRecord, -): Promise { +): Promise<{ pluginId: string; result: PluginExternalObjectResolveResult } | null> { if (!pluginWorkerManager) return null; const providers = await readyObjectReferencePlugins(db, object.companyId); for (const provider of providers) { @@ -354,23 +354,27 @@ async function resolveViaPluginProvider( if (!manifest.capabilities.includes("external.objects.read")) continue; if (!manifestProvidesObject(manifest, object)) continue; try { - return await pluginWorkerManager.call(provider.id, "resolveExternalObject", { + const result = await pluginWorkerManager.call(provider.id, "resolveExternalObject", { companyId: object.companyId, providerKey: object.providerKey, objectType: object.objectType, externalId: object.externalId, object: objectSnapshot(object), }); + return { pluginId: provider.id, result }; } catch (err) { logger.warn( { err, pluginId: provider.id, pluginKey: provider.pluginKey, objectId: object.id }, "plugin external object resolver failed", ); return { - ok: false, - liveness: "unreachable", - errorCode: "plugin_resolver_failed", - errorMessage: err instanceof Error ? err.message : String(err), + pluginId: provider.id, + result: { + ok: false, + liveness: "unreachable", + errorCode: "plugin_resolver_failed", + errorMessage: err instanceof Error ? err.message : String(err), + }, }; } } @@ -402,12 +406,9 @@ export function externalObjectService( ]; 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( + function pluginRefreshWhere(database: Db, selectedPluginId?: string) { + return exists(database.select({ id: plugins.id }).from(plugins).where(and( + selectedPluginId ? eq(plugins.id, selectedPluginId) : undefined, eq(plugins.status, "ready"), sql`${plugins.manifestJson}->'capabilities' ? 'external.objects.read'`, sql`exists (select 1 from jsonb_array_elements(coalesce(${plugins.manifestJson}->'objectReferences', '[]'::jsonb)) as reference @@ -418,7 +419,15 @@ export function externalObjectService( eq(pluginCompanySettings.companyId, externalObjects.companyId), eq(pluginCompanySettings.enabled, false), )))), - )))); + ))); + } + + function automaticRefreshWhere(database: Db = db) { + const supported = resolvers.map(resolver => and( + eq(externalObjects.providerKey, resolver.providerKey), + resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined, + )!); + if (opts.pluginWorkerManager) supported.push(pluginRefreshWhere(database)); return and( refreshableObjectWhere(), isNotNull(externalObjects.nextRefreshAt), supported.length ? or(...supported) : sql`false`, @@ -863,12 +872,14 @@ export function externalObjectService( } async function persistRefresh(object: ExternalObjectRecord, token: string, - patch: Partial, scheduled = false) { + patch: Partial, scheduled = false, selectedPluginId?: string) { return db.transaction(async tx => { await notifyExternalObjectRefresh(tx, object.companyId, object.id); const rows = await tx.update(externalObjects).set(patch) - .where(and(refreshOwnerWhere(object, token), scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined)).returning(); - if (scheduled && rows.length === 0) await tx.update(externalObjects) + .where(and(refreshOwnerWhere(object, token), + scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined, + selectedPluginId ? pluginRefreshWhere(tx as unknown as Db, selectedPluginId) : undefined)).returning(); + if ((scheduled || selectedPluginId) && rows.length === 0) await tx.update(externalObjects) .set({ refreshStartedAt: null, refreshToken: null }).where(refreshOwnerWhere(object, token)); return rows; }); @@ -881,9 +892,9 @@ export function externalObjectService( refreshToken: string, ) { 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 pluginResolution = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal); + const resolver = pluginResolution ? null : resolverRegistry.find(object); + if (!pluginResolution && !resolver) { const [updated] = await persistRefresh(object, refreshToken, { liveness: object.isTerminal ? object.liveness : visibleLiveness(object, now) === "fresh" ? "stale" : object.liveness, nextRefreshAt: addSeconds(now, DEFAULT_RETRY_AFTER_SECONDS), @@ -895,7 +906,7 @@ export function externalObjectService( return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const }; } - const result = pluginResult ?? await observeRefreshCancellation( + const result = pluginResolution?.result ?? await observeRefreshCancellation( resolver!.resolve({ companyId: object.companyId, object, signal: input.signal }), input.signal); input.signal?.throwIfAborted(); if (!result.ok) { @@ -913,7 +924,7 @@ export function externalObjectService( refreshStartedAt: null, refreshToken: null, updatedAt: now, - }, input.scheduled); + }, input.scheduled, pluginResolution?.pluginId); if (!updated) return refreshSupersededResult(object, now); publishLiveEvent({ companyId: object.companyId, @@ -950,7 +961,7 @@ export function externalObjectService( const [updated] = await persistRefresh(object, refreshToken, { ...patch, lastChangedAt: objectChanged(object, { ...object, ...patch }) ? now : object.lastChangedAt, - }, input.scheduled); + }, input.scheduled, pluginResolution?.pluginId); if (!updated) return refreshSupersededResult(object, now); const next = updated ?? object; if (objectChanged(object, next) && input.actor) {