diff --git a/doc/plans/2026-10-30-github-connection-sleep.md b/doc/plans/2026-10-30-github-connection-sleep.md new file mode 100644 index 00000000000..11271110a7a --- /dev/null +++ b/doc/plans/2026-10-30-github-connection-sleep.md @@ -0,0 +1,83 @@ +# GitHub connection delivery and credential continuity + +GitHub event delivery and credential continuity have separate obligations. +The event worker uses the shared delivery coordinator. It performs a startup +reconciliation and responds to committed configuration changes or authenticated +Cloud hints. Credential work registers the actual next grant expiry, provider +continuity check, or saved retry deadline. The aggregate heartbeat no longer +calls either service, including when the heartbeat scheduler is disabled. + +## Event delivery contract + +The Cloud webhook target queue remains durable truth. The tenant leases sealed +events, validates their bindings, persists company-scoped receipts, applies the +existing installation or pull-request behavior, and acknowledges completed work. +A receipt row lock prevents overlapping receivers from applying an installation +delta twice. Completed receipts remain passive history. The existing normalized +merged-pull-request callback remains unchanged. + +A new tenant sends signed event-lease claim `ewv=1`. A compatible Cloud broker +records delivery version 1 on the existing instance registration before returning +`eventWakeVersion=1`. Only then does the tenant stop configured-empty polling. +Cloud reconstructs unacknowledged targets after restart and retries the exact +registered tenant when its current queue lease permits another delivery. A +successful HTTP hint does not acknowledge the target. Missing, stopped or drained +tenants therefore retain their work. The hint carries no provider payload or +client-selected destination. The tenant checks a signed, stack-bound, single-use +`connector-events:wake` control assertion on the exact POST endpoint. + +Old Cloud and self-hosted registrations retain scoped reconciliation with the +existing backoff. No connector configuration or no registration history settles +to no recurring owner SQL or provider traffic. Disabled consumers can acknowledge +queued events as ignored using their trusted local registration history; they +cannot apply events to an inactive connection. Removing polling does not remove +the host sleep safety blockers. Worker quiet is not proof of host suspension. + +## Durable credential debt + +A grant stores a unique `bindingSync` revision and removed installation IDs before +Cloud registration calls. Successful delivery clears only the locked current +revision. Failure retains the saved intent and an exponential retry deadline, +including provider Retry-After advice. A restart loads that deadline from the +grant. Binding-only retries use the existing credential refresh lease and current +secret resolver without repeating account/repository discovery. The initial OAuth +callback uses the same saved debt and settlement path after connection activation. + +Credential expiry remains enforced by the existing use path. Scheduled continuity +checks run one hour before recorded access expiry or after the existing 30-day +continuity window. A successful renewal that leaves the credential short-lived +or unchanged retains a one-hour retry obligation instead of causing rapid calls. +Changed credential metadata invalidates the old retry revision, so an earlier new +expiry can preempt it. Disabled connections and archived companies do not perform +credential work. Reactivation uses the normal committed configuration signal. + +Supported local producers are the owning connection, OAuth, grant and company +services in one tenant process. Their notifications belong to the outer database +transaction. A savepoint does not commit a wake, and rollback cancels the write. +Arbitrary SQL writers must use these notification contracts or restart the owner; +process-local signals do not promise independent-writer or replica notification. +External event delivery remains durable in Cloud across processes and restarts. + +## Paired rollout and rollback + +1. Apply the companion Cloud registration-column migration and deploy its delivery + worker with the control signer present. Keep new enrollment disabled initially. +2. Deploy this tenant version. It keeps fallback polling against older Cloud. +3. Enable Cloud enrollment. Version 1 activation is durable and idempotent. Turning + the new-enrollment flag off must still service all existing version 1 records. +4. Before removing the Cloud delivery worker, restart affected tenants with + `PAPERCLIP_GITHUB_EVENT_POLLING=1`. They arm fallback before sending signed + `ewv=0`. Confirm the Cloud registration is cleared. A failed handoff leaves + fallback active and Cloud retry intact. Alternatively, restore the old polling + tenant image first while retaining the compatible Cloud worker. +5. Do not deploy a Cloud downgrade that removes responsibility for still-enrolled + tenants. There is no capability expiry, renewal poll, or periodic tenant hint. + +## Evidence + +Focused tests use isolated PostgreSQL plus controlled HTTP provider simulations. +They cover first service write after quiet, saved results, terminal receipts, +concurrent replay, signed callbacks, outer commit/rollback, drain/resume, startup +wiring in both scheduler branches, credential backoff and changed deadlines. +Cloud acceptance uses its actual broker/store/notifier boundaries and a controlled +HTTP tenant. These tests do not establish a deployed fleet rollout or cost savings. diff --git a/packages/db/src/schema/tool_access.ts b/packages/db/src/schema/tool_access.ts index b8b317ddb91..bbe6c30e584 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -231,6 +231,8 @@ export const connectionGrants = pgTable( appSlug?: string; accessRevision?: string; lastAccessRefreshAt?: string; + bindingSync?: { id: string; removeInstallationIds: string[] }; + continuityRetry?: { at: string; attempts: number; revision: string }; lastWebhookAt?: string; webhookHealth?: "pending" | "healthy" | "unhealthy"; }; diff --git a/packages/shared/src/types/tool-access.ts b/packages/shared/src/types/tool-access.ts index 0fbd422fdad..5627ca6a174 100644 --- a/packages/shared/src/types/tool-access.ts +++ b/packages/shared/src/types/tool-access.ts @@ -266,6 +266,9 @@ export interface ConnectionGrant { appSlug?: string; accessRevision?: string; lastAccessRefreshAt?: string; + /** Unsettled Cloud binding changes; retained until the exact revision is delivered. */ + bindingSync?: { id: string; removeInstallationIds: string[] }; + continuityRetry?: { at: string; attempts: number; revision: string }; lastWebhookAt?: string; webhookHealth?: "pending" | "healthy" | "unhealthy"; }; diff --git a/packages/shared/src/validators/tool-access.ts b/packages/shared/src/validators/tool-access.ts index bafe22e80e1..914af74e7ab 100644 --- a/packages/shared/src/validators/tool-access.ts +++ b/packages/shared/src/validators/tool-access.ts @@ -240,6 +240,8 @@ export const connectionGrantSchema = z.object({ managementUrl: z.string().url().max(2000).optional(), appSlug: z.string().regex(/^[a-z0-9-]{1,100}$/).optional(), lastAccessRefreshAt: z.string().datetime().optional(), + bindingSync: z.object({ id: z.string().uuid(), removeInstallationIds: z.array(z.string().regex(/^[1-9][0-9]{0,30}$/)).max(100) }).optional(), + continuityRetry: z.object({ at: z.string().datetime(), attempts: z.number().int().min(1).max(16), revision: z.string().max(2000) }).optional(), lastWebhookAt: z.string().datetime().optional(), webhookHealth: z.enum(["pending", "healthy", "unhealthy"]).optional(), }).optional(), diff --git a/server/src/__tests__/cloud-control-task-drain.test.ts b/server/src/__tests__/cloud-control-task-drain.test.ts index 3f0ab8e1951..46c2962edc5 100644 --- a/server/src/__tests__/cloud-control-task-drain.test.ts +++ b/server/src/__tests__/cloud-control-task-drain.test.ts @@ -1,7 +1,9 @@ +import { connectionEventWakeRoutes } from "../routes/connection-event-wake.js"; +import { registerConnectionEventWake } from "../services/connection-event-wake.js"; import { generateKeyPairSync, sign } from "node:crypto"; import express from "express"; import request from "supertest"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { CLOUD_CONTROL_AUDIENCE, CLOUD_CONTROL_HEADER, @@ -184,6 +186,7 @@ describe("verifyCloudControlAssertion", () => { }); describe("cloudControlMiddleware", () => { + const eventOwner = {}; const savedEnv: Record = {}; beforeEach(() => { @@ -208,6 +211,7 @@ describe("cloudControlMiddleware", () => { next(); }); app.use(cloudControlMiddleware()); + app.use(connectionEventWakeRoutes(eventOwner)); app.all("/api/instance/task-drain", (req, res) => { res.json({ actor: req.actor }); }); @@ -241,6 +245,22 @@ describe("cloudControlMiddleware", () => { return `${header}.${payload}.${signature}`; } + it("accepts only signed exact-action event hints and leaves unavailable receivers retryable", async () => { + const app = createApp(); const path = "/api/instance/connector-events/wake"; + expect((await request(app).post(path)).status).toBe(403); + expect((await request(app).post(path).set(CLOUD_CONTROL_HEADER, freshAssertion("lifecycle:read"))).status).toBe(401); + expect((await request(app).post(path).set(CLOUD_CONTROL_HEADER, freshAssertion("connector-events:wake"))).status).toBe(503); + const wake = vi.fn(); const stop = registerConnectionEventWake(eventOwner, wake); + try { + const token = freshAssertion("connector-events:wake"); + expect((await request(app).post(path).set(CLOUD_CONTROL_HEADER, token)).status).toBe(202); + expect((await request(app).post(path).set(CLOUD_CONTROL_HEADER, token)).status).toBe(401); + expect((await request(app).post(path).set(CLOUD_CONTROL_HEADER, freshAssertion("connector-events:wake"))).status).toBe(202); + expect(wake).toHaveBeenCalledTimes(2); + } finally { stop(); } + expect((await request(app).post(path).set(CLOUD_CONTROL_HEADER, freshAssertion("connector-events:wake"))).status).toBe(503); + }); + it("installs a synthetic instance-admin board actor for a valid assertion, per method", async () => { const app = createApp(); for (const [method, action] of [ diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 7c7b7025243..52cc835a444 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -35,7 +35,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: 4 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { @@ -343,7 +343,11 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { throw new Error("fixture live subscriber failed"); }); try { - await expect(service.pollOnce()).resolves.toMatchObject({ processed: 1, duplicate: 0, failed: 0 }); + const competing = githubConnectionEventService(db, { connector, now: () => currentTime }); + const outcomes = await Promise.all([service.pollOnce(), competing.pollOnce()]); + expect(outcomes.reduce((sum, result) => sum + result.processed, 0)).toBe(1); + expect(outcomes.reduce((sum, result) => sum + result.duplicate, 0)).toBe(1); + expect(outcomes.every(result => result.failed === 0)).toBe(true); } finally { unsubscribe(); } await health.expectDeadline(currentTime.getTime() + 900_000 - (refreshed ? 60_000 : 0)); await health.expectQuiet(); diff --git a/server/src/__tests__/github-connection-work.test.ts b/server/src/__tests__/github-connection-work.test.ts new file mode 100644 index 00000000000..4e789eedb2b --- /dev/null +++ b/server/src/__tests__/github-connection-work.test.ts @@ -0,0 +1,192 @@ +import { randomUUID, generateKeyPairSync, verify, createHash } from "node:crypto"; +import { createServer, type Server } from "node:http"; +import { once } from "node:events"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { applyPendingMigrations, companies, connectionGrants, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createGitHubGrantContinuity, githubGrantContinuityDeadline } from "../services/tool-access/github-grant-continuity.js"; +import { createPaperclipCloudConnector } from "../services/paperclip-cloud-connector.js"; +import { githubConnectionEventService } from "../services/github-connection-events.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerGitHubConnectionContinuityWork, registerGitHubConnectionEventWork } from "../services/github-connection-work.js"; +import { notifyGitHubCredentialChange } from "../services/github-connection-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_GITHUB_WORK_TEST_DATABASE_URL"); +const describeDatabase = databaseUrl ? describe : describe.skip; + +describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (simulation)", () => { + let db: ReturnType; + let server: Server, origin: string; + let bindings: Array>, leases: number, failBindings: boolean, githubRequests: number; + const signing = generateKeyPairSync("ed25519"), sealing = generateKeyPairSync("x25519"); + const realFetch = globalThis.fetch; + let companyIds: string[] = []; + beforeAll(async () => { + await applyPendingMigrations(databaseUrl!); + db = createDb(databaseUrl!, { maxConnections: 4 }); + server = createServer(async (req, res) => { + const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(Buffer.from(chunk)); + const raw = Buffer.concat(chunks).toString(); + if (req.url!.startsWith("/v1/connector/")) { + const body = JSON.parse(raw), [header, payload, signature] = body.request.split("."); + expect(verify(null, Buffer.from(`${header}.${payload}`), signing.publicKey, Buffer.from(signature, "base64url"))).toBe(true); + const claims = JSON.parse(Buffer.from(payload, "base64url").toString()); + expect(claims.iss).toBe("instance"); expect(claims.env).toBe("test"); + expect(claims.aud).toBe(`${origin}${req.url}`); + if (req.url!.endsWith("/events/lease")) { + leases++; res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ events: [], ...(claims.ewv === 1 ? { eventWakeVersion: 1 } : {}) })); return; + } + expect(claims.sh).toBe(createHash("sha256").update(body.binding).digest("base64url")); + bindings.push(JSON.parse(body.binding)); + res.writeHead(failBindings ? 503 : 200, { "content-type": "application/json" }).end("{}"); return; + } + githubRequests++; + const body = req.url === "/user" ? { id: 42, login: "octocat" } + : req.url!.startsWith("/user/installations?") ? { installations: [{ id: 101, repository_selection: "all", account: { login: "octocat" } }] } + : { total_count: 1, repositories: [{ id: 999, full_name: "octocat/repo" }] }; + res.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify(body)); + }); + server.listen(0, "127.0.0.1"); await once(server, "listening"); + const address = server.address(); if (!address || typeof address === "string") throw new Error("HTTP fixture did not start"); + origin = `http://127.0.0.1:${address.port}`; + }); + afterEach(async () => { + vi.restoreAllMocks(); vi.useRealTimers(); + for (const id of companyIds) await db.delete(companies).where(eq(companies.id, id)); companyIds = []; + }); + afterAll(async () => { server.close(); await once(server, "close"); await db.$client.end({ timeout: 1 }); }); + + async function fixture(ids: string[] = []) { + bindings = []; leases = 0; failBindings = false; githubRequests = 0; + const companyId = randomUUID(); companyIds.push(companyId); + const [company] = await db.insert(companies).values({ id: companyId, name: "GitHub work test", issuePrefix: "GWT" }).returning(); + const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_server", status: "active" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app!.id, uid: randomUUID(), name: "GitHub", status: "active", enabled: true, + transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } }, transportConfig: {} }).returning(); + const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId: connection!.id, kind: "user", subjectUserId: "owner", status: "active", + credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: randomUUID(), label: "Access", provider: "local_encrypted" }], + providerTenant: { oauth: { accessTokenExpiresAt: new Date(Date.now() + 8 * 3600_000).toISOString(), refreshedAt: new Date().toISOString() }, github: { + userId: "42", login: "octocat", installationCount: ids.length, repositoryCount: 1, repositorySelection: "all", installationIds: ids, + installationOwnerLogins: ["octocat"], accessRevision: randomUUID(), lastAccessRefreshAt: new Date().toISOString(), webhookHealth: "healthy", + } } }).returning(); + const connector = createPaperclipCloudConnector({ config: { baseUrl: origin, instanceId: "instance", environment: "test", + signPrivateKey: signing.privateKey.export({ format: "jwk" }).d!, sealPrivateKey: sealing.privateKey.export({ format: "jwk" }).d! } }); + type Dependencies = Parameters[1]; + const dependencies: Dependencies = { + now: () => new Date(), currentCloudConnector: () => connector, + refreshOAuthGrantCredentials: async input => (await db.select().from(connectionGrants).where(eq(connectionGrants.id, input.grantId)))[0]!, + resolveOAuthGrantSecret: (async () => ({ value: "controlled-token" })) as Dependencies["resolveOAuthGrantSecret"], + }; + const service = createGitHubGrantContinuity(db, dependencies); + return { company: company!, connection: connection!, grant: grant!, connector, service, dependencies }; + } + const fakeClock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + async function settled(coordinator: ReturnType) { + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(coordinator.nextWakeAt()).not.toBeNull(); + } + + it("ordinary grant refresh activates events from quiet, commits durable result, then performs no owner SQL/HTTP across old cadence windows", async () => { + const f = await fixture(); fakeClock(); + vi.spyOn(globalThis, "fetch").mockImplementation((url, init) => String(url).startsWith("https://api.github.com") + ? realFetch(`${origin}${new URL(String(url)).pathname}${new URL(String(url)).search}`, init) : realFetch(url, init)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const events = githubConnectionEventService(db, { connector: f.connector }); + const ew = registerGitHubConnectionEventWork(db, coordinator, events, () => true); + const cw = registerGitHubConnectionContinuityWork(db, coordinator, f.service, () => true); + try { + await Promise.all([ew.ready, cw.ready]); expect(leases).toBe(0); + await f.service.refreshManagedGitHubGrantAccess(f.connection, f.grant, { actorType: "system", actorId: "test" }); + await vi.waitFor(() => expect(leases).toBeGreaterThan(0)); + await vi.advanceTimersByTimeAsync(31_000); await settled(coordinator); + const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); + expect(saved!.providerTenant!.github!.installationIds).toEqual(["101"]); + expect(saved!.providerTenant!.github!.bindingSync).toBeUndefined(); + expect(bindings).toContainEqual(expect.objectContaining({ active: true, installationId: "101", id: `${f.grant.id}_101` })); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + const before = { leases, githubRequests, bindings: bindings.length }; + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); + expect({ leases, githubRequests, bindings: bindings.length }).toEqual(before); expect(errors).toEqual([]); + } finally { await ew.stop(); await cw.stop(); await coordinator.stop(); } + }); + + it("failed binding removal persists retry/debt through restart, uses current lease deadline and settles without refetching GitHub metadata", async () => { + const f = await fixture(["101"]); fakeClock(); + await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant, github: { ...f.grant.providerTenant!.github!, + bindingSync: { id: randomUUID(), removeInstallationIds: ["202"] } } } }).where(eq(connectionGrants.id, f.grant.id)); + failBindings = true; + expect(await f.service.sweepGitHubConnectionContinuity()).toMatchObject({ due: 1, failed: 1 }); + const [failed] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); + expect(failed!.providerTenant!.github!.bindingSync!.removeInstallationIds).toEqual(["202"]); + const retryAt = Date.parse(failed!.providerTenant!.github!.continuityRetry!.at); + const restarted = createGitHubGrantContinuity(db, f.dependencies); + expect(await restarted.nextGitHubConnectionContinuityAt()).toBe(retryAt); + await vi.advanceTimersByTimeAsync(29_000); + expect(await restarted.sweepGitHubConnectionContinuity()).toMatchObject({ due: 0 }); + failBindings = false; await vi.advanceTimersByTimeAsync(2_000); + await Promise.all([restarted.sweepGitHubConnectionContinuity(), restarted.sweepGitHubConnectionContinuity()]); + const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); + expect(saved!.providerTenant!.github!.bindingSync).toBeUndefined(); expect(saved!.providerTenant!.github!.continuityRetry).toBeUndefined(); + expect(bindings.filter(binding => binding.installationId === "202")).toHaveLength(1); + expect(githubRequests).toBe(0); + expect(await restarted.nextGitHubConnectionContinuityAt()).toBeGreaterThan(Date.now()); + await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, f.connection.id)); + expect(await restarted.nextGitHubConnectionContinuityAt()).toBeNull(); + await db.update(toolConnections).set({ enabled: true }).where(eq(toolConnections.id, f.connection.id)); + await db.update(companies).set({ status: "archived" }).where(eq(companies.id, f.company.id)); + expect(await restarted.nextGitHubConnectionContinuityAt()).toBeNull(); + }); + + it("retains a real deadline when successful renewal leaves a short-lived credential unchanged", async () => { + const f = await fixture(["101"]); fakeClock(); + vi.spyOn(globalThis, "fetch").mockImplementation((url, init) => String(url).startsWith("https://api.github.com") + ? realFetch(`${origin}${new URL(String(url)).pathname}${new URL(String(url)).search}`, init) : realFetch(url, init)); + await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant, + oauth: { ...f.grant.providerTenant!.oauth, accessTokenExpiresAt: new Date(Date.now() + 30 * 60_000).toISOString() } } }).where(eq(connectionGrants.id, f.grant.id)); + expect(await f.service.sweepGitHubConnectionContinuity()).toMatchObject({ due: 1, refreshed: 1 }); + const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); + const retryAt = Date.parse(saved!.providerTenant!.github!.continuityRetry!.at); + expect(retryAt).toBeGreaterThanOrEqual(Date.now() + 59 * 60_000); + const restarted = createGitHubGrantContinuity(db, f.dependencies); + expect(await restarted.nextGitHubConnectionContinuityAt()).toBe(retryAt); + const requests = githubRequests; await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(await restarted.sweepGitHubConnectionContinuity()).toMatchObject({ due: 0 }); expect(githubRequests).toBe(requests); + }); + + it("outer commit, rollback and first producer during drain retain the deadline and resume obligation without cross-company work", async () => { + const f = await fixture(["101"]); fakeClock(); + let enabled = false; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + const worker = registerGitHubConnectionContinuityWork(db, coordinator, f.service, () => enabled); + try { + await worker.ready; + const dueAt = Date.now() + 90 * 60_000; + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const transaction = db.transaction(async tx => { + await tx.transaction(async nested => { + await notifyGitHubCredentialChange(nested, f.connection); + await nested.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant, + oauth: { ...f.grant.providerTenant!.oauth, accessTokenExpiresAt: new Date(dueAt + 3600_000).toISOString() } } }).where(eq(connectionGrants.id, f.grant.id)); + }); + await held; + }); + await new Promise(resolve => setImmediate(resolve)); + expect(bindings).toHaveLength(0); release(); await transaction; + await vi.advanceTimersByTimeAsync(31_000); + expect(idleWorkSnapshot().active).toBe(0); expect(bindings).toHaveLength(0); + enabled = true; await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(dueAt)); + await expect(db.transaction(async tx => { await notifyGitHubCredentialChange(tx, f.connection); + await tx.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, f.grant.id)); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(dueAt)); expect(errors).toEqual([]); + } finally { await worker.stop(); await coordinator.stop(); } + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bbb020c865d..d73175ee556 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -11,7 +11,7 @@ import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import * as idleLocalWork from "../services/idle-local-work.js"; import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; -import { decisionService, decisionRetentionService, toolAccessService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; +import { decisionService, decisionRetentionService, toolAccessService, githubConnectionEventService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; import { companies, runDatabaseBackup, type Db } from "@paperclipai/db"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -36,6 +36,13 @@ vi.mock("../services/cloud-configuration-migration.js", () => ({ migrateCloudCon vi.mock("../services/cloud-configuration-qualification.js", () => ({ qualifyCloudConfigurationForCurrentImage: cloudMigrationMocks.qualify })); vi.mock("../telemetry.js", () => ({ initTelemetry: cloudRuntimeStartupMocks.telemetry, getTelemetryClient: vi.fn(() => null) })); +// Producer transaction/notification behavior is covered by the real PostgreSQL +// GitHub worker suite; this startup seam supplies a lightweight database mock. +vi.mock("../services/delivery-work-notifications.js", async () => ({ + ...await vi.importActual("../services/delivery-work-notifications.js"), + subscribeDeliveryWork: vi.fn(() => () => {}), +})); + const ORIGINAL_PAPERCLIP_API_URL = process.env.PAPERCLIP_API_URL; const ORIGINAL_PAPERCLIP_RUNTIME_API_URL = process.env.PAPERCLIP_RUNTIME_API_URL; const ORIGINAL_PAPERCLIP_RUNTIME_API_CANDIDATES_JSON = process.env.PAPERCLIP_RUNTIME_API_CANDIDATES_JSON; @@ -432,6 +439,7 @@ vi.mock("../services/index.js", () => ({ externalObjectService: externalObjectsServiceFactoryMock, heartbeatService: heartbeatServiceFactoryMock, githubConnectionEventService: vi.fn(() => ({ + nextPollAt: vi.fn(() => null), pollOnce: vi.fn(async () => ({ leased: 0, processed: 0, @@ -483,6 +491,7 @@ vi.mock("../services/index.js", () => ({ needsAttention: 0, failed: 0, })), + nextGitHubConnectionContinuityAt: vi.fn(async () => null), sweepGitHubConnectionContinuity: vi.fn(async () => ({ checked: 0, due: 0, @@ -1280,6 +1289,30 @@ describe("startServer feedback export wiring", () => { expect(completionSweepMock).not.toHaveBeenCalled(); }); + it.each([true, false])("registers GitHub delivery/deadline workers once without aggregate polling (scheduler=%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 30_000 })); + const ticks: Array<() => void> = []; + const intervals = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, ms: number) => { + if (ms === 30_000) ticks.push(callback); + return 1 as unknown as ReturnType; + }) as typeof setInterval); + try { + await startServer(); await settleStartupHousekeeping(); + for (const queue of [DELIVERY_QUEUES.connectionEvents, DELIVERY_QUEUES.connectionContinuity]) { + expect(deliveryWorkRegisterMock.mock.calls.filter(([registered]) => registered === queue)).toHaveLength(1); + } + const polls = vi.mocked(githubConnectionEventService).mock.results.map(result => result.value.pollOnce); + const sweeps = vi.mocked(toolAccessService).mock.results.map(result => result.value.sweepGitHubConnectionContinuity); + const count = () => [...polls, ...sweeps].reduce((sum, call) => sum + call.mock.calls.length, 0); + const startup = count(); + expect(startup).toBe(2); + for (let i = 0; i < 3; i++) for (const tick of ticks) tick(); + await new Promise(resolve => setImmediate(resolve)); + expect(count()).toBe(startup); + expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.connectionEvents || error.queue === DELIVERY_QUEUES.connectionContinuity)).toEqual([]); + } finally { intervals.mockRestore(); } + }); + it("does not rescan idle tool health from the aggregate maintenance tick", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, heartbeatSchedulerIntervalMs: 30_000 })); let tick: (() => void) | undefined; diff --git a/server/src/app.ts b/server/src/app.ts index 42ab549d582..d374063f4b2 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1,3 +1,4 @@ +import { connectionEventWakeRoutes } from "./routes/connection-event-wake.js"; import { registerGitHubReviewProjectionWork } from "./services/chat-channels/github/review-projection-work.js"; import { finishServiceShutdown } from "./shutdown.js"; import { promptInjectionPreventionRoutes } from "./routes/prompt-injection-prevention.js"; @@ -577,6 +578,7 @@ export async function createApp( // REPLACES whatever actor the request otherwise resolved to, and only on // the one endpoint it authorizes (see the middleware for the contract). app.use(cloudControlMiddleware()); + app.use(connectionEventWakeRoutes(db)); app.use(skillAccessRequestContext); app.use("/api/auth", authRoutes(db)); if (opts.betterAuthHandler) { diff --git a/server/src/index.ts b/server/src/index.ts index c127753889c..b2db762a808 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerGitHubConnectionEventWork, registerGitHubConnectionContinuityWork } from "./services/github-connection-work.js"; import { registerProviderTraceRetentionWork } from "./services/provider-trace-retention-work.js"; import { registerRunEvidenceMigrationWork } from "./services/run-evidence-migration-work.js"; import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js"; @@ -1490,35 +1491,13 @@ async function startServerWithDatabaseTeardown( ?? process.env.PAPERCLIP_TOOL_RUNTIME_TRUSTED_HOST ?? null, }); - const scheduleGitHubConnectionEventPoll = (forSetup = false) => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - const work = githubConnectionEvents.pollOnce({ forSetup }) - .then((result) => { - if (result.leased > 0 || result.failed > 0) { - logger.info(result, "GitHub connection event poll completed"); - } - }) - .catch((err) => { - logger.error({ err }, "GitHub connection event poll failed"); - }); - trackHeartbeatSchedulerWork(work); - return work; - }; + const githubCanRun = () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped; + const githubEventWorker = registerGitHubConnectionEventWork(db, deliveryWork, githubConnectionEvents, githubCanRun); + const githubContinuityWorker = registerGitHubConnectionContinuityWork(db, deliveryWork, tools, githubCanRun); + server.on("close", () => { void githubEventWorker.stop(); void githubContinuityWorker.stop(); }); server.on("close", registerGitHubBotCloudSetupPoller(db, async () => { - await scheduleGitHubConnectionEventPoll(true); + if (githubCanRun()) await githubConnectionEvents.pollOnce({ forSetup: true }); })); - const scheduleGitHubConnectionContinuitySweep = () => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(tools.sweepGitHubConnectionContinuity() - .then((result) => { - if (result.due > 0 || result.failed > 0) { - logger.info(result, "GitHub connection continuity sweep completed"); - } - }) - .catch((err) => { - logger.error({ err }, "GitHub connection continuity sweep failed"); - })); - }; await timeStartupRecovery("tool_reviews", async () => { if (config.heartbeatSchedulerEnabled) { @@ -1549,8 +1528,7 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready : app.locals.toolGateway.cleanupExpiredSessions()).catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed")); await timeStartupRecovery("delivery_workers", () => Promise.all([budgetEnforcementWorker, accountingWorker, fastResponseWorker, ...deliveryWorkers].map(worker => worker.ready))); - scheduleGitHubConnectionEventPoll(); - scheduleGitHubConnectionContinuitySweep(); + await Promise.all([githubEventWorker.ready, githubContinuityWorker.ready]); if (heartbeat) { const secretProposals = createSecretProposalsService(db as any); @@ -2011,8 +1989,6 @@ async function startServerWithDatabaseTeardown( if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleMergedPullRequestConfirmationSweep(); - scheduleGitHubConnectionEventPoll(); - scheduleGitHubConnectionContinuitySweep(); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork((async () => { @@ -2134,8 +2110,6 @@ async function startServerWithDatabaseTeardown( await startEnvironmentCleanup(); startHeartbeatSchedulerInterval(() => { scheduleExternalObjectRefreshSweep(new Date()); - scheduleGitHubConnectionEventPoll(); - scheduleGitHubConnectionContinuitySweep(); }); } diff --git a/server/src/middleware/cloud-control.ts b/server/src/middleware/cloud-control.ts index f0e44503e8b..b726aae3092 100644 --- a/server/src/middleware/cloud-control.ts +++ b/server/src/middleware/cloud-control.ts @@ -8,6 +8,7 @@ import { /** Endpoint → method → the one action a control assertion must name to take it. */ const ACTIONS_BY_ENDPOINT: Record> = { + "/api/instance/connector-events/wake": { POST: "connector-events:wake" }, "/api/instance/task-drain": { GET: "task-drain:read", POST: "task-drain:start", diff --git a/server/src/routes/connection-event-wake.ts b/server/src/routes/connection-event-wake.ts new file mode 100644 index 00000000000..c4a71e87717 --- /dev/null +++ b/server/src/routes/connection-event-wake.ts @@ -0,0 +1,13 @@ +import { Router } from "express"; +import { wakeConnectionEvents } from "../services/connection-event-wake.js"; + +export function connectionEventWakeRoutes(owner: object) { + const router = Router(); + router.post("/api/instance/connector-events/wake", (req, res) => { + if (req.actor.source !== "cloud_control") { res.status(403).json({ error: "cloud_control_required" }); return; } + res.set("Cache-Control", "no-store"); + if (!wakeConnectionEvents(owner)) { res.status(503).json({ error: "connection_event_receiver_unavailable" }); return; } + res.status(202).json({ accepted: true }); + }); + return router; +} diff --git a/server/src/services/cloud-runtime-identity.ts b/server/src/services/cloud-runtime-identity.ts index 435c7f7562b..80b3333b36f 100644 --- a/server/src/services/cloud-runtime-identity.ts +++ b/server/src/services/cloud-runtime-identity.ts @@ -451,6 +451,7 @@ export const CLOUD_CONTROL_ACTIONS = [ "task-drain:stop", "lifecycle:read", "lifecycle:unarchive-primary", + "connector-events:wake", ] as const; export type CloudControlAction = (typeof CLOUD_CONTROL_ACTIONS)[number]; diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 0f2ed547a79..cce62d9c934 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -332,6 +332,8 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch); if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connectionEvents); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connectionContinuity); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); @@ -488,6 +490,8 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const wasAlreadyArchived = existing.status === "archived"; if (!wasAlreadyArchived) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connectionEvents); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connectionContinuity); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/connection-event-wake.ts b/server/src/services/connection-event-wake.ts new file mode 100644 index 00000000000..87840eb0979 --- /dev/null +++ b/server/src/services/connection-event-wake.ts @@ -0,0 +1,12 @@ +/** Process-local hint only; Cloud retains the durable target until event acknowledgment. */ +const receivers = new WeakMap void>(); +export function registerConnectionEventWake(owner: object, wake: () => void) { + receivers.set(owner, wake); + return () => { if (receivers.get(owner) === wake) receivers.delete(owner); }; +} +export function wakeConnectionEvents(owner: object): boolean { + const wake = receivers.get(owner); + if (!wake) return false; + wake(); + return true; +} diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..5e7648c8703 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -107,6 +107,8 @@ export const DELIVERY_QUEUES = { feedback: "feedback-exports", chatCompletion: "chat-completions", connection: "connection-continuations", + connectionEvents: "connection-events", + connectionContinuity: "connection-continuity", gitRestorePoints: "git-restore-points", nativeWorkspaceSyncRetention: "native-workspace-sync-retention", question: "question-responses", diff --git a/server/src/services/github-connection-events.test.ts b/server/src/services/github-connection-events.test.ts index 93031ecae62..2a7a15dbffc 100644 --- a/server/src/services/github-connection-events.test.ts +++ b/server/src/services/github-connection-events.test.ts @@ -3,7 +3,7 @@ import { chatEndpoints, type Db } from "@paperclipai/db"; import { githubConnectionEventService } from "./github-connection-events.js"; import type { PaperclipCloudConnector } from "./paperclip-cloud-connector.js"; -function fixture() { +function fixture(push = false) { let time = Date.parse("2026-09-04T12:00:00Z"); const bindings = [{ grant: { id: "grant", companyId: "company", kind: "user", subjectUserId: "owner", status: "active", @@ -13,13 +13,14 @@ function fixture() { const bots: Array<{ companyId: string }> = []; // Only SQL is stubbed. Backoff, lease admission and concurrent execution use // the real service; relational persistence has its own Postgres suite. + const result = (rows: T[]) => Object.assign(Promise.resolve(rows), { limit: async () => rows.slice(0, 1) }); const select = vi.fn(() => ({ from: (table: unknown) => table === chatEndpoints - ? { where: async () => bots } - : { innerJoin: () => ({ where: async () => bindings }) } })); + ? { where: () => result(bots) } + : { innerJoin: () => ({ where: () => result(bindings) }) } })); const leaseEvents = vi.fn().mockResolvedValue(null); - const connector = { leaseEvents } as unknown as PaperclipCloudConnector; + const connector = { leaseEvents, eventWakeVersion: () => push ? 1 : null } as unknown as PaperclipCloudConnector; const service = githubConnectionEventService({ select } as unknown as Db, { connector, now: () => new Date(time) }); - return { service, leaseEvents, select, bots, advance: (ms: number) => { time += ms; } }; + return { service, leaseEvents, select, bots, bindings, advance: (ms: number) => { time += ms; } }; } describe("GitHub event polling during active App setup", () => { @@ -72,3 +73,27 @@ describe("GitHub event polling during active App setup", () => { expect(f.leaseEvents).toHaveBeenCalledTimes(2); }); }); + +describe("GitHub external delivery activation", () => { + it("does no SQL/network with no connector configuration after startup or cadence windows", async () => { + const select = vi.fn(() => { throw new Error("unconfigured worker touched SQL"); }); + const service = githubConnectionEventService({ select } as unknown as Db, { env: {} }); + for (let i = 0; i < 4; i++) await service.pollOnce({ forDelivery: true }); + expect(service.nextPollAt()).toBeNull(); expect(select).not.toHaveBeenCalled(); + }); + it("settles configured-idle after enrollment and wakes only on explicit delivery hints", async () => { + const f = fixture(true); + await f.service.pollOnce(); expect(f.service.nextPollAt()).toBeNull(); + const reads = f.select.mock.calls.length; + for (let i = 0; i < 4; i++) { f.advance(300_000); await f.service.pollOnce(); } + expect(f.select).toHaveBeenCalledTimes(reads); expect(f.leaseEvents).toHaveBeenCalledTimes(1); + await f.service.pollOnce({ forDelivery: true }); expect(f.leaseEvents).toHaveBeenCalledTimes(2); + expect(f.service.nextPollAt()).toBeNull(); + }); + it("retains old-Cloud fallback and settles once all bindings are disabled", async () => { + const f = fixture(); await f.service.pollOnce(); expect(f.service.nextPollAt()).not.toBeNull(); + f.bindings.length = 0; f.advance(10_000); await f.service.pollOnce(); + expect(f.service.nextPollAt()).toBeNull(); const reads = f.select.mock.calls.length; + f.advance(3600_000); await f.service.pollOnce(); expect(f.select).toHaveBeenCalledTimes(reads); + }); +}); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 47bdac122a2..17fff9a993c 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "./github-connection-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; @@ -221,6 +222,9 @@ export function githubConnectionEventService( ) { const now = options.now ?? (() => new Date()); let nextPollAt = 0; + const config = options.connector ? null : paperclipCloudConnectorConfigFromEnv(options.env); + const connector = options.connector ?? (config ? createPaperclipCloudConnector({ config }) : null); + const forcePolling = (options.env ?? process.env).PAPERCLIP_GITHUB_EVENT_POLLING === "1"; let emptyPolls = 0; let nextSetupPollAt = 0; let inFlight: Promise | null = null; @@ -236,6 +240,7 @@ export function githubConnectionEventService( eq(connectionGrants.status, "active"), eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), + sql`exists (select 1 from companies where id = ${connectionGrants.companyId} and status <> 'archived')`, )); return bindingRows(rows); } @@ -318,6 +323,11 @@ export function githubConnectionEventService( // Lifecycle webhooks carry IDs, not the user token’s complete repository view. // Discard the snapshot until Refresh access verifies it again. accessRevision: randomUUID(), + ...(unavailable ? { bindingSync: { + id: randomUUID(), removeInstallationIds: [...new Set([ + ...(github.bindingSync?.removeInstallationIds ?? []), binding.installationId, + ])].filter(id => !installationIds.includes(id)), + } } : {}), repositories: undefined, installationIds, installationCount: installationIds.length, @@ -327,6 +337,7 @@ export function githubConnectionEventService( webhookHealth: unavailable ? "unhealthy" as const : "healthy" as const, }, }; + await notifyGitHubCredentialChange(database, { config: { sourceTemplateKey: "github" } }); await database.update(connectionGrants).set({ providerTenant, updatedAt: now() }) .where(and(eq(connectionGrants.id, binding.grantId), eq(connectionGrants.companyId, binding.companyId))); await notifyToolConnectionHealth(database, { companyId: binding.companyId, id: binding.connectionId }); @@ -366,16 +377,20 @@ export function githubConnectionEventService( eq(connectionEventDeliveries.providerDeliveryId, event.id), )).limit(1); if (existing?.status === "processed") return "duplicate" as const; - await db.update(connectionEventDeliveries).set({ - status: "received", - attempts: sql`${connectionEventDeliveries.attempts} + 1`, - lastError: null, - updatedAt: receiptAt, - }).where(eq(connectionEventDeliveries.id, existing!.id)); + } const postCommitPublications: ActivityPublication[] = []; + let applied = true; try { const applyAndFinalize = async (database: Db) => { + const [locked] = await database.select().from(connectionEventDeliveries).where(and( + eq(connectionEventDeliveries.companyId, companyId), eq(connectionEventDeliveries.provider, event.provider), + eq(connectionEventDeliveries.providerDeliveryId, event.id), + )).for("update").limit(1); + if (locked!.status === "processed") { applied = false; return; } + if (!receipt) await database.update(connectionEventDeliveries).set({ + status: "received", attempts: sql`${connectionEventDeliveries.attempts} + 1`, lastError: null, updatedAt: receiptAt, + }).where(eq(connectionEventDeliveries.id, locked!.id)); if (event.event === "pull_request") await applyPullRequestEvent(companyId, normalizedEvent); if (event.event === "installation" || event.event === "installation_repositories") { for (const binding of bindings) await applyInstallationEvent(database, binding, normalizedEvent); @@ -428,11 +443,7 @@ export function githubConnectionEventService( }, }, postCommitPublications); }; - if (event.event === "installation" || event.event === "installation_repositories") { - await db.transaction(async (tx) => applyAndFinalize(tx as unknown as Db)); - } else { - await applyAndFinalize(db); - } + await db.transaction(async (tx) => applyAndFinalize(tx as unknown as Db)); } catch (error) { await db.update(connectionEventDeliveries).set({ status: "failed", @@ -459,31 +470,45 @@ export function githubConnectionEventService( }, "GitHub webhook activity publication failed after commit"); } } - return "processed" as const; + return applied ? "processed" as const : "duplicate" as const; } async function runPoll(): Promise { - const bindings = await activeBindings(); - const bots = await db.select().from(chatEndpoints).where(and(eq(chatEndpoints.provider, "github"), sql`${chatEndpoints.setup}->'github'->>'cloudRegistrationId' is not null`, sql`${chatEndpoints.status} <> 'archived'`)); - if (bindings.length === 0 && bots.length === 0) { - nextPollAt = now().getTime() + 5_000; - return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 }; - } - const config = options.connector ? null : paperclipCloudConnectorConfigFromEnv(options.env); - const connector = options.connector ?? (config ? createPaperclipCloudConnector({ config }) : null); if (!connector) { - nextPollAt = now().getTime() + 5 * 60_000; + nextPollAt = Number.POSITIVE_INFINITY; return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 }; } - const first = bindings[0] ?? { subject: "github-bot-events", companyId: bots[0]!.companyId }; - const lease = await connector.leaseEvents({ subject: first.subject, companyId: first.companyId }); + const bindings = await activeBindings(); + const bots = await db.select().from(chatEndpoints).where(and(eq(chatEndpoints.provider, "github"), sql`${chatEndpoints.setup}->'github'->>'cloudRegistrationId' is not null`, sql`${chatEndpoints.status} not in ('archived', 'disabled')`, sql`exists (select 1 from companies where id = ${chatEndpoints.companyId} and status <> 'archived')`)); + let identity = bindings[0] ?? (bots[0] ? { subject: "github-bot-events", companyId: bots[0].companyId } : null); + const hasConsumer = bindings.length > 0 || bots.length > 0; + if (!identity) { + const [history] = await db.select({ grant: connectionGrants, connection: toolConnections }).from(connectionGrants) + .innerJoin(toolConnections, and(eq(toolConnections.id, connectionGrants.connectionId), eq(toolConnections.companyId, connectionGrants.companyId))) + .where(and(sql`${toolConnections.config}->>'sourceTemplateKey' = 'github'`, sql`${toolConnections.config}->'oauth'->>'connectorProfile' = 'github.code'`)).limit(1); + const grant = history?.grant; + const subject = grant?.kind === "agent" && grant.subjectAgentId ? `agent:${grant.subjectAgentId}` + : grant?.kind === "user" ? grant.subjectUserId : null; + if (grant && subject) identity = { subject, companyId: grant.companyId }; + } + if (!identity) { + const [history] = await db.select().from(chatEndpoints).where(and(eq(chatEndpoints.provider, "github"), sql`${chatEndpoints.setup}->'github'->>'cloudRegistrationId' is not null`)).limit(1); + if (history) identity = { subject: "github-bot-events", companyId: history.companyId }; + } + if (!identity) { + nextPollAt = Number.POSITIVE_INFINITY; + return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 }; + } + const first = identity; + const lease = await connector.leaseEvents({ subject: first.subject, companyId: first.companyId, eventWakeVersion: forcePolling ? 0 : 1 }); if (!lease) { emptyPolls += 1; - nextPollAt = now().getTime() + Math.min(bots.length ? 15_000 : 5 * 60_000, 5_000 * (2 ** Math.min(emptyPolls, 6))); + nextPollAt = !hasConsumer || (!forcePolling && connector.eventWakeVersion?.() === 1) ? Number.POSITIVE_INFINITY + : now().getTime() + Math.min(bots.length ? 15_000 : 5 * 60_000, 5_000 * (2 ** Math.min(emptyPolls, 6))); return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 }; } emptyPolls = 0; - nextPollAt = now().getTime() + 5_000; + nextPollAt = !hasConsumer || (!forcePolling && connector.eventWakeVersion?.() === 1) ? Number.POSITIVE_INFINITY : now().getTime() + 5_000; const result: GitHubConnectionEventPollResult = { leased: lease.events.length + (lease.failedEventIds?.length ?? 0), processed: 0, @@ -512,17 +537,7 @@ export function githubConnectionEventService( result[status] += 1; } acknowledge.push(event.id); - if (event.event === "installation" && (event.action === "deleted" || event.action === "suspend")) { - await Promise.all(matched.map((binding) => connector.setWebhookBinding({ - subject: binding.subject, - companyId: binding.companyId, - id: binding.id, - installationId: binding.installationId, - connectionId: binding.connectionId, - grantId: binding.grantId, - active: false, - }))); - } + } catch { result.failed += 1; } @@ -543,13 +558,14 @@ export function githubConnectionEventService( } return { - pollOnce(input: { forSetup?: boolean } = {}): Promise { + nextPollAt: () => Number.isFinite(nextPollAt) ? nextPollAt : null, + pollOnce(input: { forSetup?: boolean; forDelivery?: boolean } = {}): Promise { // A manager actively awaiting a dedicated App's signed ping must not // inherit a personal connection's old empty-queue backoff. This is a // bounded readiness probe, not an additional background polling loop. if (inFlight) return inFlight; const time = now().getTime(); - if (time < (input.forSetup ? nextSetupPollAt : nextPollAt)) { + if (!input.forDelivery && time < (input.forSetup ? nextSetupPollAt : nextPollAt)) { return Promise.resolve({ leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 }); } nextSetupPollAt = time + 5_000; diff --git a/server/src/services/github-connection-notifications.ts b/server/src/services/github-connection-notifications.ts new file mode 100644 index 00000000000..79f3c067aba --- /dev/null +++ b/server/src/services/github-connection-notifications.ts @@ -0,0 +1,7 @@ +import type { toolConnections } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +export async function notifyGitHubCredentialChange(transaction: object, connection: Pick) { + if (connection.config.sourceTemplateKey !== "github") return; + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionContinuity); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionEvents); +} diff --git a/server/src/services/github-connection-work.ts b/server/src/services/github-connection-work.ts new file mode 100644 index 00000000000..7849cf3bda8 --- /dev/null +++ b/server/src/services/github-connection-work.ts @@ -0,0 +1,36 @@ +import { logger } from "../middleware/logger.js"; +import type { Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "./delivery-work-notifications.js"; +import { registerConnectionEventWake } from "./connection-event-wake.js"; +import type { githubConnectionEventService } from "./github-connection-events.js"; +import type { createGitHubGrantContinuity } from "./tool-access/github-grant-continuity.js"; + +type Coordinator = ReturnType; +export function registerGitHubConnectionEventWork(db: Db, coordinator: Coordinator, + service: ReturnType, canRun: () => boolean) { + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.connectionEvents, { + retryMs: 5_000, canRun, + run: () => service.pollOnce({ forDelivery: true }), + next: async () => service.nextPollAt(), + }); + const stops = [registerConnectionEventWake(db, worker.wake), + subscribeDeliveryWork(db, DELIVERY_QUEUES.toolConnectionHealth, worker.wake), + subscribeDeliveryWork(db, DELIVERY_QUEUES.chatReceivers, worker.wake)]; + return { ...worker, stop: async () => { for (const stop of stops) stop(); await worker.stop(); } }; +} +export function registerGitHubConnectionContinuityWork(db: Db, coordinator: Coordinator, + service: Pick, "sweepGitHubConnectionContinuity" | "nextGitHubConnectionContinuityAt">, + canRun: () => boolean) { + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.connectionContinuity, { + retryMs: 30_000, canRun, + run: async () => { + const result = await service.sweepGitHubConnectionContinuity(); + if (result.failed) logger.warn(result, "GitHub credential continuity retained retry obligations"); + }, + next: () => service.nextGitHubConnectionContinuityAt(), + }); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.toolConnectionHealth, worker.wake); + return { ...worker, stop: async () => { unsubscribe(); await worker.stop(); } }; +} diff --git a/server/src/services/paperclip-cloud-connector.ts b/server/src/services/paperclip-cloud-connector.ts index b1461d3e14d..6159051c5d3 100644 --- a/server/src/services/paperclip-cloud-connector.ts +++ b/server/src/services/paperclip-cloud-connector.ts @@ -1,3 +1,4 @@ +import { providerRetryAt } from "./provider-retry-after.js"; import { SLACK_MANAGER_SCOPES, ASANA_CONNECTOR_SCOPES, isAsanaConnectorProfileId, type AsanaConnectorProfileId } from "@paperclipai/shared"; import { createDecipheriv, @@ -114,6 +115,7 @@ type ConnectorResponse = { events?: unknown; acknowledged?: unknown; githubApps?: { version: number }; + eventWakeVersion?: unknown; }; const ENDPOINTS: Record = { @@ -189,6 +191,7 @@ export class PaperclipCloudConnectorError extends Error { message: string, readonly code: string, readonly status?: number, + readonly retryAt?: number, ) { super(message); this.name = "PaperclipCloudConnectorError"; @@ -272,10 +275,11 @@ export function createPaperclipCloudConnector(input: { const now = input.now ?? Date.now; const signingKey = privateKey(config.signPrivateKey, "ed25519"); const sealKey = privateKey(config.sealPrivateKey, "x25519"); + let eventWakeVersion: 1 | null = null; async function call( operation: PaperclipCloudConnectorOperation, - claims: { subject: string; companyId: string; profile?: PaperclipCloudConnectorProfileId; scopes?: string[]; returnUri?: string; returnState?: string; claimId?: string; redemptionId?: string }, + claims: { subject: string; companyId: string; profile?: PaperclipCloudConnectorProfileId; scopes?: string[]; returnUri?: string; returnState?: string; claimId?: string; redemptionId?: string; eventWakeVersion?: 0 | 1 }, secret?: { field: "refreshToken" | "token" | "binding" | "acknowledgement"; value: string }, ): Promise { const endpoint = new URL(ENDPOINTS[operation], `${config.baseUrl}/`).toString(); @@ -291,6 +295,7 @@ export function createPaperclipCloudConnector(input: { exp: issuedAt + 60, jti: randomUUID(), }; + if (claims.eventWakeVersion !== undefined) payload.ewv = claims.eventWakeVersion; if (claims.returnUri !== undefined) payload.ruri = claims.returnUri; if (claims.returnState !== undefined) payload.rst = claims.returnState; if (claims.claimId !== undefined) payload.cl = claims.claimId; @@ -327,6 +332,7 @@ export function createPaperclipCloudConnector(input: { `Paperclip Cloud connector rejected the request (operation=${operation}, status=${response.status}, reason=${reason})`, response.status === 409 ? "REAUTHORIZATION_REQUIRED" : "CONNECTOR_REQUEST_FAILED", response.status, + providerRetryAt(response.headers.get("retry-after"), now()), ); } try { @@ -374,6 +380,7 @@ export function createPaperclipCloudConnector(input: { } return { + eventWakeVersion: () => eventWakeVersion, async githubAppsAvailable() { const response = await call("status", { subject: "instance-capabilities", companyId: "instance-capabilities" }); return response.active === true && response.githubApps?.version === 2; @@ -520,8 +527,9 @@ export function createPaperclipCloudConnector(input: { }); await call("webhook-bind", { ...values, profile: "github.code" }, { field: "binding", value: binding }); }, - async leaseEvents(values: { subject: string; companyId: string }): Promise<{ leaseId: string; events: SealedConnectorEvents["events"]; failedEventIds?: string[] } | null> { + async leaseEvents(values: { subject: string; companyId: string; eventWakeVersion?: 0 | 1 }): Promise<{ leaseId: string; events: SealedConnectorEvents["events"]; failedEventIds?: string[] } | null> { const response = await call("event-lease", values); + eventWakeVersion = response.eventWakeVersion === 1 && values.eventWakeVersion !== 0 ? 1 : null; if (Array.isArray(response.events) && response.events.length === 0) return null; if (typeof response.leaseId !== "string") { throw new PaperclipCloudConnectorError("Paperclip Cloud connector returned an invalid event lease", "CONNECTOR_BAD_RESPONSE"); diff --git a/server/src/services/provider-retry-after.test.ts b/server/src/services/provider-retry-after.test.ts new file mode 100644 index 00000000000..b10fa764f27 --- /dev/null +++ b/server/src/services/provider-retry-after.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import { providerRetryAt } from "./provider-retry-after.js"; +import { githubContinuityRevision, githubGrantContinuityDeadline } from "./tool-access/github-grant-continuity.js"; +const now = Date.parse("2026-01-01T00:00:00Z"); +describe("provider and credential deadlines", () => { + it("accepts bounded Retry-After seconds/dates and ignores malformed or expired advice", () => { + expect(providerRetryAt("120", now)).toBe(now + 120_000); + expect(providerRetryAt(new Date(now + 60_000).toUTCString(), now)).toBe(now + 60_000); + expect(providerRetryAt("999999999999", now)).toBe(now + 24 * 3600_000); + for (const value of [null, "", "unknown", "-3", new Date(now - 1).toUTCString()]) expect(providerRetryAt(value, now)).toBeUndefined(); + }); + it("reconstructs expiry, continuity and saved retries and preempts a retry when credentials change", () => { + const connection = { config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } } }; + const github = { userId: "42", login: "owner", installationCount: 1, repositoryCount: 1, repositorySelection: "all" as const, + installationIds: ["101"], installationOwnerLogins: ["owner"], lastAccessRefreshAt: new Date(now).toISOString() }; + const grant = { providerTenant: { github, oauth: { accessTokenExpiresAt: new Date(now + 8 * 3600_000).toISOString(), refreshedAt: new Date(now).toISOString() } } }; + expect(githubGrantContinuityDeadline(connection, grant)).toBe(now + 7 * 3600_000); + const retried = { providerTenant: { ...grant.providerTenant, github: { ...github, + continuityRetry: { at: new Date(now + 10 * 3600_000).toISOString(), attempts: 2, revision: githubContinuityRevision(grant) } } } }; + expect(githubGrantContinuityDeadline(connection, retried)).toBe(now + 10 * 3600_000); + retried.providerTenant.oauth.accessTokenExpiresAt = new Date(now + 2 * 3600_000).toISOString(); + expect(githubGrantContinuityDeadline(connection, retried)).toBe(now + 3600_000); + expect(githubGrantContinuityDeadline(connection, { providerTenant: { github } })).toBe(now + 30 * 24 * 3600_000); + expect(githubGrantContinuityDeadline({ config: {} }, grant)).toBeNull(); + }); +}); diff --git a/server/src/services/provider-retry-after.ts b/server/src/services/provider-retry-after.ts new file mode 100644 index 00000000000..c90364ea7e6 --- /dev/null +++ b/server/src/services/provider-retry-after.ts @@ -0,0 +1,6 @@ +export function providerRetryAt(value: string | null, now = Date.now()): number | undefined { + if (!value?.trim()) return undefined; + const seconds = /^\d+$/.test(value.trim()) ? Number(value) : Number.NaN; + const deadline = Number.isFinite(seconds) ? now + seconds * 1000 : Date.parse(value); + return Number.isFinite(deadline) && deadline > now ? Math.min(deadline, now + 24 * 3600_000) : undefined; +} diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 81b6f646c3c..7d4cfe8cf0d 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -787,7 +787,7 @@ export function toolAccessService( const refreshLeases = createOAuthRefreshLeases(db, now, getConnectionRow, getOAuthGrantRow); const maybeRefreshOAuthCredentials = createLegacyOAuthRefresh(db, { now, secrets, store: credentialStore, leases: refreshLeases, getConnectionRow, exchangeOAuthToken }); const refreshOAuthGrantCredentials = createOAuthGrantRefresh(db, { now, store: credentialStore, leases: refreshLeases, currentCloudConnector, getConnectionRow, exchangeOAuthToken }); - const { refreshManagedGitHubGrantAccess, sweepGitHubConnectionContinuity } = createGitHubGrantContinuity(db, { now, currentCloudConnector, refreshOAuthGrantCredentials, resolveOAuthGrantSecret }); + const { refreshManagedGitHubGrantAccess, syncGitHubGrantBindings, sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt } = createGitHubGrantContinuity(db, { now, currentCloudConnector, refreshOAuthGrantCredentials, resolveOAuthGrantSecret }); const vercelCredentials = createVercelCredentialService(db, now, vercelConnect); const { resolveVercelCredentialHeaders, vercelGrantForConnection } = vercelCredentials; const credentialAccess = createConnectionCredentialAccess(db, { secrets, audit, vercel: vercelCredentials, resolveOAuthGrantSecret, maybeRefreshOAuthCredentials, refreshOAuthGrantCredentials }); @@ -811,7 +811,7 @@ export function toolAccessService( const finishOAuthCatalogWithRecommendedDefaults = createOAuthCatalogCompletion(db, finishGalleryAppConnection); const startOAuth = createOAuthAuthorizationStart(db, { now, getConnectionRow, setupPolicy, fixedPersonalIdentityForReconnect, currentCloudConnector, vercelConnect, ensureOAuthClient, oauthEndpointsForConnection, assertOAuthEndpointUrl }); const completeOAuthCallback = createDirectOAuthCallback(db, { now, getConnectionRow, state: oauthState, store: credentialStore, setupPolicy, refreshCatalog, finishOAuthCatalogWithRecommendedDefaults, checkConnectionHealth, ensureDefaultOrganizationGrant, oauthEndpointsForConnection, exchangeOAuthToken }); - const completePaperclipCloudConnectorCallback = createCloudOAuthCallback(db, { now, getConnectionRow, state: oauthState, store: credentialStore, refreshCatalog, finishOAuthCatalogWithRecommendedDefaults, currentCloudConnector }); + const completePaperclipCloudConnectorCallback = createCloudOAuthCallback(db, { now, getConnectionRow, state: oauthState, store: credentialStore, refreshCatalog, finishOAuthCatalogWithRecommendedDefaults, currentCloudConnector, syncGitHubGrantBindings }); const completeVercelConnectCallback = createVercelOAuthCallback(db, { now, getConnectionRow, state: oauthState, refreshCatalog, finishOAuthCatalogWithRecommendedDefaults, vercelConnect }); const finalizeOAuthAccess = createOAuthAccessFinalization(db, { getConnectionRow, secrets, store: credentialStore, assertAgentsInCompany, finishGalleryAppConnection }); const runtimeHealth = createToolRuntimeHealth(db, options, now, trustedRuntimeHost); @@ -4732,6 +4732,7 @@ export function toolAccessService( scheduledHealth: connectionHealthSchedule(db, checkConnectionHealth), sweepGitHubConnectionContinuity, + nextGitHubConnectionContinuityAt, ...catalogReader, ...toolActionReviewQueue, ...profileService, diff --git a/server/src/services/tool-access/connections/grants.ts b/server/src/services/tool-access/connections/grants.ts index 1836a6a52fc..572bb0e8856 100644 --- a/server/src/services/tool-access/connections/grants.ts +++ b/server/src/services/tool-access/connections/grants.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "../../github-connection-notifications.js"; import type { ConnectionManagementContext } from "../../connection-management.js"; import { connectionActorUserId } from "../actor.js"; import { toConnectionGrant } from "./projection.js"; @@ -537,6 +538,7 @@ replaceConnectionGrantMembers: async ( providerRevocation = "local_only_shared_client"; } const grant = await db.transaction(async (tx) => { + await notifyGitHubCredentialChange(tx, connection); const [lockedGrant] = await tx.select({ id: connectionGrants.id }).from(connectionGrants) .where(and(eq(connectionGrants.id, grantId), eq(connectionGrants.companyId, connection.companyId), eq(connectionGrants.connectionId, connection.id))) .for("update"); diff --git a/server/src/services/tool-access/github-access-discovery.ts b/server/src/services/tool-access/github-access-discovery.ts index 4f0912b000e..2d45377ab6e 100644 --- a/server/src/services/tool-access/github-access-discovery.ts +++ b/server/src/services/tool-access/github-access-discovery.ts @@ -1,3 +1,4 @@ +import { providerRetryAt } from "../provider-retry-after.js"; import { randomUUID } from "node:crypto"; import { unprocessable } from "../../errors.js"; @@ -106,6 +107,7 @@ export async function loadGitHubGrantMetadata( throw unprocessable( "GitHub could not verify this account's installation access", { + retryAt: providerRetryAt(response.headers.get("retry-after")), code: response.status === 401 ? "oauth_reauthorization_required" diff --git a/server/src/services/tool-access/github-grant-continuity.ts b/server/src/services/tool-access/github-grant-continuity.ts index 568a0a00a99..0398c2a79be 100644 --- a/server/src/services/tool-access/github-grant-continuity.ts +++ b/server/src/services/tool-access/github-grant-continuity.ts @@ -1,12 +1,14 @@ +import { randomUUID } from "node:crypto"; +import { notifyGitHubCredentialChange } from "../github-connection-notifications.js"; import type { ActorInfo } from "./types.js"; import { loadGitHubGrantMetadata } from "./github-access-discovery.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { measureRunTransaction } from "../execution-transaction-timing.js"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { connectionGrants, toolConnections } from "@paperclipai/db"; import { conflict, HttpError, notFound, unprocessable } from "../../errors.js"; -import { type PaperclipCloudConnector } from "../paperclip-cloud-connector.js"; +import { PaperclipCloudConnectorError, type PaperclipCloudConnector } from "../paperclip-cloud-connector.js"; import type { createOAuthGrantRefresh } from "./oauth/grant-refresh.js"; import type { OAuthCredentialStore } from "./oauth/credential-store.js"; @@ -24,7 +26,6 @@ type GitHubContinuityDependencies = { }; export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector, refreshOAuthGrantCredentials, resolveOAuthGrantSecret }: GitHubContinuityDependencies) { - let nextGitHubContinuitySweepAt = 0; async function refreshManagedGitHubGrantAccess( connection: typeof toolConnections.$inferSelect, initialGrant: typeof connectionGrants.$inferSelect, @@ -138,7 +139,8 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector throw retryError; } } - const { updated, previousGitHub } = await measureRunTransaction(db, "connectors.github.refresh", async (tx) => { + const { updated } = await measureRunTransaction(db, "connectors.github.refresh", async (tx) => { + await notifyGitHubCredentialChange(tx, connection); const [currentGrant] = await measureRunPhase("connectors.github.grant_lock", () => tx .select() .from(connectionGrants) @@ -170,6 +172,13 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector ...(currentGrant.providerTenant ?? {}), github: { ...metadata, + bindingSync: { + id: randomUUID(), + removeInstallationIds: [...new Set([ + ...(previousGitHub?.bindingSync?.removeInstallationIds ?? []), + ...(previousGitHub?.installationIds ?? []), + ])].filter(id => !metadata.installationIds.includes(id)), + }, ...(previousGitHub?.lastWebhookAt ? { lastWebhookAt: previousGitHub.lastWebhookAt } : {}), @@ -195,111 +204,161 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector return { updated, previousGitHub }; }); + return syncGitHubGrantBindings(connection, updated, actor, accessToken); + } + + async function syncGitHubGrantBindings( + connection: typeof toolConnections.$inferSelect, + initialGrant: typeof connectionGrants.$inferSelect, + actor?: ActorInfo, + accessToken?: string, + ) { + await actor?.managementContext?.reauthorize(); const cloudConnector = currentCloudConnector(); - const subject = - updated.kind === "agent" && updated.subjectAgentId - ? `agent:${updated.subjectAgentId}` - : updated.kind === "user" && updated.subjectUserId - ? updated.subjectUserId - : null; - if (cloudConnector && subject) { - const previous = new Set(previousGitHub?.installationIds ?? []); - const current = new Set(metadata.installationIds); - await measureRunPhase("connectors.github.webhooks", () => Promise.all([ - ...metadata.installationIds.map((installationId) => - cloudConnector.setWebhookBinding({ - subject, - companyId: connection.companyId, - id: `${updated.id}_${installationId}`, - installationId, - connectionId: connection.id, - grantId: updated.id, - active: true, - accessToken, - }), - ), - ...[...previous] - .filter((installationId) => !current.has(installationId)) - .map((installationId) => - cloudConnector.setWebhookBinding({ - subject, - companyId: connection.companyId, - id: `${updated.id}_${installationId}`, - installationId, - connectionId: connection.id, - grantId: updated.id, - active: false, - }), - ), - ])); - } - return updated; + if (!initialGrant.providerTenant?.github?.bindingSync) return initialGrant; + if (!cloudConnector) throw unprocessable("GitHub webhook delivery is not configured"); + const grant = accessToken || initialGrant.providerTenant.github.installationIds.length === 0 ? initialGrant : await refreshOAuthGrantCredentials({ + companyId: connection.companyId, connectionId: connection.id, grantId: initialGrant.id, actor, + }); + const accessRef = grant.credentialSecretRefs.find(ref => ref.configPath === "oauth.access_token"); + if (!accessRef && grant.providerTenant?.github?.installationIds.length) throw unprocessable("GitHub authorization must be reconnected", { code: "oauth_reauthorization_required" }); + const token = accessToken ?? (accessRef ? (await resolveOAuthGrantSecret(connection, grant, accessRef, actor, undefined)).value : undefined); + // Serialize external binding changes with grant/connection lifecycle writes. + // Requests occur only for saved debt; no idle lock or connection is retained. + return db.transaction(async tx => { + const [current] = await tx.select().from(connectionGrants).where(and( + eq(connectionGrants.id, grant.id), eq(connectionGrants.companyId, connection.companyId), + eq(connectionGrants.connectionId, connection.id), + )).for("update").limit(1); + if (!current || current.status !== "active") return current ?? grant; + const [liveConnection] = await tx.select().from(toolConnections).where(and( + eq(toolConnections.id, connection.id), eq(toolConnections.companyId, connection.companyId), + )).for("update").limit(1); + if (!liveConnection || liveConnection.status !== "active" || !liveConnection.enabled) return current; + const github = current.providerTenant?.github; + if (!github?.bindingSync || github.bindingSync.id !== grant.providerTenant?.github?.bindingSync?.id) return current; + const subject = current.kind === "agent" && current.subjectAgentId ? `agent:${current.subjectAgentId}` + : current.kind === "user" ? current.subjectUserId : null; + if (!subject) throw unprocessable("GitHub webhook subject is unavailable"); + for (const installationId of github.installationIds) await cloudConnector.setWebhookBinding({ + subject, companyId: connection.companyId, id: `${current.id}_${installationId}`, + installationId, connectionId: connection.id, grantId: current.id, active: true, accessToken: token!, + }); + for (const installationId of github.bindingSync.removeInstallationIds) await cloudConnector.setWebhookBinding({ + subject, companyId: connection.companyId, id: `${current.id}_${installationId}`, + installationId, connectionId: connection.id, grantId: current.id, active: false, + }); + await notifyGitHubCredentialChange(tx, connection); + const [settled] = await tx.update(connectionGrants).set({ + providerTenant: sql`${connectionGrants.providerTenant} #- '{github,bindingSync}' #- '{github,continuityRetry}'`, + updatedAt: now(), + }).where(and(eq(connectionGrants.id, current.id), eq(connectionGrants.companyId, connection.companyId))).returning(); + return settled!; + }); + } + + async function retainRetry(connection: typeof toolConnections.$inferSelect, grantId: string, error?: unknown, minimumMs = 30_000) { + await db.transaction(async tx => { + const [current] = await tx.select().from(connectionGrants).where(and( + eq(connectionGrants.id, grantId), eq(connectionGrants.companyId, connection.companyId), + eq(connectionGrants.connectionId, connection.id), eq(connectionGrants.status, "active"), + )).for("update").limit(1); + if (!current?.providerTenant?.github) return; + const attempts = Math.min(16, (current.providerTenant.github.continuityRetry?.attempts ?? 0) + 1); + const advised = error instanceof PaperclipCloudConnectorError ? error.retryAt + : error instanceof HttpError ? asRecord(error.details).retryAt : undefined; + const at = new Date(Math.max(now().getTime() + Math.max(minimumMs, Math.min(60 * 60_000, 30_000 * 2 ** Math.min(attempts - 1, 7))), + typeof advised === "number" ? advised : 0)).toISOString(); + await notifyGitHubCredentialChange(tx, connection); + await tx.update(connectionGrants).set({ + providerTenant: sql`jsonb_set(${connectionGrants.providerTenant}, '{github,continuityRetry}', ${JSON.stringify({ + at, attempts, revision: githubContinuityRevision(current), + })}::jsonb)`, + updatedAt: now(), + }).where(and(eq(connectionGrants.id, current.id), eq(connectionGrants.companyId, connection.companyId))); + }); + } + + async function continuityRows() { + return db + .select({ grant: connectionGrants, connection: toolConnections }) + .from(connectionGrants) + .innerJoin(toolConnections, and( + eq(toolConnections.id, connectionGrants.connectionId), + eq(toolConnections.companyId, connectionGrants.companyId), + )) + .where(and( + eq(connectionGrants.status, "active"), + eq(toolConnections.status, "active"), + eq(toolConnections.enabled, true), + sql`${toolConnections.config}->>'sourceTemplateKey' = 'github'`, + sql`${toolConnections.config}->'oauth'->>'connectorProfile' = 'github.code'`, + sql`${connectionGrants.providerTenant}->'github' is not null`, + sql`exists (select 1 from companies where id = ${connectionGrants.companyId} and status = 'active')`, + )); + } + + async function nextGitHubConnectionContinuityAt(): Promise { + const deadlines = (await continuityRows()).flatMap(({ grant, connection }) => { + const deadline = githubGrantContinuityDeadline(connection, grant); + return deadline === null ? [] : [deadline]; + }); + return deadlines.length ? Math.min(...deadlines) : null; } async function sweepGitHubConnectionContinuity() { - if (now().getTime() < nextGitHubContinuitySweepAt) { - return { checked: 0, due: 0, refreshed: 0, failed: 0 }; - } - nextGitHubContinuitySweepAt = now().getTime() + 60 * 60_000; - const cutoff = now().getTime() - 30 * 24 * 60 * 60_000; - const rows = await db - .select({ grant: connectionGrants, connection: toolConnections }) - .from(connectionGrants) - .innerJoin( - toolConnections, - and( - eq(toolConnections.id, connectionGrants.connectionId), - eq(toolConnections.companyId, connectionGrants.companyId), - ), - ) - .where( - and( - eq(connectionGrants.status, "active"), - eq(toolConnections.status, "active"), - eq(toolConnections.enabled, true), - ), - ); + const rows = await continuityRows(); const due = rows.filter(({ grant, connection }) => { - const config = asRecord(connection.config); - const oauth = asRecord(config.oauth); - if ( - config.sourceTemplateKey !== "github" || - oauth.connectorProfile !== "github.code" || - !grant.providerTenant?.github - ) - return false; - const expiresAt = grant.providerTenant.oauth?.accessTokenExpiresAt; - const refreshedAt = grant.providerTenant.oauth?.refreshedAt - ? Date.parse(grant.providerTenant.oauth.refreshedAt) - : Number.NaN; - const accessCheckedAt = grant.providerTenant.github.lastAccessRefreshAt - ? Date.parse(grant.providerTenant.github.lastAccessRefreshAt) - : Number.NaN; - if (typeof expiresAt === "string") { - const expiry = Date.parse(expiresAt); - return ( - !Number.isFinite(expiry) || - expiry <= now().getTime() + 60 * 60_000 || - !Number.isFinite(refreshedAt) || - refreshedAt <= cutoff - ); - } - return !Number.isFinite(accessCheckedAt) || accessCheckedAt <= cutoff; + const deadline = githubGrantContinuityDeadline(connection, grant); + return deadline !== null && deadline <= now().getTime(); }); let refreshed = 0; let failed = 0; for (const row of due) { try { - await refreshManagedGitHubGrantAccess(row.connection, row.grant, { - actorType: "system", - actorId: "github-connection-continuity", - }); + const actor: ActorInfo = { actorType: "system", actorId: "github-connection-continuity" }; + const updated = row.grant.providerTenant?.github?.bindingSync + ? await syncGitHubGrantBindings(row.connection, row.grant, actor) + : await refreshManagedGitHubGrantAccess(row.connection, row.grant, actor); + // A successful provider response can still leave an unchanged/short-lived + // credential due. Retain that obligation with bounded durable backoff. + const next = githubGrantContinuityDeadline(row.connection, updated); + if (next !== null && next <= now().getTime()) await retainRetry(row.connection, updated.id, undefined, 60 * 60_000); refreshed += 1; - } catch { + } catch (error) { + await retainRetry(row.connection, row.grant.id, error); failed += 1; } } return { checked: rows.length, due: due.length, refreshed, failed }; } - return { refreshManagedGitHubGrantAccess, sweepGitHubConnectionContinuity }; + return { refreshManagedGitHubGrantAccess, syncGitHubGrantBindings, sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt }; +} + +/** Credential expiry and provider continuity are obligations, not a scan cadence. */ +export function githubGrantContinuityDeadline( + connection: Pick, + grant: Pick, +): number | null { + const config = asRecord(connection.config); + if (config.sourceTemplateKey !== "github" || asRecord(config.oauth).connectorProfile !== "github.code" + || !grant.providerTenant?.github) return null; + const oauth = grant.providerTenant.oauth; + const access = grant.providerTenant.github; + const parsed = (value: string | undefined) => { + const time = value ? Date.parse(value) : Number.NaN; + return Number.isFinite(time) ? time : 0; + }; + const continuityMs = 30 * 24 * 60 * 60_000; + const credentialDeadline = typeof oauth?.accessTokenExpiresAt === "string" + ? Math.min(parsed(oauth.accessTokenExpiresAt) - 60 * 60_000, oauth.refreshedAt ? parsed(oauth.refreshedAt) + continuityMs : 0) + : access.lastAccessRefreshAt ? parsed(access.lastAccessRefreshAt) + continuityMs : 0; + const deadline = access.bindingSync ? 0 : credentialDeadline; + return access.continuityRetry?.revision === githubContinuityRevision(grant) + ? Math.max(deadline, parsed(access.continuityRetry.at)) : deadline; +} + +export function githubContinuityRevision(grant: Pick): string { + return JSON.stringify([grant.providerTenant?.oauth?.accessTokenExpiresAt, grant.providerTenant?.oauth?.refreshedAt, + grant.providerTenant?.github?.lastAccessRefreshAt, grant.providerTenant?.github?.bindingSync?.id]); } diff --git a/server/src/services/tool-access/oauth/cloud-callback.ts b/server/src/services/tool-access/oauth/cloud-callback.ts index fac5bce24ba..7381c1687a4 100644 --- a/server/src/services/tool-access/oauth/cloud-callback.ts +++ b/server/src/services/tool-access/oauth/cloud-callback.ts @@ -1,3 +1,5 @@ +import type { createGitHubGrantContinuity } from "../github-grant-continuity.js"; +import { randomUUID } from "node:crypto"; import { notifyIssueReviewState } from "../../issue-review-state-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { managedConnectorProfile } from "./client-settings.js"; @@ -32,6 +34,7 @@ type Dependencies = { finishOAuthCatalogWithRecommendedDefaults: OAuthCatalogCompletion; store: Pick; currentCloudConnector: () => PaperclipCloudConnector | null; + syncGitHubGrantBindings: ReturnType["syncGitHubGrantBindings"]; }; export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { @@ -301,6 +304,7 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { accessRef, ...(refreshRef ? [refreshRef] : []), ]; + await notifyToolConnectionHealth(tx, connection); const grantValues = { providerTenant: { name: githubMetadata?.login ?? providerName, @@ -319,7 +323,12 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { ? { refreshedAt: now().toISOString() } : {}), }, - ...(githubMetadata ? { github: githubMetadata } : {}), + ...(githubMetadata ? { github: { ...githubMetadata, bindingSync: { + id: randomUUID(), removeInstallationIds: [...new Set([ + ...(existingCredentialGrant?.providerTenant?.github?.bindingSync?.removeInstallationIds ?? []), + ...(existingCredentialGrant?.providerTenant?.github?.installationIds ?? []), + ])].filter(id => !githubMetadata.installationIds.includes(id)), + } } } : {}), }, credentialSecretRefs, status: "active" as const, @@ -358,7 +367,6 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { scopes: [...profile.scopes], }, }; - await notifyToolConnectionHealth(tx, connection); [connection] = await tx .update(toolConnections) .set({ @@ -438,42 +446,8 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { // Paperclip Cloud connector callback is the managed variant of an OAuth // callback completion. emitConnectionUpdated(connection, preCloudCallbackLifecycle, "oauth_callback"); - if (githubMetadata) { - const [githubGrant] = await db - .select({ id: connectionGrants.id }) - .from(connectionGrants) - .where( - and( - eq(connectionGrants.companyId, connection.companyId), - eq(connectionGrants.connectionId, connection.id), - subjectAgentId - ? and( - eq(connectionGrants.kind, "agent"), - eq(connectionGrants.subjectAgentId, subjectAgentId), - ) - : and( - eq(connectionGrants.kind, "user"), - eq(connectionGrants.subjectUserId, subjectUserId!), - ), - ), - ) - .limit(1); - if (!githubGrant) throw new Error("GitHub grant was not persisted"); - await Promise.all( - githubMetadata.installationIds.map((installationId) => - cloudConnector.setWebhookBinding({ - subject: connectorSubject, - companyId: connection.companyId, - id: `${githubGrant.id}_${installationId}`, - installationId, - connectionId: connection.id, - grantId: githubGrant.id, - active: true, - accessToken: credentials.accessToken, - }), - ), - ); - } + // Binding registration is saved above before Cloud I/O. The continuity + // worker drains it after activation, including after a callback/server crash. const refresh = await refreshCatalog(connection.id, input.actor, { enableAllByDefault: false, quarantineManagedOAuthDraft: shouldFinalizeManagedDefaults, @@ -494,6 +468,15 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { actor: input.actor, }) : null; + if (githubMetadata) { + const [grant] = await db.select().from(connectionGrants).where(and( + eq(connectionGrants.companyId, connection.companyId), eq(connectionGrants.connectionId, connection.id), + subjectAgentId ? and(eq(connectionGrants.kind, "agent"), eq(connectionGrants.subjectAgentId, subjectAgentId)) + : and(eq(connectionGrants.kind, "user"), eq(connectionGrants.subjectUserId, subjectUserId!)), + )).limit(1); + if (!grant) throw new Error("GitHub grant was not persisted"); + await dependencies.syncGitHubGrantBindings(await getConnectionRow(connection.id, connection.companyId), grant, input.actor, credentials.accessToken); + } const activatedCatalogEntryIds = new Set( shouldFinalizeManagedDefaults ? refresh.catalog diff --git a/server/src/services/tool-access/oauth/grant-refresh.ts b/server/src/services/tool-access/oauth/grant-refresh.ts index 48e09cf27b9..ce0ca102352 100644 --- a/server/src/services/tool-access/oauth/grant-refresh.ts +++ b/server/src/services/tool-access/oauth/grant-refresh.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "../../github-connection-notifications.js"; import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import type { ActorInfo } from "../types.js"; import { createOAuthClientProtocol, storedOAuthTokenEndpointAuthMethod } from "./client.js"; @@ -193,6 +194,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo }, }; const updated = await db.transaction(async (tx) => { + await notifyGitHubCredentialChange(tx, connection); const [current] = await tx.select({ id: connectionGrants.id }).from(connectionGrants) .where(and(eq(connectionGrants.id, grant.id), eq(connectionGrants.companyId, grant.companyId), eq(connectionGrants.status, "active"), @@ -539,7 +541,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo refreshedAt: now().toISOString(), }, }; - const [updated] = await db + const [updated] = await writeToolConnectionHealthChange(db, connection, tx => tx .update(connectionGrants) .set({ providerTenant, @@ -555,7 +557,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo sql`${connectionGrants.providerTenant} -> 'oauth' -> 'refreshLease' ->> 'id' = ${lease.leaseId}`, ), ) - .returning(); + .returning()); if (!updated) { throw conflict( "OAuth credentials changed while refresh was in progress. Retry the request.",