diff --git a/doc/idle-incoming-wake.md b/doc/idle-incoming-wake.md new file mode 100644 index 00000000000..05dfac779c8 --- /dev/null +++ b/doc/idle-incoming-wake.md @@ -0,0 +1,118 @@ +# Authenticated incoming-wake proof + +The Cloud control signer can attach `idleCapabilities.incomingWake` to an owned +`task-drain:read` assertion. The tenant verifies the existing stack, action, +audience, expiry and one-use request ID, then validates the closed capability +shape. A tenant query, browser session, API key or forwarded identity header +cannot install this capability. Unknown versions and routes fail closed. + +The report uses outer version 3 and includes the exact version-1 incoming proof. +Cloud compares its requested capability and the initial/final proof before stop. +Without this capability, versions 1/2 and all existing ingress blockers remain. +Older Cloud/app combinations retain their conservative behavior. + +## Supported profiles + +`tenant-api-http-v2` allows saved agent and board API key rows to remain while +sleeping. Cloud resolves the target from its verified host registry, coalesces a +bounded wake, waits for tenant readiness, refreshes routing, then forwards the +original HTTP request once. Ordinary `/api/instance` administration routes are +included. The separately reserved Muse protocol, non-API MCP routes, upgrades, +encoded/noncanonical paths and methods outside GET/HEAD/OPTIONS/POST/PUT/PATCH/ +DELETE do not gain this wake path. + +The Cloud capability authenticates the **host's ability to receive work**. Opaque +tenant keys are authenticated only by the awakened tenant, under current key, +expiry, user, membership, company, agent and resource permissions. No raw key or original +credential hash is copied into Cloud. A signed final-drain projection supplies +only per-cycle derived wake fingerprints. Unknown inputs cannot consume +provider-start quota. Tenant authentication still rejects any later revocation. + +Successful current board/agent key authentication followed by a successful HTTP +response retains ten minutes of local API activity, measured with a monotonic +clock. Further successful use extends the deadline. Rejection, expiry, revoked +membership, failed requests and drain refusals do not extend it. The existing +local-work probe observes this timestamp; there is no timer, SQL, renewal or +network work from the hold, and browser-entry metrics are unchanged. + +`github-broker-events-v1` uses the durable Cloud connector queue from the existing +connector-event wake protocol. The signed proof binds the current enrolled +connector instance and canonical fingerprint of all current GitHub bindings. +The tenant reconstructs those bindings through the original connection-event +owner and exact completed GitHub bot registration ownership. An active broker +bot can retain its endpoint and paired channel connection only when setup has +finished and every binding matches. Missing, foreign, changed, unsupported or +unsettled binding state prevents sleep. Original GitHub signature verification +and current endpoint/connection authorization still run on delivery. + +Personal GitHub MCP connections additionally owe credential continuity work. +This change exposes their required connection IDs but keeps them blocked. The +stacked domain-deadline change must supply its inspected future continuity proof +before those rows can be exempted. A signed catalog name alone is insufficient. +Legacy scheduled alarms remain conservative when combined with ingress until +that downstream composition is present. + +## Delivery and admission guarantees + +Synchronous HTTP is not durable acceptance. A request arriving during stopping +or cold start is held only before forwarding, under finite capacity/time bounds. +An active-route race with tenant drain returns its real 503 and Retry-After; +clients retry through their normal API/domain idempotency contract. A streamed +mutation is never automatically replayed after an ambiguous upstream outcome. +Wake/readiness failure, manual suspension, deployment recovery holds, wrong +hosts and exhausted admission return non-success without acknowledging work. + +Broker events have stronger persistence: the existing queue target survives a +lost wake hint, failed wake, process restart and lease expiry. A 202 hint does not +settle delivery. Only the current event lease acknowledgment does. Duplicate +transport delivery uses the domain's existing receipts/atomic claims. + +Cloud API admission is explicitly bounded per process: 64 waiting requests, +8 per stack, 4 per source, 32 pending provider preparations; source request +budget 24/minute, source wake budget 3/10 minutes, stack wake budget 8/hour, +and aggregate wake budget 32/minute. Readiness waiting is at most 90 seconds. +A timed-out provider preparation remains coalesced until it actually settles. +The stack budget accommodates ordinary ten-minute quiet cycles with boundary +headroom. These are availability limits, not an unrestricted API promise. +Only the existing explicitly configured trusted-proxy hop contract may resolve +a forwarded source IP; otherwise the socket address is used. A shared proxy/NAT +therefore shares its source limits and can receive truthful retry refusals. + +## Explicit remaining coverage + +Direct GitHub, routine, Slack, voice and X webhooks retain their existing +configured-ingress blockers until their exact response budget, retry/retention, +authenticated wake and replay contracts are qualified. Cloud X forwarding by +itself does not establish wake. Saved outbound HTTP MCP remains passive under +its existing narrow rule; public MCP/SSE subscriptions, Dot, Muse and gateway +resources retain their own blockers. Discord Gateway, local MCP stdio, and other +required persistent sockets need owner-proved stop, reconnect and sequence replay. +Telegram/Teams/Photon transports require their own actual producer/lifecycle +classification. Hosted previews belong to their separate ingress changes. + +The source/loopback/PostgreSQL tests establish the exercised protocol boundaries; +they do not establish a fleet deployment, actual provider/Neon suspension, +provider startup latency or cost savings. Deployed combined qualification remains +separate and must include every enabled transport. + + +## Pre-wake API admission + +The v2 API profile requires an authenticated per-sleep-cycle key projection before compute stops. The signed challenge is derived from the newly random owned idle hold and stack. In the same read-only snapshot, the app projects up to 1000 active key rows into domain-separated salted fingerprints; original key hashes and raw keys stay private. Board-key expiry is carried and enforced before wake. Unknown, duplicate, malformed or overflow projection fails closed. The complete initial and final receipts must match; changing a credential or challenge refuses stop. + +Cloud stores the final receipt in one internal stack column under the sleep operation lock. Unknown requests cannot consume provider-start quota. Possession of a matching snapshot key only admits a bounded wake; tenant authentication and current permissions still run on delivery. A key revoked after the snapshot can therefore receive no tenant authority, even if it once qualified for wake. Wake claims recheck the exact cycle, matching key and expiry atomically. Normal next sleep re-reads revocations. Successful wake, manual/archive/delete or a known sleep abort clears the receipt; uncertain stop and a same-cycle failed wake retain recovery. A restarted host reads the durable receipt. No global TTL strands nonexpiring saved keys during long sleep, and no periodic credential refresh worker is added. + + +Removed GitHub endpoints are retained history only after the ordinary removal +writer archives the endpoint with its timestamp and generation, archives and +disables the same-company channel connection, clears its credential references, +and releases endpoint leases. Any retained lease, incomplete registration or +malformed cleanup state still blocks. This exemption covers only GitHub; archived +persistent transports remain blocked. Actions, deliveries and publications keep +their separate completion checks even when their endpoint has been removed. + +The service also keeps its existing local-work probe active while runtime +initialization or retirement is unsettled. Both supported runtime registries +report failed retirement even after `get` and `list` hide the runtime. Successful +owner retry clears that debt; a failed full service shutdown retains the probe. +These checks read process-local state and add no polling, network or SQL. diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index a07fc4d9f6f..9b71b4cf034 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -644,3 +644,13 @@ retry dates and package/object/native-asset retention timestamps are earliest maintenance eligibility, not promised delivery deadlines. Workers reconstruct persisted rows and filesystem access leases on startup, and use the existing coordinator's drain/standby/shutdown controls. No external Cloud alarm or provider suspension is established by these tests. + + +## Negotiated incoming work + +A signed, stack/owner-scoped Cloud incoming-wake capability can produce a version-3 +safety report for saved API keys and exactly bound brokered GitHub bot endpoints. +Tenant authorization still runs after wake. Personal GitHub credential continuity +and unsupported transports retain their blockers. See [authenticated incoming +wake proof](idle-incoming-wake.md) for scope, HTTP versus durable-event guarantees, +rate/availability bounds, compatibility and remaining transport dependencies. diff --git a/maps/agent-access-map/instance-operations.md b/maps/agent-access-map/instance-operations.md index db0c87834f0..cebe15a8df6 100644 --- a/maps/agent-access-map/instance-operations.md +++ b/maps/agent-access-map/instance-operations.md @@ -168,6 +168,14 @@ API: `present` — representative operation IDs: status remains an in-memory read. [Route coverage](../../server/src/__tests__/instance-settings-routes.test.ts) and [PostgreSQL coverage](../../server/src/services/heartbeat/upgrade-readiness.test.ts) establish this local boundary; hosted active-run continuity requires separate proof. +- `GET /api/instance/task-drain?idleSleepSafety=1` — the negotiated incoming-wake + extension requires a verified stack/action/owner-scoped Cloud control capability; + query parameters and tenant credentials cannot grant it. Its version-3 proof + permits only saved API keys and exactly bound brokered GitHub bot configuration. + Agent and board API requests authenticate under current tenant authority after + bounded host wake. Personal GitHub continuity remains blocked until the stacked + deadline proof is present. See [protocol scope](../../doc/idle-incoming-wake.md) + and [local auth/PostgreSQL evidence](../../server/src/__tests__/idle-incoming-wake.test.ts). - `GET /api/ready` and `HEAD /api/ready` — public deployment readiness only; 200 after recovery, 503 while starting, recovering, or draining. No SQL, session resolution, company identity, or control authority. diff --git a/maps/feature-map/instance-operations.md b/maps/feature-map/instance-operations.md index 29387666067..6ff9a9b0633 100644 --- a/maps/feature-map/instance-operations.md +++ b/maps/feature-map/instance-operations.md @@ -154,3 +154,15 @@ Scope: cli Storybook: No authored story required for doctor/health command results. Production: Run doctor against an owned data directory; check readiness and nonzero failures. A listening port alone does not prove readiness; repair requires observing the repaired result. Sources: Source: [doctor.ts](../../cli/src/commands/doctor.ts). Tests: [doctor.test.ts](../../cli/src/__tests__/doctor.test.ts) + + +Incoming API wake is an optional Cloud deployment capability. On an owned qualified +stack, close the browser, retain valid saved credentials, let the stack sleep, +and send an ordinary API request. Check the persisted result and current +revocation/company authorization after wake, truthful retry failures, and return +to idle after successful API activity ends. Brokered GitHub bot wake uses the +existing durable connector queue and exact registration ownership. Direct +webhooks, persistent transports and personal GitHub credential continuity retain +their explicit dependencies. [Source protocol](../../doc/idle-incoming-wake.md) +and [local fixtures](../../server/src/__tests__/idle-incoming-wake.test.ts) cover +these boundaries; they do not establish deployed host/database suspension. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 89f7c9741d8..5364202d735 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -11,6 +11,7 @@ import { retainGitHubReviewReceipt } from "../services/chat-channels/github/task import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js"; import { recordRunPresentationDecision } from "../services/heartbeat/run-completion.js"; import { environmentService } from "../services/environments.js"; +import { reconcilePersistedRuntimeServicesOnStartup, readWorkspaceRuntimeIdleWork } from "../services/workspace-runtime.js"; import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js"; import { issueTreeControlService } from "../services/issue-tree-control.js"; @@ -33,6 +34,10 @@ import * as githubTaskPreparation from "../services/chat-channels/github/task-pr import { budgetPolicies } from "@paperclipai/db"; import { budgetService } from "../services/budgets.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; +import { readIdleLocalWork } from "../services/idle-local-work.js"; +import { incomingWakeWorkQuery } from "../services/idle-incoming-wake.js"; +import { createChatRuntime } from "../services/chat-runtime.js"; +import { createChatSdkRuntime } from "../services/chat-sdk-runtime.js"; import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js"; import { requestTaskRetry, reconcileTaskChatRetryReceipts } from "../services/task-retry.js"; import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js"; @@ -809,6 +814,8 @@ class FakeChatSdkRuntime { return this.endpoints.get(endpointId) ?? null; } + hasUnsettledLifecycle() { return false; } + async replaceEndpoint(options: CreateChatSdkEndpointRuntimeOptions) { this.replaceCount += 1; this.configurations.set(options.endpointId, options); @@ -1128,8 +1135,20 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { // Do not expire or time out its connections on those virtual clock jumps. db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, { idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0, - ...(historyTestDatabaseUrl ? { maxConnections: 4 } : {}), + ...(externalTestDatabaseUrl ? { maxConnections: 4 } : {}), }); + // The composed local-probe inventory includes the workspace owner. Match + // server startup on this fresh fixture; never adopt or start a leftover. + const [runtimeState] = await db.execute<{ configured: boolean }>(sql`SELECT + EXISTS (SELECT 1 FROM workspace_runtime_services) + OR EXISTS (SELECT 1 FROM project_workspaces) + OR EXISTS (SELECT 1 FROM execution_workspaces) AS configured`); + expect(runtimeState.configured).toBe(false); + expect(await reconcilePersistedRuntimeServicesOnStartup(db)).toEqual({ + reconciled: 0, adopted: 0, stopped: 0, backfilled: 0, restarted: 0, + restartFailed: 0, exposureReservationDrift: [], + }); + expect(readWorkspaceRuntimeIdleWork()).toBe("none"); }, 90_000); afterAll(async () => { @@ -4185,10 +4204,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .toEqual({ next: null, active: 0 }); expect(f.responseWrites).toHaveLength(baseline); vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); - const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + const readStacks: string[] = [], select = db.select; + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + if (readStacks.length < 5) readStacks.push(new Error("Unexpected idle database read").stack ?? "Stack unavailable"); + return Reflect.apply(select, db, args); + }), writes = vi.spyOn(db, "transaction"); try { await vi.advanceTimersByTimeAsync(120_000); - expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + expect(reads, readStacks.join("\n\n")).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(f.responseWrites).toHaveLength(baseline); } finally { reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); } const settlePresentation = async (tx: Parameters[0]>[0]) => { @@ -29957,13 +29980,23 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }), ); await initializationEntered; + expect(runtime.hasUnsettledLifecycle()).toBe(false); // Fake registry tail is already settled. + expect(await registeredIdleProbeState()).toBe("present"); await service.configure(endpoint.id, { action: "pause" }, "owner-user"); + let shutdownSettled = false; + const shutdown = service.shutdown().then(() => { shutdownSettled = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(shutdownSettled).toBe(false); + expect(await registeredIdleProbeState()).toBe("present"); releaseInitialization(); await expect(initializingWebhook).rejects.toMatchObject({ status: 409, details: { code: "chat_endpoint_runtime_superseded" }, }); + await shutdown; + expect(shutdownSettled).toBe(true); + expect(await registeredIdleProbeState()).toBe("none"); expect(runtime.endpoints.has(endpoint.id)).toBe(false); await expect(service.get(endpoint.id)).resolves.toMatchObject({ status: "paused", @@ -49515,7 +49548,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .then((rows) => rows.length); expect(deliveryCount).toBeGreaterThan(0); + const endpointBlocksSleep = async () => (await db.execute(sql` + SELECT EXISTS (${incomingWakeWorkQuery("chat_endpoints")} AND endpoint.id = ${endpoint.id}::uuid) AS blocked`))[0].blocked; + if (provider === "github") expect(await endpointBlocksSleep()).toBe(true); await service.configure(endpoint.id, { action: "remove" }, "owner-user"); + if (provider === "github") expect(await endpointBlocksSleep()).toBe(false); + else expect(await endpointBlocksSleep()).toBe(true); // No persistent/provider coverage expansion. await expect(service.get(endpoint.id)).resolves.toMatchObject({ status: "archived", @@ -49557,6 +49595,44 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }, ); + async function registeredIdleProbeState() { + let state: string | undefined; + await readIdleLocalWork({ diagnostic: (reason, value) => { if (reason === "local.registered_probes") state = value; } }); + return state; + } + + it.each([ + { name: "SDK", create: createChatSdkRuntime }, + { name: "shared", create: createChatRuntime }, + ])("retains the service idle probe after failed $name retirement and shutdown until owner retry", async ({ create }) => { + const fixture = await seedCompany(); + const f = await configuredGitHubEndpoint(fixture); + const registry = create(); + const first = await registry.replaceEndpoint(f.runtime.configurations.get(f.endpoint.id)!); + const shutdown = vi.spyOn(first, "shutdown") + .mockRejectedValueOnce(new Error("remove retirement uncertain")) + .mockRejectedValueOnce(new Error("shutdown retirement uncertain")) + .mockResolvedValue(undefined); + const service = chatChannelService(db, { runtime: registry, fetch: f.providerFetch, publicBaseUrl: "https://paperclip.example" }); + fixtureServices.add(service); + expect(await registeredIdleProbeState()).toBe("none"); + await service.configure(f.endpoint.id, { action: "remove" }, "owner-user"); + expect(registry.get(f.endpoint.id)).toBeNull(); + expect(registry.list()).toEqual([]); + expect(registry.hasUnsettledLifecycle()).toBe(true); + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + expect(await registeredIdleProbeState()).toBe("present"); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + await expect(service.shutdown()).rejects.toThrow("shutdown retirement uncertain"); + expect(await registeredIdleProbeState()).toBe("present"); + await service.shutdown(); + expect(shutdown).toHaveBeenCalledTimes(3); + expect(registry.hasUnsettledLifecycle()).toBe(false); + expect(await registeredIdleProbeState()).toBe("none"); + }); + it("retries credential cleanup after archival made a remove partially complete", async () => { const fixture = await seedCompany(); const { endpoint, service } = await configuredSlackEndpoint(fixture); @@ -60587,7 +60663,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const commentsBefore = await db .select({ id: issueComments.id }) .from(issueComments) - .where(eq(issueComments.issueId, conversation.issueId)); + .where(eq(issueComments.issueId, conversation.issueId)) + .orderBy(asc(issueComments.id)); const wakesBefore = wakeup.mock.calls.length; const reactionsBefore = providerRuntime.reactions.length; const postsBefore = providerRuntime.posts.length; @@ -60617,7 +60694,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await db .select({ id: issueComments.id }) .from(issueComments) - .where(eq(issueComments.issueId, conversation.issueId)), + .where(eq(issueComments.issueId, conversation.issueId)) + .orderBy(asc(issueComments.id)), ).toEqual(commentsBefore); expect(wakeup).toHaveBeenCalledTimes(wakesBefore); expect(cancelRun).not.toHaveBeenCalled(); diff --git a/server/src/__tests__/chat-failed-run-retry-boundary.test.ts b/server/src/__tests__/chat-failed-run-retry-boundary.test.ts index ad62169924d..67cbd7f863a 100644 --- a/server/src/__tests__/chat-failed-run-retry-boundary.test.ts +++ b/server/src/__tests__/chat-failed-run-retry-boundary.test.ts @@ -14,7 +14,7 @@ function createService(db: Db) { const runtimeShutdown = vi.fn().mockResolvedValue(undefined); const service = chatChannelService(db, { heartbeat: { wakeup }, - runtime: { shutdown: runtimeShutdown } as unknown as ChatChannelServiceOptions["runtime"], + runtime: { shutdown: runtimeShutdown, hasUnsettledLifecycle: () => false } as unknown as ChatChannelServiceOptions["runtime"], }); services.add(service); return { service, wakeup, runtimeShutdown }; diff --git a/server/src/__tests__/idle-incoming-wake.test.ts b/server/src/__tests__/idle-incoming-wake.test.ts new file mode 100644 index 00000000000..18628e4c35b --- /dev/null +++ b/server/src/__tests__/idle-incoming-wake.test.ts @@ -0,0 +1,700 @@ +import { matchesApiWakeAdmission } from "../services/idle-api-wake-admission.js"; +import { generateKeyPairSync, randomUUID, sign } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { eq, sql } from "drizzle-orm"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; +import { + agents, + agentApiKeys, + authUsers, + boardApiKeys, + companies, + companyMemberships, + createDb, + toolApplications, + toolConnections, + connectionGrants, + chatEndpoints, + chatGitHubRegistrations, + chatSlackRegistrations, + chatEndpointLeases, + chatActions, + chatDeliveries, + chatPublications, + chatConversations, + issues, + type Db, +} from "@paperclipai/db"; +import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js"; +import { actorMiddleware } from "../middleware/auth.js"; +import { + idleAdmissionMiddleware, + trackIdleRequestHandlers, +} from "../middleware/idle-admission.js"; +import { errorHandler } from "../middleware/error-handler.js"; +import { companyRoutes } from "../routes/companies.js"; +import { agentService } from "../services/agents.js"; +import { boardAuthService } from "../services/board-auth.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; +import { + readIdleIncomingWakeProof, + incomingWakeFingerprint, + incomingWakeWorkQuery, +} from "../services/idle-incoming-wake.js"; +import { registerConnectionEventWake } from "../services/connection-event-wake.js"; +import { idleApiActivity } from "../services/idle-api-activity.js"; +import { startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; +import { + verifyCloudControlAssertion, + CLOUD_CONTROL_AUDIENCE, + CLOUD_CONTROL_JWS_TYPE, + CLOUD_RUNTIME_IDENTITY_ISSUER, +} from "../services/cloud-runtime-identity.js"; +import type { + IdleIncomingWakeCapability, + VerifiedIdleCapabilities, +} from "../services/idle-incoming-wake-capability.js"; + +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +const url = resolveExternalTestDatabaseUrl( + "PAPERCLIP_INCOMING_WAKE_TEST_DATABASE_URL", +); +const support = url + ? { supported: true } + : await getEmbeddedPostgresTestSupport(); +const ownerId = "00000000-0000-4000-8000-000000000001"; +const now = Date.now(); +const hold = () => ({ + ownerId, + draining: true, + startedAt: new Date(now - 1000), + expiresAt: new Date(now + 300_000), + activeRuns: 0, + pendingWakes: 0, +}); +const capability: IdleIncomingWakeCapability = { + version: 1, + ownerId, + generation: "a".repeat(64), + apiKeyProof: { version: 1 as const, challenge: "c".repeat(64) }, + routes: ["tenant-api-http-v2"], +}; +function verified(incomingWake = capability): VerifiedIdleCapabilities { + const pair = generateKeyPairSync("ed25519"), + kid = randomUUID(), + issued = Math.floor(Date.now() / 1000); + const encode = (v: unknown) => + Buffer.from(JSON.stringify(v)).toString("base64url"); + const header = encode({ alg: "EdDSA", typ: CLOUD_CONTROL_JWS_TYPE, kid }); + const payload = encode({ + v: 1, + iss: CLOUD_RUNTIME_IDENTITY_ISSUER, + aud: CLOUD_CONTROL_AUDIENCE, + sub: "fixture-stack", + action: "task-drain:read", + requestId: randomUUID(), + iat: issued, + exp: issued + 60, + idleCapabilities: { incomingWake }, + }); + const token = `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`), pair.privateKey).toString("base64url")}`; + return verifyCloudControlAssertion({ + compactJws: token, + expectedAction: "task-drain:read", + env: { + PAPERCLIP_CLOUD_STACK_ID: "fixture-stack", + PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: JSON.stringify({ + keys: [ + { + ...pair.publicKey.export({ format: "jwk" }), + kid, + use: "sig", + alg: "EdDSA", + }, + ], + }), + }, + }).idleCapabilities as VerifiedIdleCapabilities; +} + +(support.supported ? describe : describe.skip)( + "scoped incoming wake PostgreSQL contract", + () => { + let db: Db; + let database: + Awaited> | undefined; + beforeAll(async () => { + if (!url) + database = await startEmbeddedPostgresTestDatabase( + "idle-incoming-wake-", + ); + db = createDb(url ?? database!.connectionString, { maxConnections: 4 }); + await db.execute(sql`SELECT 1 FROM agent_api_keys LIMIT 1`); // Coordinator migrates the fresh fixture; no embedded fallback. + vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "fixture-only"); + vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "fixture-stack"); + setCloudRuntimeConfigSnapshot({ version: 1, revision: 0, config: null }); + }); + afterAll(async () => { + setCloudRuntimeConfigSnapshot(null); + stopTaskDrain(); + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + await db?.$client.end({ timeout: 5 }); + await database?.cleanup(); + }); + // The externally supplied database is dedicated to this suite. Preserve all + // rows during every eligibility scenario, then isolate the next scenario. + const resetFixture = async () => { + await db.execute( + sql`TRUNCATE companies, "user", chat_github_registrations CASCADE`, + ); + }; + beforeEach(resetFixture); + afterEach(async () => { + stopTaskDrain(); + vi.restoreAllMocks(); + await resetFixture(); + }); + + const read = (incomingWake?: VerifiedIdleCapabilities) => + readIdleSleepSafety( + db, + hold, + () => now, + ownerId, + async () => "none", + undefined, + undefined, + false, + false, + { incomingWake }, + ); + async function seed() { + const companyId = randomUUID(), + agentId = randomUUID(), + userId = randomUUID(), + foreignId = randomUUID(); + await db.insert(companies).values([ + { + id: companyId, + name: "Incoming fixture", + issuePrefix: `I${companyId.slice(0, 6)}`, + }, + { + id: foreignId, + name: "Foreign fixture", + issuePrefix: `F${foreignId.slice(0, 6)}`, + }, + ]); + await db + .insert(authUsers) + .values({ + id: userId, + name: "Fixture owner", + email: `${userId}@example.test`, + createdAt: new Date(), + updatedAt: new Date(), + }); + await db + .insert(companyMemberships) + .values({ + companyId, + principalType: "user", + principalId: userId, + membershipRole: "owner", + status: "active", + }); + await db + .insert(agents) + .values({ + id: agentId, + companyId, + name: "Fixture agent", + role: "engineer", + adapterType: "process", + status: "idle", + }); + const board = await boardAuthService(db).createNamedBoardApiKey({ + userId, + name: "incoming board", + expiresAt: null, + }); + const agent = await agentService(db).createApiKey( + agentId, + "incoming agent", + { kind: "standard" }, + { responsibleUserId: userId }, + ); + return { companyId, foreignId, userId, agentId, board, agent }; + } + it("retains real saved keys and exempts them only for the verified exact owner/profile", async () => { + const f = await seed(); + expect((await read()).backgroundWork).toBe("present"); + expect(await read(verified())).toMatchObject({ + version: 3, + backgroundWork: "none", + incomingWake: capability, + }); + const receipt = (await read(verified())).apiWakeAdmission!; + expect(receipt.keys).toHaveLength(2); + expect(matchesApiWakeAdmission(receipt, { stackId: "fixture-stack", token: f.board.token, now })).toBe(true); + expect(matchesApiWakeAdmission(receipt, { stackId: "fixture-stack", token: f.agent.token, now })).toBe(true); + const serialized = JSON.stringify(receipt); + expect(serialized).not.toContain(f.board.token); + expect(serialized).not.toContain(f.agent.token); + await db.update(boardApiKeys).set({ revokedAt: new Date() }).where(eq(boardApiKeys.id, f.board.id)); + const changedReceipt = (await read(verified())).apiWakeAdmission!; + expect(changedReceipt.generation).not.toBe(receipt.generation); + expect(matchesApiWakeAdmission(changedReceipt, { stackId: "fixture-stack", token: f.board.token, now })).toBe(false); + expect( + (await read(verified({ ...capability, ownerId: randomUUID() }))) + .backgroundWork, + ).toBe("unknown"); + expect( + await db + .select() + .from(agentApiKeys) + .where(eq(agentApiKeys.id, f.agent.id)), + ).toHaveLength(1); + expect( + await db + .select() + .from(boardApiKeys) + .where(eq(boardApiKeys.id, f.board.id)), + ).toHaveLength(1); + }); + it("delivers through actual tenant auth/routes, denies current revoked/expired/foreign authority and holds only successful API use", async () => { + const f = await seed(); + let monotonic = 1; + vi.spyOn(performance, "now").mockImplementation(() => monotonic); + let release: (() => void) | undefined, arrived: (() => void) | undefined; + const app = express(); + // Named Cloud wake boundary: preserve request until the simulated compute handoff finishes. + app.use(async (_req, _res, next) => { + if (release) { + await new Promise((resolve) => { + release = resolve; + arrived?.(); + }); + } + next(); + }); + app.use(idleAdmissionMiddleware); + app.use(express.json()); + app.use( + actorMiddleware(db, { + deploymentMode: "authenticated", + resolveSession: async () => null, + }), + ); + app.use("/api/companies", companyRoutes(db)); + app.use(errorHandler); + trackIdleRequestHandlers(app); + expect(idleApiActivity.active()).toBe(false); + expect( + ( + await request(app) + .get(`/api/companies/${f.companyId}`) + .set("Authorization", "Bearer unknown") + ).status, + ).toBe(401); + expect(idleApiActivity.active()).toBe(false); + expect( + ( + await request(app) + .get(`/api/companies/${f.foreignId}`) + .set("Authorization", `Bearer ${f.board.token}`) + ).status, + ).toBe(403); + expect(idleApiActivity.active()).toBe(false); + const own = await request(app) + .patch(`/api/companies/${f.companyId}`) + .set("Authorization", `Bearer ${f.board.token}`) + .send({ name: "Persisted after wake" }); + expect(own.status).toBe(200); + expect(idleApiActivity.active()).toBe(true); + expect( + ( + await db.select().from(companies).where(eq(companies.id, f.companyId)) + )[0]!.name, + ).toBe("Persisted after wake"); + monotonic += 300_000; + expect( + ( + await request(app) + .get(`/api/companies/${f.companyId}`) + .set("Authorization", `Bearer ${f.agent.token}`) + ).status, + ).toBe(200); + monotonic += 300_001; + expect(idleApiActivity.active()).toBe(true); + monotonic += 299_999; + expect(idleApiActivity.active()).toBe(false); + const arrival = new Promise((resolve) => { + arrived = resolve; + }); + release = () => {}; + const held = request(app) + .patch(`/api/companies/${f.companyId}`) + .set("Authorization", `Bearer ${f.board.token}`) + .send({ name: "Must not commit" }) + .then((r) => r); + await arrival; + await db + .update(boardApiKeys) + .set({ revokedAt: new Date() }) + .where(eq(boardApiKeys.id, f.board.id)); + release!(); + release = undefined; + expect((await held).status).toBe(401); + expect(idleApiActivity.active()).toBe(false); + const expired = await boardAuthService(db).createNamedBoardApiKey({ + userId: f.userId, + name: "expired", + expiresAt: new Date(0), + }); + expect( + ( + await request(app) + .get(`/api/companies/${f.companyId}`) + .set("Authorization", `Bearer ${expired.token}`) + ).status, + ).toBe(401); + expect(idleApiActivity.active()).toBe(false); + startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + expect( + ( + await request(app) + .patch(`/api/companies/${f.companyId}`) + .set("Authorization", `Bearer ${f.agent.token}`) + .send({ name: "Draining" }) + ).status, + ).toBe(503); + expect(idleApiActivity.active()).toBe(false); + stopTaskDrain(); + expect( + ( + await db.select().from(companies).where(eq(companies.id, f.companyId)) + )[0]!.name, + ).toBe("Persisted after wake"); + }); + // Fault injection below supplies only malformed/incomplete persisted states; + // the positive archived transition is covered by the real configure(remove) + // producer in chat-channels.integration.test.ts, without deleting history. + async function removedGitHubFixture() { + const f = await seed(), applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID(); + await db.insert(toolApplications).values({ id: applicationId, companyId: f.companyId, name: "Removed GitHub", type: "chat" }); + await db.insert(toolConnections).values({ id: connectionId, companyId: f.companyId, applicationId, + name: "Removed GitHub", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel", + status: "archived", enabled: false, credentialSecretRefs: [] }); + await db.insert(chatEndpoints).values({ id: endpointId, companyId: f.companyId, connectionId, + assignedAgentId: f.agentId, provider: "github", publicId: randomUUID(), status: "archived", + archivedAt: new Date(), setup: { step: "complete", runtimeGeneration: 1 } }); + const blocked = async () => (await db.execute(sql`SELECT EXISTS ( + ${incomingWakeWorkQuery("chat_endpoints")} AND endpoint.id = ${endpointId}::uuid) AS blocked`))[0].blocked; + return { ...f, applicationId, connectionId, endpointId, blocked }; + } + it.each(["stamp", "active", "paused", "revoked", "persistent_provider", "enabled_connection", "active_connection", + "non_sdk_connection", "credentials", "malformed_credentials", "malformed_setup", "missing_generation", "fractional_generation", + "lease", "expired_lease", "pending_registration", "unconsumed_registration", "wrong_registry"])( + "keeps incomplete removed GitHub proof blocked: %s", async fault => { + const f = await removedGitHubFixture(); + if (fault === "stamp") await db.update(chatEndpoints).set({ archivedAt: null }).where(eq(chatEndpoints.id, f.endpointId)); + if (["active", "paused", "revoked"].includes(fault)) await db.update(chatEndpoints) + .set({ status: fault as "active" | "paused" | "revoked" }).where(eq(chatEndpoints.id, f.endpointId)); + if (fault === "persistent_provider") await db.update(chatEndpoints).set({ provider: "discord" }).where(eq(chatEndpoints.id, f.endpointId)); + if (fault === "enabled_connection") await db.update(toolConnections).set({ enabled: true }).where(eq(toolConnections.id, f.connectionId)); + if (fault === "active_connection") await db.update(toolConnections).set({ status: "active" }).where(eq(toolConnections.id, f.connectionId)); + if (fault === "non_sdk_connection") await db.update(toolConnections).set({ transport: "voice" }).where(eq(toolConnections.id, f.connectionId)); + if (fault === "credentials" || fault === "malformed_credentials") await db.execute(sql`UPDATE tool_connections + SET credential_secret_refs = ${fault === "credentials" ? '[{"secretId":"retained"}]' : '{}'}::jsonb WHERE id = ${f.connectionId}::uuid`); + if (fault === "malformed_setup") await db.execute(sql`UPDATE chat_endpoints SET setup = '"unknown"'::jsonb WHERE id = ${f.endpointId}::uuid`); + if (fault === "missing_generation" || fault === "fractional_generation") await db.update(chatEndpoints) + .set({ setup: fault === "missing_generation" ? { step: "complete" } : { step: "complete", runtimeGeneration: 1.5 } }) + .where(eq(chatEndpoints.id, f.endpointId)); + if (fault === "lease" || fault === "expired_lease") await db.insert(chatEndpointLeases).values({ companyId: f.companyId, + endpointId: f.endpointId, leaseKey: "credentials", token: randomUUID(), expiresAt: new Date(fault === "lease" ? Date.now() + 60_000 : 0) }); + if (fault === "pending_registration" || fault === "unconsumed_registration") await db.insert(chatGitHubRegistrations).values({ + companyId: f.companyId, endpointId: f.endpointId, userId: f.userId, stateHash: randomUUID(), trustedOrigin: "https://tenant.example.test", + status: fault === "pending_registration" ? "pending" : "completed", expiresAt: new Date(Date.now() + 60_000) }); + if (fault === "wrong_registry") await db.insert(chatSlackRegistrations).values({ endpointId: f.endpointId, companyId: f.companyId, + requestId: randomUUID(), status: "uncertain", manifest: {}, manifestHash: "a".repeat(64), callbackUri: "https://tenant.example.test", + createdByUserId: f.userId, secretIds: { clientSecret: randomUUID() } }); + expect(await f.blocked()).toBe(true); + expect(await db.select({ id: chatEndpoints.id }).from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId))).toHaveLength(1); + }); + it("retains tenant boundaries and rejects unfamiliar archived state at the schema boundary", async () => { + const f = await removedGitHubFixture(); + await expect(db.execute(sql`UPDATE chat_endpoints SET connection_id = ${randomUUID()}::uuid WHERE id = ${f.endpointId}::uuid`)).rejects.toThrow(); + const foreignApplication = randomUUID(), foreignConnection = randomUUID(); + await db.insert(toolApplications).values({ id: foreignApplication, companyId: f.foreignId, name: "Foreign", type: "chat" }); + await db.insert(toolConnections).values({ id: foreignConnection, companyId: f.foreignId, applicationId: foreignApplication, + name: "Foreign", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel", status: "archived", enabled: false }); + await expect(db.execute(sql`UPDATE chat_endpoints SET connection_id = ${foreignConnection}::uuid WHERE id = ${f.endpointId}::uuid`)).rejects.toThrow(); + await expect(db.execute(sql`UPDATE chat_endpoints SET status = 'unknown' WHERE id = ${f.endpointId}::uuid`)).rejects.toThrow(); + expect((await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId)))[0].connectionId).toBe(f.connectionId); + }); + it("does not let removed GitHub endpoint history discharge pending delivery, publication or action obligations", async () => { + const f = await removedGitHubFixture(); + await db.insert(chatGitHubRegistrations).values({ companyId: f.companyId, endpointId: f.endpointId, userId: f.userId, + stateHash: randomUUID(), trustedOrigin: "https://tenant.example.test", status: "completed", consumedAt: new Date(), expiresAt: new Date(0) }); + expect(await f.blocked()).toBe(false); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Retained chat debt", status: "done" }).returning(); + const [conversation] = await db.insert(chatConversations).values({ companyId: f.companyId, endpointId: f.endpointId, + issueId: issue.id, externalConversationId: "github:retained", externalLabel: "retained", state: "endpoint_removed" }).returning(); + await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: conversation.id, + providerEventId: "retained", deduplicationKey: "retained", eventKind: "message", normalizedEvent: {}, state: "retry", nextAttemptAt: new Date() }); + await db.insert(chatPublications).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: conversation.id, + issueId: issue.id, idempotencyKey: "retained", payload: { text: "Retained" }, state: "retry", nextAttemptAt: new Date() }); + await db.insert(chatActions).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: conversation.id, + kind: "provider_effect", providerActionId: "retained", payload: {}, status: "received" }); + const report = await readIdleSleepSafety(db, hold, () => now, ownerId, async () => "none", undefined, undefined, + false, false, { incomingWake: verified(), collectDiagnostics: true }); + expect(report.backgroundWork).toBe("present"); + expect(report.diagnostics?.checks).toEqual(expect.arrayContaining([ + { reason: "chat_endpoints", state: "none" }, { reason: "chat_deliveries", state: "present" }, + { reason: "chat_publications", state: "present" }, { reason: "chat_actions", state: "present" }, + ])); + expect(await db.select().from(chatDeliveries)).toHaveLength(1); + expect(await db.select().from(chatPublications)).toHaveLength(1); + expect(await db.select().from(chatActions)).toHaveLength(1); + }); + + it("requires current broker instance and exact owner binding, preserving continuity and unsupported transport blockers", async () => { + const f = await seed(); + const applicationId = randomUUID(), + connectionId = randomUUID(), + grantId = randomUUID(); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID", "fixture-instance"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY", "fixture"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY", "fixture"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT", "test"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_BASE_URL", "http://127.0.0.1"); + const botConnectionId = randomUUID(), + endpointId = randomUUID(), + registration = "b".repeat(43); + const botApplication = randomUUID(); + await db + .insert(toolApplications) + .values({ + id: botApplication, + companyId: f.companyId, + name: "GitHub bot", + type: "chat", + }); + await db + .insert(toolConnections) + .values({ + id: botConnectionId, + companyId: f.companyId, + applicationId: botApplication, + name: "GitHub bot", + uid: randomUUID(), + transport: "chat_sdk", + connectionPurpose: "channel", + status: "active", + enabled: true, + config: { provider: "github" }, + }); + await db + .insert(chatEndpoints) + .values({ + id: endpointId, + companyId: f.companyId, + connectionId: botConnectionId, + assignedAgentId: f.agentId, + provider: "github", + publicId: "p".repeat(43), + status: "active", + activatedAt: new Date(), + providerAccountId: "fixture-app", + botExternalId: "fixture-bot", + setup: { + step: "complete", + github: { + stage: "test", + registrationStatus: "completed", + cloudRegistrationId: registration, + }, + }, + }); + await db + .insert(chatGitHubRegistrations) + .values({ + companyId: f.companyId, + endpointId, + userId: f.userId, + stateHash: randomUUID(), + trustedOrigin: "https://tenant.example.test", + status: "completed", + expiresAt: new Date(0), + handoff: { + cloudId: registration, + redemptionId: randomUUID(), + returnState: randomUUID(), + }, + }); + const botTuple = [ + `github-app:${registration}`, + f.companyId, + f.userId, + "github.bot", + `gateway:${registration}`, + registration, + registration, + ]; + const stop = registerConnectionEventWake(db, () => {}); + const botCap: IdleIncomingWakeCapability = { + ...capability, + routes: ["github-broker-events-v1", "tenant-api-http-v2"], + githubBroker: { + instanceId: "fixture-instance", + bindingsFingerprint: incomingWakeFingerprint([botTuple]), + }, + }; + expect((await read(verified())).backgroundWork).toBe("present"); + expect(await read(verified(botCap))).toMatchObject({ + version: 3, + backgroundWork: "none", + }); + const botProof = await readIdleIncomingWakeProof(db, { + capabilities: verified(botCap), + ownerId, + receiverOwner: db, + }); + expect(botProof?.endpointIds).toEqual([endpointId]); + expect(botProof?.connectionIds).toEqual([botConnectionId]); + expect( + await db + .select() + .from(chatEndpoints) + .where(eq(chatEndpoints.id, endpointId)), + ).toHaveLength(1); + + await db + .insert(toolApplications) + .values({ + id: applicationId, + companyId: f.companyId, + name: "GitHub", + type: "mcp_http", + }); + await db + .insert(toolConnections) + .values({ + id: connectionId, + companyId: f.companyId, + applicationId, + name: "GitHub", + uid: randomUUID(), + transport: "mcp_remote", + connectionPurpose: "tool", + status: "active", + enabled: true, + config: { + sourceTemplateKey: "github", + oauth: { connectorProfile: "github.code" }, + }, + }); + const github = { + userId: "42", + login: "fixture", + installationCount: 1, + repositoryCount: 1, + repositorySelection: "all" as const, + installationIds: ["123"], + installationOwnerLogins: ["fixture"], + lastAccessRefreshAt: new Date().toISOString(), + }; + await db + .insert(connectionGrants) + .values({ + id: grantId, + companyId: f.companyId, + connectionId, + kind: "user", + subjectUserId: f.userId, + providerTenant: { github }, + status: "active", + }); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID", "fixture-instance"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY", "fixture"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY", "fixture"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT", "test"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_BASE_URL", "http://127.0.0.1"); + const brokerCap: IdleIncomingWakeCapability = { + ...capability, + routes: ["github-broker-events-v1", "tenant-api-http-v2"], + githubBroker: { + instanceId: "fixture-instance", + bindingsFingerprint: incomingWakeFingerprint([ + botTuple, + [ + `${grantId}_123`, + f.companyId, + f.userId, + "github.code", + "123", + connectionId, + grantId, + ], + ]), + }, + }; + try { + const proof = await readIdleIncomingWakeProof(db, { + capabilities: verified(brokerCap), + ownerId, + receiverOwner: db, + }); + expect(proof?.requiredContinuityConnectionIds).toEqual([connectionId]); + expect(proof?.connectionIds).toEqual([botConnectionId]); + expect((await read(verified(brokerCap))).backgroundWork).toBe( + "present", + ); // transport alone never discharges real credential deadline + await db + .update(connectionGrants) + .set({ + providerTenant: { + github: { + ...github, + bindingSync: { id: randomUUID(), removeInstallationIds: [] }, + }, + }, + }) + .where(eq(connectionGrants.id, grantId)); + expect((await read(verified(brokerCap))).backgroundWork).toBe( + "unknown", + ); + await db + .update(connectionGrants) + .set({ status: "revoked" }) + .where(eq(connectionGrants.id, grantId)); + expect((await read(verified(brokerCap))).backgroundWork).toBe( + "unknown", + ); + await db + .update(toolConnections) + .set({ transport: "local_stdio" }) + .where(eq(toolConnections.id, connectionId)); + expect((await read(verified())).backgroundWork).toBe("present"); + } finally { + stop(); + } + }); + }, +); diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index 08123e23a55..eb9fb0cb38e 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -1,3 +1,4 @@ +import { generateKeyPairSync, randomUUID, sign } from "node:crypto"; import express from "express"; import request from "supertest"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -81,21 +82,24 @@ function localAdminActor() { describe("instance settings routes", () => { const routeModules = hoistModuleGraph(registerModuleMocks, async () => { - const [{ errorHandler }, { instanceSettingsRoutes }] = await Promise.all([ + const [{ errorHandler }, { instanceSettingsRoutes }, { cloudControlMiddleware }, controlIdentity] = await Promise.all([ vi.importActual("../middleware/index.js"), vi.importActual("../routes/instance-settings.js"), + vi.importActual("../middleware/cloud-control.js"), + vi.importActual("../services/cloud-runtime-identity.js"), ]); - return { errorHandler, instanceSettingsRoutes }; + return { errorHandler, instanceSettingsRoutes, cloudControlMiddleware, controlIdentity }; }); - function createApp(actor: any, pluginWorkers?: PluginWorkerManager, prepareBackup?: () => Promise) { - const { errorHandler, instanceSettingsRoutes } = routeModules.value; + function createApp(actor: any, pluginWorkers?: PluginWorkerManager, prepareBackup?: () => Promise, verifyControl = false) { + const { errorHandler, instanceSettingsRoutes, cloudControlMiddleware } = routeModules.value; const app = express(); app.use(express.json()); app.use((req, _res, next) => { req.actor = actor; next(); }); + if (verifyControl) app.use(cloudControlMiddleware()); app.use("/api", instanceSettingsRoutes(mockDb as any, pluginWorkers, prepareBackup)); app.use(errorHandler); return app; @@ -1101,7 +1105,7 @@ describe("instance settings routes", () => { await request(createApp(adminActor)).post("/api/instance/task-drain").send({ purpose: "idle", ttlMs: 60_000 }).expect(200); expect(mockHeartbeatService.computeTaskDrain).toHaveBeenCalledWith({ purpose: "idle", ttlMs: 60_000 }); await request(createApp(adminActor)).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200); - expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false); + expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined }); }); it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => { @@ -1128,7 +1132,7 @@ describe("instance settings routes", () => { const res = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1"); expect(res.status).toBe(200); expect(res.body).toEqual({ ...idleStatus, idleSleepSafety: report }); - expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false); + expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined }); expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus); }); @@ -1142,7 +1146,7 @@ describe("instance settings routes", () => { .get("/api/instance/task-drain?idleSleepSafety=1&idleSleepDiagnostics=1&ownerId=fixture-owner").expect(200); expect(response.body.idleSleepSafety.diagnostics.checks).toEqual([{ reason: "agent_api_keys", state: "present" }]); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, undefined, false, false, { collectDiagnostics: true }); + "fixture-owner", expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined, collectDiagnostics: true }); mockReadIdleSleepSafety.mockClear(); for (const actor of [nonAdminActor, { type: "agent", agentId: "agent-1", companyId: "company-1", source: "api_key" }]) { await request(createApp(actor)) @@ -1164,7 +1168,93 @@ describe("instance settings routes", () => { await request(createApp({ ...adminActor, source: "cloud_control" })) .get(`/api/instance/task-drain?idleSleepSafety=1&scheduledWake=${version}&ownerId=fixture-owner`).expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, undefined, true, version === "2"); + "fixture-owner", expect.any(Function), undefined, undefined, true, version === "2", { incomingWake: undefined, deadlineCapability: undefined }); + }); + + describe("signed incoming capability handoff", () => { + const createControlApp = (actor: any) => createApp(actor, undefined, undefined, true); + const keyPair = generateKeyPairSync("ed25519"); + const publicJwk = { ...keyPair.publicKey.export({ format: "jwk" }), kid: "idle-route-test", alg: "EdDSA", use: "sig" }; + const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895"; + const catalog = { + incomingWake: { version: 1, ownerId, generation: "a".repeat(64), apiKeyProof: { version: 1, challenge: "c".repeat(64) }, routes: ["tenant-api-http-v2"] }, + deadlineCatalog: "domain_deadlines_v1", + }; + const path = `/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`; + function assertion(claims: Record = {}) { + const identity = routeModules.value.controlIdentity; + const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url"); + const iat = Math.floor(Date.now() / 1000); + const header = encode({ alg: "EdDSA", typ: identity.CLOUD_CONTROL_JWS_TYPE, kid: publicJwk.kid }); + const payload = encode({ v: 1, iss: identity.CLOUD_RUNTIME_IDENTITY_ISSUER, aud: identity.CLOUD_CONTROL_AUDIENCE, + sub: "stack-idle-route-test", action: "task-drain:read", requestId: randomUUID(), iat, exp: iat + 60, + idleCapabilities: catalog, ...claims }); + return `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`, "ascii"), keyPair.privateKey).toString("base64url")}`; + } + beforeEach(() => { + vi.stubEnv("PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS", JSON.stringify({ keys: [publicJwk] })); + vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "stack-idle-route-test"); + routeModules.value.controlIdentity.resetCloudControlReplayFenceForTests(); + mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus); + mockReadIdleSleepSafety.mockResolvedValue({ version: 3, backgroundWork: "none" }); + }); + afterEach(() => vi.unstubAllEnvs()); + + it("passes only the immutable signed catalog through real middleware and the production route", async () => { + const token = assertion(); + const response = await request(createControlApp({ type: "none" })) + .get(`${path}&incomingWake=${encodeURIComponent(JSON.stringify({ ...catalog.incomingWake, generation: "b".repeat(64) }))}`) + .set("x-paperclip-cloud-control", token) + .send({ idleCapabilities: { incomingWake: { ...catalog.incomingWake, ownerId: randomUUID() } } }).expect(200); + expect(response.body.idleSleepSafety).toEqual({ version: 3, backgroundWork: "none" }); + expect(mockReadIdleSleepSafety).toHaveBeenCalledExactlyOnceWith(mockDb, expect.any(Function), Date.now, + ownerId, expect.any(Function), undefined, undefined, false, false, + { incomingWake: catalog, deadlineCapability: catalog }); + const options = mockReadIdleSleepSafety.mock.calls[0][9]; + expect(options.incomingWake).toBe(options.deadlineCapability); + expect(Object.isFrozen(options.incomingWake)).toBe(true); + expect(Object.isFrozen(options.incomingWake.incomingWake)).toBe(true); + expect(Object.isFrozen(options.incomingWake.incomingWake.routes)).toBe(true); + await request(createControlApp({ type: "none" })).get(path).set("x-paperclip-cloud-control", token).expect(401); + expect(mockReadIdleSleepSafety).toHaveBeenCalledTimes(1); + }); + + it.each([ + ["wrong stack", () => ({ sub: "other-stack" })], + ["wrong action", () => ({ action: "task-drain:start" })], + ["expired", () => ({ iat: Math.floor(Date.now() / 1000) - 600, exp: Math.floor(Date.now() / 1000) - 300 })], + ["unsupported catalog", () => ({ idleCapabilities: { incomingWake: { ...catalog.incomingWake, routes: ["unverified-path"] } } })], + ])("rejects %s before safety inspection", async (_name, claims) => { + await request(createControlApp(adminActor)).get(path).set("x-paperclip-cloud-control", assertion(claims())).expect(401); + expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); + }); + + it("rejects a tampered catalog signature before safety inspection", async () => { + const [header, payload, signature] = assertion().split("."); + const changed = JSON.parse(Buffer.from(payload!, "base64url").toString()); + changed.idleCapabilities.incomingWake.generation = "b".repeat(64); + const tampered = `${header}.${Buffer.from(JSON.stringify(changed)).toString("base64url")}.${signature}`; + await request(createControlApp(adminActor)).get(path).set("x-paperclip-cloud-control", tampered).expect(401); + expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); + }); + + it("leaves legacy signed reads without a catalog conservative", async () => { + await request(createControlApp({ type: "none" })).get(path) + .set("x-paperclip-cloud-control", assertion({ idleCapabilities: undefined })).expect(200); + expect(mockReadIdleSleepSafety.mock.calls[0][9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined }); + }); + + it("cannot install capabilities from tenant, actor, query or body values", async () => { + for (const source of ["session", "cloud_tenant", "local_implicit", "cloud_control"]) { + const actor = { ...adminActor, source, idleCapabilities: catalog, incomingWake: catalog }; + await request(createControlApp(actor)).get(`${path}&incomingWake=${encodeURIComponent(JSON.stringify(catalog))}`) + .send({ idleCapabilities: catalog }).expect(200); + expect(mockReadIdleSleepSafety.mock.lastCall?.[9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined }); + } + mockReadIdleSleepSafety.mockClear(); + await request(createControlApp(nonAdminActor)).get(path).send({ idleCapabilities: catalog }).expect(403); + expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); + }); }); it("uses the live worker manager for inspection and releases only the owned hold", async () => { @@ -1199,7 +1289,7 @@ describe("instance settings routes", () => { expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); await request(app).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, backup, false, false); + "fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined }); mockReadIdleSleepSafety.mockClear(); await request(createApp(nonAdminActor, undefined, backup)).get("/api/instance/task-drain?idleSleepSafety=1").expect(403); expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); @@ -1221,7 +1311,7 @@ describe("instance settings routes", () => { const control = createApp({ ...adminActor, source: "cloud_control" }, undefined, backup); await request(control).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, backup, false, false); + "fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined }); } finally { delete process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN; } diff --git a/server/src/__tests__/live-events-toast.test.ts b/server/src/__tests__/live-events-toast.test.ts index c877b70f090..ffa96702e50 100644 --- a/server/src/__tests__/live-events-toast.test.ts +++ b/server/src/__tests__/live-events-toast.test.ts @@ -8,6 +8,7 @@ import { describe, expect, it, vi } from "vitest"; import { setupLiveEventsWebSocketServer } from "../realtime/live-events-ws.js"; import { publishLiveEvent } from "../services/live-events.js"; import { taskToastEligible } from "../realtime/task-toast-eligibility.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; vi.mock("../realtime/task-toast-eligibility.js", () => ({ taskToastEligible: vi.fn(async () => false) })); @@ -39,6 +40,48 @@ function database(onPredicate: (kind: "where" | "join", condition: SQL) => void } }; } describe("authenticated activity notifications", () => { + it("retains live delivery idle work until the identifier read settles after the last socket closes", async () => { + const baseline = idleWorkSnapshot().active; + let release!: () => void, entered!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const db = database(), select = db.select; + db.select = fields => { + const query = select(fields); + if (fields.identifier) { + const limit = query.limit; + query.limit = async () => { entered(); await pending; return limit(); }; + } + return query; + }; + const server = createServer(); + const wss = setupLiveEventsWebSocketServer(server, db as never, { deploymentMode: "local_trusted" }); + const closed = new Promise(resolve => wss.on("connection", socket => socket.on("close", resolve))); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const socket = new WebSocket(`ws://127.0.0.1:${(server.address() as { port: number }).port}/api/companies/company/events/ws`); + try { + await once(socket, "open"); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(baseline)); + // System activity has no actor lookup: this gate is the new identifier SQL. + publishLiveEvent({ companyId: "company", type: "activity.logged", payload: { + action: "issue.updated", entityType: "issue", entityId: "visible", actorType: "system", + } }); + await started; + socket.terminate(); + await closed; + expect(wss.clients.size).toBe(0); + expect(idleWorkSnapshot().active).toBe(baseline + 1); + release(); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(baseline)); + } finally { + release(); + socket.terminate(); + await new Promise(resolve => server.close(() => resolve())); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(baseline)); + } + }); + it.each(["local_trusted", "authenticated"] as const)("keeps authorized notifications when optional identifier reads fail (%s)", async deploymentMode => { const db = database(); const select = db.select; diff --git a/server/src/middleware/auth.ts b/server/src/middleware/auth.ts index f9bd7446947..bf8a05f22d0 100644 --- a/server/src/middleware/auth.ts +++ b/server/src/middleware/auth.ts @@ -1,3 +1,4 @@ +import { idleApiActivity } from "../services/idle-api-activity.js"; import { createHash, timingSafeEqual } from "node:crypto"; import type { Request, RequestHandler } from "express"; import { and, eq, isNull, sql } from "drizzle-orm"; @@ -43,7 +44,7 @@ import { ensureHumanRoleDefaultGrants } from "../services/principal-access-compa import { forbidden, unauthorized, unprocessable } from "../errors.js"; export { isCloudManagedInstance } from "../services/cloud-instance.js"; -import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js"; +import { cloudTenantPrimaryCompanyId, isCloudManagedInstance } from "../services/cloud-instance.js"; function hashToken(token: string) { return createHash("sha256").update(token).digest("hex"); @@ -329,6 +330,9 @@ export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHa runId: runIdHeader || undefined, source: "board_key", }; + if (isCloudManagedInstance() && (access.isInstanceAdmin || access.companyIds.length)) { + _res.once("finish", () => { if (_res.statusCode >= 200 && _res.statusCode < 400) idleApiActivity.record(); }); + } next(); return; } @@ -521,7 +525,9 @@ export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHa runId: runIdHeader || undefined, source: "agent_key", }; - + if (isCloudManagedInstance() && req.actor.onBehalfOfMemberships?.length) { + _res.once("finish", () => { if (_res.statusCode >= 200 && _res.statusCode < 400) idleApiActivity.record(); }); + } next(); }; return async (req, res, next) => { diff --git a/server/src/middleware/cloud-control.ts b/server/src/middleware/cloud-control.ts index b726aae3092..6165af6b0b6 100644 --- a/server/src/middleware/cloud-control.ts +++ b/server/src/middleware/cloud-control.ts @@ -1,3 +1,5 @@ +import type { Request } from "express"; +import type { VerifiedIdleCapabilities } from "../services/idle-incoming-wake-capability.js"; import type { RequestHandler } from "express"; import { logger } from "./logger.js"; import { @@ -6,6 +8,12 @@ import { type CloudControlAction, } from "../services/cloud-runtime-identity.js"; +const idleCapabilities = new WeakMap(); +/** Only this middleware's verified assertion can install request capabilities. */ +export function verifiedIdleCapabilities(req: Request): VerifiedIdleCapabilities | undefined { + return idleCapabilities.get(req); +} + /** 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" }, @@ -50,7 +58,8 @@ export function cloudControlMiddleware(): RequestHandler { return; } try { - verifyCloudControlAssertion({ compactJws: assertion, expectedAction }); + const claims = verifyCloudControlAssertion({ compactJws: assertion, expectedAction }); + if (claims.idleCapabilities) idleCapabilities.set(req, claims.idleCapabilities as VerifiedIdleCapabilities); } catch (error) { logger.warn({ err: error }, "Rejected Cloud control assertion"); res.status(401).json({ error: "invalid_cloud_control_assertion" }); diff --git a/server/src/realtime/live-event-delivery.test.ts b/server/src/realtime/live-event-delivery.test.ts index 02e8c536c83..2f8b4c4eac4 100644 --- a/server/src/realtime/live-event-delivery.test.ts +++ b/server/src/realtime/live-event-delivery.test.ts @@ -1,6 +1,7 @@ import { afterEach, expect, it, vi } from "vitest"; import type { LiveEvent } from "@paperclipai/shared"; import { createLiveEventDelivery, LIVE_DELIVERY_MAX_BYTES, LIVE_DELIVERY_MAX_ITEMS } from "./live-event-delivery.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; const event = (seq = 1, runId = "run-1"): LiveEvent => ({ id: seq, companyId: "company", type: "heartbeat.run.log", createdAt: "2026-10-24T00:00:00Z", payload: { runId, agentId: "agent", seq, chunk: "output" } }); @@ -116,7 +117,7 @@ it.each([false, true])("preserves the overflow deadline across resubscription un await vi.advanceTimersByTimeAsync(1_000); if (settled) expect(f.socket.close).not.toHaveBeenCalled(); else expect(f.socket.close).toHaveBeenCalledWith(1013, "persistent event backlog"); - } finally { f.delivery.dispose(); } + } finally { f.delivery.dispose(); release(event()); await settle(); } expect(vi.getTimerCount()).toBe(0); }); it("recovers failed delivery reads through resync rather than silently claiming healthy streaming", async () => { @@ -146,6 +147,7 @@ it("rejects unbounded interest sets and stale revision commands", () => { it.each([false, true].flatMap(negotiated => ["authorize", "decorate"].map(stage => ({ negotiated, stage }))))( "closes a stalled $stage read below the queue bound (negotiated=$negotiated)", async ({ negotiated, stage }) => { vi.useFakeTimers(); + const baseline = idleWorkSnapshot().active; let release!: (value: LiveEvent) => void; const pending = new Promise(resolve => { release = resolve; }); const f = fixture(stage === "authorize" ? vi.fn(() => pending) : undefined); @@ -168,11 +170,13 @@ it.each([false, true].flatMap(negotiated => ["authorize", "decorate"].map(stage expect(f.socket.close).toHaveBeenCalledWith(1013, "persistent event backlog"); expect(f.delivery.metrics().retainedItems).toBe(1); expect(vi.getTimerCount()).toBe(0); + expect(idleWorkSnapshot().active).toBe(baseline + 1); release(source); await settle(); + expect(idleWorkSnapshot().active).toBe(baseline); expect(f.delivery.metrics()).toMatchObject({ retainedItems: 0, retainedBytes: 0 }); expect(f.frames.filter(frame => frame.type.startsWith("heartbeat."))).toEqual([]); expect(f.authorize).toHaveBeenCalledOnce(); - } finally { f.delivery.dispose(); other.delivery.dispose(); } + } finally { f.delivery.dispose(); other.delivery.dispose(); release(event()); await settle(); } }); it.each(["resolve", "dispose"])("clears a pending-read deadline on %s without idle work", async ending => { diff --git a/server/src/realtime/live-event-delivery.ts b/server/src/realtime/live-event-delivery.ts index 3d7f4371779..588eb654a35 100644 --- a/server/src/realtime/live-event-delivery.ts +++ b/server/src/realtime/live-event-delivery.ts @@ -1,4 +1,5 @@ import type { LiveEvent } from "@paperclipai/shared"; +import { beginIdleTrackedWork } from "../services/task-admission.js"; export const LIVE_DELIVERY_MAX_ITEMS = 1_000; export const LIVE_DELIVERY_MAX_BYTES = 1024 * 1024; @@ -90,6 +91,7 @@ export function createLiveEventDelivery(input: { async function drain() { if (draining) return; draining = true; + const finishIdleWork = beginIdleTrackedWork(); try { while (queued.length && isOpen()) { const item = queued.shift()!; @@ -121,7 +123,10 @@ export function createLiveEventDelivery(input: { retainedBytes -= item.bytes; retainedItems -= 1; } } - } finally { draining = false; resync(); } + } finally { + try { draining = false; resync(); } + finally { finishIdleWork(); } + } } return { receive(raw: string) { diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index 66123a51e52..e5d9c9f648e 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -1,3 +1,4 @@ +import { verifiedIdleCapabilities } from "../middleware/cloud-control.js"; import { trackIdleWork } from "../services/task-admission.js"; import { Router, type Request } from "express"; import { companies, type Db } from "@paperclipai/db"; @@ -336,7 +337,8 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag typeof req.query.ownerId === "string" ? req.query.ownerId : undefined, diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic }), pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors, - ...(req.query.idleSleepDiagnostics === "1" ? [{ collectDiagnostics: true }] as const : [] as const)); + { incomingWake: verifiedIdleCapabilities(req), deadlineCapability: verifiedIdleCapabilities(req), + ...(req.query.idleSleepDiagnostics === "1" ? { collectDiagnostics: true } : {}) }); res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety }); return; } diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 804359109e5..010e9670202 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -136,6 +136,7 @@ import { notifyChatDeliveryWork, notifyChatPublicationWork } from "./chat-work-n import { voiceSessionStore } from "./voice/voice-session-store.js"; import { createChatRuntime, type ChatRuntime } from "./chat-runtime.js"; +import { registerIdleLocalWorkProbe } from "./idle-local-work.js"; import type { ChatEndpointRuntime } from "./voice/voice-runtime.js"; import { fastResponseService, fastResponseSourceCurrent } from "./fast-responses.js"; @@ -517,6 +518,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { version: string; } >(); + // SDK initialize() and the final authority read continue after the registry's + // replacement tail settles. Neither that work nor a failed retirement may + // disappear just because the archived endpoint is no longer returned by get(). + const unregisterRuntimeIdleProbe = registerIdleLocalWorkProbe(() => + runtimeInitializations.size > 0 || runtime.hasUnsettledLifecycle() ? "present" : "none"); const { recordCurrentMicrosoftTeamsRoute } = createTeamsRoutes(db, { runtime, runtimeCallbackEndpoint }); const persistence = createChatSdkStatePersistence(db); const fetchImpl = options.fetch ?? globalThis.fetch; @@ -3622,12 +3628,17 @@ return { // Stop every provider runtime before releasing leader rows. This gives a // standby an immediate takeover path without overlapping Gateway sockets. await runtime.shutdown(); + await Promise.allSettled([...runtimeInitializations.values()].map(initialization => initialization.promise)); await slackRegistration.close(); await Promise.all( ownedDiscordGateways.map((ownership) => releaseDiscordGatewayLeaseRow(ownership), ), ); + if (runtimeInitializations.size > 0 || runtime.hasUnsettledLifecycle()) { + throw new Error("Chat runtime lifecycle is still unsettled after shutdown"); + } + unregisterRuntimeIdleProbe(); }, }; } diff --git a/server/src/services/chat-runtime.ts b/server/src/services/chat-runtime.ts index b404a229e9b..881810ae54d 100644 --- a/server/src/services/chat-runtime.ts +++ b/server/src/services/chat-runtime.ts @@ -23,6 +23,11 @@ export class ChatRuntime { return [...this.endpoints.values()]; } + /** Retirement failures retain their owner even after get/list hide it. */ + hasUnsettledLifecycle(): boolean { + return this.retiringEndpoints.size > 0 || this.lifecycleTails.size > 0; + } + private enqueueLifecycle( endpointId: string, operation: () => Promise, diff --git a/server/src/services/chat-sdk-runtime.test.ts b/server/src/services/chat-sdk-runtime.test.ts index 89e9a9c3587..a956946dc6e 100644 --- a/server/src/services/chat-sdk-runtime.test.ts +++ b/server/src/services/chat-sdk-runtime.test.ts @@ -241,6 +241,8 @@ import { type CreateChatSdkEndpointRuntimeOptions, } from "./chat-sdk-runtime.js"; +import { createChatRuntime } from "./chat-runtime.js"; + const persistence: ChatSdkStatePersistence = { async compareAndSet() { return true; @@ -1152,6 +1154,31 @@ describe("Chat SDK endpoint runtime", () => { await registry.shutdown(); }); + it.each([ + { name: "SDK", create: createChatSdkRuntime }, + { name: "shared", create: createChatRuntime }, + ])("keeps $name lifecycle work visible while hidden retirement fails, until its owner retries", async ({ create }) => { + const registry = create(); + const first = await registry.replaceEndpoint(registryOptions()); + expect(registry.hasUnsettledLifecycle()).toBe(false); + const held = deferred(); + const shutdown = vi.spyOn(first, "shutdown").mockReturnValueOnce(held.promise).mockResolvedValue(undefined); + const removal = registry.removeEndpoint("endpoint-1"); + const result = removal.catch((error: Error) => error.message); + expect(registry.hasUnsettledLifecycle()).toBe(true); + await vi.waitFor(() => expect(shutdown).toHaveBeenCalledOnce()); + expect(registry.get("endpoint-1")).toBeNull(); + expect(registry.list()).toEqual([]); + expect(registry.hasUnsettledLifecycle()).toBe(true); + held.reject(new Error("retirement uncertain")); + expect(await result).toBe("retirement uncertain"); + expect(registry.hasUnsettledLifecycle()).toBe(true); + await registry.removeEndpoint("endpoint-1"); + expect(shutdown).toHaveBeenCalledTimes(2); + expect(registry.hasUnsettledLifecycle()).toBe(false); + await registry.shutdown(); + }); + it("retains failed retirement ownership and only constructs a replacement after an explicit successful retry", async () => { const registry = createChatSdkRuntime(); const first = await registry.replaceEndpoint(registryOptions()); diff --git a/server/src/services/chat-sdk-runtime.ts b/server/src/services/chat-sdk-runtime.ts index cffdd09f5e4..6afd8fb844c 100644 --- a/server/src/services/chat-sdk-runtime.ts +++ b/server/src/services/chat-sdk-runtime.ts @@ -3195,6 +3195,11 @@ export class ChatSdkRuntime { return [...this.endpoints.values()]; } + /** Retirement failures retain their owner even after get/list hide it. */ + hasUnsettledLifecycle(): boolean { + return this.retiringEndpoints.size > 0 || this.lifecycleTails.size > 0; + } + private enqueueLifecycle( endpointId: string, operation: () => Promise, diff --git a/server/src/services/cloud-runtime-identity.ts b/server/src/services/cloud-runtime-identity.ts index 80b3333b36f..cffe6a8dbfd 100644 --- a/server/src/services/cloud-runtime-identity.ts +++ b/server/src/services/cloud-runtime-identity.ts @@ -1,3 +1,4 @@ +import { parseIdleCapabilities, type IdleCapabilities } from "./idle-incoming-wake-capability.js"; import { createPublicKey, timingSafeEqual, verify, type JsonWebKey } from "node:crypto"; import { eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -491,6 +492,7 @@ export type CloudControlClaims = { sub: string; action: CloudControlAction; requestId: string; + idleCapabilities?: IdleCapabilities; iat: number; exp: number; }; @@ -572,6 +574,10 @@ export function verifyCloudControlAssertion(input: { ) { throw new Error("Cloud control assertion is expired or has an invalid lifetime"); } + if (payload.idleCapabilities !== undefined) { + if (payload.action !== "task-drain:read") throw new Error("Idle capabilities require task-drain:read"); + payload.idleCapabilities = parseIdleCapabilities(payload.idleCapabilities); + } // Consumed LAST, only after every other check passed: a rejected // assertion must not burn its request id, or an attacker could deny a // legitimate call by replaying a mangled copy of it first. diff --git a/server/src/services/connection-event-wake.ts b/server/src/services/connection-event-wake.ts index 87840eb0979..cd7bd24da51 100644 --- a/server/src/services/connection-event-wake.ts +++ b/server/src/services/connection-event-wake.ts @@ -10,3 +10,6 @@ export function wakeConnectionEvents(owner: object): boolean { wake(); return true; } + +/** Inspection never emits a wake or polls the broker. */ +export const hasConnectionEventWakeReceiver = (owner: object): boolean => receivers.has(owner); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index af936353bdd..e84092ed826 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -146,7 +146,7 @@ function normalizeLeasedPayload(event: LeasedEvent): Record { return compact(base); } -function bindingRows(rows: Array<{ +export function githubConnectionEventBindings(rows: Array<{ grant: typeof connectionGrants.$inferSelect; connection: typeof toolConnections.$inferSelect; }>): GitHubBinding[] { @@ -259,7 +259,7 @@ export function githubConnectionEventService( eq(toolConnections.enabled, true), sql`exists (select 1 from companies where id = ${connectionGrants.companyId} and status <> 'archived')`, )); - return bindingRows(rows); + return githubConnectionEventBindings(rows); } async function applyPullRequestEvent(database: Db, companyId: string, event: LeasedEvent, effects: Array<(committedDb: Db) => Promise>) { diff --git a/server/src/services/idle-api-activity.test.ts b/server/src/services/idle-api-activity.test.ts new file mode 100644 index 00000000000..e0e54ddfe83 --- /dev/null +++ b/server/src/services/idle-api-activity.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from "vitest"; +import { IdleApiActivity } from "./idle-api-activity.js"; + +describe("authenticated API idle activity", () => { + it("is passive without successful authentication and expires without timers or dependency work", () => { + let monotonic = 1; + const activity = new IdleApiActivity(() => monotonic); + expect(activity.active()).toBe(false); + activity.record(); + expect(activity.active()).toBe(true); + monotonic += 599_999; + expect(activity.active()).toBe(true); + monotonic++; + expect(activity.active()).toBe(false); + }); + it("extends only from the actual latest successful use and ignores wall clock changes", () => { + let monotonic = 1; + const activity = new IdleApiActivity(() => monotonic); + activity.record(); + monotonic += 300_000; + activity.record(); + monotonic += 300_001; + expect(activity.active()).toBe(true); + monotonic += 299_999; + expect(activity.active()).toBe(false); + }); +}); diff --git a/server/src/services/idle-api-activity.ts b/server/src/services/idle-api-activity.ts new file mode 100644 index 00000000000..6c4879201f0 --- /dev/null +++ b/server/src/services/idle-api-activity.ts @@ -0,0 +1,12 @@ +/** Authenticated API use is real activity; rejecting an unknown key is not. */ +export class IdleApiActivity { + private until = 0; + constructor(private readonly now: () => number = () => performance.now()) {} + record(): void { + this.until = Math.max(this.until, this.now() + 10 * 60_000); + } + active(): boolean { + return this.now() < this.until; + } +} +export const idleApiActivity = new IdleApiActivity(); diff --git a/server/src/services/idle-api-wake-admission.test.ts b/server/src/services/idle-api-wake-admission.test.ts new file mode 100644 index 00000000000..604eb2cb4ef --- /dev/null +++ b/server/src/services/idle-api-wake-admission.test.ts @@ -0,0 +1,33 @@ +import { createHash } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { apiWakeFingerprint, createApiWakeAdmission, parseApiWakeAdmission, matchesApiWakeAdmission } from "./idle-api-wake-admission.js"; + +const token = "api-wake-vector-token"; +const keyHash = createHash("sha256").update(token).digest("hex"); +const input = { stackId: "vector-stack", ownerId: "00000000-0000-4000-8000-000000000001", capabilityGeneration: "a".repeat(64), challenge: "c".repeat(64) }; +const fingerprint = apiWakeFingerprint({ ...input, keyHash }); +const receipt = createApiWakeAdmission({ ...input, keys: [{ fingerprint, expiresAt: null }] }); +describe("API wake receipt wire vectors", () => { + it("uses exact domain-separated vectors and exports neither token nor database hash", () => { + expect(fingerprint).toBe("6478041b90f4665eeca8403186fefedc456bf1322f38dab9fb89212eecaafd43"); + expect(JSON.stringify(receipt)).not.toContain(token); + expect(JSON.stringify(receipt)).not.toContain(keyHash); + expect(matchesApiWakeAdmission(receipt, { stackId: input.stackId, token, now: new Date("2099-01-01").getTime() })).toBe(true); + expect(matchesApiWakeAdmission(receipt, { stackId: "other", token, now: 0 })).toBe(false); + expect(matchesApiWakeAdmission(receipt, { stackId: input.stackId, token: "unknown", now: 0 })).toBe(false); + }); + it("enforces board expiry and canonical unique bounded complete receipts", () => { + const changedChallenge = createApiWakeAdmission({ ...input, challenge: "b".repeat(64), keys: receipt.keys }); + expect(matchesApiWakeAdmission(changedChallenge, { stackId: input.stackId, token, now: 0 })).toBe(false); + const expiring = createApiWakeAdmission({ ...input, keys: [{ fingerprint, expiresAt: new Date(1000).toISOString() }] }); + expect(matchesApiWakeAdmission(expiring, { stackId: input.stackId, token, now: 999 })).toBe(true); + expect(matchesApiWakeAdmission(expiring, { stackId: input.stackId, token, now: 1000 })).toBe(false); + for (const invalid of [ + { ...receipt, generation: "b".repeat(64) }, { ...receipt, challenge: "b".repeat(64) }, + { ...receipt, extra: true }, { ...receipt, keys: [...receipt.keys, ...receipt.keys] }, + { ...receipt, keys: Array.from({ length: 1001 }, () => receipt.keys[0]) }, + { ...receipt, keys: [{ fingerprint, expiresAt: "2099-01-01" }] }, + { ...receipt, keys: [{ fingerprint: "d".repeat(64), expiresAt: null }, { fingerprint: "a".repeat(64), expiresAt: null }] }, + ]) expect(() => parseApiWakeAdmission(invalid)).toThrow(); + }); +}); diff --git a/server/src/services/idle-api-wake-admission.ts b/server/src/services/idle-api-wake-admission.ts new file mode 100644 index 00000000000..26ed499a231 --- /dev/null +++ b/server/src/services/idle-api-wake-admission.ts @@ -0,0 +1,56 @@ +import { createHash } from "node:crypto"; + +export type ApiWakeClaim = { generation: string; fingerprint: string }; +export interface ApiWakeAdmission { + version: 1; + stackId: string; + ownerId: string; + capabilityGeneration: string; + challenge: string; + generation: string; + keys: { fingerprint: string; expiresAt: string | null }[]; +} +const digest = (value: unknown): value is string => typeof value === "string" && /^[a-f0-9]{64}$/.test(value); +const object = (value: unknown): value is Record => Boolean(value && typeof value === "object" && !Array.isArray(value)); +const exact = (value: Record, keys: string[]) => Object.keys(value).length === keys.length && Object.keys(value).every(key => keys.includes(key)); +const generation = (value: Omit) => createHash("sha256").update(JSON.stringify([ + "paperclip-api-wake-receipt-v1", value.version, value.stackId, value.ownerId, value.capabilityGeneration, value.challenge, + value.keys.map(key => [key.fingerprint, key.expiresAt]), +])).digest("hex"); +export function apiWakeFingerprint(input: { stackId: string; ownerId: string; challenge: string; keyHash: string }): string { + if (!digest(input.keyHash) || !digest(input.challenge)) throw new Error("Invalid API wake fingerprint input"); + return createHash("sha256").update(JSON.stringify(["paperclip-api-wake-v1", input.stackId, input.ownerId, input.challenge, input.keyHash])).digest("hex"); +} +/** Strict bounded canonical wire contract. Parsing never grants tenant permissions. */ +export function parseApiWakeAdmission(value: unknown): ApiWakeAdmission { + if (!object(value) || !exact(value, ["version", "stackId", "ownerId", "capabilityGeneration", "challenge", "generation", "keys"]) + || value.version !== 1 || typeof value.stackId !== "string" || !/^[A-Za-z0-9_-]{1,200}$/.test(value.stackId) + || typeof value.ownerId !== "string" || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(value.ownerId) + || !digest(value.capabilityGeneration) || !digest(value.challenge) || !digest(value.generation) + || !Array.isArray(value.keys) || value.keys.length > 1000) throw new Error("Invalid API wake receipt"); + let previous = ""; + for (const key of value.keys) { + if (!object(key) || !exact(key, ["fingerprint", "expiresAt"]) || !digest(key.fingerprint) || key.fingerprint <= previous + || !(key.expiresAt === null || typeof key.expiresAt === "string" && key.expiresAt.length <= 30 && Number.isFinite(Date.parse(key.expiresAt)) + && new Date(key.expiresAt).toISOString() === key.expiresAt)) throw new Error("Invalid API wake receipt key"); + previous = key.fingerprint; + } + const parsed = value as unknown as ApiWakeAdmission; + if (generation(parsed) !== parsed.generation) throw new Error("API wake receipt generation changed"); + return { version: 1, stackId: parsed.stackId, ownerId: parsed.ownerId, capabilityGeneration: parsed.capabilityGeneration, + challenge: parsed.challenge, generation: parsed.generation, + keys: parsed.keys.map(key => ({ fingerprint: key.fingerprint, expiresAt: key.expiresAt })) }; +} +export function createApiWakeAdmission(input: Omit): ApiWakeAdmission { + const canonical = { version: 1 as const, ...input, keys: [...input.keys].sort((a, b) => a.fingerprint.localeCompare(b.fingerprint)) }; + return parseApiWakeAdmission({ ...canonical, generation: generation(canonical) }); +} +export function authorizeApiWakeAdmission(receipt: ApiWakeAdmission, input: { stackId: string; token: string; now: number }): ApiWakeClaim | undefined { + if (receipt.stackId !== input.stackId || input.token.length < 1 || input.token.length > 8192 || /\s/.test(input.token)) return; + const fingerprint = apiWakeFingerprint({ ...receipt, keyHash: createHash("sha256").update(input.token).digest("hex") }); + return receipt.keys.some(key => key.fingerprint === fingerprint && (key.expiresAt === null || Date.parse(key.expiresAt) > input.now)) + ? { generation: receipt.generation, fingerprint } : undefined; +} +export function matchesApiWakeAdmission(receipt: ApiWakeAdmission, input: { stackId: string; token: string; now: number }): boolean { + return authorizeApiWakeAdmission(receipt, input) !== undefined; +} diff --git a/server/src/services/idle-incoming-wake-capability.test.ts b/server/src/services/idle-incoming-wake-capability.test.ts new file mode 100644 index 00000000000..c3b9cd70da6 --- /dev/null +++ b/server/src/services/idle-incoming-wake-capability.test.ts @@ -0,0 +1,113 @@ +import { generateKeyPairSync, sign } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { parseIdleCapabilities } from "./idle-incoming-wake-capability.js"; +import { + verifyCloudControlAssertion, + resetCloudControlReplayFenceForTests, + CLOUD_CONTROL_AUDIENCE, + CLOUD_CONTROL_JWS_TYPE, + CLOUD_RUNTIME_IDENTITY_ISSUER, +} from "./cloud-runtime-identity.js"; +const incoming = { + version: 1 as const, + ownerId: "00000000-0000-4000-8000-000000000001", + generation: "a".repeat(64), + apiKeyProof: { version: 1 as const, challenge: "c".repeat(64) }, + routes: ["tenant-api-http-v2" as const], +}; +const invalid: unknown[] = [ + null, + {}, + { other: true }, + { deadlineCatalog: "other" }, + { incomingWake: { ...incoming, version: 2 } }, + { incomingWake: { ...incoming, routes: ["all"] } }, + { + incomingWake: { + ...incoming, + routes: ["tenant-api-http-v2", "tenant-api-http-v2"], + }, + }, + { incomingWake: { ...incoming, ownerId: "other" } }, + { incomingWake: { ...incoming, extra: true } }, + { + incomingWake: { + ...incoming, + githubBroker: { + instanceId: "instance", + bindingsFingerprint: "b".repeat(64), + }, + }, + }, + { incomingWake: { ...incoming, routes: ["github-broker-events-v1"] } }, +]; + +describe("signed idle capabilities", () => { + it.each(invalid)("rejects unknown/partial capabilities: %j", (value) => { + expect(() => parseIdleCapabilities(value)).toThrow(); + }); + it("detaches and freezes the exact known catalogs", () => { + const source = { + incomingWake: { ...incoming, routes: [...incoming.routes] }, + deadlineCatalog: "domain_deadlines_v1", + }; + const parsed = parseIdleCapabilities(source); + source.incomingWake.routes.length = 0; + expect(parsed.incomingWake?.routes).toEqual(["tenant-api-http-v2"]); + expect(Object.isFrozen(parsed.incomingWake?.routes)).toBe(true); + }); + it("binds capability to signed stack, action and replay fence", () => { + resetCloudControlReplayFenceForTests(); + const pair = generateKeyPairSync("ed25519"); + const now = new Date("2099-01-01T00:00:00Z"); + const jwk = { + ...pair.publicKey.export({ format: "jwk" }), + kid: "idle", + use: "sig", + alg: "EdDSA", + }; + const env = { + PAPERCLIP_CLOUD_STACK_ID: "stack", + PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: JSON.stringify({ keys: [jwk] }), + }; + const encode = (v: unknown) => + Buffer.from(JSON.stringify(v)).toString("base64url"); + const token = (overrides: object = {}) => { + const header = encode({ + alg: "EdDSA", + typ: CLOUD_CONTROL_JWS_TYPE, + kid: "idle", + }); + const body = encode({ + v: 1, + iss: CLOUD_RUNTIME_IDENTITY_ISSUER, + aud: CLOUD_CONTROL_AUDIENCE, + sub: "stack", + action: "task-drain:read", + requestId: "same", + iat: now.getTime() / 1000, + exp: now.getTime() / 1000 + 60, + idleCapabilities: { + incomingWake: incoming, + deadlineCatalog: "domain_deadlines_v1", + }, + ...overrides, + }); + return `${header}.${body}.${sign(null, Buffer.from(`${header}.${body}`), pair.privateKey).toString("base64url")}`; + }; + const verify = (compactJws: string) => + verifyCloudControlAssertion({ + compactJws, + env, + now, + expectedAction: "task-drain:read", + }); + expect(() => verify(token({ sub: "foreign" }))).toThrow(); + expect(() => verify(token({ action: "task-drain:start" }))).toThrow(); + expect(() => + verify(token({ idleCapabilities: { deadlineCatalog: "unknown" } })), + ).toThrow(); + expect(verify(token()).idleCapabilities?.incomingWake).toEqual(incoming); + expect(() => verify(token())).toThrow(/already been used/); + }); +}); diff --git a/server/src/services/idle-incoming-wake-capability.ts b/server/src/services/idle-incoming-wake-capability.ts new file mode 100644 index 00000000000..48dfe19c654 --- /dev/null +++ b/server/src/services/idle-incoming-wake-capability.ts @@ -0,0 +1,93 @@ +/** Closed Cloud-owned ingress catalog. No credential material belongs in this proof. */ +export interface IdleIncomingWakeCapability { + version: 1; + ownerId: string; + generation: string; + routes: ("tenant-api-http-v2" | "github-broker-events-v1")[]; + apiKeyProof?: { version: 1; challenge: string }; + githubBroker?: { instanceId: string; bindingsFingerprint: string }; +} +export interface IdleCapabilities { + incomingWake?: IdleIncomingWakeCapability; + deadlineCatalog?: "domain_deadlines_v1"; +} +const object = (v: unknown): v is Record => + Boolean(v && typeof v === "object" && !Array.isArray(v)); +const exact = (v: Record, keys: string[]) => + Object.keys(v).every((k) => keys.includes(k)); +const digest = (v: unknown): v is string => + typeof v === "string" && /^[a-f0-9]{64}$/.test(v); +export function parseIdleCapabilities(value: unknown): IdleCapabilities { + if ( + !object(value) || + !exact(value, ["incomingWake", "deadlineCatalog"]) || + !Object.keys(value).length + ) + throw new Error("Invalid idle capabilities"); + if ( + value.deadlineCatalog !== undefined && + value.deadlineCatalog !== "domain_deadlines_v1" + ) + throw new Error("Unknown idle deadline catalog"); + if (value.incomingWake !== undefined) { + const v = value.incomingWake; + if ( + !object(v) || + !exact(v, [ + "version", + "ownerId", + "generation", + "routes", + "githubBroker", + "apiKeyProof", + ]) || + v.version !== 1 || + typeof v.ownerId !== "string" || + !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test( + v.ownerId, + ) || + !digest(v.generation) || + !Array.isArray(v.routes) || + v.routes.length < 1 || + v.routes.length > 2 || + v.routes.some( + (r) => !["tenant-api-http-v2", "github-broker-events-v1"].includes(r), + ) || + new Set(v.routes).size !== v.routes.length || + [...v.routes].sort().join() !== v.routes.join() + ) + throw new Error("Invalid incoming wake capability"); + if (v.routes.includes("tenant-api-http-v2")) { + const proof = v.apiKeyProof; + if (!object(proof) || Object.keys(proof).length !== 2 || !exact(proof, ["version", "challenge"]) + || proof.version !== 1 || !digest(proof.challenge)) throw new Error("Invalid API wake challenge"); + } else if (v.apiKeyProof !== undefined) throw new Error("Unexpected API wake challenge"); + if (v.routes.includes("github-broker-events-v1")) { + const b = v.githubBroker; + if ( + !object(b) || + !exact(b, ["instanceId", "bindingsFingerprint"]) || + typeof b.instanceId !== "string" || + !/^[A-Za-z0-9_-]{1,200}$/.test(b.instanceId) || + !digest(b.bindingsFingerprint) + ) + throw new Error("Invalid GitHub wake binding"); + } else if (v.githubBroker !== undefined) + throw new Error("Unexpected GitHub wake binding"); + } + // Return a detached immutable snapshot; request callers cannot mutate a verified claim. + const parsed = JSON.parse(JSON.stringify(value)) as IdleCapabilities; + if (parsed.incomingWake) { + Object.freeze(parsed.incomingWake.routes); + if (parsed.incomingWake.apiKeyProof) Object.freeze(parsed.incomingWake.apiKeyProof); + if (parsed.incomingWake.githubBroker) + Object.freeze(parsed.incomingWake.githubBroker); + Object.freeze(parsed.incomingWake); + } + return Object.freeze(parsed); +} + +declare const verified: unique symbol; +export type VerifiedIdleCapabilities = IdleCapabilities & { + readonly [verified]: true; +}; diff --git a/server/src/services/idle-incoming-wake.test.ts b/server/src/services/idle-incoming-wake.test.ts new file mode 100644 index 00000000000..d432a3083fe --- /dev/null +++ b/server/src/services/idle-incoming-wake.test.ts @@ -0,0 +1,184 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { + readIdleIncomingWakeProof, + incomingWakeFingerprint, +} from "./idle-incoming-wake.js"; +import { registerConnectionEventWake } from "./connection-event-wake.js"; +import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js"; + +const ownerId = "00000000-0000-4000-8000-000000000001"; +function fixture(count: number) { + const endpoints = Array.from({ length: count }, (_, i) => ({ + endpoint: { + id: `endpoint-${i}`, + companyId: "company", + status: "active", + activatedAt: new Date(0), + botExternalId: `bot-${i}`, + providerAccountId: `app-${i}`, + setup: { + step: "complete", + github: { + cloudRegistrationId: String(i).padStart(43, "a"), + registrationStatus: "completed", + }, + }, + }, + connection: { id: `connection-${i}` }, + })); + const registrations = endpoints.map(({ endpoint }) => ({ + endpointId: endpoint.id, + companyId: endpoint.companyId, + userId: "owner", + cloudId: endpoint.setup.github.cloudRegistrationId, + })); + const tuples = registrations.map((row) => [ + `github-app:${row.cloudId}`, + row.companyId, + row.userId, + "github.bot", + `gateway:${row.cloudId}`, + row.cloudId, + row.cloudId, + ]); + const capability = { + incomingWake: { + version: 1, + ownerId, + generation: "a".repeat(64), + routes: ["github-broker-events-v1"], + githubBroker: { + instanceId: "fixture-instance", + bindingsFingerprint: incomingWakeFingerprint(tuples), + }, + }, + } as VerifiedIdleCapabilities; + return { endpoints, registrations, capability }; +} +/** Named DB boundary for cardinality/elapsed guards; PostgreSQL fixture covers real SQL separately. */ +function database(replies: Array<() => unknown[]>) { + let index = 0; + const select = vi.fn(() => { + const reply = replies[index++]; + const query = { + from: () => query, + innerJoin: () => query, + where: () => query, + limit: async () => reply?.() ?? [], + }; + return query; + }); + const execute = vi.fn(async () => []); + return { + db: { select, execute } as unknown as Pick, + select, + execute, + }; +} +const stops: Array<() => void> = []; +function input( + db: object, + capabilities: VerifiedIdleCapabilities, + now?: () => number, +) { + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID", "fixture-instance"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY", "fixture"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY", "fixture"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT", "test"); + vi.stubEnv("PAPERCLIP_CLOUD_CONNECTOR_BASE_URL", "http://127.0.0.1"); + stops.push(registerConnectionEventWake(db, () => {})); + return { ownerId, receiverOwner: db, capabilities, now }; +} +afterEach(() => { + for (const stop of stops.splice(0)) stop(); + vi.unstubAllEnvs(); +}); + +describe("bounded incoming wake owner projection", () => { + it("uses three bounded selects for a thousand bots and retains exact per-bot authority", async () => { + const f = fixture(1000), + sql = database([() => [], () => f.endpoints, () => f.registrations]); + const result = await readIdleIncomingWakeProof( + sql.db, + input(sql.db, f.capability), + ); + expect(sql.select).toHaveBeenCalledTimes(3); + expect(sql.execute).toHaveBeenCalledTimes(4); + expect(result?.endpointIds).toHaveLength(1000); + expect(result?.connectionIds).toHaveLength(1000); + }); + it.each(["duplicate", "foreign", "missing"])( + "rejects %s registration authority", + async (fault) => { + const f = fixture(1); + let registrations = f.registrations; + if (fault === "duplicate") + registrations = [...registrations, ...registrations]; + if (fault === "foreign") + registrations = registrations.map((r) => ({ + ...r, + companyId: "foreign", + })); + if (fault === "missing") registrations = []; + const sql = database([() => [], () => f.endpoints, () => registrations]); + await expect( + readIdleIncomingWakeProof(sql.db, input(sql.db, f.capability)), + ).rejects.toThrow("registration authority"); + }, + ); + it("stops the complete inspection when its monotonic budget expires before another query", async () => { + const f = fixture(1); + let now = 0; + const sql = database([ + () => { + now = 2001; + return []; + }, + () => f.endpoints, + () => f.registrations, + ]); + await expect( + readIdleIncomingWakeProof( + sql.db, + input(sql.db, f.capability, () => now), + ), + ).rejects.toThrow("budget exhausted"); + expect(sql.select).toHaveBeenCalledTimes(1); + expect(sql.execute).toHaveBeenCalledTimes(1); + }); + it("fails closed at row limits before expanding work or issuing registration queries", async () => { + const f = fixture(1001), + sql = database([() => [], () => f.endpoints, () => f.registrations]); + await expect( + readIdleIncomingWakeProof(sql.db, input(sql.db, f.capability)), + ).rejects.toThrow("Too many GitHub endpoints"); + expect(sql.select).toHaveBeenCalledTimes(2); + }); +}); + +describe("bounded API admission projection", () => { + const api = { incomingWake: { version: 1, ownerId, generation: "a".repeat(64), routes: ["tenant-api-http-v2"], + apiKeyProof: { version: 1, challenge: "c".repeat(64) } } } as VerifiedIdleCapabilities; + it.each(["overflow", "duplicate", "malformed"])("refuses complete eligibility for %s key state", async kind => { + vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "fixture-stack"); + const row = { keyHash: "a".repeat(64), expiresAt: null }; + const rows = kind === "overflow" ? Array.from({ length: 1001 }, () => row) + : kind === "duplicate" ? [row, row] : [{ keyHash: "not-a-digest", expiresAt: null }]; + const execute = vi.fn().mockResolvedValueOnce([]).mockResolvedValueOnce(rows); + await expect(readIdleIncomingWakeProof({ execute } as unknown as Pick, + { ownerId, receiverOwner: {}, capabilities: api })).rejects.toThrow(); + expect(execute).toHaveBeenCalledTimes(2); + }); + it("has an explicit complete empty-key receipt and shares the inspection time budget", async () => { + vi.stubEnv("PAPERCLIP_CLOUD_STACK_ID", "fixture-stack"); + const execute = vi.fn(async () => []); + const proof = await readIdleIncomingWakeProof({ execute } as unknown as Pick, + { ownerId, receiverOwner: {}, capabilities: api }); + expect(proof?.apiWakeAdmission?.keys).toEqual([]); + let clock = 0; + const slow = vi.fn(async () => { clock += 1100; return []; }); + await expect(readIdleIncomingWakeProof({ execute: slow } as unknown as Pick, + { ownerId, receiverOwner: {}, capabilities: api, now: () => clock })).rejects.toThrow(/budget exhausted/); + }); +}); diff --git a/server/src/services/idle-incoming-wake.ts b/server/src/services/idle-incoming-wake.ts new file mode 100644 index 00000000000..f0aa26e1b13 --- /dev/null +++ b/server/src/services/idle-incoming-wake.ts @@ -0,0 +1,359 @@ +import { apiWakeFingerprint, createApiWakeAdmission, type ApiWakeAdmission } from "./idle-api-wake-admission.js"; +import { createHash } from "node:crypto"; +import { and, eq, inArray, sql, type SQL } from "drizzle-orm"; +import { + chatEndpoints, + chatGitHubRegistrations, + connectionGrants, + toolConnections, + type Db, +} from "@paperclipai/db"; +import type { + IdleIncomingWakeCapability, + VerifiedIdleCapabilities, +} from "./idle-incoming-wake-capability.js"; +import { githubConnectionEventBindings } from "./github-connection-events.js"; +import { paperclipCloudConnectorConfigFromEnv } from "./paperclip-cloud-connector.js"; +import { hasConnectionEventWakeReceiver } from "./connection-event-wake.js"; + +export interface IdleIncomingWakeProof { + capability: IdleIncomingWakeCapability; + apiWakeAdmission?: ApiWakeAdmission; + connectionIds: string[]; + endpointIds: string[]; + /** A broker enrollment cannot discharge these credential continuity obligations. */ + requiredContinuityConnectionIds: string[]; +} +export const incomingWakeFingerprint = (bindings: string[][]) => + createHash("sha256") + .update( + JSON.stringify( + bindings + .map((b) => [...b]) + .sort((a, b) => + JSON.stringify(a) < JSON.stringify(b) + ? -1 + : JSON.stringify(a) > JSON.stringify(b) + ? 1 + : 0, + ), + ), + ) + .digest("hex"); + +/** Read current binding authority in the safety transaction. No network call, renewal or write. */ +export async function readIdleIncomingWakeProof( + tx: Pick, + input: { + capabilities?: VerifiedIdleCapabilities; + ownerId: string; + receiverOwner: object; + now?: () => number; + }, +): Promise { + const capability = input.capabilities?.incomingWake; + if (!capability) return; + if (capability.ownerId !== input.ownerId) + throw new Error("Incoming wake owner changed"); + const proof: IdleIncomingWakeProof = { + capability, + connectionIds: [], + endpointIds: [], + requiredContinuityConnectionIds: [], + }; + const monotonic = input.now ?? (() => performance.now()); + const started = monotonic(); + const remaining = () => { + const ms = 2_000 - (monotonic() - started); + if (!Number.isFinite(ms) || ms <= 0) + throw new Error("Incoming wake inspection budget exhausted"); + return Math.max(1, Math.floor(ms)); + }; + const beforeRead = async () => { + // One shared owner budget, not five seconds multiplied by the row count. + await tx.execute( + sql`SELECT set_config('statement_timeout', ${`${remaining()}ms`}, true)`, + ); + remaining(); + }; + if (capability.routes.includes("tenant-api-http-v2")) { + const stackId = process.env.PAPERCLIP_CLOUD_STACK_ID; + if (!stackId || !capability.apiKeyProof) throw new Error("API wake challenge is missing"); + await beforeRead(); + const rows = await tx.execute<{ keyHash: string; expiresAt: Date | null }>(sql` + SELECT key_hash AS "keyHash", NULL::timestamptz AS "expiresAt" FROM agent_api_keys WHERE revoked_at IS NULL + UNION ALL SELECT key_hash AS "keyHash", expires_at AS "expiresAt" FROM board_api_keys + WHERE revoked_at IS NULL AND (expires_at IS NULL OR expires_at > now()) LIMIT 1001`); + remaining(); + if (rows.length > 1000) throw new Error("Too many API keys to certify wake admission"); + proof.apiWakeAdmission = createApiWakeAdmission({ stackId, ownerId: input.ownerId, + capabilityGeneration: capability.generation, challenge: capability.apiKeyProof.challenge, + keys: rows.map(row => ({ fingerprint: apiWakeFingerprint({ stackId, ownerId: input.ownerId, + challenge: capability.apiKeyProof!.challenge, keyHash: row.keyHash }), + expiresAt: row.expiresAt === null ? null : new Date(row.expiresAt).toISOString() })), + }); + } + if (!capability.routes.includes("github-broker-events-v1")) { + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + remaining(); + return proof; + } + const broker = capability.githubBroker; + const config = paperclipCloudConnectorConfigFromEnv(); + if ( + !broker || + config?.instanceId !== broker.instanceId || + !hasConnectionEventWakeReceiver(input.receiverOwner) + ) + throw new Error("Incoming GitHub receiver is not bound"); + await beforeRead(); + const rows = await tx + .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`exists (select 1 from companies where id=${connectionGrants.companyId} and status <> 'archived')`, + ), + ) + .limit(1001); + if (rows.length > 1000) + throw new Error("Too many GitHub bindings to inspect"); + remaining(); + let bindingCount = 0; + for (const { grant, connection } of rows) { + const github = grant.providerTenant?.github; + if ( + !github || + !Array.isArray(github.installationIds) || + !github.installationIds.length || + github.installationIds.length > 1000 || + github.bindingSync || + github.continuityRetry || + github.installationIds.some( + (id) => typeof id !== "string" || !/^[0-9]+$/.test(id), + ) + ) + throw new Error("Unsettled GitHub authority"); + bindingCount += github.installationIds.length; + if (bindingCount > 1000) + throw new Error("Too many GitHub bindings to inspect"); + proof.requiredContinuityConnectionIds.push(connection.id); + } + const bindings = githubConnectionEventBindings(rows); + if ( + bindings.length !== + rows.reduce( + (n, r) => n + r.grant.providerTenant!.github!.installationIds.length, + 0, + ) + ) + throw new Error("Incomplete GitHub binding"); + const tuples = bindings.map((b) => [ + b.id, + b.companyId, + b.subject, + "github.code", + b.installationId, + b.connectionId, + b.grantId, + ]); + await beforeRead(); + const endpoints = await tx + .select({ endpoint: chatEndpoints, connection: toolConnections }) + .from(chatEndpoints) + .innerJoin( + toolConnections, + and( + eq(toolConnections.id, chatEndpoints.connectionId), + eq(toolConnections.companyId, chatEndpoints.companyId), + ), + ) + .where( + and( + eq(chatEndpoints.provider, "github"), + eq(chatEndpoints.status, "active"), + eq(toolConnections.status, "active"), + eq(toolConnections.enabled, true), + eq(toolConnections.transport, "chat_sdk"), + eq(toolConnections.connectionPurpose, "channel"), + sql`exists (select 1 from companies where id=${chatEndpoints.companyId} and status <> 'archived')`, + ), + ) + .limit(1001); + if (endpoints.length > 1000) + throw new Error("Too many GitHub endpoints to inspect"); + remaining(); + const brokerEndpointIds = endpoints + .filter( + ({ endpoint }) => + typeof endpoint.setup.github?.cloudRegistrationId === "string", + ) + .map(({ endpoint }) => endpoint.id); + const registrationUsers = new Map(); + if (brokerEndpointIds.length) { + await beforeRead(); + const registrations = await tx + .select({ + endpointId: chatGitHubRegistrations.endpointId, + companyId: chatGitHubRegistrations.companyId, + userId: chatGitHubRegistrations.userId, + cloudId: sql`${chatGitHubRegistrations.handoff}->>'cloudId'`, + }) + .from(chatGitHubRegistrations) + .where( + and( + inArray(chatGitHubRegistrations.endpointId, brokerEndpointIds), + eq(chatGitHubRegistrations.status, "completed"), + ), + ) + .limit(10001); + remaining(); + if (registrations.length > 10000) + throw new Error("Too many GitHub registrations to inspect"); + for (const registration of registrations) { + const key = JSON.stringify([ + registration.endpointId, + registration.companyId, + registration.cloudId, + ]); + const users = registrationUsers.get(key) ?? []; + users.push(registration.userId); + registrationUsers.set(key, users); + } + } + for (const { endpoint, connection } of endpoints) { + const registration = endpoint.setup.github?.cloudRegistrationId; + if (!registration || !/^[A-Za-z0-9_-]{43}$/.test(registration)) continue; + if ( + endpoint.setup.step !== "complete" || + endpoint.setup.github?.registrationStatus !== "completed" || + endpoint.setup.github.initialSetupPending || + endpoint.setup.github.initialRepositoryImportPending || + !endpoint.activatedAt || + !endpoint.botExternalId || + !endpoint.providerAccountId + ) + throw new Error("GitHub bot setup has unfinished work"); + const users = registrationUsers.get( + JSON.stringify([endpoint.id, endpoint.companyId, registration]), + ); + if (users?.length !== 1 || !users[0]) + throw new Error("GitHub registration authority changed"); + tuples.push([ + `github-app:${registration}`, + endpoint.companyId, + users[0], + "github.bot", + `gateway:${registration}`, + registration, + registration, + ]); + proof.connectionIds.push(connection.id); + proof.endpointIds.push(endpoint.id); + } + remaining(); + if (tuples.length > 1000) + throw new Error("Too many GitHub bindings to inspect"); + if (incomingWakeFingerprint(tuples) !== broker.bindingsFingerprint) + throw new Error("GitHub wake bindings changed"); + remaining(); + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + remaining(); + return proof; +} + +/** Exact domain predicates selected by stable inventory reason, never SQL-text identity. */ +export function incomingWakeWorkQuery( + reason: string, + proof?: IdleIncomingWakeProof, + continuityConnectionIds: readonly string[] = [], +): SQL | undefined { + const connectionIds = [ + ...(proof?.connectionIds ?? []), + ...(proof?.requiredContinuityConnectionIds.filter((id) => + continuityConnectionIds.includes(id), + ) ?? []), + ]; + if (reason === "agent_api_keys") + return proof?.apiWakeAdmission !== undefined + ? sql`SELECT 1 WHERE false` + : sql`SELECT 1 FROM agent_api_keys WHERE revoked_at IS NULL`; + if (reason === "board_api_keys") + return proof?.apiWakeAdmission !== undefined + ? sql`SELECT 1 WHERE false` + : sql`SELECT 1 FROM board_api_keys WHERE revoked_at IS NULL + AND (expires_at IS NULL OR expires_at > now())`; + if (reason === "chat_endpoints") { + const ids = proof?.endpointIds ?? []; + // Only the supported GitHub webhook path gains a removed-history proof. + // Archival commits before cleanup, so status alone cannot discharge work. + return sql`SELECT 1 FROM chat_endpoints endpoint WHERE (((endpoint.provider = 'github' + AND endpoint.status = 'archived' AND endpoint.archived_at IS NOT NULL + AND jsonb_typeof(endpoint.setup) = 'object' + AND CASE WHEN jsonb_typeof(endpoint.setup->'runtimeGeneration') = 'number' + THEN (endpoint.setup->>'runtimeGeneration')::numeric >= 1 + AND trunc((endpoint.setup->>'runtimeGeneration')::numeric) = (endpoint.setup->>'runtimeGeneration')::numeric + ELSE false END + AND EXISTS (SELECT 1 FROM tool_connections removed_connection + WHERE removed_connection.id = endpoint.connection_id AND removed_connection.company_id = endpoint.company_id + AND removed_connection.connection_purpose = 'channel' AND removed_connection.transport = 'chat_sdk' + AND removed_connection.status = 'archived' AND removed_connection.enabled = false + AND removed_connection.credential_secret_refs = '[]'::jsonb) + AND NOT EXISTS (SELECT 1 FROM chat_endpoint_leases lease WHERE lease.endpoint_id = endpoint.id) + AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id) + AND NOT EXISTS (SELECT 1 FROM chat_github_registrations registration WHERE registration.endpoint_id = endpoint.id + AND (registration.company_id = endpoint.company_id AND registration.status = 'completed' + AND registration.consumed_at IS NOT NULL) IS NOT TRUE)) + OR (endpoint.provider = 'slack' + AND endpoint.status = 'draft' AND endpoint.setup->>'step' = 'provider_setup' + AND (endpoint.setup - ARRAY['step', 'command']) = '{}'::jsonb + AND (endpoint.setup->'command' IS NULL OR endpoint.setup->'command' = 'null'::jsonb + OR jsonb_typeof(endpoint.setup->'command') = 'string') + AND endpoint.activated_at IS NULL AND endpoint.provider_account_id IS NULL + AND endpoint.bot_external_id IS NULL AND endpoint.bot_username IS NULL + AND endpoint.last_event_at IS NULL AND endpoint.last_publication_at IS NULL + AND endpoint.health_message IS NULL AND endpoint.last_error IS NULL + AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id) + AND EXISTS (SELECT 1 FROM tool_connections draft_connection + WHERE draft_connection.id = endpoint.connection_id AND draft_connection.company_id = endpoint.company_id + AND draft_connection.connection_purpose = 'channel' AND draft_connection.transport = 'chat_sdk' + AND draft_connection.status = 'draft' AND draft_connection.enabled = false))) IS NOT TRUE) AND ${ + ids.length + ? sql`endpoint.id NOT IN (${sql.join( + ids.map((id) => sql`${id}::uuid`), + sql`, `, + )})` + : sql`true` + }`; + } + if (reason === "tool_connections") { + const ids = connectionIds; + return sql`SELECT 1 FROM tool_connections c WHERE (c.status <> 'archived' AND ( + (c.status = 'draft' AND c.enabled = false) + OR (c.status = 'active' AND c.connection_purpose = 'ai' AND c.transport = 'runtime_auth') + OR (c.status = 'active' AND c.connection_purpose = 'tool' AND c.transport = 'mcp_remote' + AND NOT EXISTS (SELECT 1 FROM connection_grants g WHERE g.connection_id = c.id + AND (g.company_id <> c.company_id OR g.provider_tenant->'github' IS NOT NULL))) + ) IS NOT TRUE) AND ${ + ids.length + ? sql`c.id NOT IN (${sql.join( + ids.map((id) => sql`${id}::uuid`), + sql`, `, + )})` + : sql`true` + }`; + } + return undefined; +} diff --git a/server/src/services/idle-local-work.ts b/server/src/services/idle-local-work.ts index 6034d6515d8..70b8cd6a921 100644 --- a/server/src/services/idle-local-work.ts +++ b/server/src/services/idle-local-work.ts @@ -1,3 +1,4 @@ +import { idleApiActivity } from "./idle-api-activity.js"; import { promises as fs } from "node:fs"; import path from "node:path"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; @@ -83,7 +84,7 @@ export async function readIdleLocalWork(options: { backupCheckpoint?: boolean; d } let probes: IdleLocalWork; try { - const local = [...localWorkProbes].map(probe => probe()); + const local = [idleApiActivity.active() ? "present" : "none", ...[...localWorkProbes].map(probe => probe())]; probes = local.some(state => state !== "none" && state !== "present") ? "unknown" : local.includes("present") ? "present" : "none"; } catch { probes = "unknown"; } diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index 96d06daae0e..265fb61a6fd 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -1,3 +1,6 @@ +import type { ApiWakeAdmission } from "./idle-api-wake-admission.js"; +import { readIdleIncomingWakeProof, incomingWakeWorkQuery, type IdleIncomingWakeProof } from "./idle-incoming-wake.js"; +import type { VerifiedIdleCapabilities, IdleIncomingWakeCapability } from "./idle-incoming-wake-capability.js"; import { idleWorkSnapshot } from "./task-admission.js"; import { readIdleLocalWork, type IdleLocalWork } from "./idle-local-work.js"; import { readIdleRoutineSchedule, type ScheduledWake } from "./idle-scheduled-work.js"; @@ -38,10 +41,12 @@ export interface IdleSleepDrainStatus { } export interface IdleSleepSafety { - version: 1 | 2; + version: 1 | 2 | 3; backgroundWork: "none" | "present" | "unknown"; scheduledWake?: ScheduledWake; diagnostics?: IdleSleepDiagnostics; + incomingWake?: IdleIncomingWakeCapability; + apiWakeAdmission?: ApiWakeAdmission; } export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) => Promise<{ @@ -84,9 +89,8 @@ const WORK_CHECKS = [ workCheck("prompt_injection_invocations", `SELECT 1 FROM prompt_injection_invocations WHERE status = 'running'`), // Persistent machine credentials can receive work without a human opening // the instance. They need an ingress wake contract before we can sleep. - workCheck("agent_api_keys", `SELECT 1 FROM agent_api_keys WHERE revoked_at IS NULL`), - workCheck("board_api_keys", `SELECT 1 FROM board_api_keys WHERE revoked_at IS NULL - AND (expires_at IS NULL OR expires_at > now())`), + workCheck("agent_api_keys", incomingWakeWorkQuery("agent_api_keys")!), + workCheck("board_api_keys", incomingWakeWorkQuery("board_api_keys")!), // Coalescing consumes a request into another run. Retain it as work until // both the request and that exact company/agent's run have finished. workCheck("agent_wakeup_requests", `SELECT 1 FROM agent_wakeup_requests w WHERE @@ -137,20 +141,7 @@ const WORK_CHECKS = [ // An unconfigured Slack draft has no provider identity or inbound runtime. // Setup still wakes through normal browser admission. Retain every enabled, // previously active, partially connected, foreign, or unsupported endpoint. - workCheck("chat_endpoints", `SELECT 1 FROM chat_endpoints endpoint WHERE (endpoint.provider = 'slack' - AND endpoint.status = 'draft' AND endpoint.setup->>'step' = 'provider_setup' - AND (endpoint.setup - ARRAY['step', 'command']) = '{}'::jsonb - AND (endpoint.setup->'command' IS NULL OR endpoint.setup->'command' = 'null'::jsonb - OR jsonb_typeof(endpoint.setup->'command') = 'string') - AND endpoint.activated_at IS NULL AND endpoint.provider_account_id IS NULL - AND endpoint.bot_external_id IS NULL AND endpoint.bot_username IS NULL - AND endpoint.last_event_at IS NULL AND endpoint.last_publication_at IS NULL - AND endpoint.health_message IS NULL AND endpoint.last_error IS NULL - AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id) - AND EXISTS (SELECT 1 FROM tool_connections draft_connection - WHERE draft_connection.id = endpoint.connection_id AND draft_connection.company_id = endpoint.company_id - AND draft_connection.connection_purpose = 'channel' AND draft_connection.transport = 'chat_sdk' - AND draft_connection.status = 'draft' AND draft_connection.enabled = false)) IS NOT TRUE`), + workCheck("chat_endpoints", incomingWakeWorkQuery("chat_endpoints")!), workCheck("routines", `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`), workCheck("routine_runs", `SELECT 1 FROM routine_runs WHERE status NOT IN ('completed', 'failed', 'skipped', 'cancelled')`), workCheck("plugin_jobs", `SELECT 1 FROM plugin_jobs WHERE status <> 'paused'`), @@ -159,13 +150,7 @@ const WORK_CHECKS = [ // Saved outbound MCP configuration and disabled drafts have no independent // listener. Keep local processes, channels and GitHub webhook bindings awake; // accepted calls, credential recovery and delivery work have separate fences. - workCheck("tool_connections", `SELECT 1 FROM tool_connections c WHERE c.status <> 'archived' AND ( - (c.status = 'draft' AND c.enabled = false) - OR (c.status = 'active' AND c.connection_purpose = 'ai' AND c.transport = 'runtime_auth') - OR (c.status = 'active' AND c.connection_purpose = 'tool' AND c.transport = 'mcp_remote' - AND NOT EXISTS (SELECT 1 FROM connection_grants g WHERE g.connection_id = c.id - AND (g.company_id <> c.company_id OR g.provider_tenant->'github' IS NOT NULL))) - ) IS NOT TRUE`), + workCheck("tool_connections", incomingWakeWorkQuery("tool_connections")!), // An explicit board escalation has no automatic retry. Retain the visible // action and all execution/cleanup checks while the board decides what to do. workCheck("issue_recovery_actions", `SELECT 1 FROM issue_recovery_actions a WHERE a.status NOT IN ('resolved', 'cancelled') @@ -294,6 +279,8 @@ function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus * Only a caller holding the exact idle owner may receive an empty report. */ export interface IdleSleepSafetyOptions { + incomingWake?: VerifiedIdleCapabilities; + deadlineCapability?: VerifiedIdleCapabilities; collectDiagnostics?: true; } @@ -310,12 +297,15 @@ export async function readIdleSleepSafety( scheduledIssueMonitors = false, options: IdleSleepSafetyOptions = {}, ): Promise { + let incomingProof: IdleIncomingWakeProof | undefined; const diagnostic = options.collectDiagnostics ? createIdleSleepDiagnostics() : undefined; const report = (backgroundWork: IdleLocalWork, scheduledWake?: ScheduledWake): IdleSleepSafety => { const diagnostics = diagnostic?.report(); if (backgroundWork === "none" && diagnostics?.checks.some(check => check.state === "unknown" || check.state === "not_checked")) backgroundWork = "unknown"; - return { version: backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork, + return { version: backgroundWork === "none" && incomingProof ? 3 : backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork, + ...(backgroundWork === "none" && incomingProof ? { incomingWake: incomingProof.capability, + ...(incomingProof.apiWakeAdmission ? { apiWakeAdmission: incomingProof.apiWakeAdmission } : {}) } : {}), ...(backgroundWork === "none" && scheduledWake ? { scheduledWake } : {}), ...(diagnostics ? { diagnostics } : {}) }; }; @@ -360,9 +350,14 @@ export async function readIdleSleepSafety( let scheduledWake: ScheduledWake | undefined; const blocked = await db.transaction(async (tx) => { await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + if (options.incomingWake?.incomingWake) { + if (!ownerId || ownerId !== before.ownerId) return undefined; + incomingProof = await readIdleIncomingWakeProof(tx, { capabilities: options.incomingWake, ownerId, receiverOwner: db }); + } const checks = WORK_CHECKS.filter(check => (!scheduledRoutines || check.reason !== "routines") - && (!scheduledIssueMonitors || check.reason !== "issues")); + && (!scheduledIssueMonitors || check.reason !== "issues")) + .map(check => ({ ...check, query: incomingWakeWorkQuery(check.reason, incomingProof) ?? check.query })); checks.push(workCheck("agents.company_hold", sql`SELECT 1 FROM agents JOIN companies ON companies.id = agents.company_id WHERE ${companyHoldMismatch()}`)); checks.push(workCheck("heartbeat_runs.budget_cancellation", sql`SELECT 1 FROM heartbeat_runs WHERE ${pendingBudgetRunCancellationPredicate()}`)); @@ -447,6 +442,9 @@ export async function readIdleSleepSafety( diagnostic?.record("inspection.hold", "unknown"); return report("present"); } + // Legacy alarm kinds cannot be silently combined with a new ingress proof. + // The negotiated deadline catalog supplies the composed v3 alarm separately. + if (incomingProof && scheduledWake) return report("unknown"); return report(local, scheduledWake); } catch { // Missing migrations, unknown state, malformed configuration, timeouts and