diff --git a/doc/plans/2026-10-30-aggregate-maintenance-sleep.md b/doc/plans/2026-10-30-aggregate-maintenance-sleep.md index 19a68c6db66..58ef7de3f7f 100644 --- a/doc/plans/2026-10-30-aggregate-maintenance-sleep.md +++ b/doc/plans/2026-10-30-aggregate-maintenance-sleep.md @@ -66,7 +66,7 @@ backups, live operation keepalives or authorized external reconciliation disappe One positive control uses a real 31-second observation after actual startup settlement. Longer windows use only child-local Date/setTimeout/setInterval -control. Boot and ordinary HTTP producer phases advance at real speed. The +control. Boot, ordinary HTTP/service producer phases and graceful shutdown advance at real speed. The clock freezes after readiness and real socket settlement, then advances at most 30 seconds per step, settling real I/O between steps. Parent deadlines, setImmediate and database sockets remain usable. Receipts distinguish real @@ -97,6 +97,27 @@ Current assembled cases cover: - A real HTTP 503 from the controlled provider, durable retry deadline, no SQL/provider retry during drain, successful resume and quiet after settlement. - Actual Cloud warm standby, signed identity claim and durable canonical identity. +- Ordinary interaction creation with a disclosed fixture-only activity timestamp + older than 90 days, followed by actual retention archive and system audit. +- Cloud routine creation/run through HTTP, actual issue-service completion held + across drain, and registered projection after resume. Service ingress avoids + counting the HTTP route's existing redundant direct projection hook as worker + proof. Production worktree suppression prevents agent execution. +- Ordinary connection activation followed by a simulated Cloud broker's actual + Ed25519-verified request and X25519/AES-GCM sealed event. The real runtime + decrypts it, persists a processed receipt, acknowledges it and stays quiet + across two minutes of child clock before the real credential deadline. +- Ordinary merge-confirmation creation, actual provider HTTP observation, + persisted acceptance and system actor audit. +- Status-card creation and authorized query/summary writes through real HTTP + authentication, then an ordinary blocked-task source write, debounce, + generation and persisted result. Agent key, responsible user, membership and + execution rows form an explicitly controlled summarizer boundary; no model + provider runs and no card result is seeded. +- Actual native backup publication with the runtime-owned database, ordinary + task closure, physical backup deletion and resume-pointer cleanup while run + history remains. File modification time and closure timestamp are fixture-aged + ten days; cleanup results are created only by the production owner. Warm standby retains exactly one identified startup intent: the committed `companyTransferRunService.recoverStrandedApplyingRuns` notification to import @@ -108,7 +129,15 @@ existing bounded 30-second retry deadline. It does not call standby sleep-eligib while this hold exists. Temporary diagnostic begin/end tracing used to identify this owner is not part of the production change. -Remaining acceptance before a removal PR: final recovery integration, all owner +An assembled admitted completion exposed a separate decision-target idle hold: +its owner retained a reconciled write hold while drain prevented its sweep. +The owner now reports pending dispatch only when admitted, retains service +scopes and parks a 30-second deadline only for those scopes. The generic +coordinator still holds unresolved transactions. PostgreSQL regressions cover +committed and uncommitted changes, failed expiry, zero SQL/active work during +drain, exactly-once resume and existing shutdown/restart behavior. + +Remaining acceptance before a removal PR: final recovery integration, final recovery first-producer paths through assembled wiring, enabled empty/configured/history quiet including the real 90-second window, scheduling suppression, startup and shutdown races, relevant contracts across the combined tree, current-head diff --git a/server/src/__tests__/helpers/maintenance/server-child.ts b/server/src/__tests__/helpers/maintenance/server-child.ts index 029b0762f0c..218f7a5bb74 100644 --- a/server/src/__tests__/helpers/maintenance/server-child.ts +++ b/server/src/__tests__/helpers/maintenance/server-child.ts @@ -1,5 +1,6 @@ import { channel } from "node:diagnostics_channel"; import { mock } from "node:test"; +import type { Db } from "@paperclipai/db"; import type { Socket } from "node:net"; const realTimeout = globalThis.setTimeout, realInterval = globalThis.setInterval, clearRealInterval = globalThis.clearInterval, realNow = Date.now; @@ -44,8 +45,9 @@ try { const { idleWorkSnapshot, startTaskDrain, stopTaskDrain } = await import("../../../services/task-admission.js"); const { getStartupRecoveryState } = await import("../../../startup-recovery-state.js"); const server = await startServer(); + const app = server.server.listeners("request")[0] as unknown as { locals: { paperclipDb: Db } }; process.send?.({ kind: "ready", port: server.listenPort }); - process.on("message", async (message: { id: number; command: string; ms?: number }) => { + process.on("message", async (message: { id: number; command: string; ms?: number; issueUpdate?: { id: string; status: "done" }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } }) => { try { if (message.command === "snapshot" || message.command === "advance" || message.command === "drain" || message.command === "resume" || message.command === "freeze" || message.command === "realtime") { if (message.command === "freeze" && bootClock) { clearRealInterval(bootClock); bootClock = realInterval(() => mock.timers.tick(0), 10); } @@ -58,9 +60,20 @@ try { if (message.command === "resume") stopTaskDrain(); if (fakeClock) mock.timers.tick(0); process.send?.({ id: message.id, kind: "snapshot", idle: idleWorkSnapshot(), startup: getStartupRecoveryState(), outgoingHttpRequests, network: network(), now: Date.now() }); + } else if (message.command === "updateIssue") { + if (!message.issueUpdate) throw new Error("Missing issue update"); + const { issueService } = await import("../../../services/issues.js"); + await issueService(app.locals.paperclipDb).update(message.issueUpdate.id, { status: message.issueUpdate.status }); + process.send?.({ id: message.id, kind: "updated" }); + } else if (message.command === "publishBackup") { + if (!message.publication) throw new Error("Missing backup publication"); + const { publishNativeHarnessBackup } = await import("../../../services/native-runtime/native-harness-backup-publication.js"); + await publishNativeHarnessBackup({ db: app.locals.paperclipDb, ...message.publication }); + process.send?.({ id: message.id, kind: "published" }); } else if (message.command === "shutdown") { - if (bootClock) { clearRealInterval(bootClock); bootClock = undefined; } + if (fakeClock) runClock(); await server.shutdown(); + if (bootClock) { clearRealInterval(bootClock); bootClock = undefined; } process.send?.({ id: message.id, kind: "stopped", outgoingHttpRequests }); if (fakeClock) mock.timers.reset(); await new Promise(resolve => realTimeout(resolve, 20)); diff --git a/server/src/__tests__/helpers/maintenance/server-harness.ts b/server/src/__tests__/helpers/maintenance/server-harness.ts index e438420bb6b..169ededa625 100644 --- a/server/src/__tests__/helpers/maintenance/server-harness.ts +++ b/server/src/__tests__/helpers/maintenance/server-harness.ts @@ -18,7 +18,7 @@ async function withDeadline(promise: Promise, ms: number, message: () => s type Snapshot = { idle: { active: number; generation: number }; startup: { phase: string }; outgoingHttpRequests: number; network: { nonDatabaseConnections: number; databaseConnections: number; nonDatabaseBytesWritten: number }; now: number }; -export async function startMaintenanceServer(databaseUrl: string, options: { enabled?: boolean; clock?: boolean; suppressed?: boolean; githubOrigin?: string; cloud?: { jwks: string; s3Origin: string; standby: boolean } } = {}) { +export async function startMaintenanceServer(databaseUrl: string, options: { enabled?: boolean; clock?: boolean; suppressed?: boolean; githubOrigin?: string; connector?: { origin: string; signPrivateKey: string; sealPrivateKey: string }; cloud?: { jwks: string; s3Origin: string; standby: boolean } } = {}) { const proxy = await startMaintenancePostgresProxy(databaseUrl); const directory = await mkdtemp(path.join(tmpdir(), "paperclip-maintenance-")); const reservation = createServer(); @@ -43,6 +43,7 @@ export async function startMaintenanceServer(databaseUrl: string, options: { ena env: { PATH: process.env.PATH, TMPDIR: process.env.TMPDIR, NODE_ENV: "production", PAPERCLIP_HOME: directory, PAPERCLIP_CONFIG: configPath, PAPERCLIP_INSTANCE_ID: "maintenance-test", + PAPERCLIP_RUNNER_STATE_DIR: path.join(directory, "durable-sessions"), DATABASE_URL: proxy.url, DATABASE_POOL_MAX: "6", DATABASE_APPLICATION_NAME: "maintenance-integration-test", PAPERCLIP_AGENT_JWT_SECRET: randomUUID(), PAPERCLIP_DECISION_SIGNING_SECRET: randomUUID(), PAPERCLIP_TOOL_ACTION_SIGNING_SECRET: randomUUID(), BETTER_AUTH_SECRET: randomUUID(), @@ -52,6 +53,12 @@ export async function startMaintenanceServer(databaseUrl: string, options: { ena PAPERCLIP_MAINTENANCE_FAKE_CLOCK: String(options.clock ?? false), ...(options.suppressed ? { PAPERCLIP_IN_WORKTREE: "true" } : {}), ...(options.githubOrigin ? { PAPERCLIP_MAINTENANCE_GITHUB_ORIGIN: options.githubOrigin } : {}), + ...(options.connector ? { + PAPERCLIP_CLOUD_CONNECTOR_INSTANCE_ID: "maintenance-connector", PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT: "test", + PAPERCLIP_CLOUD_CONNECTOR_SIGN_PRIVATE_KEY: options.connector.signPrivateKey, + PAPERCLIP_CLOUD_CONNECTOR_SEAL_PRIVATE_KEY: options.connector.sealPrivateKey, + PAPERCLIP_CLOUD_CONNECTOR_BASE_URL: options.connector.origin, + } : {}), ...(options.cloud ? { PAPERCLIP_MANAGED_CONFIG: JSON.stringify({ v: 1, mode: "cloud", catalogVersion: "2026.720.0", features: {}, plugins: { autoInstall: [] } }), PAPERCLIP_SECRETS_MASTER_KEY: randomBytes(32).toString("hex"), @@ -86,7 +93,7 @@ export async function startMaintenanceServer(databaseUrl: string, options: { ena for (const request of requests.values()) request.reject(error); requests.clear(); }); - async function request(command: string, payload: { ms?: number } = {}): Promise { + async function request(command: string, payload: { ms?: number; issueUpdate?: { id: string; status: "done" }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } } = {}): Promise { const id = ++requestId; const result = new Promise((resolve, reject) => requests.set(id, { resolve, reject })); child.send({ id, command, ...payload }); @@ -107,16 +114,22 @@ export async function startMaintenanceServer(databaseUrl: string, options: { ena const port = await withDeadline(ready, 90_000, () => `Maintenance boot timed out: ${output.slice(-12_000)}`); return { proxy, directory, port, stop, - async api(method: string, route: string, body?: unknown): Promise { + async api(method: string, route: string, body?: unknown, headers: Record = {}): Promise { if (options.clock) await request("realtime"); try { const response = await fetch(`http://127.0.0.1:${port}/api${route}`, { method, signal: AbortSignal.timeout(30_000), - headers: { "Content-Type": "application/json" }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }); + headers: { "Content-Type": "application/json", ...headers }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }); const value = await response.json(); if (!response.ok) throw new Error(`Maintenance API ${method} ${route}: ${response.status} ${JSON.stringify(value)}`); return value as T; } finally { if (options.clock) await request("freeze"); } }, + async updateIssue(issueUpdate: { id: string; status: "done" }) { + if (options.clock) await request("realtime"); + try { return await request("updateIssue", { issueUpdate }); } + finally { if (options.clock) await request("freeze"); } + }, + publishBackup: (publication: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string }) => request("publishBackup", { publication }), snapshot: () => request("snapshot"), drain: () => request("drain"), resume: () => request("resume"), advance: (ms: number) => request("advance", { ms }), freeze: () => request("freeze"), diff --git a/server/src/__tests__/maintenance-server-routes.test.ts b/server/src/__tests__/maintenance-server-routes.test.ts index 0f98f927163..ac5b4e448d2 100644 --- a/server/src/__tests__/maintenance-server-routes.test.ts +++ b/server/src/__tests__/maintenance-server-routes.test.ts @@ -1,7 +1,7 @@ -import { generateKeyPairSync, sign } from "node:crypto"; +import { createCipheriv, createHash, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, sign, verify } from "node:crypto"; import { createServer } from "node:http"; -import { createDb } from "@paperclipai/db"; -import { writeFile } from "node:fs/promises"; +import { agents, agentApiKeys, agentTaskSessions, authUsers, companies, companyMemberships, connectionGrants, createDb, heartbeatRuns, toolApplications, toolConnections } from "@paperclipai/db"; +import { access, mkdir, readdir, utimes, writeFile } from "node:fs/promises"; import path from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; @@ -21,6 +21,14 @@ async function clearDatabase() { if (tables.length) await fixture.$client.unsafe(`truncate ${tables.map(table => `"${table.tablename.replaceAll('"', '""')}"`).join(",")} cascade`); } finally { await fixture.$client.end(); } } +async function settleClock(server: Awaited>) { + for (let step = 0; step < 60; step++) { + if (server.proxy.pending === 0 && (await server.snapshot()).idle.active === 0) return server.settled(100); + if (server.proxy.pending === 0) await server.advance(1_000); + await delay(50); + } + return server.settled(); +} function receipt(queries: Array<{ protocol: string; sql: string; fingerprint: string }>) { const groups = new Map(); for (const query of queries) { @@ -103,6 +111,56 @@ describe("assembled maintenance server production wiring", () => { } finally { await server.stop(); } }, 150_000); + it("settles an in-flight provider claim on shutdown and reconstructs its unfinished object after restart", async () => { + await clearDatabase(); + let arrived!: () => void, cancelled = false; + const firstRequest = new Promise(resolve => { arrived = resolve; }); + const providerRequests: string[] = []; + const provider = createServer((req, res) => { + providerRequests.push(req.url!); + if (providerRequests.length === 1) { res.once("close", () => { cancelled = true; }); arrived(); return; } + res.setHeader("Content-Type", "application/json"); + res.end(JSON.stringify({ id: 8, number: 8, state: "closed", merged: true, merged_at: new Date().toISOString(), + title: "Recovered object", html_url: "https://github.com/maintenance/example/pull/8", user: { login: "fixture" } })); + }); + await new Promise(resolve => provider.listen(0, "127.0.0.1", resolve)); + const address = provider.address(); + if (!address || typeof address === "string") throw new Error("Simulated GitHub failed to bind"); + const options = { enabled: false, clock: true, githubOrigin: `http://127.0.0.1:${address.port}` }; + let server: Awaited> | undefined = await startMaintenanceServer(databaseUrl!, options); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await server.settled(); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Shutdown recovery" }); + await server.api("PATCH", "/instance/settings/experimental", { enableExternalObjects: true }); + const issue = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { + title: "Object survives shutdown", status: "done", description: "https://github.com/maintenance/example/pull/8", + }); + await server.advance(1_000); + await Promise.race([firstRequest, delay(10_000).then(() => { throw new Error("Provider claim did not arrive"); })]); + const [claimed] = await fixture.$client`select is_terminal, refresh_started_at is not null as claimed from external_objects + where company_id=${company.id}`; + expect(claimed).toMatchObject({ is_terminal: false, claimed: true }); + await server.stop(); server = undefined; + expect(cancelled).toBe(true); + const [stopped] = await fixture.$client`select is_terminal, refresh_started_at, refresh_token from external_objects where company_id=${company.id}`; + expect(stopped).toMatchObject({ is_terminal: false, refresh_started_at: null, refresh_token: null }); + server = await startMaintenanceServer(databaseUrl!, options); + await settleClock(server); + const [recovered] = await fixture.$client`select is_terminal, status_key from external_objects where company_id=${company.id}`; + expect(recovered).toMatchObject({ is_terminal: true, status_key: "merged" }); + expect(providerRequests).toEqual(["/repos/maintenance/example/pulls/8", "/repos/maintenance/example/pulls/8"]); + await writeFile(path.join(server.directory, "shutdown-measurement.json"), JSON.stringify({ phase: "actual-shutdown-inflight-and-restart", + companyId: company.id, issueId: issue.id, claimed, cancelled, stopped, recovered, providerRequests, + boundary: "first simulated HTTP response deliberately withheld; actual shutdown abort/settlement and real restart reconstruction", + frames: server.proxy.frames, + }, null, 2)); + } finally { + await server?.stop(); await fixture.$client.end(); provider.closeAllConnections(); + await new Promise(resolve => provider.close(() => resolve())); + } + }, 150_000); + it("processes the first ordinary external-object producer committed during idle drain and becomes quiet again", async () => { await clearDatabase(); const providerRequests: string[] = []; @@ -206,6 +264,328 @@ describe("assembled maintenance server production wiring", () => { } }, 150_000); + it("discovers and acknowledges a sealed GitHub event after the first ordinary connection activation", async () => { + await clearDatabase(); + const signing = generateKeyPairSync("ed25519"), sealing = generateKeyPairSync("x25519"); + const requests: string[] = [], acknowledgements: unknown[] = []; + let event: Record | undefined, leased = false; + const broker = createServer(async (req, res) => { + const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(Buffer.from(chunk)); + const body = JSON.parse(Buffer.concat(chunks).toString()); + const [header, payload, signature] = body.request.split("."); + if (!verify(null, Buffer.from(`${header}.${payload}`), signing.publicKey, Buffer.from(signature, "base64url"))) { + res.statusCode = 401; res.end("{}"); return; + } + requests.push(req.url!); res.setHeader("Content-Type", "application/json"); + if (req.url === "/v1/connector/events/ack") { acknowledgements.push(JSON.parse(body.acknowledgement)); res.end("{}"); return; } + if (req.url !== "/v1/connector/events/lease") { res.statusCode = 500; res.end("{}"); return; } + if (!event || leased) { res.end(JSON.stringify({ events: [], eventWakeVersion: 1 })); return; } + leased = true; + const ephemeral = generateKeyPairSync("x25519"), algorithm = "X25519-HKDF-SHA256-A256GCM"; + const epk = Buffer.from(ephemeral.publicKey.export({ format: "jwk" }).x!, "base64url"); + const recipient = Buffer.from(sealing.publicKey.export({ format: "jwk" }).x!, "base64url"); + const aad = Buffer.from([1, algorithm, "events", "maintenance-connector", "test", "github", "github.code", ""].join("\n")); + const key = Buffer.from(hkdfSync("sha256", diffieHellman({ privateKey: ephemeral.privateKey, publicKey: sealing.publicKey }), Buffer.concat([epk, recipient]), aad, 32)); + const iv = randomBytes(12), cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad); + const clear = { v: 1, instanceId: "maintenance-connector", environment: "test", leaseId: "maintenance-lease", events: [event] }; + const ct = Buffer.concat([cipher.update(JSON.stringify(clear)), cipher.final(), cipher.getAuthTag()]); + res.end(JSON.stringify({ leaseId: "maintenance-lease", eventWakeVersion: 1, sealed: { v: 1, alg: algorithm, + purpose: "events", provider: "github", profile: "github.code", epk: epk.toString("base64url"), iv: iv.toString("base64url"), ct: ct.toString("base64url") } })); + }); + await new Promise(resolve => broker.listen(0, "127.0.0.1", resolve)); + const address = broker.address(); + if (!address || typeof address === "string") throw new Error("Simulated Cloud broker failed to bind"); + const server = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true, + githubOrigin: `http://127.0.0.1:${address.port}`, connector: { + origin: `http://127.0.0.1:${address.port}`, signPrivateKey: signing.privateKey.export({ format: "jwk" }).d!, + sealPrivateKey: sealing.privateKey.export({ format: "jwk" }).d!, + } }); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await server.settled(); expect(requests).toEqual([]); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "GitHub event ingress" }); + const [application] = await fixture.insert(toolApplications).values({ companyId: company.id, applicationKey: randomUUID(), name: "GitHub", type: "mcp_http", status: "active" }).returning(); + const [connection] = await fixture.insert(toolConnections).values({ companyId: company.id, applicationId: application!.id, + uid: randomUUID(), name: "GitHub", status: "active", enabled: false, transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", + config: { url: "https://api.github.com/mcp", sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } }, transportConfig: {} }).returning(); + const now = (await server.snapshot()).now; + const [grant] = await fixture.insert(connectionGrants).values({ companyId: company.id, connectionId: connection!.id, + kind: "user", subjectUserId: "maintenance-owner", status: "active", providerTenant: { + oauth: { accessTokenExpiresAt: new Date(now + 8 * 3_600_000).toISOString(), refreshedAt: new Date(now).toISOString() }, + github: { userId: "42", login: "fixture", installationIds: ["101"], installationCount: 1, repositoryCount: 1, + repositorySelection: "all", installationOwnerLogins: ["fixture"], accessRevision: randomUUID(), lastAccessRefreshAt: new Date(now).toISOString(), webhookHealth: "healthy" }, + } }).returning(); + event = { id: "maintenance-delivery", provider: "github", event: "pull_request", action: "closed", installationId: "101", repositoryId: "99", + createdAt: new Date(now).toISOString(), bindingIds: [`${grant!.id}_101`], payload: { repository: "maintenance/example", number: 7, + url: "https://github.com/maintenance/example/pull/7", state: "closed", merged: true, mergedAt: new Date(now).toISOString() } }; + await server.api("PATCH", `/tool-connections/${connection!.id}`, { enabled: true }); + await server.advance(5_000); await settleClock(server); + const [delivery] = await fixture.$client`select company_id, status, attempts, processed_at is not null as processed, normalized_payload + from connection_event_deliveries where provider_delivery_id='maintenance-delivery'`; + expect(delivery).toMatchObject({ company_id: company.id, status: "processed", attempts: 1, processed: true, + normalized_payload: expect.objectContaining({ repository: "maintenance/example", number: 7, merged: true }) }); + expect(acknowledgements).toHaveLength(1); + const baseline = server.proxy.queries.length, requestCount = requests.length; + for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(baseline); expect(requests).toHaveLength(requestCount); + await writeFile(path.join(server.directory, "github-measurement.json"), JSON.stringify({ phase: "ordinary-connection-activation-sealed-event", + companyId: company.id, connectionId: connection!.id, grantId: grant!.id, delivery, requests, acknowledgements, + quietClockMs: 120_000, quietQueries: 0, frames: server.proxy.frames, + providerBoundary: "simulated Cloud HTTP validates real Ed25519 request and seals X25519/AES-GCM event; actual runtime decrypts/persists/acks", + }, null, 2)); + } finally { + await fixture.$client.end(); await server.stop(); broker.closeAllConnections(); + await new Promise(resolve => broker.close(() => resolve())); + } + }, 150_000); + + it("reacts to the first ordinary status-card source write and accepts authorized controlled summarizer output", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true }); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await server.settled(); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Status-card ingress" }); + const [agent] = await fixture.insert(agents).values({ companyId: company.id, name: "Controlled summarizer", + status: "idle", adapterType: "process", adapterConfig: { command: "/usr/bin/false" } }).returning(); + const userId = randomUUID(); + await fixture.insert(authUsers).values({ id: userId, name: "Controlled writer owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await fixture.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: userId, status: "active", membershipRole: "member" }); + const token = randomUUID(); + await fixture.insert(agentApiKeys).values({ companyId: company.id, agentId: agent!.id, name: "Controlled writer", responsibleUserId: userId, + keyHash: createHash("sha256").update(token).digest("hex") }); + await server.api("PATCH", "/instance/settings/experimental", { enableStatusCards: true }); + const card = await server.api<{ id: string; generatingIssueId: string }>("POST", `/companies/${company.id}/status-cards`, { + interestPrompt: "Blocked tasks", agentId: agent!.id, refreshPolicy: { mode: "reactive", debounceSeconds: 60 }, + }); + async function writeback(generationIssueId: string, compile: boolean) { + const [run] = await fixture.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent!.id, + status: "running", responsibleUserId: userId, contextSnapshot: { issueId: generationIssueId } }).returning(); + await fixture.$client`update issues set checkout_run_id=${run!.id} where id=${generationIssueId}`; + const headers = { authorization: `Bearer ${token}`, "x-paperclip-run-id": run!.id }; + if (compile) await server.api("PUT", `/status-cards/${card.id}/query`, { queries: [{ scope: "issues", status: ["blocked"] }], + title: "Blocked tasks", changeSummary: "Controlled compilation", generationIssueId }, headers); + await server.api("PUT", `/status-cards/${card.id}/summary`, { markdown: "Controlled current blockers", + changeSummary: "Controlled summary", generationIssueId }, headers); + await fixture.$client`update heartbeat_runs set status='succeeded', finished_at=now() where id=${run!.id}`; + await server.api("PATCH", `/issues/${generationIssueId}`, { status: "done" }); + } + await writeback(card.generatingIssueId, true); await server.advance(30_000); await settleClock(server); + const [compiled] = await fixture.$client`select state, query_version, generating_issue_id from status_cards where id=${card.id}`; + expect(compiled).toMatchObject({ state: "active", query_version: 1, generating_issue_id: null }); + const changed = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { + title: "Ordinary new blocker", status: "blocked", + }); + await server.advance(1_000); await settleClock(server); + const [pending] = await fixture.$client`select pending_change_count, next_eval_at is not null as scheduled from status_cards where id=${card.id}`; + expect(pending!.pending_change_count).toBeGreaterThan(0); expect(pending!.scheduled).toBe(true); + for (let step = 0; step < 3; step++) { await server.advance(30_000); await settleClock(server); } + const [generating] = await fixture.$client`select generating_issue_id from status_cards where id=${card.id}`; + expect(generating!.generating_issue_id).toBeTypeOf("string"); + expect(generating!.generating_issue_id).not.toBe(card.generatingIssueId); + await writeback(generating!.generating_issue_id, false); await server.advance(30_000); await settleClock(server); + const [result] = await fixture.$client`select status, finished_at is not null as finished from status_card_updates + where card_id=${card.id} and generation_issue_id=${generating!.generating_issue_id}`; + expect(result).toMatchObject({ status: "ok", finished: true }); + const [saved] = await fixture.$client`select pending_change_count, generating_issue_id, state from status_cards where id=${card.id}`; + expect(saved).toMatchObject({ pending_change_count: 0, generating_issue_id: null, state: "active" }); + await writeFile(path.join(server.directory, "status-card-measurement.json"), JSON.stringify({ + phase: "ordinary-reactive-card-source-and-authorized-writeback", companyId: company.id, cardId: card.id, issueId: changed.id, + compiled, pending, generating, result, saved, frames: server.proxy.frames, + executionBoundary: "controlled summarizer execution rows/key only; real authenticated query/summary routes; production execution suppressed; no model provider", + }, null, 2)); + } finally { await fixture.$client.end(); await server.stop(); } + }, 150_000); + + it("expires an ordinarily published backup after its ordinary task close without losing run history", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true }); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await server.settled(); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Backup ingress" }); + const [agent] = await fixture.insert(agents).values({ companyId: company.id, name: "Backup owner", status: "paused" }).returning(); + const issue = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { title: "Close aged backup work", status: "todo" }); + const sessionId = randomUUID(), scopeDigest = "b".repeat(64); + const [run] = await fixture.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent!.id, + status: "succeeded", issueId: issue.id, nativeIssueId: issue.id, nativeSessionId: sessionId, finishedAt: new Date() }).returning(); + await fixture.insert(agentTaskSessions).values({ companyId: company.id, agentId: agent!.id, adapterType: "paperclip_runner", + taskKey: issue.id, sessionParamsJson: { sessionId }, lastRunId: run!.id }); + const root = path.join(server.directory, "durable-sessions", scopeDigest), backupRoot = path.join(root, "failover-backups"); + const pendingRoot = path.join(backupRoot, "pending-publication"), current = path.join(backupRoot, "current"); + await mkdir(path.join(root, "control-plane"), { recursive: true }); await mkdir(pendingRoot, { recursive: true }); + await writeFile(path.join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ identity: { runId: run!.id, normalizedSessionId: sessionId } })); + await writeFile(path.join(pendingRoot, "manifest.json"), "{}"); await writeFile(path.join(pendingRoot, "session-bytes"), "retained session bytes"); + const old = new Date((await server.snapshot()).now - 10 * 86_400_000); await utimes(pendingRoot, old, old); + await server.publishBackup({ companyId: company.id, scopeDigest, backupRoot, pendingRoot }); + await settleClock(server); await access(current); + await fixture.$client.unsafe(`create or replace function maintenance_age_close() returns trigger language plpgsql as $$begin + if new.title = 'Close aged backup work' and new.status = 'done' then new.completed_at = now() - interval '10 days'; end if; + return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_age_close before update on issues for each row execute function maintenance_age_close()`); + await server.api("PATCH", `/issues/${issue.id}`, { status: "done" }); + await server.advance(30_000); await settleClock(server); + await expect(access(current)).rejects.toMatchObject({ code: "ENOENT" }); + expect(await readdir(backupRoot)).toEqual([]); + const [pointers] = await fixture.$client`select count(*)::integer as count from agent_task_sessions where agent_id=${agent!.id}`; + const [history] = await fixture.$client`select status from heartbeat_runs where id=${run!.id}`; + expect(pointers!.count).toBe(0); expect(history!.status).toBe("succeeded"); + await writeFile(path.join(server.directory, "backup-measurement.json"), JSON.stringify({ phase: "ordinary-publication-and-task-close", + companyId: company.id, issueId: issue.id, runId: run!.id, scopeDigest, pointers: pointers!.count, history, + timestampBoundary: "fixture ages backup mtime and ordinary close timestamp10days; cleanup outcome not seeded", + publicationBoundary: "actual publishNativeHarnessBackup with actual runtime-owned DB; no provider execution", frames: server.proxy.frames, + }, null, 2)); + } finally { + await fixture.$client.unsafe("drop trigger if exists maintenance_age_close on issues").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_age_close()").catch(() => {}); + await fixture.$client.end(); await server.stop(); + } + }, 150_000); + + it("archives the first ordinary overdue attention source through the assembled retention worker", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true }); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await server.settled(); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Attention ingress" }); + const issue = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { title: "Await human decision", status: "todo" }); + await fixture.$client.unsafe(`create or replace function maintenance_age_attention() returns trigger language plpgsql as $$begin + if new.title = 'Overdue ordinary human decision' then new.updated_at = now() - interval '91 days'; end if; + return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_age_attention before insert on issue_thread_interactions for each row execute function maintenance_age_attention()`); + const card = await server.api<{ id: string }>("POST", `/issues/${issue.id}/interactions`, { + kind: "request_confirmation", title: "Overdue ordinary human decision", resolverPolicy: "human_only", + continuationPolicy: "none", payload: { version: 1, prompt: "Choose the preferred color?" }, + }); + await server.advance(30_000); await settleClock(server); + const [archived] = await fixture.$client`select company_id, source_kind, archived_reason, archived_by_type, + archived_at is not null as archived, archive_version from decision_retention where source_id=${card.id}`; + expect(archived).toMatchObject({ company_id: company.id, source_kind: "issue_thread_interaction", archived_reason: "idle_ttl", + archived_by_type: "system", archived: true, archive_version: 1 }); + const [source] = await fixture.$client`select status, updated_at < now() - interval '90 days' as aged from issue_thread_interactions where id=${card.id}`; + expect(source).toMatchObject({ status: "pending", aged: true }); + const audit = await fixture.$client`select actor_id, details from activity_log where company_id=${company.id} + and action='decision_retention.auto_archived'`; + expect(audit).toHaveLength(1); + await writeFile(path.join(server.directory, "attention-measurement.json"), JSON.stringify({ + phase: "ordinary-overdue-attention-ingress", companyId: company.id, issueId: issue.id, interactionId: card.id, + timestampBoundary: "fixture-only INSERT trigger ages source activity91days; no retention/outcome rows seeded", + archived, source, audit, frames: server.proxy.frames, + }, null, 2)); + } finally { + await fixture.$client.unsafe("drop trigger if exists maintenance_age_attention on issue_thread_interactions").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_age_attention()").catch(() => {}); + await fixture.$client.end(); await server.stop(); + } + }, 150_000); + + it("accepts the first ordinary merge confirmation through the assembled provider worker", async () => { + await clearDatabase(); + const providerRequests: string[] = []; + const provider = createServer((req, res) => { + providerRequests.push(req.url!); res.setHeader("Content-Type", "application/json"); + if (req.url !== "/repos/maintenance/example/pulls/39") { res.statusCode = 500; res.end("{}"); return; } + res.end(JSON.stringify({ id: 39, number: 39, state: "closed", merged: true, merged_at: new Date().toISOString(), + title: "Controlled merged confirmation", html_url: "https://github.com/maintenance/example/pull/39", user: { login: "fixture" } })); + }); + await new Promise(resolve => provider.listen(0, "127.0.0.1", resolve)); + const address = provider.address(); + if (!address || typeof address === "string") throw new Error("Simulated GitHub failed to bind"); + const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true, + githubOrigin: `http://127.0.0.1:${address.port}` }); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await server.settled(); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Merge ingress" }); + const issue = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { title: "Merge task", status: "todo" }); + const card = await server.api<{ id: string }>("POST", `/issues/${issue.id}/interactions`, { + kind: "request_confirmation", title: "Merge maintenance/example#39?", resolverPolicy: "human_only", + continuationPolicy: "none", payload: { version: 1, prompt: "Merge the PR?" }, + }); + await server.advance(30_000); await settleClock(server); + const [accepted] = await fixture.$client`select company_id, status, resolved_at is not null as resolved + from issue_thread_interactions where id=${card.id}`; + expect(accepted).toMatchObject({ company_id: company.id, status: "accepted", resolved: true }); + const audit = await fixture.$client`select actor_id, details from activity_log where company_id=${company.id} + and action='issue.thread_interaction_accepted'`; + expect(audit).toEqual([expect.objectContaining({ actor_id: "system:pr-merged", details: expect.objectContaining({ interactionId: card.id }) })]); + expect(providerRequests).toEqual(["/repos/maintenance/example/pulls/39"]); + await writeFile(path.join(server.directory, "confirmation-measurement.json"), JSON.stringify({ + phase: "ordinary-merge-confirmation", companyId: company.id, issueId: issue.id, interactionId: card.id, + accepted, audit, providerRequests, frames: server.proxy.frames, + providerBoundary: "simulated GitHub HTTP; actual production worker and interaction acceptance", + }, null, 2)); + } finally { + await fixture.$client.end(); await server.stop(); provider.closeAllConnections(); + await new Promise(resolve => provider.close(() => resolve())); + } + }, 150_000); + + it("projects the first ordinary routine completion in the actual suppressed Cloud runtime", async () => { + await clearDatabase(); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + const companyId = randomUUID(), agentId = randomUUID(); + await fixture.insert(companies).values({ id: companyId, name: "Routine ingress", issuePrefix: "MRT", + defaultResponsibleUserId: "maintenance-board", requireBoardApprovalForNewAgents: false }); + await fixture.insert(agents).values({ id: agentId, companyId, name: "Suppressed routine worker", status: "active", + lifecycleState: "ready", adapterType: "process", adapterConfig: { command: "/usr/bin/false" } }); + const pair = generateKeyPairSync("ed25519"); + const publicJwk = { ...pair.publicKey.export({ format: "jwk" }), kid: "maintenance-routine-key", use: "sig", alg: "EdDSA" }; + const storageRequests: string[] = []; + const storage = createServer((req, res) => { storageRequests.push(req.url!); res.statusCode = 500; res.end("Unexpected S3 request in routine fixture"); }); + await new Promise(resolve => storage.listen(0, "127.0.0.1", resolve)); + const address = storage.address(); + if (!address || typeof address === "string") throw new Error("Simulated S3 failed to bind"); + const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true, + cloud: { jwks: JSON.stringify({ keys: [publicJwk] }), s3Origin: `http://127.0.0.1:${address.port}`, standby: false } }); + try { + await server.settled(); + const routine = await server.api<{ id: string }>("POST", `/companies/${companyId}/routines`, { + title: "Ordinary projected routine", assigneeAgentId: agentId, concurrencyPolicy: "always_enqueue", + }); + const run = await server.api<{ id: string; linkedIssueId: string; status: string }>("POST", `/routines/${routine.id}/run`, { + source: "manual", idempotencyKey: randomUUID(), + }); + expect(run.status).toBe("issue_created"); + const [before] = await fixture.$client`select status, completed_at from routine_runs where id=${run.id}`; + expect(before).toMatchObject({ status: "issue_created", completed_at: null }); + await fixture.$client.unsafe(`create or replace function maintenance_hold_completion() returns trigger language plpgsql as $$begin + if new.id = '${run.linkedIssueId}'::uuid and new.status = 'done' then perform pg_advisory_xact_lock(913702); end if; + return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_hold_completion before update on issues for each row execute function maintenance_hold_completion()`); + const reserved = await fixture.$client.reserve(); + await reserved`select pg_advisory_lock(913702)`; + const writesBefore = server.proxy.queries.length; + const completing = server.updateIssue({ id: run.linkedIssueId, status: "done" }); + void completing.catch(() => {}); + try { + await expect.poll(() => server.proxy.queries.slice(writesBefore).some(query => /update "issues"/.test(query.sql)), { timeout: 10_000 }).toBe(true); + await server.drain(); + } finally { await reserved`select pg_advisory_unlock(913702)`; reserved.release(); } + await completing; await server.advance(30_000); await settleClock(server); + const [drained] = await fixture.$client`select status, completed_at from routine_runs where id=${run.id}`; + expect(drained).toMatchObject({ status: "issue_created", completed_at: null }); + await server.resume(); await server.advance(30_000); await settleClock(server); + const [projected] = await fixture.$client`select rr.company_id, rr.status, rr.completed_at is not null as completed, + i.status as issue_status from routine_runs rr join issues i on i.id=rr.linked_issue_id where rr.id=${run.id}`; + expect(projected).toMatchObject({ company_id: companyId, status: "completed", completed: true, issue_status: "done" }); + const [executions] = await fixture.$client`select count(*)::integer as count from heartbeat_runs where company_id=${companyId}`; + expect(executions!.count).toBe(0); + expect(storageRequests).toEqual([]); + await writeFile(path.join(server.directory, "routine-measurement.json"), JSON.stringify({ + phase: "ordinary-routine-service-completion-during-drain", completionBoundary: "actual issueService.update with runtime-owned DB; excludes redundant route projection hook", companyId, routineId: routine.id, runId: run.id, + before, drained, projected, executions: executions!.count, storageRequests, + suppression: "PAPERCLIP_IN_WORKTREE; actual production policy", frames: server.proxy.frames, + }, null, 2)); + } finally { + await fixture.$client.unsafe("drop trigger if exists maintenance_hold_completion on issues").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_hold_completion()").catch(() => {}); + await server.stop(); await fixture.$client.end(); storage.closeAllConnections(); + await new Promise(resolve => storage.close(() => resolve())); + } + }, 150_000); + it("keeps an actual Cloud warm standby quiet and activates after a normally signed durable claim", async () => { await clearDatabase(); const pair = generateKeyPairSync("ed25519");