diff --git a/doc/plans/2026-10-30-github-connection-sleep.md b/doc/plans/2026-10-30-github-connection-sleep.md index 3e6de1148bb..287b479b07c 100644 --- a/doc/plans/2026-10-30-github-connection-sleep.md +++ b/doc/plans/2026-10-30-github-connection-sleep.md @@ -114,3 +114,28 @@ concurrent replay, signed callbacks, outer commit/rollback, drain/resume, startu 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. + +## Enrollment after worker startup + +The event owner resolves the current connector only when a delivery or setup pass +is admitted. A service created before self-host enrollment can therefore use a +later completed enrollment without being recreated. The full normalized identity +and target configuration keys the connector cache, preserving negotiated event +wake support while it is unchanged. An absent identity clears the cache; a +replacement identity or changed key creates a new connector. Explicitly injected +connectors keep their original lifetime. + +Enrollment does not add a timer, database scan or network probe. An ordinary +credential/configuration producer, authenticated control wake or active setup +request admits the next pass. Unconfigured passes return before owner SQL or +HTTP. A pass already in flight keeps its connector for that lease and its matching +acknowledgment; identity discovery never changes authority halfway through a pass. + +The real PostgreSQL/loopback regression starts the registered owner unenrolled, +completes the canonical signed enrollment flow, then applies and acknowledges an +encrypted installation event. Removing the identity parks the same owner, and a +second enrollment proves the replacement signing/sealing identity is used. Each +applied installation update has one bounded empty follow-up from its credential +hint; the test joins that saved deadline once before checking 25-minute SQL/HTTP +quiet windows. This is process-local enrollment recovery, not a deployed Cloud +rollout or a new cancellation policy. diff --git a/server/src/__tests__/github-connection-enrollment-work.test.ts b/server/src/__tests__/github-connection-enrollment-work.test.ts new file mode 100644 index 00000000000..c888565b3d9 --- /dev/null +++ b/server/src/__tests__/github-connection-enrollment-work.test.ts @@ -0,0 +1,139 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createServer, type Server } from "node:http"; +import { once } from "node:events"; +import { createCipheriv, createPublicKey, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, verify } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { companies, connectionEventDeliveries, connectionGrants, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { githubConnectionEventService } from "../services/github-connection-events.js"; +import { registerGitHubConnectionEventWork } from "../services/github-connection-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { wakeConnectionEvents } from "../services/connection-event-wake.js"; +import { completePaperclipCloudConnectorEnrollment, loadPaperclipCloudConnectorIdentity, startPaperclipCloudConnectorEnrollment } from "../services/paperclip-cloud-connector-enrollment.js"; +import { paperclipCloudConnectorIdentityPath } from "../services/paperclip-cloud-connector-enrollment.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("GitHub enrollment on a live event worker (real PostgreSQL and signed loopback Cloud)", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("github-live-enrollment-"); + db = createDb(temporary.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 90_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + it("enrolls without recreation, drains an authenticated event, forgets removed identity and re-enrolls from quiet", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "github-live-enrollment-")); + const previous = { home: process.env.PAPERCLIP_HOME, instance: process.env.PAPERCLIP_INSTANCE_ID }; + process.env.PAPERCLIP_HOME = root; process.env.PAPERCLIP_INSTANCE_ID = "enrollment-worker"; + let server: Server | undefined; + let worker: ReturnType | undefined; + let coordinator: ReturnType | undefined; + const errors: unknown[] = [], httpErrors: unknown[] = []; + let requests = 0, acknowledged = 0, leases = 0, enrollment: Record | undefined; + let origin = "", eventId = randomUUID(), grantId = "", companyId = ""; + const identities: string[] = []; + const delivered = new Set(); + try { + server = createServer(async (req, res) => { + try { + requests++; + const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(Buffer.from(chunk)); + const body = JSON.parse(Buffer.concat(chunks).toString()); + if (req.url === "/v1/connector/enrollments") { + enrollment = body; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ enrollmentId: "enrollment", verificationUrl: `${origin}/connections/enroll?id=enrollment`, expiresAt: new Date(Date.now() + 60_000).toISOString() })); return; + } + expect(enrollment).toBeDefined(); + const [header, payload, signature] = body.request.split("."); + expect(verify(null, Buffer.from(`${header}.${payload}`), createPublicKey({ key: { kty: "OKP", crv: "Ed25519", x: enrollment!.signPublicKey }, format: "jwk" }), Buffer.from(signature, "base64url"))).toBe(true); + const claims = JSON.parse(Buffer.from(payload, "base64url").toString()); + expect(claims.iss).toBe(enrollment!.instanceId); expect(claims.env).toBe("test"); expect(claims.aud).toBe(`${origin}${req.url}`); + res.setHeader("content-type", "application/json"); + if (req.url === "/v1/connector/enrollment-claims") { + res.end(JSON.stringify({ id: enrollment!.instanceId, environment: "test", origins: [enrollment!.origin] })); return; + } + identities.push(claims.iss); + if (req.url === "/v1/connector/events/ack") { acknowledged++; delivered.add(eventId); res.end(JSON.stringify({ acknowledged: 1 })); return; } + expect(req.url).toBe("/v1/connector/events/lease"); leases++; + if (delivered.has(eventId)) { res.end(JSON.stringify({ events: [], eventWakeVersion: 1 })); return; } + const ephemeral = generateKeyPairSync("x25519"), epk = ephemeral.publicKey.export({ format: "jwk" }).x!; + const recipient = createPublicKey({ key: { kty: "OKP", crv: "X25519", x: enrollment!.sealPublicKey }, format: "jwk" }); + const aad = Buffer.from([1, "X25519-HKDF-SHA256-A256GCM", "events", claims.iss, "test", "github", "github.code", ""].join("\n")); + const key = Buffer.from(hkdfSync("sha256", diffieHellman({ privateKey: ephemeral.privateKey, publicKey: recipient }), Buffer.concat([Buffer.from(epk, "base64url"), Buffer.from(enrollment!.sealPublicKey, "base64url")]), aad, 32)); + const iv = randomBytes(12), cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad); + const packet = { v: 1, instanceId: claims.iss, environment: "test", leaseId: "lease", events: [{ id: eventId, provider: "github", event: "installation", action: "unsuspend", installationId: "101", repositoryId: null, createdAt: new Date().toISOString(), bindingIds: [`${grantId}_101`], payload: { installationId: "101" } }] }; + const ct = Buffer.concat([cipher.update(JSON.stringify(packet)), cipher.final(), cipher.getAuthTag()]); + res.end(JSON.stringify({ leaseId: "lease", eventWakeVersion: 1, sealed: { v: 1, alg: "X25519-HKDF-SHA256-A256GCM", purpose: "events", provider: "github", profile: "github.code", epk, iv: iv.toString("base64url"), ct: ct.toString("base64url") } })); + } catch (error) { httpErrors.push(error); res.writeHead(500).end(); } + }); + const listening = once(server, "listening"); server.listen(0, "127.0.0.1"); await listening; + const address = server.address(); if (!address || typeof address === "string") throw new Error("No loopback listener"); + origin = `http://127.0.0.1:${address.port}`; + companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Live enrollment", issuePrefix: "GLE" }); + const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_http", 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", providerTenant: { github: { userId: "42", login: "octocat", installationIds: ["101"], installationCount: 1, repositoryCount: 0, repositorySelection: "all", installationOwnerLogins: [], webhookHealth: "unhealthy" } } }).returning(); + grantId = grant!.id; + const env = { PAPERCLIP_CLOUD_CONNECTOR_BASE_URL: origin, PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT: "test" }; + const service = githubConnectionEventService(db, { env }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: e => errors.push(e) }); + const selects = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + worker = registerGitHubConnectionEventWork(db, coordinator, service, () => true); + await worker.ready; + async function quiet(stage: string) { + await vi.waitFor(() => { expect(idleWorkSnapshot().active, stage).toBe(0); expect(coordinator!.nextWakeAt(), stage).toBeNull(); }); + selects.mockClear(); transactions.mockClear(); executes.mockClear(); const before = requests; + await vi.advanceTimersByTimeAsync(25 * 60_000); + expect(selects).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(requests).toBe(before); + } + await quiet("unenrolled startup"); expect(requests).toBe(0); + async function enroll() { + await startPaperclipCloudConnectorEnrollment({ origin: "https://self-hosted.example.test", env, companyId }); + const pending = loadPaperclipCloudConnectorIdentity()!; + await completePaperclipCloudConnectorEnrollment({ enrollmentId: pending.pending!.enrollmentId, state: pending.pending!.returnState, approvalCode: "approved", env }); + return loadPaperclipCloudConnectorIdentity()!.instanceId; + } + const firstIdentity = await enroll(); + expect(wakeConnectionEvents(db)).toBe(true); + await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(acknowledged).toBe(1)); + expect(await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.providerDeliveryId, eventId))).toMatchObject([{ status: "processed" }]); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grantId)))[0]!.providerTenant!.github!.webhookHealth).toBe("healthy"); + // The applied installation update publishes one credential-change hint. + // Join that existing minimum-spacing pass once; the broker now has no debt. + const followup = coordinator.nextWakeAt(); expect(followup).not.toBeNull(); + expect(followup! - Date.now()).toBeLessThanOrEqual(5_000); + await vi.advanceTimersByTimeAsync(Math.max(0, followup! - Date.now()) + 1); + await vi.waitFor(() => expect(leases).toBe(2)); + await quiet("first event plus its one empty follow-up"); + await rm(paperclipCloudConnectorIdentityPath()); + const oldLeases = leases; expect(wakeConnectionEvents(db)).toBe(true); await vi.advanceTimersByTimeAsync(6_000); + await quiet("removed identity"); expect(leases).toBe(oldLeases); + eventId = randomUUID(); const secondIdentity = await enroll(); expect(secondIdentity).not.toBe(firstIdentity); + expect(wakeConnectionEvents(db)).toBe(true); await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(acknowledged).toBe(2)); + expect(identities).toContain(firstIdentity); expect(identities).toContain(secondIdentity); + expect(identities.slice(identities.indexOf(secondIdentity))).toEqual(expect.arrayContaining([secondIdentity, secondIdentity])); + expect(identities.slice(identities.indexOf(secondIdentity)).every(id => id === secondIdentity)).toBe(true); + const replacementFollowup = coordinator.nextWakeAt(); expect(replacementFollowup).not.toBeNull(); + expect(replacementFollowup! - Date.now()).toBeLessThanOrEqual(5_000); + await vi.advanceTimersByTimeAsync(Math.max(0, replacementFollowup! - Date.now()) + 1); + await vi.waitFor(() => expect(leases).toBe(oldLeases + 2)); + await quiet("replacement event plus its one empty follow-up"); expect(errors).toEqual([]); expect(httpErrors).toEqual([]); + } finally { + await worker?.stop(); await coordinator?.stop(); vi.restoreAllMocks(); vi.useRealTimers(); + if (server) { const closed = once(server, "close"); server.close(); server.closeAllConnections(); await closed; } + if (previous.home === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previous.home; + if (previous.instance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = previous.instance; + await rm(root, { recursive: true, force: true }); + } + }, 30_000); +}); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index a70213dc44b..af936353bdd 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -4,6 +4,7 @@ import { notifyToolConnectionHealth } from "./tool-connection-health-notificatio import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import { connectionEventDeliveries, connectionGrants, @@ -223,8 +224,21 @@ 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); + let configuredConnector: { + config: NonNullable>; + connector: PaperclipCloudConnector; + } | undefined; + function currentConnector() { + if (options.connector) return options.connector; + // Enrollment and identity replacement can happen after this owner starts. + // Read only on an admitted pass; preserve negotiated delivery while unchanged. + const config = paperclipCloudConnectorConfigFromEnv(options.env); + if (!config) { configuredConnector = undefined; return null; } + if (!configuredConnector || !isDeepStrictEqual(configuredConnector.config, config)) { + configuredConnector = { config, connector: createPaperclipCloudConnector({ config }) }; + } + return configuredConnector.connector; + } const forcePolling = (options.env ?? process.env).PAPERCLIP_GITHUB_EVENT_POLLING === "1"; let emptyPolls = 0; let nextSetupPollAt = 0; @@ -484,6 +498,7 @@ export function githubConnectionEventService( } async function runPoll(): Promise { + const connector = currentConnector(); if (!connector) { nextPollAt = Number.POSITIVE_INFINITY; return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 };