From b21a1d0f0edfb80d8e417b398e712c4a0422bc75 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 2 Nov 2026 13:30:07 -0600 Subject: [PATCH] Compose negotiated deadline receipts and exercise the original domain owners Require an unchanged catalog receipt even when no alarm remains. Preserve active work, verify stored credential ownership, and cover real plugin dispatch, trace cleanup and credential continuity producers. Co-Authored-By: Paperclip --- doc/idle-domain-deadlines.md | 24 +- maps/agent-access-map/instance-operations.md | 12 +- maps/agent-access-map/routines.md | 3 + .../fixtures/plugin-worker-deadline.cjs | 13 + .../idle-domain-deadlines-postgres.test.ts | 264 +++++++++++++++--- .../src/__tests__/plugin-idle-manager.test.ts | 28 +- server/src/routes/instance-settings.ts | 12 +- server/src/services/connection-credentials.ts | 40 ++- .../services/heartbeat/idle-timer-schedule.ts | 3 - server/src/services/heartbeat/queue.ts | 11 +- server/src/services/heartbeat/timer-policy.ts | 4 - .../src/services/idle-deadline-proof.test.ts | 1 - server/src/services/idle-deadline-proof.ts | 8 +- server/src/services/idle-deadline-types.ts | 8 +- server/src/services/idle-sleep-safety.ts | 65 ++++- server/src/services/plugin-idle-catalog.ts | 1 - .../src/services/plugin-job-idle-schedule.ts | 1 - server/src/services/plugin-worker-manager.ts | 3 +- .../services/provider-trace-idle-schedule.ts | 2 - .../services/provider-trace-retention-work.ts | 2 - .../tool-access/github-grant-continuity.ts | 14 +- 21 files changed, 424 insertions(+), 95 deletions(-) create mode 100644 server/src/__tests__/fixtures/plugin-worker-deadline.cjs diff --git a/doc/idle-domain-deadlines.md b/doc/idle-domain-deadlines.md index c9bf863f862..73b537ee223 100644 --- a/doc/idle-domain-deadlines.md +++ b/doc/idle-domain-deadlines.md @@ -6,7 +6,12 @@ Cloud requests `scheduledWake=domain-deadlines-v1` and signs the fixed catalog i A query parameter or an ordinary credential cannot grant this capability. The response uses idle-safety version 3 with a scheduled proof of version 2, kind `domain_deadlines`, catalog `domain_deadlines_v1`, opaque generation and -canonical UTC `nextWakeAt`. Legacy requests and versions keep their prior rules. +canonical UTC `nextWakeAt`. Every requested catalog also returns an outer +`deadlineProof: { catalog: "domain_deadlines_v1", generation }`, including when +there is no future alarm. Any scheduled proof must use that same generation. +Cloud rejects an omitted receipt, an older ingress-only version 3 response, and +a changed no-alarm generation on the final read. Legacy requests and versions +keep their prior rules. The proof combines existing routine and issue-monitor obligations with: @@ -25,8 +30,10 @@ The proof combines existing routine and issue-monitor obligations with: - Personal GitHub grant continuity, using the owner's token-expiry-minus-one-hour and 30-day access-refresh rules. Broker enrollment alone is insufficient: exact ingress binding and actual grant deadline must both be proven. Missing - credentials, refresh leases, binding synchronization and unresolved retry - records remain blockers. + credentials, inactive/missing owner-scoped vault definitions or current token + versions, refresh leases, binding synchronization and unresolved retry records + remain blockers. Inspection reads credential ownership/version metadata and + never resolves plaintext token material. Pending watchdog first-run grace (15 seconds) and automatic merge-confirmation checks (five minutes) remain awake. They provide no useful safe sleep window @@ -45,12 +52,15 @@ ordinary routine/monitor proof, exact domain rows and plugin worker identities. Any changed state, cancellation, moved deadline, owner replacement or restart invalidates the old proof. Unknown or malformed states fail closed. -The additional domain projection is bounded to 64 SQL reads, two seconds of -monotonic inspection time, 128 eligible timer candidates and 10,000 rows per +The additional domain projection is bounded to 64 data reads, a two-second monotonic budget for +starting further reads, 128 eligible timer candidates and 10,000 rows per family, while reserving 60 seconds of hold headroom. Overflow refuses sleep; it never truncates the proof or changes a schedule. These limits supplement the -transaction's statement timeout. A query already in progress remains bounded -by that timeout, and no subsequent query starts after budget exhaustion. +transaction's statement timeout. A query already in progress may still take its five-second statement timeout; +SQL execution can therefore finish about seven seconds after the projection +starts, plus bounded local projection/driver overhead. No subsequent data query +starts after budget exhaustion. This is not a strict two-second wall-clock +promise. The final hold and generation checks remain mandatory. Cloud persists one existing registry alarm under its current sleep lease before final proof validation and provider stop. It compares every proof field on the diff --git a/maps/agent-access-map/instance-operations.md b/maps/agent-access-map/instance-operations.md index cebe15a8df6..2c547c620b1 100644 --- a/maps/agent-access-map/instance-operations.md +++ b/maps/agent-access-map/instance-operations.md @@ -173,9 +173,17 @@ API: `present` — representative operation IDs: 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) + bounded host wake. Personal GitHub also requires the signed domain deadline + receipt and its actual stored-credential continuity proof. See [protocol scope](../../doc/idle-incoming-wake.md) and [local auth/PostgreSQL evidence](../../server/src/__tests__/idle-incoming-wake.test.ts). +- `GET /api/instance/task-drain?idleSleepSafety=1&scheduledWake=domain-deadlines-v1` + additionally requires the signed fixed deadline catalog. The version-3 + `deadlineProof` receipt is mandatory even without a future alarm. It binds + actual timer, plugin, trace expiry and credential continuity state to the + owned hold. Unknown/due/recovery work and pending watchdog/merge polling + remain blocked. It grants no business execution authority. See + [domain protocol](../../doc/idle-domain-deadlines.md); local source and + fixture checks do not establish deployed sleep or external wake. - `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/agent-access-map/routines.md b/maps/agent-access-map/routines.md index 01fb9b378f2..0212749f79b 100644 --- a/maps/agent-access-map/routines.md +++ b/maps/agent-access-map/routines.md @@ -77,6 +77,9 @@ idle hold. It requires signed Cloud control authority; company agents and tenant admin tokens cannot use it. It does not grant routine management or execution authority. See the [hosting protocol](../../doc/idle-sleep-safety.md#scheduled-routine-wake-protocol) and [route authorization tests](../../server/src/__tests__/instance-settings-routes.test.ts). +The signed `scheduledWake=domain-deadlines-v1` catalog can combine this routine +obligation with other supported domain deadlines. Its generation receipt is +mandatory even when no future alarm remains. See [domain scope](../../doc/idle-domain-deadlines.md). These are local contract checks, not a live provider sleep/wake qualification. ## `run-now` diff --git a/server/src/__tests__/fixtures/plugin-worker-deadline.cjs b/server/src/__tests__/fixtures/plugin-worker-deadline.cjs new file mode 100644 index 00000000000..a60a6f735a8 --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-deadline.cjs @@ -0,0 +1,13 @@ +void Promise.all([ + import("../../../../packages/plugins/sdk/src/define-plugin.ts"), + import("../../../../packages/plugins/sdk/src/worker-rpc-host.ts"), +]).then(([{ definePlugin }, { startWorkerRpcHost }]) => { + startWorkerRpcHost({ plugin: definePlugin({ + async setup(ctx) { + ctx.jobs.register("hourly", async job => { + await ctx.state.set({ scopeKind: "instance", stateKey: job.runId }, { runId: job.runId, scheduledAt: job.scheduledAt }); + }); + }, + async onIdleDrain() { return "none"; }, + }) }); +}); diff --git a/server/src/__tests__/idle-domain-deadlines-postgres.test.ts b/server/src/__tests__/idle-domain-deadlines-postgres.test.ts index aad1b98eb5a..04d77e63fc9 100644 --- a/server/src/__tests__/idle-domain-deadlines-postgres.test.ts +++ b/server/src/__tests__/idle-domain-deadlines-postgres.test.ts @@ -1,9 +1,29 @@ -import { pluginIdleManifestGeneration } from "../services/plugin-idle-catalog.js"; +import { createGitHubGrantContinuity, readIdleGitHubGrantContinuity } from "../services/tool-access/github-grant-continuity.js"; +import { createOAuthCredentialStore } from "../services/tool-access/oauth/credential-store.js"; +import { createOAuthRefreshLeases } from "../services/tool-access/oauth/refresh-leases.js"; +import { createOAuthGrantRefresh } from "../services/tool-access/oauth/grant-refresh.js"; +import { secretService } from "../services/secrets.js"; +import { writeConnectionCredential } from "../services/connection-credentials.js"; +import type { PaperclipCloudConnector } from "../services/paperclip-cloud-connector.js"; +import { GITHUB_CONNECTOR_PROFILES } from "@paperclipai/shared"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { createRequire } from "node:module"; +import { setTimeout as delay } from "node:timers/promises"; +import { createPluginWorkerManager, type PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { createPluginJobScheduler } from "../services/plugin-job-scheduler.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { pluginStateStore } from "../services/plugin-state-store.js"; +import { startTaskDrain, stopTaskDrain, beginIdleTrackedWork } from "../services/task-admission.js"; +import { providerTraceStore } from "../services/provider-trace-store.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; -import { agents, companies, heartbeatRuns, plugins, pluginJobs, pluginJobRuns, providerTraceRecords, +import { agents, companies, heartbeatRuns, plugins, pluginJobs, pluginJobRuns, pluginState, providerTraceRecords, issues, issueWatchdogs, issueThreadInteractions, toolApplications, toolConnections, connectionGrants, companySecrets, applyPendingMigrations, createDb, type Db } from "@paperclipai/db"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { readIdleDeadlineProof } from "../services/idle-deadline-proof.js"; @@ -16,30 +36,46 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu (support.supported ? describe : describe.skip)("durable domain deadline snapshot with PostgreSQL", () => { let temporary: Awaited> | undefined; let db: Db; + let directory: string; beforeAll(async () => { if (external) await applyPendingMigrations(external); else temporary = await startEmbeddedPostgresTestDatabase("idle-domain-deadlines-"); db = createDb(external ?? temporary!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + directory = await fs.mkdtemp(path.join(os.tmpdir(), "idle-deadline-producers-")); + vi.stubEnv("PAPERCLIP_CONFIG", path.join(directory, "config.json")); + vi.stubEnv("PROVIDER_TRACE_BASE_PATH", path.join(directory, "traces")); + vi.stubEnv("PAPERCLIP_STORAGE_PROVIDER", "local_disk"); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "d".repeat(64)); }, 60_000); - afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + afterAll(async () => { vi.useRealTimers(); vi.unstubAllEnvs(); await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); + if (directory) await fs.rm(directory, { recursive: true, force: true }); }); - it("reconstructs timers and plugin schedules, preserves actual claims, and retains capture/cleanup debt", async () => { + it("projects durable timers across client reconnection and executes the original plugin and trace owners", async () => { const now = Date.now(), companyId = randomUUID(), agentId = randomUUID(), pluginId = randomUUID(), runId = randomUUID(); - const hold = { ownerId: randomUUID(), expiresAt: now + 300_000 }; + let hold: { ownerId: string; expiresAt: number } = { ownerId: randomUUID(), expiresAt: now + 300_000 }; + let pluginReceipt: Awaited>> | undefined; const capability = { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities; const read = (quietPluginIds: string[] = []) => db.transaction(async tx => { await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); - return readIdleDeadlineProof(tx, hold, { now, verifiedCapability: capability, quietPluginIds, - ...(quietPluginIds.length ? { quietPluginGeneration: "fixture-running-worker", quietPluginCatalog: await tx.select().from(plugins).where(eq(plugins.id, pluginId)).then(rows => rows.map(plugin => ({ pluginId: plugin.id, manifestGeneration: pluginIdleManifestGeneration(plugin.manifestJson) }))) } : {}) }); + return readIdleDeadlineProof(tx, hold, { now: Date.now(), verifiedCapability: capability, quietPluginIds, + quietPluginGeneration: pluginReceipt?.catalogGeneration, quietPluginCatalog: pluginReceipt?.catalog }); }, { isolationLevel: "repeatable read", accessMode: "read only" }); await db.insert(companies).values({ id: companyId, name: "Deadline fixture", issuePrefix: `D${companyId.slice(0, 6)}` }); await db.insert(agents).values({ id: agentId, companyId, name: "Timer", role: "engineer", status: "idle", adapterType: "codex_local", runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } }, createdAt: new Date(now), lastHeartbeatAt: new Date(now) }); + const central = (enabled: boolean, diagnostics = false) => readIdleSleepSafety(db, () => ({ ...hold, ownerId: hold.ownerId, + draining: true, startedAt: new Date(now - 1000), expiresAt: new Date(hold.expiresAt), activeRuns: 0, pendingWakes: 0 }), + () => now, hold.ownerId, async () => "none", undefined, undefined, enabled, enabled, + enabled ? { deadlineCapability: capability, ...(diagnostics ? { collectDiagnostics: true as const } : {}) } : {}); + expect((await central(false)).backgroundWork).toBe("present"); + const negotiated = await central(true); + expect(negotiated).toMatchObject({ version: 3, backgroundWork: "none", deadlineProof: { catalog: "domain_deadlines_v1" }, + scheduledWake: { version: 2, kind: "domain_deadlines" } }); + expect(negotiated.deadlineProof?.generation).toBe(negotiated.scheduledWake?.generation); const initial = await read(); expect(initial.blocked).toBe(false); expect(initial.scheduledWake?.nextWakeAt).toBe(new Date(now + 600_000).toISOString()); expect((await db.select().from(agents).where(eq(agents.id, agentId)))[0].lastHeartbeatAt?.getTime()).toBe(now); - // A new database client reconstructs the obligation without any scheduler cache. await db.$client.end({ timeout: 1 }); db = createDb(external ?? temporary!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); expect((await read()).scheduledWake).toEqual(initial.scheduledWake); @@ -64,38 +100,194 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await read()).blocked).toBe(true); await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: false } } }).where(eq(agents.id, agentId)); expect((await read()).scheduledWake).toBeUndefined(); + const noAlarm = await central(true); + expect(noAlarm).toMatchObject({ version: 3, backgroundWork: "none", deadlineProof: { catalog: "domain_deadlines_v1" } }); + expect(noAlarm.scheduledWake).toBeUndefined(); + expect(noAlarm.deadlineProof?.generation).not.toBe(negotiated.deadlineProof?.generation); + await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0.5 } }, lastHeartbeatAt: new Date(now) }).where(eq(agents.id, agentId)); + let checkpointClock = now; + const becameDue = await readIdleSleepSafety(db, () => ({ ownerId: hold.ownerId, draining: true, startedAt: new Date(now - 1000), + expiresAt: new Date(hold.expiresAt), activeRuns: 0, pendingWakes: 0 }), () => checkpointClock, hold.ownerId, async () => "none", undefined, + async () => { const finish = beginIdleTrackedWork(); try { checkpointClock += 1000; return true; } finally { finish(); } }, + true, true, { deadlineCapability: capability }); + expect(becameDue.backgroundWork).toBe("present"); + await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: false } } }).where(eq(agents.id, agentId)); - await db.insert(plugins).values({ id: pluginId, pluginKey: `deadline-${pluginId}`, packageName: "fixture-plugin", version: "1.0.0", status: "ready", - manifestJson: { id: `deadline-${pluginId}`, apiVersion: 1, version: "1.0.0", displayName: "Fixture", description: "Deadline fixture", author: "Test", categories: [], capabilities: ["jobs.schedule"], - entrypoints: { worker: "worker.js" }, jobs: [{ jobKey: "hourly", displayName: "Hourly", schedule: "0 * * * *" }] } }); - const store = pluginJobStore(db); - await store.syncJobDeclarations(pluginId, [{ jobKey: "hourly", displayName: "Hourly", schedule: "0 * * * *" }]); - const [job] = await db.select().from(pluginJobs).where(eq(pluginJobs.pluginId, pluginId)); - expect((await read([pluginId])).blocked).toBe(true); // startup has not initialized the schedule - await store.updateRunTimestamps(job.id, new Date(now - 3_600_000), new Date(now + 900_000)); - expect((await read()).blocked).toBe(true); // missing current quiet worker - expect((await read([pluginId])).scheduledWake?.nextWakeAt).toBe(new Date(now + 900_000).toISOString()); - const pending = await store.createRun({ jobId: job.id, pluginId, trigger: "schedule" }); - expect((await read([pluginId])).blocked).toBe(true); - await store.completeRun(pending.id, { status: "succeeded", durationMs: 1 }); - expect((await read([pluginId])).blocked).toBe(false); - await db.update(pluginJobs).set({ status: "paused" }).where(eq(pluginJobs.id, job.id)); + const manifest = { id: `deadline-${pluginId}`, apiVersion: 1 as const, version: "1.0.0", displayName: "Fixture", description: "Deadline fixture", author: "Test", + categories: [], capabilities: ["jobs.schedule" as const, "plugin.state.write" as const], entrypoints: { worker: "worker.cjs" }, + jobs: [{ jobKey: "hourly", displayName: "Hourly", schedule: "0 * * * *" }] }; + await db.insert(plugins).values({ id: pluginId, pluginKey: manifest.id, packageName: "fixture-plugin", version: manifest.version, status: "ready", manifestJson: manifest }); + const store = pluginJobStore(db), state = pluginStateStore(db), manager = createPluginWorkerManager(); + let allowed = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => allowed, onError: error => errors.push(error) }); + const scheduler = createPluginJobScheduler({ db, jobStore: store, workerManager: manager, isBackgroundWorkEnabled: () => allowed, tickIntervalMs: 10 }); + const prepare = async () => { + allowed = false; + const lease = startTaskDrain({ purpose: "idle", ttlMs: 300_000 }); + hold = { ownerId: lease.ownerId!, expiresAt: lease.expiresAt!.getTime() }; + pluginReceipt = await manager.inspectIdleSleep!({ ...hold }); + expect(pluginReceipt.backgroundWork).toBe("none"); + return read([pluginId]); + }; + const resume = () => { stopTaskDrain(); manager.releaseIdleSleep!(); allowed = true; }; + try { + await manager.startWorker(pluginId, { manifest, apiVersion: 1, config: {}, instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, + entrypointPath: fileURLToPath(new URL("./fixtures/plugin-worker-deadline.cjs", import.meta.url)), + execArgv: ["--import", createRequire(import.meta.url).resolve("tsx")], + hostHandlers: { "state.set": async input => { + if (input.scopeKind !== "instance") throw new Error("Fixture job must publish instance-scoped state"); + await state.set(pluginId, { ...input, scopeKind: "instance" }); + } } }); + await store.syncJobDeclarations(pluginId, manifest.jobs); + const [job] = await db.select().from(pluginJobs).where(eq(pluginJobs.pluginId, pluginId)); + expect((await prepare()).blocked).toBe(true); + resume(); + await scheduler.start(coordinator); + await scheduler.tick(); + await store.updateRunTimestamps(job.id, new Date(now - 3_600_000), new Date(now + 900_000)); + expect((await read()).blocked).toBe(true); + const transferred = await prepare(); + expect(transferred.scheduledWake?.nextWakeAt).toBe(new Date(now + 900_000).toISOString()); + resume(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(now + 900_001); + await Promise.all([scheduler.tick(), scheduler.tick()]); + await expect.poll(async () => (await db.select().from(pluginJobRuns).where(eq(pluginJobRuns.jobId, job.id))).map(run => run.status), { timeout: 5_000 }).toEqual(["succeeded"]); + const effects = await db.select().from(pluginState).where(eq(pluginState.pluginId, pluginId)); + const runs = await db.select().from(pluginJobRuns).where(eq(pluginJobRuns.jobId, job.id)); + expect(effects).toHaveLength(1); expect(effects[0].stateKey).toBe(runs[0].id); + expect(runs[0].finishedAt).not.toBeNull(); + const [after] = await db.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)); + expect(after.nextRunAt!.getTime()).toBeGreaterThan(now + 900_001); + expect((await prepare()).scheduledWake?.nextWakeAt).toBe(after.nextRunAt!.toISOString()); + resume(); vi.useRealTimers(); + await store.updateJobStatus(job.id, "paused"); + expect(errors).toEqual([]); + } finally { resume(); vi.useRealTimers(); await scheduler.stop(); await coordinator.stop(); await manager.stopAll(); } + hold = { ownerId: randomUUID(), expiresAt: Date.now() + 300_000 }; expect((await read()).scheduledWake).toBeUndefined(); - - const traceId = randomUUID(); - await db.insert(providerTraceRecords).values({ id: traceId, companyId, runId, provider: "fixture", requestedBy: "fixture", status: "complete", - traceRef: `${randomUUID()}.ndjson`, expiresAt: new Date(now + 86_400_000) }); - expect((await read()).scheduledWake?.nextWakeAt).toBe(new Date(now + 86_400_000).toISOString()); - await db.update(providerTraceRecords).set({ status: "capturing" }).where(eq(providerTraceRecords.id, traceId)); + const traces = providerTraceStore(db); + const capture = await traces.prepare({ companyId, runId, provider: "opencode", requestedBy: "fixture" }); expect((await read()).blocked).toBe(true); - await db.update(providerTraceRecords).set({ status: "deleted", deletedAt: new Date(now), traceRef: `s3-segments-v1:${companyId}/${runId}/${randomUUID()}.ndjson` }).where(eq(providerTraceRecords.id, traceId)); + const bytes = JSON.stringify({ kind: "trace_status", status: "complete", debugChannel: "typescript_opencode_native", debugSequence: 1, acknowledgedDebugSequence: 0 }) + "\n"; + await fs.writeFile(capture.path, bytes); + expect((await traces.finalize(runId, companyId))?.status).toBe("complete"); + expect((await read()).scheduledWake?.nextWakeAt).toBe(new Date(capture.metadata.expiresAt).toISOString()); + const savedCapture = (await traces.getByRun(runId, companyId))!; + await db.update(providerTraceRecords).set({ status: "deleted", deletedAt: new Date(), traceRef: `s3-segments-v1:${companyId}/${runId}/${randomUUID()}.ndjson` }).where(eq(providerTraceRecords.runId, runId)); expect((await read()).blocked).toBe(true); - // The completed deletion receipt is the existing owner marker, not terminal UI status. - await db.update(providerTraceRecords).set({ traceRef: "00000000-0000-0000-0000-000000000000.ndjson" }).where(eq(providerTraceRecords.id, traceId)); + await db.update(providerTraceRecords).set({ status: "complete", deletedAt: null, traceRef: savedCapture.traceRef }).where(eq(providerTraceRecords.runId, runId)); + await traces.remove(runId, companyId); + await expect(fs.stat(capture.path)).rejects.toMatchObject({ code: "ENOENT" }); expect((await read()).blocked).toBe(false); - expect(await db.select().from(providerTraceRecords).where(eq(providerTraceRecords.id, traceId))).toHaveLength(1); - // Fixture rows stay retained for receipt inspection; root drops only the owned test DB after closing all clients. - expect(await db.select().from(pluginJobRuns).where(eq(pluginJobRuns.id, pending.id))).toHaveLength(1); + + const expiringRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: expiringRunId, companyId, agentId, status: "succeeded", invocationSource: "timer", startedAt: new Date(now), finishedAt: new Date(), costAccountingPending: false }); + const expiring = await traces.prepare({ companyId, runId: expiringRunId, provider: "opencode", requestedBy: "fixture" }); + await fs.writeFile(expiring.path, bytes); + expect((await traces.finalize(expiringRunId, companyId))?.status).toBe("complete"); + await fs.unlink(expiring.path); await fs.mkdir(expiring.path); + const expiry = new Date(new Date(expiring.metadata.expiresAt).getTime() + 1); + await expect(traces.cleanupExpired(expiry)).rejects.toThrow(); + expect((await read()).blocked).toBe(true); + await fs.rmdir(expiring.path); await fs.writeFile(expiring.path, bytes); + await traces.cleanupExpired(expiry); + const cleaned = await traces.getByRun(expiringRunId, companyId); + expect(cleaned).toMatchObject({ status: "expired", traceRef: "00000000-0000-0000-0000-000000000000.ndjson" }); + await expect(fs.stat(expiring.path)).rejects.toMatchObject({ code: "ENOENT" }); + expect((await read()).blocked).toBe(false); + const watched = randomUUID(); + await db.insert(issues).values({ id: watched, companyId, title: "Watchdog grace fixture", status: "done" }); + await db.insert(issueWatchdogs).values({ companyId, issueId: watched, watchdogAgentId: agentId, status: "active" }); + expect((await central(true, true)).diagnostics?.checks).toContainEqual({ reason: "issue_watchdogs", state: "present" }); + const mergeIssue = randomUUID(); + await db.insert(issues).values({ id: mergeIssue, companyId, title: "Merge confirmation fixture", status: "in_review", assigneeAgentId: agentId }); + await db.insert(issueThreadInteractions).values({ companyId, issueId: mergeIssue, kind: "request_confirmation", status: "pending", + title: "Merge fixture/repo#39?", payload: { version: 1, prompt: "Merge fixture/repo#39?" } }); + expect((await central(true, true)).diagnostics?.checks).toContainEqual({ reason: "issues.schedule", state: "present" }); }, 30_000); + + it("uses real vault refresh leases, rotates credentials and settles GitHub binding debt after wake", async () => { + const now = Date.now(), companyId = randomUUID(), owner = "deadline-owner"; + await db.insert(companies).values({ id: companyId, name: "GitHub deadline", issuePrefix: `G${companyId.slice(0, 6)}` }); + const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_http", status: "active" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app.id, uid: randomUUID(), name: "GitHub", status: "active", enabled: true, + transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", credentialSource: "paperclip_vault", transportConfig: {}, + config: { sourceTemplateKey: "github", oauth: { strategy: "paperclip_cloud_connector", connectorProfile: "github.code" } } }).returning(); + const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId: connection.id, kind: "user", subjectUserId: owner, status: "active", + credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: randomUUID(), versionSelector: "latest" }], + providerTenant: { oauth: { strategy: "paperclip_cloud_connector", accessTokenExpiresAt: new Date(now + 8 * 3_600_000).toISOString(), refreshedAt: new Date(now).toISOString() }, + github: { userId: "42", login: "fixture", installationCount: 1, repositoryCount: 1, repositorySelection: "all", installationIds: ["101"], + installationOwnerLogins: ["fixture"], accessRevision: randomUUID(), lastAccessRefreshAt: new Date(now).toISOString(), webhookHealth: "healthy" } } }).returning(); + const read = () => db.transaction(async tx => { + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + return readIdleGitHubGrantContinuity(tx, Date.now()); + }, { isolationLevel: "repeatable read", accessMode: "read only" }); + expect((await read()).blocked).toBe(true); + const refs = []; + for (const [configPath, value] of [["oauth.access_token", "saved-access"], ["oauth.refresh_token", "saved-refresh"]] as const) { + const saved = await writeConnectionCredential(db, { companyId, connectionName: "GitHub", configPath, label: configPath, value, ownerUserId: owner }); + refs.push({ configPath, secretId: saved.secret.id, versionSelector: "latest" as const, required: true }); + } + await db.update(connectionGrants).set({ credentialSecretRefs: refs }).where(eq(connectionGrants.id, grant.id)); + const vault = secretService(db), credentials = createOAuthCredentialStore(db, vault); + await credentials.syncCredentialBindings(connection, refs); + const baseline = await read(); + expect(baseline.blocked).toBe(false); + expect(baseline.connectionIds).toContain(connection.id); + expect(new Date(baseline.deadlines[0]).toISOString()).toBe(new Date(now + 7 * 3_600_000).toISOString()); + const getConnection = async () => (await db.select().from(toolConnections).where(eq(toolConnections.id, connection.id)))[0]; + const leases = createOAuthRefreshLeases(db, () => new Date(), getConnection, credentials.getOAuthGrantRow); + let failBindings = true, rotations = 0; + const successfulBindings = new Map(); + const connector = { + refresh: async (input: Parameters[0]) => { + rotations++; expect(input.refreshToken).toBe("saved-refresh"); expect(input.subject).toBe(owner); + return { v: 1 as const, accessToken: "renewed-access", refreshToken: "renewed-refresh", tokenType: "Bearer", accessTokenExpiresAt: new Date(Date.now() + 8 * 3_600_000).toISOString(), + refreshTokenExpiresAt: null, scopes: [...GITHUB_CONNECTOR_PROFILES["github.code"].scopes], subject: owner, companyId, instanceId: "fixture", environment: "test" as const, provider: "github" as const, profile: "github.code" }; + }, + setWebhookBinding: async (input: Parameters[0]) => { + expect(input.accessToken).toBe("renewed-access"); expect(input.companyId).toBe(companyId); + if (failBindings) throw new Error("named Cloud binding provider unavailable"); + successfulBindings.set(input.id, { active: input.active, grantId: input.grantId }); + }, + } as PaperclipCloudConnector; + const refresh = createOAuthGrantRefresh(db, { now: () => new Date(), store: credentials, leases, getConnectionRow: getConnection, + currentCloudConnector: () => connector, exchangeOAuthToken: async () => { throw new Error("Direct OAuth must not run for managed GitHub"); } }); + const continuity = createGitHubGrantContinuity(db, { now: () => new Date(), currentCloudConnector: () => connector, + refreshOAuthGrantCredentials: refresh, resolveOAuthGrantSecret: credentials.resolveOAuthGrantSecret }); + vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => { + const url = new URL(String(input)); expect(url.origin).toBe("https://api.github.com"); + expect(new Headers(init?.headers).get("authorization")).toBe("Bearer renewed-access"); + const body = url.pathname === "/user" ? { id: 42, login: "fixture" } + : url.pathname === "/user/installations" ? { installations: [{ id: 101, repository_selection: "all", account: { login: "fixture" } }] } + : { repositories: [{ id: 999, full_name: "fixture/repo" }] }; + return new Response(JSON.stringify(body), { status: 200, headers: { "content-type": "application/json" } }); + }); + try { + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(now + 7 * 3_600_000 + 1); + expect(await continuity.sweepGitHubConnectionContinuity(companyId)).toMatchObject({ due: 1, failed: 1 }); + expect(rotations).toBe(1); + const [debt] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)); + expect(debt.providerTenant?.github?.bindingSync).toBeDefined(); + expect(debt.providerTenant?.github?.continuityRetry).toBeDefined(); + expect((await read()).blocked).toBe(true); + failBindings = false; vi.setSystemTime(Date.now() + 31_000); + const reconstructed = createGitHubGrantContinuity(db, { now: () => new Date(), currentCloudConnector: () => connector, + refreshOAuthGrantCredentials: refresh, resolveOAuthGrantSecret: credentials.resolveOAuthGrantSecret }); + expect(await reconstructed.sweepGitHubConnectionContinuity(companyId)).toMatchObject({ due: 1, failed: 0 }); + expect(rotations).toBe(1); expect(successfulBindings.size).toBe(1); + const [settled] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)); + expect(settled.providerTenant?.github?.bindingSync).toBeUndefined(); + expect(settled.providerTenant?.github?.continuityRetry).toBeUndefined(); + const [stored] = await db.select().from(companySecrets).where(eq(companySecrets.id, refs[0].secretId)); + expect(stored.latestVersion).toBe(2); + expect((await credentials.resolveOAuthGrantSecret(connection, settled, refs[0], { actorType: "system", actorId: "fixture" }, undefined)).value).toBe("renewed-access"); + const after = await read(); + expect(after.blocked).toBe(false); expect(after.connectionIds).toContain(connection.id); + expect(Math.min(...after.deadlines)).toBeGreaterThan(Date.now()); + } finally { vi.useRealTimers(); vi.restoreAllMocks(); } + }, 30_000); + }); diff --git a/server/src/__tests__/plugin-idle-manager.test.ts b/server/src/__tests__/plugin-idle-manager.test.ts index 1ea93bc172b..ef0bb175d24 100644 --- a/server/src/__tests__/plugin-idle-manager.test.ts +++ b/server/src/__tests__/plugin-idle-manager.test.ts @@ -1,8 +1,9 @@ +import { pluginIdleManifestGeneration } from "../services/plugin-idle-catalog.js"; import { fileURLToPath } from "node:url"; import { createRequire } from "node:module"; import { setTimeout as delay } from "node:timers/promises"; import { afterEach, describe, expect, it } from "vitest"; -import { createPluginWorkerHandle } from "../services/plugin-worker-manager.js"; +import { createPluginWorkerHandle, createPluginWorkerManager } from "../services/plugin-worker-manager.js"; import { startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; const manifest = { id: "test.idle", apiVersion: 1 as const, version: "1.0.0", displayName: "Idle fixture", @@ -23,6 +24,31 @@ function hold() { describe("plugin manager idle receipts", () => { afterEach(() => stopTaskDrain()); + it("binds a quiet catalog to the actual started manifest and replacement worker", async () => { + const manager = createPluginWorkerManager(); + const options = (version: string) => ({ + entrypointPath: fileURLToPath(new URL("./fixtures/plugin-worker-idle.cjs", import.meta.url)), + execArgv: ["--import", createRequire(import.meta.url).resolve("tsx")], + manifest: { ...manifest, version }, config: {}, instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, apiVersion: 1, + hostHandlers: { "events.subscribe": async () => ({}) }, + }); + try { + await manager.startWorker("test.idle", options("1.0.0")); + const lease = hold(); + const first = await manager.inspectIdleSleep!(lease); + expect(first.backgroundWork).toBe("none"); + expect(first.catalog).toEqual([{ pluginId: "test.idle", manifestGeneration: pluginIdleManifestGeneration(manifest) }]); + expect((await manager.inspectIdleSleep!(lease)).catalogGeneration).toBe(first.catalogGeneration); + stopTaskDrain(); manager.releaseIdleSleep!(); + await manager.stopWorker("test.idle"); + await manager.startWorker("test.idle", options("1.0.1")); + const replacement = await manager.inspectIdleSleep!(hold()); + expect(replacement.backgroundWork).toBe("none"); + expect(replacement.catalogGeneration).not.toBe(first.catalogGeneration); + expect(replacement.catalog?.[0].manifestGeneration).not.toBe(first.catalog?.[0].manifestGeneration); + } finally { stopTaskDrain(); await manager.stopAll(); } + }); + it("retains timed-out worker operations and admits work after the owned hold releases", async () => { const handle = worker(); await handle.start(); diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index e5d9c9f648e..75d918bb358 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -320,7 +320,12 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag // The report covers every company in this process. Ordinary company // members may read process counters, but not instance-wide work state. assertCanManageInstanceSettings(req); - const scheduledIssueMonitors = req.query.scheduledWake === "2"; + const domainDeadlines = req.query.scheduledWake === "domain-deadlines-v1"; + const capabilities = verifiedIdleCapabilities(req); + if (domainDeadlines && capabilities?.deadlineCatalog !== "domain_deadlines_v1") { + throw forbidden("Domain deadline reports require the signed Cloud deadline catalog"); + } + const scheduledIssueMonitors = req.query.scheduledWake === "2" || domainDeadlines; const scheduledRoutines = req.query.scheduledWake === "1" || scheduledIssueMonitors; if (scheduledRoutines && req.actor.source !== "cloud_control") { throw forbidden("Scheduled wake reports require Cloud control authority"); @@ -335,9 +340,10 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag } const idleSleepSafety = await readIdleSleepSafety(db, () => heartbeat.getTaskDrainStatus(), Date.now, typeof req.query.ownerId === "string" ? req.query.ownerId : undefined, - diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic }), + diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic, + deadlineCapability: domainDeadlines ? capabilities : undefined }), pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors, - { incomingWake: verifiedIdleCapabilities(req), deadlineCapability: verifiedIdleCapabilities(req), + { incomingWake: capabilities, deadlineCapability: domainDeadlines ? capabilities : undefined, ...(req.query.idleSleepDiagnostics === "1" ? { collectDiagnostics: true } : {}) }); res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety }); return; diff --git a/server/src/services/connection-credentials.ts b/server/src/services/connection-credentials.ts index ed87874a85a..a8e3407705e 100644 --- a/server/src/services/connection-credentials.ts +++ b/server/src/services/connection-credentials.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import { and, eq, inArray, isNull, ne, or } from "drizzle-orm"; -import { companySecrets, companySecretProposals, managedAgentProfiles, routineTriggers, userSecretDefinitions, type connectionGrants, type toolConnections } from "@paperclipai/db"; +import { companySecrets, companySecretVersions, companySecretProposals, managedAgentProfiles, routineTriggers, userSecretDefinitions, type Db, type connectionGrants, type toolConnections } from "@paperclipai/db"; import type { ToolCredentialSecretRef } from "@paperclipai/shared"; import { connectionCredentialConfigPath } from "@paperclipai/shared"; import { unprocessable } from "../errors.js"; @@ -119,7 +119,7 @@ export async function writeConnectionCredential( /** Validate metadata without reading secret values, including for connection pickers. */ export async function validateConnectionGrantSecretOwnership( - db: CredentialDb, + db: Pick, connection: Pick, grant: Pick, ref: ToolCredentialSecretRef, @@ -165,3 +165,39 @@ export async function resolveConnectionGrantSecret( }); return { value: resolved.value, latestVersion: secret.latestVersion }; } + +export async function readConnectionGrantCredentialState( + db: Pick, + connection: Pick, + grant: Pick, +): Promise { + const refs = grant.credentialSecretRefs.filter(ref => ["oauth.access_token", "oauth.refresh_token"].includes(ref.configPath)); + if (refs.filter(ref => ref.configPath === "oauth.access_token").length !== 1 + || new Set(refs.map(ref => ref.configPath)).size !== refs.length) return; + const state: unknown[] = []; + for (const ref of refs) { + let secret; + try { secret = await validateConnectionGrantSecretOwnership(db, connection, grant, ref); } + catch (error) { + if (error && typeof error === "object" && "status" in error && error.status === 422) return; + throw error; + } + if (secret.status !== "active" || secret.deletedAt || !Number.isSafeInteger(secret.latestVersion) || secret.latestVersion <= 0) return; + let definitionState: unknown; + if (secret.scope === "user") { + const [definition] = await db.select({ id: userSecretDefinitions.id, companyId: userSecretDefinitions.companyId, + status: userSecretDefinitions.status, deletedAt: userSecretDefinitions.deletedAt, updatedAt: userSecretDefinitions.updatedAt }) + .from(userSecretDefinitions).where(eq(userSecretDefinitions.id, secret.userSecretDefinitionId!)).limit(1); + if (!definition || definition.companyId !== connection.companyId || definition.status !== "active" || definition.deletedAt) return; + definitionState = definition; + } + const version = ref.versionSelector === undefined || ref.versionSelector === "latest" ? secret.latestVersion : ref.versionSelector; + if (typeof version !== "number" || !Number.isSafeInteger(version) || version <= 0) return; + const [saved] = await db.select({ id: companySecretVersions.id, version: companySecretVersions.version, + status: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt }) + .from(companySecretVersions).where(and(eq(companySecretVersions.secretId, secret.id), eq(companySecretVersions.version, version))).limit(1); + if (!saved || saved.status !== "current" || saved.revokedAt) return; + state.push([ref.configPath, secret.id, secret.scope, secret.ownerUserId, secret.latestVersion, secret.updatedAt, definitionState, saved]); + } + return state; +} diff --git a/server/src/services/heartbeat/idle-timer-schedule.ts b/server/src/services/heartbeat/idle-timer-schedule.ts index 11d3f15bce1..ce822586b69 100644 --- a/server/src/services/heartbeat/idle-timer-schedule.ts +++ b/server/src/services/heartbeat/idle-timer-schedule.ts @@ -6,7 +6,6 @@ import { supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js"; import { createHeartbeatTimerPolicy, heartbeatTimerPolicyIsWellFormed } from "./timer-policy.js"; import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "../idle-deadline-types.js"; -/** Normal Cloud scheduling only. Worktree overrides/restore are not a transferable execution authority. */ export async function readIdleAgentTimerSchedule(tx: Pick, now: number, env: Record = process.env, checkBudget: () => void = () => {}): Promise { const rows = await tx.select({ agent: getTableColumns(agents), companyStatus: companies.status, @@ -32,8 +31,6 @@ export async function readIdleAgentTimerSchedule(tx: Pick, now: nu const policy = owner.parseHeartbeatPolicy(agent); if (!Number.isFinite(policy.intervalSec)) return blocked(); if (!policy.enabled || policy.intervalSec <= 0 || !supportsUnscopedTimerHeartbeat(agent)) continue; - // Org-chain evaluation belongs to its owner. Bound its candidate count - // even when every timer needs no additional SQL (for example zero caps). if (++inspectedTimers > 128) return blocked(); const invokability = evaluateAgentInvokability(agent, byCompany.get(agent.companyId)!); if (!invokability.invokable) { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index e3fd4495b23..b8dfdbba6fd 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -35,9 +35,7 @@ import { CHAT_COMPLETION_WAKE_REASON, isCompletedOnboardingHandoffWake, } from "../chat-completion-delivery.js"; -import { - hasWorkspaceRestoreFailure, -} from "@paperclipai/shared"; +import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; import { readQueuedInteractionResponse } from "../queued-interaction-response.js"; import { isConversation, @@ -121,9 +119,7 @@ import { withQueuedCommentIdsInWakePayload, withQueuedCommentIdsInRunContext, } from "../issue-queued-comment-queue.js"; -import { - parseObject, -} from "../../adapters/utils.js"; +import { parseObject } from "../../adapters/utils.js"; import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; import { @@ -343,7 +339,6 @@ function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; } -/** Exact durable timer claim used by ordinary dispatch and restart qualification. */ export async function claimDueTimerHeartbeatForAgent( db: Db, agent: typeof agents.$inferSelect, @@ -372,7 +367,7 @@ export async function claimDueTimerHeartbeatForAgent( } -/** Queue admission and dispatch share lifecycle callbacks and execution ownership. */ +/** Queue admission and dispatch share the service lifecycle callbacks and execution ownership. */ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDependencies) { const { setRunStatus, diff --git a/server/src/services/heartbeat/timer-policy.ts b/server/src/services/heartbeat/timer-policy.ts index 56dd0803fe3..161a5994836 100644 --- a/server/src/services/heartbeat/timer-policy.ts +++ b/server/src/services/heartbeat/timer-policy.ts @@ -21,7 +21,6 @@ function normalizeMaxConcurrentRuns(value: unknown) { ); } -/** Sleep proof rejects malformed stored inputs rather than borrowing parser fallbacks. */ export function heartbeatTimerPolicyIsWellFormed(agent: typeof agents.$inferSelect): boolean { const record = (v: unknown): v is Record => v !== null && typeof v === "object" && !Array.isArray(v); if (!record(agent.runtimeConfig)) return false; @@ -48,8 +47,6 @@ export function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) { enabled: asBoolean(heartbeat.enabled, false), intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)), wakeOnDemand: isHeartbeatWakeOnDemandEnabled(agent), - // A Dot binding has one external turn. Competing assignments must retain - // their queue position instead of claiming a second run that cannot bind. maxConcurrentRuns: agent.adapterType === "paperclip_runner" && ["openai_dot", "grok_bot", "muse"].includes(String(parseObject(agent.adapterConfig).provider)) ? 1 : normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns), @@ -108,7 +105,6 @@ function currentUtcDayWindow(now = new Date()) { -/** Shared by dispatch and owned, read-only sleep inspections. */ export function createHeartbeatTimerPolicy(db: Pick) { async function getHeartbeatDailyCapBlock( agent: typeof agents.$inferSelect, diff --git a/server/src/services/idle-deadline-proof.test.ts b/server/src/services/idle-deadline-proof.test.ts index ff79d9bdedb..deb0142779f 100644 --- a/server/src/services/idle-deadline-proof.test.ts +++ b/server/src/services/idle-deadline-proof.test.ts @@ -9,7 +9,6 @@ import { projectIdleProviderTraceSchedule, type IdleProviderTraceRow } from "./p import { projectIdleGitHubGrantContinuity } from "./tool-access/github-grant-continuity.js"; const now = Date.parse("2099-01-01T00:00:00Z"); -// The caller's authentication boundary has its own real signed-capability suite. const verifiedCapability = { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities; const options: IdleDeadlineProofOptions = { now, verifiedCapability, quietPluginIds: [] }; const hold = { ownerId: "owned", expiresAt: now + 300_000 }; diff --git a/server/src/services/idle-deadline-proof.ts b/server/src/services/idle-deadline-proof.ts index d700f96beeb..d9176c05698 100644 --- a/server/src/services/idle-deadline-proof.ts +++ b/server/src/services/idle-deadline-proof.ts @@ -11,7 +11,6 @@ import { readIdleGitHubGrantContinuity } from "./tool-access/github-grant-contin const processGeneration = randomUUID(); export class IdleDeadlineInspectionBudgetExceeded extends Error {} -/** Global bounds supplement the transaction's per-statement timeout. */ export function createIdleDeadlineInspectionBudget(now: number, expiresAt: number, monotonicNow: () => number = () => performance.now()) { const started = monotonicNow(); @@ -43,7 +42,6 @@ export type IdleDeadlineProofOptions = { }; const generation = (state: unknown) => createHash("sha256").update(JSON.stringify(state)).digest("hex"); -/** Proof composition only. Domain ownership, durable reads and execution stay with their services. */ export function composeIdleDeadlineProof(hold: { ownerId: string; expiresAt: number }, options: IdleDeadlineProofOptions, snapshots: { timers: IdleDeadlineSnapshot; jobs: IdleDeadlineSnapshot; traces: IdleDeadlineSnapshot; github: IdleDeadlineSnapshot & { connectionIds: string[] } }): IdleDeadlineProof { @@ -73,7 +71,6 @@ export async function readIdleDeadlineProof(tx: Pick, hold: { owne return Reflect.apply(tx.select, tx, args); } } as Pick; try { - // Sequential reads share the caller's snapshot and statement timeout. const timers = await readIdleAgentTimerSchedule(guarded, options.now, process.env, budget.check); budget.check(); if (timers.blocked) return { blocked: true }; @@ -85,14 +82,15 @@ export async function readIdleDeadlineProof(tx: Pick, hold: { owne if (traces.blocked) return { blocked: true }; const github = await readIdleGitHubGrantContinuity(guarded, options.now); budget.check(); - return composeIdleDeadlineProof(hold, options, { timers, jobs, traces, github }); + const proof = composeIdleDeadlineProof(hold, options, { timers, jobs, traces, github }); + budget.check(); + return proof; } catch (error) { if (error instanceof IdleDeadlineInspectionBudgetExceeded) return { blocked: true }; throw error; } } -/** Caller retains the original reason identity and applies this only after complete proof. */ export function deadlineWorkReplacement(reason: string): string | undefined { if (reason === "agents.execution_or_timer") return `SELECT 1 FROM agents WHERE status = 'running'`; if (reason === "plugin_jobs") return `SELECT 1 FROM plugin_jobs WHERE status NOT IN ('active', 'paused')`; diff --git a/server/src/services/idle-deadline-types.ts b/server/src/services/idle-deadline-types.ts index fb72e7fa3e8..4749d2fe33e 100644 --- a/server/src/services/idle-deadline-types.ts +++ b/server/src/services/idle-deadline-types.ts @@ -1,5 +1,8 @@ -/** Closed negotiated catalog; in-process scheduler entries are not wake proofs. */ export const IDLE_DEADLINE_CATALOG = "domain_deadlines_v1" as const; +export interface IdleDeadlineReceipt { + catalog: typeof IDLE_DEADLINE_CATALOG; + generation: string; +} export interface DomainScheduledWake { version: 2; kind: "domain_deadlines"; @@ -9,8 +12,7 @@ export interface DomainScheduledWake { } export interface IdleDeadlineSnapshot { blocked: boolean; - /** Exact durable state inspected by the domain; hashed, never returned to callers. */ - state: unknown; + state: unknown; deadlines: number[]; } export const idleDeadlineRowLimit = 10_000; diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index 7ad14037f45..b5502eabca0 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -1,3 +1,6 @@ +import { readIdleDeadlineProof, deadlineWorkReplacement, type IdleDeadlineProof } from "./idle-deadline-proof.js"; +import type { DomainScheduledWake, IdleDeadlineReceipt } from "./idle-deadline-types.js"; +import type { IdlePluginCatalogEntry } from "./plugin-idle-catalog.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"; @@ -25,14 +28,17 @@ export interface IdleSleepDrainStatus { export interface IdleSleepSafety { version: 1 | 2 | 3; backgroundWork: "none" | "present" | "unknown"; - scheduledWake?: ScheduledWake; + scheduledWake?: ScheduledWake | DomainScheduledWake; diagnostics?: IdleSleepDiagnostics; incomingWake?: IdleIncomingWakeCapability; + deadlineProof?: IdleDeadlineReceipt; } export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) => Promise<{ backgroundWork: "none" | "present" | "unknown"; pluginIds: string[]; + catalogGeneration?: string; + catalog?: IdlePluginCatalogEntry[]; }>; // A failed turn may have successfully torn down its ephemeral allocation. A @@ -287,7 +293,7 @@ export async function readIdleSleepSafety( now: () => number = Date.now, ownerId?: string, inspectLocalWork: (diagnostic?: IdleSleepDiagnosticSink) => Promise = - diagnostic => readIdleLocalWork({ diagnostic }), + diagnostic => readIdleLocalWork({ diagnostic, deadlineCapability: options.deadlineCapability }), inspectPlugins?: InspectIdlePlugins, prepareBackup?: () => Promise, scheduledRoutines = false, @@ -295,14 +301,20 @@ export async function readIdleSleepSafety( options: IdleSleepSafetyOptions = {}, ): Promise { let incomingProof: IdleIncomingWakeProof | undefined; + let deadlineProof: IdleDeadlineProof | undefined; + const deadlines = options.deadlineCapability?.deadlineCatalog === "domain_deadlines_v1"; 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" && incomingProof ? 3 : backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork, + if (backgroundWork === "none" && deadlines && (!deadlineProof || deadlineProof.blocked || !deadlineProof.generation)) backgroundWork = "unknown"; + const wake = deadlineProof ? deadlineProof.scheduledWake : scheduledWake; + return { version: backgroundWork === "none" && (incomingProof || deadlineProof) ? 3 : backgroundWork === "none" && wake ? 2 : 1, backgroundWork, ...(backgroundWork === "none" && incomingProof ? { incomingWake: incomingProof.capability } : {}), - ...(backgroundWork === "none" && scheduledWake ? { scheduledWake } : {}), + ...(backgroundWork === "none" && deadlineProof?.generation ? { deadlineProof: { + catalog: "domain_deadlines_v1" as const, generation: deadlineProof.generation } } : {}), + ...(backgroundWork === "none" && wake ? { scheduledWake: wake } : {}), ...(diagnostics ? { diagnostics } : {}) }; }; const before = getDrainStatus(); @@ -320,6 +332,8 @@ export async function readIdleSleepSafety( diagnostic?.record("local.inspection", localFirst); if (localFirst !== "none" && !diagnostic) return report(localFirst); let quietPlugins: string[] = []; + let quietPluginGeneration: string | undefined; + let quietPluginCatalog: IdlePluginCatalogEntry[] | undefined; let pluginState: IdleLocalWork = "none"; if (inspectPlugins) { const currentLocal = diagnostic ? idleWorkSnapshot() : localBefore; @@ -334,7 +348,11 @@ export async function readIdleSleepSafety( pluginState = pluginWork.backgroundWork; diagnostic?.record("plugins.runtime_drain", pluginState); if (pluginState !== "none" && !diagnostic) return report(pluginState); - if (pluginState === "none") quietPlugins = pluginWork.pluginIds; + if (pluginState === "none") { + quietPlugins = pluginWork.pluginIds; + quietPluginGeneration = pluginWork.catalogGeneration; + quietPluginCatalog = pluginWork.catalog; + } } } if (diagnostic && !sameQuietHold(before, getDrainStatus(), now())) { @@ -350,10 +368,35 @@ export async function readIdleSleepSafety( if (!ownerId || ownerId !== before.ownerId) return undefined; incomingProof = await readIdleIncomingWakeProof(tx, { capabilities: options.incomingWake, ownerId, receiverOwner: db }); } + if (deadlines) { + if (!owned() || !scheduledRoutines || !scheduledIssueMonitors) return undefined; + const routine = await readIdleRoutineSchedule(tx, ownerId!, now()); + const issue = await readIdleIssueSchedule(tx, ownerId!, now()); + diagnostic?.record("routines.schedule", routine.blocked ? "present" : "none"); + diagnostic?.record("issues.schedule", issue.blocked ? "present" : "none"); + diagnostic?.expect(["deadlines.catalog"]); + failureReason = "deadlines.catalog"; + if (routine.blocked || issue.blocked) deadlineProof = { blocked: true }; + else { + scheduledWake = combineIdleSchedules(routine.scheduledWake, issue.scheduledWake, issue.generation); + deadlineProof = await readIdleDeadlineProof(tx, { ownerId: ownerId!, expiresAt: before.expiresAt!.getTime() }, { + now: now(), verifiedCapability: options.deadlineCapability, + quietPluginIds: quietPlugins, quietPluginGeneration, quietPluginCatalog, + scheduledWake, passiveGeneration: issue.generation, + }); + } + diagnostic?.record("deadlines.catalog", deadlineProof.blocked ? "present" : "none"); + if (deadlineProof.blocked && !diagnostic) return true; + } + failureReason = "durable.snapshot"; const checks = WORK_CHECKS.filter(check => (!scheduledRoutines || check.reason !== "routines") && (!scheduledIssueMonitors || check.reason !== "issues")) - .map(check => ({ ...check, query: incomingWakeWorkQuery(check.reason, incomingProof) ?? check.query })); + .map(check => { + const replacement = deadlineProof?.blocked === false ? deadlineWorkReplacement(check.reason) : undefined; + return { ...check, query: incomingWakeWorkQuery(check.reason, incomingProof, + deadlineProof?.githubContinuity?.connectionIds) ?? (replacement ? sql.raw(replacement) : 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()}`)); @@ -378,8 +421,9 @@ export async function readIdleSleepSafety( if (rows.length !== 1 || typeof rows[0]?.blocked !== "boolean") return undefined; found = rows[0].blocked; } + found ||= deadlineProof?.blocked ?? false; if (found && !diagnostic) return true; - if (scheduledRoutines) { + if (scheduledRoutines && !deadlines) { if (!owned()) return undefined; const schedule = await readIdleRoutineSchedule(tx, ownerId!, now()); diagnostic?.record("routines.schedule", schedule.blocked ? "present" : "none"); @@ -387,7 +431,7 @@ export async function readIdleSleepSafety( found ||= schedule.blocked; scheduledWake = schedule.scheduledWake; } - if (scheduledIssueMonitors) { + if (scheduledIssueMonitors && !deadlines) { if (!scheduledRoutines || !owned()) return undefined; const schedule = await readIdleIssueSchedule(tx, ownerId!, now()); diagnostic?.record("issues.schedule", schedule.blocked ? "present" : "none"); @@ -434,13 +478,14 @@ export async function readIdleSleepSafety( diagnostic?.record("inspection.generation", "none"); if (diagnostic && (localFirst === "unknown" || local === "unknown" || pluginState === "unknown")) return report("unknown"); if (diagnostic && (blocked || localFirst === "present" || pluginState === "present")) return report("present"); - if (local === "none" && scheduledWake && Date.parse(scheduledWake.nextWakeAt) <= now()) { + const finalWake = deadlineProof ? deadlineProof.scheduledWake : scheduledWake; + if (local === "none" && finalWake && Date.parse(finalWake.nextWakeAt) <= now()) { 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"); + if (incomingProof && scheduledWake && !deadlineProof) return report("unknown"); return report(local, scheduledWake); } catch { // Missing migrations, unknown state, malformed configuration, timeouts and diff --git a/server/src/services/plugin-idle-catalog.ts b/server/src/services/plugin-idle-catalog.ts index b7c6bfb652f..0d7949d1178 100644 --- a/server/src/services/plugin-idle-catalog.ts +++ b/server/src/services/plugin-idle-catalog.ts @@ -1,7 +1,6 @@ import { createHash } from "node:crypto"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; export interface IdlePluginCatalogEntry { pluginId: string; manifestGeneration: string } -/** JSONB reorders object keys. Preserve array order and compare the actual manifest. */ export function pluginIdleManifestGeneration(manifest: PaperclipPluginManifestV1): string { const ordered = (value: unknown): unknown => Array.isArray(value) ? value.map(ordered) : value !== null && typeof value === "object" diff --git a/server/src/services/plugin-job-idle-schedule.ts b/server/src/services/plugin-job-idle-schedule.ts index 20a14665724..e840119bca5 100644 --- a/server/src/services/plugin-job-idle-schedule.ts +++ b/server/src/services/plugin-job-idle-schedule.ts @@ -41,7 +41,6 @@ export function projectIdlePluginJobSchedule(rows: IdlePluginJobRow[], quietPlug const declaration = declarations.get(job.jobKey); if (!declaration) return blocked(); if ((declaration.schedule ?? "") !== job.schedule) return blocked(); - // An explicitly unscheduled declaration is manual-only, not a missing timer. if (!job.schedule && job.nextRunAt === null) continue; const at = job.nextRunAt?.getTime() ?? NaN; if (validateCron(job.schedule) || !isFutureIdleDeadline(at, now)) return blocked(); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index cdad4f8d419..8058a60c9f5 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -1,5 +1,5 @@ import { pluginIdleManifestGeneration, type IdlePluginCatalogEntry } from "./plugin-idle-catalog.js"; -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; /** * PluginWorkerManager — spawns and manages out-of-process plugin worker child * processes, routes JSON-RPC 2.0 calls over stdio, and handles lifecycle @@ -21,7 +21,6 @@ import { createHash } from "node:crypto"; */ import { fork, type ChildProcess } from "node:child_process"; -import { randomUUID } from "node:crypto"; import { beginIdleTrackedWork, readTaskDrain, trackIdleWork } from "./task-admission.js"; import { EventEmitter } from "node:events"; import { createInterface, type Interface as ReadlineInterface } from "node:readline"; diff --git a/server/src/services/provider-trace-idle-schedule.ts b/server/src/services/provider-trace-idle-schedule.ts index 47c917c28dd..c409b02b515 100644 --- a/server/src/services/provider-trace-idle-schedule.ts +++ b/server/src/services/provider-trace-idle-schedule.ts @@ -19,8 +19,6 @@ export function projectIdleProviderTraceSchedule(rows: IdleProviderTraceRow[], n } const objectRef = objectProviderTraceRef(trace.traceRef); const localRef = /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}\.ndjson$/.test(trace.traceRef); - // Local remove writes its tombstone only after both sidecars are absent. - // Object remove instead persists debt first and needs the cleanup marker. if (localRef && trace.status === "deleted" && trace.deletedAt) continue; if (importingProviderTraceRef(trace.traceRef) || (objectRef && (objectRef[1] !== trace.companyId || objectRef[2] !== trace.runId)) || (!objectRef && !localRef) diff --git a/server/src/services/provider-trace-retention-work.ts b/server/src/services/provider-trace-retention-work.ts index dba9c7d591c..580f151f209 100644 --- a/server/src/services/provider-trace-retention-work.ts +++ b/server/src/services/provider-trace-retention-work.ts @@ -16,8 +16,6 @@ export function registerProviderTraceRetentionWork( const unregister = registerIdleLocalWorkProbe(options => { if (deadline === undefined) return "unknown"; if (deadline === null) return "none"; - // Only a full negotiated durable scan may transfer a future retention - // obligation. Due cleanup and failed/unknown reconciliation remain work. return options.deadlineCapability?.deadlineCatalog === "domain_deadlines_v1" && deadline > Date.now() ? "none" : "present"; }); diff --git a/server/src/services/tool-access/github-grant-continuity.ts b/server/src/services/tool-access/github-grant-continuity.ts index 69a67e7171c..8132f7ed0c5 100644 --- a/server/src/services/tool-access/github-grant-continuity.ts +++ b/server/src/services/tool-access/github-grant-continuity.ts @@ -1,3 +1,4 @@ +import { readConnectionGrantCredentialState } from "../connection-credentials.js"; import { randomUUID } from "node:crypto"; import { notifyGitHubCredentialChange } from "../github-connection-notifications.js"; import type { ActorInfo } from "./types.js"; @@ -396,14 +397,23 @@ export function githubContinuityRevision(grant: Pick, now: number) { const rows = await tx.select({ grant: connectionGrants, connection: toolConnections, companyStatus: companies.status }) .from(connectionGrants).leftJoin(toolConnections, eq(toolConnections.id, connectionGrants.connectionId)) .leftJoin(companies, eq(companies.id, connectionGrants.companyId)) .where(and(eq(connectionGrants.status, "active"), sql`${connectionGrants.providerTenant}->'github' IS NOT NULL`)) .orderBy(connectionGrants.id).limit(10_001); - return projectIdleGitHubGrantContinuity(rows, now); + const proof = projectIdleGitHubGrantContinuity(rows, now); + if (proof.blocked) return proof; + const credentials: unknown[] = []; + const eligible = new Set(proof.connectionIds); + for (const row of rows) { + if (!row.connection || !eligible.has(row.connection.id)) continue; + const state = await readConnectionGrantCredentialState(tx, row.connection, row.grant); + if (!state) return { ...proof, blocked: true, connectionIds: [], deadlines: [] }; + credentials.push([row.grant.id, state]); + } + return { ...proof, state: [rows, credentials] }; } export function projectIdleGitHubGrantContinuity(rows: { grant: typeof connectionGrants.$inferSelect;