diff --git a/doc/plans/2026-10-30-github-connection-sleep.md b/doc/plans/2026-10-30-github-connection-sleep.md index 11271110a7a..40f83708a71 100644 --- a/doc/plans/2026-10-30-github-connection-sleep.md +++ b/doc/plans/2026-10-30-github-connection-sleep.md @@ -14,7 +14,11 @@ 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. +merged-pull-request callback keeps its normalized key. Snapshot and confirmation +reads/writes share the receipt transaction, including at pool size one; existing +confirmation publications and direct wakes flush only after the outer commit. +The pre-existing non-chat direct-wake crash gap remains: chat-bound continuations +have their existing durable outbox, while an unbound direct wake is best effort. 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 @@ -51,6 +55,15 @@ Changed credential metadata invalidates the old retry revision, so an earlier ne expiry can preempt it. Disabled connections and archived companies do not perform credential work. Reactivation uses the normal committed configuration signal. +GitHub owners receive only relevant credential/configuration signals, not generic +health or receiver notifications. Continuity reconstruction is company-scoped +after the startup scan. Event draining retains instance-wide binding matching +because Cloud leases an instance queue. Event hints coalesce with a five-second +minimum pass spacing; an unrelated connection PATCH cannot trigger that drain. +OAuth rotation merges only its owned OAuth fields, lease acquisition compares +current metadata, and cleanup atomically removes only the matching lease path, +preserving concurrent GitHub installation changes and binding-removal debt. + 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. diff --git a/server/src/__tests__/github-catalog-timing.test.ts b/server/src/__tests__/github-catalog-timing.test.ts index 4ee15c6db85..5d0c4835d84 100644 --- a/server/src/__tests__/github-catalog-timing.test.ts +++ b/server/src/__tests__/github-catalog-timing.test.ts @@ -1,9 +1,10 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { context, trace } from "@opentelemetry/api"; import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; import { - authUsers, companies, companyMemberships, connectionGrants, createDb, + applyPendingMigrations, authUsers, companies, companySecrets, companyMemberships, connectionGrants, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, toolApplications, toolConnections, } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; @@ -15,15 +16,18 @@ import type { PaperclipCloudConnector } from "../services/paperclip-cloud-connec import { toolAccessService } from "../services/tool-access.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -const support = await getEmbeddedPostgresTestSupport(); +const databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_GITHUB_CATALOG_TEST_DATABASE_URL"); +const support = databaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("GitHub repository catalog execution timing", () => { let temporary: Awaited>; let db: ReturnType; + const ownedCompanies: string[] = []; let exporter: InMemorySpanExporter, provider: TracerProvider, manager: AsyncLocalStorageContextManager; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("paperclip-github-catalog-timing-"); - db = createDb(temporary.connectionString, { maxConnections: 3 }); + if (databaseUrl) await applyPendingMigrations(databaseUrl); + else temporary = await startEmbeddedPostgresTestDatabase("paperclip-github-catalog-timing-"); + db = createDb(databaseUrl ?? temporary.connectionString, { maxConnections: 3 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); beforeEach(() => { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "01234567890123456789012345678901"); @@ -38,6 +42,7 @@ const support = await getEmbeddedPostgresTestSupport(); vi.restoreAllMocks(); vi.unstubAllEnvs(); trace.disable(); context.disable(); manager.disable(); await provider.shutdown(); + for (const id of ownedCompanies.splice(0)) { await db.delete(companyMemberships).where(eq(companyMemberships.companyId, id)); await db.delete(companySecrets).where(eq(companySecrets.companyId, id)); await db.delete(companies).where(eq(companies.id, id)); } }); afterAll(async () => { await db?.$client.end({ timeout: 0 }); @@ -47,6 +52,7 @@ const support = await getEmbeddedPostgresTestSupport(); async function fixture() { const userId = `catalog-user-${randomUUID()}`; const [company] = await db.insert(companies).values({ name: "Catalog timing", issuePrefix: `GH${randomUUID().slice(0, 6)}` }).returning(); + ownedCompanies.push(company.id); await db.insert(authUsers).values({ id: userId, name: "Catalog owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" }); const [application] = await db.insert(toolApplications).values({ companyId: company.id, name: "GitHub catalog", type: "mcp_http" }).returning(); diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 52cc835a444..aa6b1545078 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -10,6 +10,8 @@ import { connectionGrants, createDb, externalObjects, + issues, + issueThreadInteractions, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, toolApplications, toolConnections, @@ -35,12 +37,14 @@ 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, { maxConnections: 4 }); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 1 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { await db.delete(activityLog); await db.delete(connectionEventDeliveries); + await db.delete(issueThreadInteractions); + await db.delete(issues); await db.delete(externalObjects); await db.delete(connectionGrants); await db.delete(toolConnections); @@ -105,6 +109,10 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { }, }, }); + const [issue] = await db.insert(issues).values({ companyId, title: "Merge release", status: "in_progress" }).returning(); + const [confirmation] = await db.insert(issueThreadInteractions).values({ companyId, issueId: issue!.id, + kind: "request_confirmation", title: "Confirm pull request merged", continuationPolicy: "none", + payload: { version: 1, prompt: "Confirm https://github.com/paperclipai/paperclip/pull/123 is merged" } }).returning(); const externalObjectId = randomUUID(); await db.insert(externalObjects).values({ id: externalObjectId, @@ -172,11 +180,15 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { } const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined); - await expect(service.pollOnce({ forSetup: activeSetup })).resolves.toMatchObject({ leased: 2, processed: 1, duplicate: 0, failed: 1 }); + const outcome = await service.pollOnce({ forSetup: activeSetup }); + const [processingReceipt] = await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.providerDeliveryId, leasedEvent.id)); + expect(processingReceipt?.lastError).toBeNull(); + expect(outcome).toMatchObject({ leased: 2, processed: 1, duplicate: 0, failed: 1 }); expect(warning.mock.calls).toEqual([[{ deliveryId: "undecryptable-delivery", reason: "inner_event_authentication_failed", }, "GitHub connector event retained for retry"]]); warning.mockRestore(); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, confirmation!.id)))[0]!.status).toBe("accepted"); const [snapshot] = await db.select().from(externalObjects).where(eq(externalObjects.id, externalObjectId)); expect(snapshot).toMatchObject({ statusKey: "merged", diff --git a/server/src/__tests__/github-connection-work.test.ts b/server/src/__tests__/github-connection-work.test.ts index f92cefd5ac1..a079c130f02 100644 --- a/server/src/__tests__/github-connection-work.test.ts +++ b/server/src/__tests__/github-connection-work.test.ts @@ -1,10 +1,12 @@ +import { notifyChatEndpointConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; +import { toolAccessService } from "../services/tool-access.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -import { randomUUID, generateKeyPairSync, verify, createHash } from "node:crypto"; +import { randomUUID, generateKeyPairSync, verify, createHash, diffieHellman, hkdfSync, randomBytes, createCipheriv } 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, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, toolApplications, toolConnections } from "@paperclipai/db"; +import { eq, sql } from "drizzle-orm"; +import { applyPendingMigrations, companies, activityLog, companySecrets, connectionGrants, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, 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"; @@ -12,6 +14,13 @@ import { githubConnectionEventService } from "../services/github-connection-even 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 { createOAuthCredentialStore } from "../services/tool-access/oauth/credential-store.js"; +import { createOAuthRefreshLeases } from "../services/tool-access/oauth/refresh-leases.js"; +import { createOAuthGrantRefresh } from "../services/tool-access/oauth/grant-refresh.js"; +import { secretService } from "../services/secrets.js"; +import { writeConnectionCredential } from "../services/connection-credentials.js"; +import { GITHUB_CONNECTOR_PROFILES } from "@paperclipai/shared"; +import { notifyToolConnectionHealth } from "../services/tool-connection-health-notifications.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; const databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_GITHUB_WORK_TEST_DATABASE_URL"); @@ -24,12 +33,16 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( let server: Server, origin: string; let bindings: Array>, leases: number, failBindings: boolean, githubRequests: number; const signing = generateKeyPairSync("ed25519"), sealing = generateKeyPairSync("x25519"); + let releaseRefresh: (() => void) | undefined, refreshArrived: (() => void) | undefined; + let refreshGate: Promise | undefined; const realFetch = globalThis.fetch; let companyIds: string[] = []; beforeAll(async () => { if (databaseUrl) await applyPendingMigrations(databaseUrl); else temporaryDatabase = await startEmbeddedPostgresTestDatabase("paperclip-github-work-"); db = createDb(databaseUrl ?? temporaryDatabase!.connectionString, { maxConnections: 4 }); + // This explicitly assigned external test database is exclusive to this suite. + await db.transaction(async tx => { await tx.execute(sql`SET LOCAL client_min_messages = warning`); await tx.execute(sql`TRUNCATE companies CASCADE`); }); 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(); @@ -39,6 +52,23 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( 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("/refresh")) { + refreshArrived?.(); await refreshGate; + const scopes = [...GITHUB_CONNECTOR_PROFILES["github.code"].scopes]; + const algorithm = "X25519-HKDF-SHA256-A256GCM"; + const purpose = "access"; + const aad = Buffer.from([1, algorithm, purpose, "instance", "test", "github", "github.code", scopes.sort().join(" ")].join("\n")); + const ephemeral = generateKeyPairSync("x25519"); + const epk = Buffer.from(ephemeral.publicKey.export({ format: "jwk" }).x!, "base64url"); + const recipient = Buffer.from(sealing.publicKey.export({ format: "jwk" }).x!, "base64url"); + const key = Buffer.from(hkdfSync("sha256", diffieHellman({ privateKey: ephemeral.privateKey, publicKey: sealing.publicKey }), Buffer.concat([epk, recipient]), aad, 32)); + const iv = randomBytes(12), cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad); + const credentials = { v: 1, accessToken: "rotated-access", refreshToken: "rotated-refresh", tokenType: "Bearer", scopes, + accessTokenExpiresAt: new Date(Date.now() + 8 * 3600_000).toISOString(), refreshTokenExpiresAt: null, + subject: claims.sub, companyId: claims.cid, instanceId: "instance", environment: "test", provider: "github", profile: "github.code" }; + const ct = Buffer.concat([cipher.update(JSON.stringify(credentials)), cipher.final(), cipher.getAuthTag()]); + res.setHeader("content-type", "application/json"); res.end(JSON.stringify({ sealed: { v: 1, alg: algorithm, purpose, provider: "github", profile: "github.code", epk: epk.toString("base64url"), iv: iv.toString("base64url"), ct: ct.toString("base64url") } })); return; + } if (req.url!.endsWith("/events/lease")) { leases++; res.setHeader("content-type", "application/json"); res.end(JSON.stringify({ events: [], ...(claims.ewv === 1 ? { eventWakeVersion: 1 } : {}) })); return; @@ -59,7 +89,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { vi.restoreAllMocks(); vi.useRealTimers(); - for (const id of companyIds) await db.delete(companies).where(eq(companies.id, id)); companyIds = []; + for (const id of companyIds) { await db.delete(activityLog).where(eq(activityLog.companyId, id)); await db.delete(companySecrets).where(eq(companySecrets.companyId, id)); await db.delete(companies).where(eq(companies.id, id)); } companyIds = []; }); afterAll(async () => { server.close(); await once(server, "close"); await db.$client.end({ timeout: 1 }); await temporaryDatabase?.cleanup(); }); @@ -106,6 +136,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( await Promise.all([ew.ready, cw.ready]); expect(leases).toBe(1); const startupLeases = leases; await f.service.refreshManagedGitHubGrantAccess(f.connection, f.grant, { actorType: "system", actorId: "test" }); + await vi.advanceTimersByTimeAsync(6_000); await vi.waitFor(() => expect(leases).toBeGreaterThan(startupLeases)); await vi.advanceTimersByTimeAsync(31_000); await settled(coordinator); const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); @@ -148,6 +179,111 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( expect(await restarted.nextGitHubConnectionContinuityAt()).toBeNull(); }); + it("preserves webhook removal debt across an actual signed, held Cloud token rotation and lease cleanup", async () => { + const f = await fixture(["101"]); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "01234567890123456789012345678901"); + const refs = []; + for (const path of ["oauth.access_token", "oauth.refresh_token"]) { + const saved = await writeConnectionCredential(db, { companyId: f.company.id, connectionName: "GitHub", configPath: path, label: path, value: path, ownerUserId: "owner" }); + refs.push({ configPath: path, secretId: saved.secret.id, versionSelector: "latest" as const }); + } + const [connection] = await db.update(toolConnections).set({ credentialSource: "paperclip_vault", config: { ...f.connection.config, + oauth: { strategy: "paperclip_cloud_connector", connectorProfile: "github.code" } } }).where(eq(toolConnections.id, f.connection.id)).returning(); + const [grant] = await db.update(connectionGrants).set({ credentialSecretRefs: refs }).where(eq(connectionGrants.id, f.grant.id)).returning(); + const getConnection = async () => (await db.select().from(toolConnections).where(eq(toolConnections.id, connection!.id)))[0]!; + const store = createOAuthCredentialStore(db, secretService(db)); + await store.syncCredentialBindings(connection!, refs); + const leases = createOAuthRefreshLeases(db, () => new Date(), getConnection, store.getOAuthGrantRow); + const rotation = createOAuthGrantRefresh(db, { now: () => new Date(), store, leases, getConnectionRow: getConnection, + currentCloudConnector: () => f.connector, exchangeOAuthToken: vi.fn() }); + const arrived = new Promise(resolve => { refreshArrived = resolve; }); + refreshGate = new Promise(resolve => { releaseRefresh = resolve; }); + const pending = rotation({ companyId: f.company.id, connectionId: connection!.id, grantId: grant!.id, forceRefresh: true }); + const result = Promise.allSettled([pending]); + try { + await Promise.race([arrived, pending.then(() => { throw new Error("Rotation returned before the Cloud HTTP boundary"); })]); + const connector = { ...f.connector, acknowledgeEvents: async () => 1, leaseEvents: async () => ({ leaseId: "removal", events: [{ id: randomUUID(), provider: "github" as const, + event: "installation", action: "deleted", installationId: "101", repositoryId: null, createdAt: new Date(Date.now() + 1000).toISOString(), bindingIds: [`${grant!.id}_101`], payload: { installationId: "101" } }] }) }; + expect(await githubConnectionEventService(db, { connector }).pollOnce()).toMatchObject({ processed: 1 }); + const [before] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant!.id)); + expect(before!.providerTenant!.github!.bindingSync!.removeInstallationIds).toEqual(["101"]); + releaseRefresh!(); + const [completion] = await result; if (completion.status === "rejected") throw completion.reason; + const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant!.id)); + expect(saved!.providerTenant!.github).toEqual(before!.providerTenant!.github); + expect(saved!.providerTenant!.oauth).toMatchObject({ tokenType: "Bearer" }); + expect(saved!.providerTenant!.oauth!.refreshLease).toBeUndefined(); + await f.service.syncGitHubGrantBindings(connection!, saved!); + expect(bindings).toContainEqual(expect.objectContaining({ installationId: "101", active: false })); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant!.id)))[0]!.providerTenant!.github!.bindingSync).toBeUndefined(); + } finally { releaseRefresh?.(); await result; refreshGate = undefined; refreshArrived = undefined; vi.unstubAllEnvs(); } + }); + + it("lease acquire and cleanup cannot replace a concurrent webhook metadata revision", async () => { + const f = await fixture(["101"]); + let interleave = true; + const removal = { id: randomUUID(), removeInstallationIds: ["202"] }; + const getGrant = async () => { + const [snapshot] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); + if (interleave) { + interleave = false; + await db.update(connectionGrants).set({ providerTenant: sql`jsonb_set(${connectionGrants.providerTenant}, '{github,bindingSync}', ${JSON.stringify(removal)}::jsonb)` }).where(eq(connectionGrants.id, f.grant.id)); + } + return snapshot!; + }; + const leaseOwner = createOAuthRefreshLeases(db, () => new Date(), async () => f.connection, getGrant); + const lease = await leaseOwner.acquireOAuthGrantRefreshLease(f.connection, f.grant, true); + expect(lease.leaseId).toBeTruthy(); + expect(lease.grant.providerTenant!.github!.bindingSync).toEqual(removal); + // Cleanup deliberately reads a stale snapshot immediately before a new delta. + const second = { ...removal, id: randomUUID(), removeInstallationIds: ["202", "303"] }; + const cleanup = createOAuthRefreshLeases(db, () => new Date(), async () => f.connection, async () => { + const snapshot = await getGrant(); + await db.update(connectionGrants).set({ providerTenant: sql`jsonb_set(${connectionGrants.providerTenant}, '{github,bindingSync}', ${JSON.stringify(second)}::jsonb)` }).where(eq(connectionGrants.id, f.grant.id)); + return snapshot; + }); + const cleared = await cleanup.clearOAuthGrantRefreshLease(lease.grant, lease.leaseId!); + expect(cleared.providerTenant!.github!.bindingSync).toEqual(second); + expect(cleared.providerTenant!.oauth!.refreshLease).toBeUndefined(); + }); + + it("ignores unrelated health write bursts and scopes credential reconstruction to its owning company", async () => { + const f = await fixture(["101"]), other = await fixture(["202"]); fakeClock(); + // A large unrelated company set must not enter a targeted credential reload. + for (let index = 0; index < 50; index++) { + const id = randomUUID(); companyIds.push(id); + await db.insert(companies).values({ id, name: "Unrelated credential owner", issuePrefix: `U${index}${id.slice(0, 6)}` }); + } + const [unrelatedConnection] = await db.update(toolConnections).set({ config: {}, transportConfig: {}, transport: "rest_api", authKind: "none", credentialRefs: [], credentialSecretRefs: [] }).where(eq(toolConnections.id, other.connection.id)).returning(); + await db.update(connectionGrants).set({ credentialSecretRefs: [] }).where(eq(connectionGrants.id, other.grant.id)); + const api = toolAccessService(db); + const load = vi.spyOn(f.service, "githubConnectionContinuitySchedule"); + const sweep = vi.spyOn(f.service, "sweepGitHubConnectionContinuity"); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true }); + const continuity = registerGitHubConnectionContinuityWork(db, coordinator, f.service, () => true); + const events = registerGitHubConnectionEventWork(db, coordinator, githubConnectionEventService(db, { connector: f.connector }), () => true); + try { + await Promise.all([continuity.ready, events.ready]); + await vi.waitFor(() => expect(leases).toBe(1)); + await vi.waitFor(() => expect(load).toHaveBeenCalledTimes(1)); + load.mockClear(); sweep.mockClear(); + for (let i = 0; i < 30; i++) { + await db.transaction(async tx => { await notifyToolConnectionHealth(tx, other.connection); await notifyChatEndpointConfiguration(tx, { id: randomUUID(), provider: "discord" }); }); + await api.updateConnection(unrelatedConnection!.id, { name: `Unrelated patch ${i}` }); + } + await vi.advanceTimersByTimeAsync(10_000); + expect(load).not.toHaveBeenCalled(); expect(sweep).not.toHaveBeenCalled(); expect(leases).toBe(1); + for (let index = 0; index < 30; index++) await db.transaction(async tx => { await notifyGitHubCredentialChange(tx, f.connection); }); + expect(leases).toBe(1); + await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(load).toHaveBeenCalledWith(f.company.id)); + expect(load.mock.calls.every(([companyId]) => companyId === f.company.id)).toBe(true); + expect(sweep).not.toHaveBeenCalled(); + await vi.waitFor(() => expect(leases).toBe(2)); + expect(load).toHaveBeenCalledTimes(1); + } finally { await events.stop(); await continuity.stop(); await coordinator.stop(); } + }); + it("settles only the company and grant that own saved binding debt", async () => { const f = await fixture(["101"]), other = await fixture(["202"]); await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant, github: { ...f.grant.providerTenant!.github!, diff --git a/server/src/__tests__/openapi-routes.test.ts b/server/src/__tests__/openapi-routes.test.ts index 0a1a220b8fd..e763cc85608 100644 --- a/server/src/__tests__/openapi-routes.test.ts +++ b/server/src/__tests__/openapi-routes.test.ts @@ -47,6 +47,7 @@ const apiPrefixes: Record = { "company-skills.ts": "/api", "company-skill-policy.ts": "/api", "connection-intents.ts": "/api", + "connection-event-wake.ts": "/", "costs.ts": "/api", "dashboard.ts": "/api", "dot-runner.ts": "/api", @@ -113,6 +114,8 @@ const HTTP_METHODS = new Set([ const explicitOpenApiCoverageExclusions = new Set(); const explicitOpenApiOperationCoverageExclusions = new Set([ + // Exact Cloud-control signed action; ordinary board/agent credentials cannot invoke it. + "POST /api/instance/connector-events/wake", // Muse uses its closed, versioned receiver protocol and dedicated credentials, // documented in doc/muse-personal-agent.md and the shared protocol contract. // Board connection/qualification management remains in this OpenAPI document. @@ -180,6 +183,7 @@ function normalizeExpressPath(routePath: string) { } function resolveMountedPath(file: string, prefix: string, routePath: string) { + if (file === "connection-event-wake.ts") return routePath; if (file === "public-mcp.ts" && (routePath.startsWith("/mcp/oauth/") || routePath.startsWith("/.well-known/"))) return routePath; if ( (file === "chat-channels.ts" || file === "email.ts") && diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index d73175ee556..29ff63b985b 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -492,6 +492,7 @@ vi.mock("../services/index.js", () => ({ failed: 0, })), nextGitHubConnectionContinuityAt: vi.fn(async () => null), + githubConnectionContinuitySchedule: vi.fn(async () => []), sweepGitHubConnectionContinuity: vi.fn(async () => ({ checked: 0, due: 0, @@ -1303,7 +1304,9 @@ describe("startServer feedback export wiring", () => { } 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 schedules = vi.mocked(toolAccessService).mock.results.map(result => result.value.githubConnectionContinuitySchedule); + expect(sweeps.reduce((sum, call) => sum + call.mock.calls.length, 0)).toBe(0); + const count = () => [...polls, ...sweeps, ...schedules].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(); diff --git a/server/src/services/chat-channels/configuration-work-notifications.ts b/server/src/services/chat-channels/configuration-work-notifications.ts index 9d3240aee12..8a301b7cd0f 100644 --- a/server/src/services/chat-channels/configuration-work-notifications.ts +++ b/server/src/services/chat-channels/configuration-work-notifications.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyGitHubWebhookRecovery } from "./github/webhook-recovery-notifications.js"; import { notifyVoiceChange } from "../voice/voice-recovery-notifications.js"; import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js"; @@ -7,7 +8,10 @@ import { notifyDiscordCommandRegistration } from "./discord/command-work-notific export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { if (endpoint.provider === "speko") await notifyVoiceChange(transaction, { kind: "endpoint", id: endpoint.id }); await notifyChatReceiver(transaction, endpoint); - if (endpoint.provider === "github") await notifyGitHubWebhookRecovery(transaction, { kind: "endpoint", id: endpoint.id }); + if (endpoint.provider === "github") { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionEvents); + await notifyGitHubWebhookRecovery(transaction, { kind: "endpoint", id: endpoint.id }); + } if (endpoint.provider === "discord") { await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); } diff --git a/server/src/services/chat-channels/configuration.ts b/server/src/services/chat-channels/configuration.ts index fa1b1d5b1ce..8ded7cbb96e 100644 --- a/server/src/services/chat-channels/configuration.ts +++ b/server/src/services/chat-channels/configuration.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import { CHAT_HUMAN_CREDENTIAL_FIELDS } from "./credentials.js"; @@ -196,6 +197,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption code: "chat_endpoint_not_active", }); } + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) @@ -334,6 +336,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption inArray(chatDeliveries.state, ["received", "retry"]), ), ); + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); [activatedEndpoint] = await tx .update(chatEndpoints) @@ -383,6 +386,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption await managementContext?.reauthorize(); await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) @@ -487,6 +491,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption code: "chat_endpoint_already_removed", }); } + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) @@ -721,6 +726,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption await managementContext?.reauthorize(); const claimed = await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); const rows = await tx .update(chatEndpoints) @@ -800,6 +806,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption await installSpekoVoiceTools(tx, endpoint, actorUserId); } const preserveSlackVerification = preservesProvisionedSlackVerification(current.setup, credentials.signingSecret ?? "", `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/slack`); + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) @@ -887,6 +894,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption await managementContext?.reauthorize(); await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyChatEndpointConfiguration(tx, endpoint); await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) diff --git a/server/src/services/chat-github-wizard.ts b/server/src/services/chat-github-wizard.ts index cd5f38a1c75..abde5141afd 100644 --- a/server/src/services/chat-github-wizard.ts +++ b/server/src/services/chat-github-wizard.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyGitHubWebhookRecovery } from "./chat-channels/github/webhook-recovery-notifications.js"; import type { AuthorizationActor } from "./authorization.js"; import { connectionManagementService, type ConnectionManagementContext } from "./connection-management.js"; @@ -173,8 +174,10 @@ export function githubChatWizardService( values: Partial>, ) { await db.transaction(async tx => { - if (values.cloudRegistrationId !== undefined) + if (values.cloudRegistrationId !== undefined) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.connectionEvents, bot.companyId); await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: bot.id }); + } await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index cce62d9c934..d26f41c761f 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCompanyChange } from "./github-connection-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -332,8 +333,7 @@ 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 notifyGitHubCompanyChange(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); @@ -490,8 +490,7 @@ 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 notifyGitHubCompanyChange(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 53834d6c4f5..5db8585ffcf 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -237,6 +237,8 @@ export function githubConnectionEventService( eq(toolConnections.companyId, connectionGrants.companyId), )) .where(and( + sql`${toolConnections.config}->>'sourceTemplateKey' = 'github'`, + sql`${toolConnections.config}->'oauth'->>'connectorProfile' = 'github.code'`, eq(connectionGrants.status, "active"), eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), @@ -245,11 +247,11 @@ export function githubConnectionEventService( return bindingRows(rows); } - async function applyPullRequestEvent(companyId: string, event: LeasedEvent) { + async function applyPullRequestEvent(database: Db, companyId: string, event: LeasedEvent, effects: Array<(committedDb: Db) => Promise>) { const snapshot = githubSnapshotUpdate(event.payload); if (!snapshot) return; const appliedAt = now(); - await db.update(externalObjects).set({ + await database.update(externalObjects).set({ statusKey: snapshot.statusKey, statusLabel: snapshot.statusLabel, statusCategory: snapshot.statusCategory, @@ -269,13 +271,13 @@ export function githubConnectionEventService( sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`, )); if (snapshot.merged && event.action === "closed") { - await issueThreadInteractionService(db, { wakeup: options.wakeup }) + await issueThreadInteractionService(database, { wakeup: options.wakeup }) .sweepMergedPullRequestConfirmations([{ companyId, owner: snapshot.owner, repo: snapshot.repo, number: snapshot.number, - }]); + }], { deferConfirmationCommitEffects: effect => effects.push(effect) }); } } @@ -337,7 +339,7 @@ export function githubConnectionEventService( webhookHealth: unavailable ? "unhealthy" as const : "healthy" as const, }, }; - await notifyGitHubCredentialChange(database, { config: { sourceTemplateKey: "github" } }); + await notifyGitHubCredentialChange(database, { companyId: binding.companyId, 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 }); @@ -380,6 +382,7 @@ export function githubConnectionEventService( } const postCommitPublications: ActivityPublication[] = []; + const postCommitEffects: Array<(committedDb: Db) => Promise> = []; let applied = true; try { const applyAndFinalize = async (database: Db) => { @@ -391,7 +394,7 @@ export function githubConnectionEventService( 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 === "pull_request") await applyPullRequestEvent(database, companyId, normalizedEvent, postCommitEffects); if (event.event === "installation" || event.event === "installation_repositories") { for (const binding of bindings) await applyInstallationEvent(database, binding, normalizedEvent); } else { @@ -470,6 +473,7 @@ export function githubConnectionEventService( }, "GitHub webhook activity publication failed after commit"); } } + for (const effect of postCommitEffects) await effect(db); return applied ? "processed" as const : "duplicate" as const; } diff --git a/server/src/services/github-connection-notifications.ts b/server/src/services/github-connection-notifications.ts index 79f3c067aba..55811a3a64d 100644 --- a/server/src/services/github-connection-notifications.ts +++ b/server/src/services/github-connection-notifications.ts @@ -1,7 +1,26 @@ -import type { toolConnections } from "@paperclipai/db"; +import { toolConnections, chatEndpoints, type Db } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyGitHubCredentialChange(transaction: object, connection: Pick) { +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); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionContinuity, connection.companyId); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionEvents, connection.companyId); +} + +/** Company lifecycle changes only activate owners with a supported registration. */ +export async function notifyGitHubCompanyChange(transaction: object, companyId: string) { + const db = transaction as Db; + const [connection] = await db.select({ id: toolConnections.id }).from(toolConnections).where(and( + eq(toolConnections.companyId, companyId), sql`${toolConnections.config}->>'sourceTemplateKey' = 'github'`, + )).limit(1); + if (connection) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionContinuity, companyId); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionEvents, companyId); + return; + } + const [bot] = await db.select({ id: chatEndpoints.id }).from(chatEndpoints).where(and( + eq(chatEndpoints.companyId, companyId), eq(chatEndpoints.provider, "github"), + sql`${chatEndpoints.setup}->'github'->>'cloudRegistrationId' is not null`, + )).limit(1); + if (bot) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.connectionEvents, companyId); } diff --git a/server/src/services/github-connection-work.ts b/server/src/services/github-connection-work.ts index 7849cf3bda8..45d2d0ee6e2 100644 --- a/server/src/services/github-connection-work.ts +++ b/server/src/services/github-connection-work.ts @@ -1,8 +1,9 @@ import { logger } from "../middleware/logger.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.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 { DELIVERY_QUEUES } 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"; @@ -11,26 +12,28 @@ 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, + retryMs: 5_000, minimumSpacingMs: 5_000, notificationCoalesceMs: 1_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)]; + const stops = [registerConnectionEventWake(db, 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">, + service: Pick, "sweepGitHubConnectionContinuity" | "nextGitHubConnectionContinuityAt" | "githubConnectionContinuitySchedule">, canRun: () => boolean) { - const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.connectionContinuity, { - retryMs: 30_000, canRun, - run: async () => { - const result = await service.sweepGitHubConnectionContinuity(); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.connectionContinuity, + minimumPassIntervalMs: 5_000, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: (entry, companyId?: string) => !companyId || entry.companyId === companyId, + load: companyId => service.githubConnectionContinuitySchedule(companyId), + check: async entry => { + const result = await service.sweepGitHubConnectionContinuity(entry.companyId); if (result.failed) logger.warn(result, "GitHub credential continuity retained retry obligations"); + const dueAt = await service.nextGitHubConnectionContinuityAt(entry.companyId); + return dueAt === null ? null : { ...entry, dueAt }; }, - next: () => service.nextGitHubConnectionContinuityAt(), - }); - const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.toolConnectionHealth, worker.wake); - return { ...worker, stop: async () => { unsubscribe(); await worker.stop(); } }; + }, canRun); + return worker; } diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 59c14618342..730a82db3a6 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -2986,6 +2986,7 @@ export function issueThreadInteractionService( repo: string; number: number; }> = [], + mutationOptions: InteractionResolutionMutationOptions = {}, ) => { const rows = await db .select({ @@ -3005,6 +3006,7 @@ export function issueThreadInteractionService( and( eq(issueThreadInteractions.kind, "request_confirmation"), eq(issueThreadInteractions.status, "pending"), + mergedHints.length ? inArray(issues.companyId, [...new Set(mergedHints.map(hint => hint.companyId))]) : undefined, ), ); @@ -3087,6 +3089,7 @@ export function issueThreadInteractionService( issue: candidate.issue, current: candidate.interaction, input: {}, + mutationOptions, actor: { systemId: "system:pr-merged", resolutionDetails: { @@ -3123,7 +3126,7 @@ export function issueThreadInteractionService( ) { continue; } - await opts.wakeup(wakeIssue.assigneeAgentId, { + const wake = async () => { await opts.wakeup!(wakeIssue.assigneeAgentId!, { source: "automation", triggerDetail: "system", reason: "issue_commented", @@ -3154,6 +3157,9 @@ export function issueThreadInteractionService( source: "merged_pull_request_sweep", }, }); + }; + if (mutationOptions.deferConfirmationCommitEffects) mutationOptions.deferConfirmationCommitEffects(wake); + else await wake(); woken += 1; } diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 7d4cfe8cf0d..23932fafbf9 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "./github-connection-notifications.js"; import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js"; import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { connectionHealthSchedule } from "./tool-access/catalog/health-schedule.js"; @@ -787,7 +788,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, syncGitHubGrantBindings, sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt } = createGitHubGrantContinuity(db, { now, currentCloudConnector, refreshOAuthGrantCredentials, resolveOAuthGrantSecret }); + const { refreshManagedGitHubGrantAccess, syncGitHubGrantBindings, sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt, githubConnectionContinuitySchedule } = 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 }); @@ -3152,6 +3153,7 @@ export function toolAccessService( await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, companyId))) .for("update"); + await notifyGitHubCredentialChange(tx, connection); await notifyToolConnectionHealth(tx, connection); const [row] = await tx.update(toolConnections).set({ config, transportConfig: config, status: "active", enabled: true, ...(input.agentInstructions !== undefined ? { agentInstructions: input.agentInstructions } : {}), updatedAt: new Date() }) .where(and(eq(toolConnections.id, connection.id), eq(toolConnections.companyId, companyId))).returning(); @@ -3510,6 +3512,7 @@ export function toolAccessService( }, tx, ); + await notifyGitHubCredentialChange(tx, connection); await notifyToolConnectionHealth(tx, connection); const [updatedConnection] = await tx .update(toolConnections) @@ -4673,6 +4676,10 @@ export function toolAccessService( return rows; }) : await writeToolConnectionHealthChange(db, existing, async tx => { + if (input.enabled !== undefined || input.status !== undefined || input.config !== undefined || input.transportConfig !== undefined || input.credentialSecretRefs !== undefined || input.credentialRefs !== undefined) { + await notifyGitHubCredentialChange(tx, existing); + await notifyGitHubCredentialChange(tx, { companyId: existing.companyId, config }); + } if (existing.transport === "voice") await notifyVoiceChange(tx, { kind: "connection", id: existing.id }); const rows = await write(tx); await syncCredentialBindings(rows[0]!, [], tx); @@ -4733,6 +4740,7 @@ export function toolAccessService( sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt, + githubConnectionContinuitySchedule, ...catalogReader, ...toolActionReviewQueue, ...profileService, diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index a9c201f9984..be9bea0dd78 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "../../github-connection-notifications.js"; import { notifyVoiceChange } from "../../voice/voice-recovery-notifications.js"; import { notifyChatEndpointWork } from "../../chat-work-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; @@ -543,6 +544,7 @@ async function removeConnection( const archived = await db.transaction(async (tx) => { if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id, connection.companyId); if (connection.transport === "voice") await notifyVoiceChange(tx, { kind: "connection", id: connection.id }); + await notifyGitHubCredentialChange(tx, connection); await notifyToolConnectionHealth(tx, connection); await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId))) diff --git a/server/src/services/tool-access/github-grant-continuity.ts b/server/src/services/tool-access/github-grant-continuity.ts index 1096d5f08e6..e3095597d4b 100644 --- a/server/src/services/tool-access/github-grant-continuity.ts +++ b/server/src/services/tool-access/github-grant-continuity.ts @@ -204,7 +204,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector return { updated, previousGitHub }; }); - return syncGitHubGrantBindings(connection, updated, actor, accessToken); + return measureRunPhase("connectors.github.webhooks", () => syncGitHubGrantBindings(connection, updated, actor, accessToken)); } async function syncGitHubGrantBindings( @@ -287,7 +287,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector }); } - async function continuityRows() { + async function continuityRows(companyId?: string) { return db .select({ grant: connectionGrants, connection: toolConnections }) .from(connectionGrants) @@ -296,6 +296,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector eq(toolConnections.companyId, connectionGrants.companyId), )) .where(and( + companyId ? eq(connectionGrants.companyId, companyId) : undefined, eq(connectionGrants.status, "active"), eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), @@ -306,16 +307,16 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector )); } - async function nextGitHubConnectionContinuityAt(): Promise { - const deadlines = (await continuityRows()).flatMap(({ grant, connection }) => { + async function nextGitHubConnectionContinuityAt(companyId?: string): Promise { + const deadlines = (await continuityRows(companyId)).flatMap(({ grant, connection }) => { const deadline = githubGrantContinuityDeadline(connection, grant); return deadline === null ? [] : [deadline]; }); return deadlines.length ? Math.min(...deadlines) : null; } - async function sweepGitHubConnectionContinuity() { - const rows = await continuityRows(); + async function sweepGitHubConnectionContinuity(companyId?: string) { + const rows = await continuityRows(companyId); const due = rows.filter(({ grant, connection }) => { const deadline = githubGrantContinuityDeadline(connection, grant); return deadline !== null && deadline <= now().getTime(); @@ -340,7 +341,15 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector } return { checked: rows.length, due: due.length, refreshed, failed }; } - return { refreshManagedGitHubGrantAccess, syncGitHubGrantBindings, sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt }; + async function githubConnectionContinuitySchedule(companyId?: string) { + const schedule = new Map(); + for (const { grant, connection } of await continuityRows(companyId)) { + const dueAt = githubGrantContinuityDeadline(connection, grant); + if (dueAt !== null) schedule.set(connection.companyId, Math.min(schedule.get(connection.companyId) ?? Infinity, dueAt)); + } + return [...schedule].map(([companyId, dueAt]) => ({ id: companyId, companyId, dueAt })); + } + return { githubConnectionContinuitySchedule, refreshManagedGitHubGrantAccess, syncGitHubGrantBindings, sweepGitHubConnectionContinuity, nextGitHubConnectionContinuityAt }; } /** Credential expiry and provider continuity are obligations, not a scan cadence. */ diff --git a/server/src/services/tool-access/oauth/access-finalization.ts b/server/src/services/tool-access/oauth/access-finalization.ts index a6ad361dde8..4d0ab9ed010 100644 --- a/server/src/services/tool-access/oauth/access-finalization.ts +++ b/server/src/services/tool-access/oauth/access-finalization.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "../../github-connection-notifications.js"; import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import type { ActorInfo } from "../types.js"; @@ -116,6 +117,7 @@ async function finalizeOAuthAccess( throw conflict("This connection already uses a company identity"); } connection = await writeToolConnectionHealthChange(db, connection, async tx => { + await notifyGitHubCredentialChange(tx, connection); const [updated] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/cloud-callback.ts b/server/src/services/tool-access/oauth/cloud-callback.ts index 7381c1687a4..ebba51b0b92 100644 --- a/server/src/services/tool-access/oauth/cloud-callback.ts +++ b/server/src/services/tool-access/oauth/cloud-callback.ts @@ -1,3 +1,4 @@ +import { notifyGitHubCredentialChange } from "../../github-connection-notifications.js"; import type { createGitHubGrantContinuity } from "../github-grant-continuity.js"; import { randomUUID } from "node:crypto"; import { notifyIssueReviewState } from "../../issue-review-state-notifications.js"; @@ -304,6 +305,7 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { accessRef, ...(refreshRef ? [refreshRef] : []), ]; + await notifyGitHubCredentialChange(tx, connection); await notifyToolConnectionHealth(tx, connection); const grantValues = { providerTenant: { diff --git a/server/src/services/tool-access/oauth/grant-refresh.ts b/server/src/services/tool-access/oauth/grant-refresh.ts index ce0ca102352..096b5d4f9c9 100644 --- a/server/src/services/tool-access/oauth/grant-refresh.ts +++ b/server/src/services/tool-access/oauth/grant-refresh.ts @@ -14,7 +14,7 @@ import { secretService } from "../../secrets.js"; import { isPaperclipCloudConnectorStrategy, PaperclipCloudConnectorError, type PaperclipCloudConnector } from "../../paperclip-cloud-connector.js"; import { actorForSecret, type OAuthCredentialStore } from "./credential-store.js"; -import { withoutOAuthRefreshLease, withoutOAuthGrantRefreshLease, type OAuthRefreshLeases } from "./refresh-leases.js"; +import { withoutOAuthRefreshLease, type OAuthRefreshLeases } from "./refresh-leases.js"; import { managedConnectorProfile, oauthGrantConfig, oauthGrantExpiresAtMs } from "./client-settings.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; import { singleFlight } from "../single-flight.js"; @@ -179,10 +179,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo grant.kind === "user" && grant.subjectUserId ? { actorType: "user", actorId: grant.subjectUserId } : input.actor; - const providerTenant = { - ...(grant.providerTenant ?? {}), - oauth: { - ...withoutOAuthGrantRefreshLease(oauthGrantConfig(grant)), + const oauthUpdates = { strategy: "paperclip_cloud_connector", accessTokenExpiresAt: credentials.accessTokenExpiresAt, scopes: credentials.scopes, @@ -191,7 +188,6 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo ...(credentials.refreshTokenExpiresAt ? { refreshTokenExpiresAt: credentials.refreshTokenExpiresAt } : {}), - }, }; const updated = await db.transaction(async (tx) => { await notifyGitHubCredentialChange(tx, connection); @@ -219,7 +215,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo const [committed] = await tx .update(connectionGrants) .set({ - providerTenant, + providerTenant: sql`jsonb_set(coalesce(${connectionGrants.providerTenant}, '{}'::jsonb), '{oauth}', (coalesce(${connectionGrants.providerTenant}->'oauth', '{}'::jsonb) || ${JSON.stringify(oauthUpdates)}::jsonb) - 'refreshLease')`, status: "active", updatedAt: now(), }) @@ -344,18 +340,11 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo ref.configPath !== "oauth.access_token" && ref.configPath !== "oauth.refresh_token", ); - const providerTenant = { - ...(grant.providerTenant ?? {}), - oauth: { - ...withoutOAuthGrantRefreshLease(grantOauth), - accessTokenExpiresAt: undefined, - }, - }; const [marked] = await db .update(connectionGrants) .set({ status: "needs_reauthorization", - providerTenant, + providerTenant: sql`${connectionGrants.providerTenant} #- '{oauth,refreshLease}' #- '{oauth,accessTokenExpiresAt}'`, credentialSecretRefs: retainedCredentialSecretRefs, updatedAt: now(), }) @@ -525,26 +514,22 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo if (connection.config.sourceTemplateKey === "google-workspace") { refreshed.scopes = googleWorkspaceEffectiveScopes(normalizeOauthScopes(grantOauth.requestedScopes ?? grantOauth.scopes), refreshed.scopes); } - const providerTenant = { - ...(grant.providerTenant ?? {}), - oauth: { - ...withoutOAuthGrantRefreshLease(grantOauth), + const oauthUpdates = { strategy: typeof grantOauth.strategy === "string" ? grantOauth.strategy : "direct_oauth", - accessTokenExpiresAt: expiresAt ?? undefined, + accessTokenExpiresAt: expiresAt ?? null, scopes: refreshed.scopes, scopeSource: refreshed.scopeSource, unrequestedScopes: refreshed.unrequestedScopes, tokenType: token.tokenType, refreshedAt: now().toISOString(), - }, }; const [updated] = await writeToolConnectionHealthChange(db, connection, tx => tx .update(connectionGrants) .set({ - providerTenant, + providerTenant: sql`jsonb_set(coalesce(${connectionGrants.providerTenant}, '{}'::jsonb), '{oauth}', (coalesce(${connectionGrants.providerTenant}->'oauth', '{}'::jsonb) || ${JSON.stringify(oauthUpdates)}::jsonb) - 'refreshLease')`, credentialSecretRefs: nextCredentialSecretRefs, status: "active", updatedAt: now(), diff --git a/server/src/services/tool-access/oauth/refresh-leases.ts b/server/src/services/tool-access/oauth/refresh-leases.ts index 8250da26e65..44dd257129d 100644 --- a/server/src/services/tool-access/oauth/refresh-leases.ts +++ b/server/src/services/tool-access/oauth/refresh-leases.ts @@ -43,15 +43,11 @@ export function createOAuthRefreshLeases(db: Db, now: () => Date, getConnectionR ) { const latest = await getConnectionRow(connection.id, connection.companyId); if (oauthRefreshLeaseId(latest) !== leaseId) return latest; - const nextConfig = { - ...latest.config, - oauth: withoutOAuthRefreshLease(oauthConfig(latest)), - }; const [updated] = await db .update(toolConnections) .set({ - config: nextConfig, - transportConfig: nextConfig, + config: sql`${toolConnections.config} #- '{oauth,refreshLease}'`, + transportConfig: sql`${toolConnections.transportConfig} #- '{oauth,refreshLease}'`, updatedAt: now(), }) .where( @@ -165,14 +161,10 @@ export function createOAuthRefreshLeases(db: Db, now: () => Date, getConnectionR }); const oauth = oauthGrantConfig(latest); if (asRecord(oauth.refreshLease).id !== leaseId) return latest; - const providerTenant = { - ...(latest.providerTenant ?? {}), - oauth: withoutOAuthGrantRefreshLease(oauth), - }; const [updated] = await db .update(connectionGrants) .set({ - providerTenant, + providerTenant: sql`${connectionGrants.providerTenant} #- '{oauth,refreshLease}'`, updatedAt: now(), }) .where( @@ -202,7 +194,7 @@ export function createOAuthRefreshLeases(db: Db, now: () => Date, getConnectionR leaseId: string | null; }> { const waitDeadline = Date.now() + OAUTH_REFRESH_LEASE_WAIT_MS; - const initialUpdatedAt = grant.updatedAt.getTime(); + const initialOAuth = JSON.stringify(withoutOAuthGrantRefreshLease(oauthGrantConfig(grant))); while (true) { const latest = await getOAuthGrantRow({ companyId: connection.companyId, @@ -221,7 +213,7 @@ export function createOAuthRefreshLeases(db: Db, now: () => Date, getConnectionR (!forceRefresh && (expiresAtMs === null || expiresAtMs > Date.now() + 60_000)) || (forceRefresh && - latest.updatedAt.getTime() > initialUpdatedAt && + JSON.stringify(withoutOAuthGrantRefreshLease(oauthGrantConfig(latest))) !== initialOAuth && !asRecord(oauthGrantConfig(latest).refreshLease).id) ) { return { grant: latest, leaseId: null }; @@ -266,6 +258,7 @@ export function createOAuthRefreshLeases(db: Db, now: () => Date, getConnectionR eq(connectionGrants.id, latest.id), eq(connectionGrants.companyId, latest.companyId), eq(connectionGrants.status, "active"), + sql`${connectionGrants.providerTenant} = ${JSON.stringify(latest.providerTenant)}::jsonb`, sql`${connectionGrants.providerTenant} #>> '{oauth,refreshLease,id}' is null`, ), )