From 681f7b77262a9df6c1b4aad55291bf960061e491 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:22:31 -0500 Subject: [PATCH] perf(server): retire aggregate maintenance polling (#18425) Reviewed and verified at 901c4c4bdc2b9345ae0c4b121b285301cb3af9d3. Co-Authored-By: Paperclip --- .../2026-10-30-aggregate-maintenance-sleep.md | 225 +++++ .../__tests__/agent-live-run-routes.test.ts | 25 +- .../__tests__/decision-target-work.test.ts | 39 +- .../helpers/maintenance/environment.ts | 24 + .../helpers/maintenance/postgres-proxy.ts | 109 +++ .../helpers/maintenance/runtime-build.ts | 21 + .../helpers/maintenance/server-child.ts | 99 ++ .../helpers/maintenance/server-harness.ts | 218 +++++ .../helpers/maintenance/tsconfig.json | 8 + .../maintenance-harness-resources.test.ts | 127 +++ .../maintenance-postgres-proxy.test.ts | 115 +++ .../maintenance-server-routes.test.ts | 926 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 43 +- server/src/index.ts | 45 - server/src/services/decision-target-work.ts | 6 +- .../heartbeat-run-runtime-status.test.ts | 40 +- .../services/heartbeat-run-runtime-status.ts | 15 + server/src/services/heartbeat.ts | 2 - .../runner-api.integration.test.ts | 28 +- 19 files changed, 2022 insertions(+), 93 deletions(-) create mode 100644 doc/plans/2026-10-30-aggregate-maintenance-sleep.md create mode 100644 server/src/__tests__/helpers/maintenance/environment.ts create mode 100644 server/src/__tests__/helpers/maintenance/postgres-proxy.ts create mode 100644 server/src/__tests__/helpers/maintenance/runtime-build.ts create mode 100644 server/src/__tests__/helpers/maintenance/server-child.ts create mode 100644 server/src/__tests__/helpers/maintenance/server-harness.ts create mode 100644 server/src/__tests__/helpers/maintenance/tsconfig.json create mode 100644 server/src/__tests__/maintenance-harness-resources.test.ts create mode 100644 server/src/__tests__/maintenance-postgres-proxy.test.ts create mode 100644 server/src/__tests__/maintenance-server-routes.test.ts diff --git a/doc/plans/2026-10-30-aggregate-maintenance-sleep.md b/doc/plans/2026-10-30-aggregate-maintenance-sleep.md new file mode 100644 index 00000000000..aaad73edf7a --- /dev/null +++ b/doc/plans/2026-10-30-aggregate-maintenance-sleep.md @@ -0,0 +1,225 @@ +# Aggregate maintenance interval integration + +## Delivery boundary + +Remove the catch-all heartbeat maintenance interval only after its domain duties +have replacement owners. Keep existing routine, agent-timer and issue-monitor +scheduling, feature toggles, plugin readiness, environment cleanup, idle tracking, +hot-restart shutdown, and conservative sleep-safety checks. This work does not +qualify actual database or host suspension, Cloud alarms, or cost savings. + +The preceding domain migrations retain the aggregate interval. This change +removes its enabled and disabled paths after those owners are assembled. The +acceptance harness first calibrated its observation boundary against the original +recurring SQL, then verified the final source without that interval. + +## Duty ownership + +| Former interval duty | Owning replacement | +| --- | --- | +| Attention retention | Attention eligibility, committed source notifications and actual archive deadlines | +| Run/task recovery | Run dispatch, issue recovery, watchdogs, locks, residual live/orphan obligations | +| Routine execution projection | Routine execution-status notifications and projection debt | +| GitHub discovery/credentials | Durable supported event ingress, enrollment and credential continuity | +| Merged PR confirmations | Actual unresolved cards, provider ingress and scoped fallback | +| External objects | Supported source/configuration producers, exact refresh and retry deadlines | +| Status cards | Feature-owned scheduled/reactive evaluation, claims and assignment sequencing | +| Native backup expiration | Existing backup/session/file records and source-owned retention deadlines | +| Ephemeral runtime status | Expiry on read, explicit terminal clear and time-gated cleanup on output writes | + +The in-memory runtime-status store performs at most one expired-entry scan per +strictly more than 90 seconds of continuing writes. It has no maintenance timer. +An otherwise inactive process can retain its last unread abandoned cohort until +its next read/write or process exit; those entries are never returned after TTL. + +## Assembled acceptance boundary + +`server/src/__tests__/maintenance-server-routes.test.ts` starts the real exported +`startServer` in a child with its real app, services, coordinator and PostgreSQL. +The suite prepares shared/SDK/Runner TypeScript artifacts, uses the existing +owned embedded PostgreSQL helper by default, and accepts validated disposable +external database overrides for constrained hosts. The file name places this +expensive suite in the existing serialized route lane. It does not skip merely +because an external database environment variable is absent. + +The test-only PostgreSQL proxy observes all frontend simple Query messages and +extended Parse/Bind/Execute messages. It counts prepared executions that reuse +an earlier parse, raw tagged SQL, transaction commands and rollback. SQL +parameters and result values are not written to measurement receipts. Receipts +contain fingerprints, static table names and separate protocol frame counts. +The observer issues no database queries and has a real-socket calibration test. +Setup and independent result inspection use another bounded client outside the +application's measured connection boundary. + +Child-local diagnostics observe outbound HTTP requests and non-database TCP +connections/bytes. A controlled GitHub HTTP endpoint is explicitly simulated; +its request log independently records provider activity. The Cloud standby case +uses actual managed bootstrap validation, an S3 HTTP guard and generated +Ed25519 signing keys, without a deployed Cloud service or real bucket. + +The absent-feature fixture disables telemetry, announcements, scheduled database +backups and UI serving. It supplies no provider secrets or configured external +channels. Other maintenance registrations remain on their production paths. +Those exclusions are fixture conditions, not a claim that configured periodic +backups, live operation keepalives or authorized external reconciliation disappear. + +## Clock and settlement + +The original pre-migration positive control recorded recurring maintenance SQL +during a real 31-second observation after actual startup settlement. Its baseline +receipt is retained; the obsolete recurring-SQL expectation is replaced by the +final enabled cases. The real-socket raw/prepared execution calibration remains +in the suite. The two enabled cases are prepared to observe 90 real seconds each +and two further hours using child-local Date/setTimeout/setInterval control. +Both cases pass on the final cutover tree. They also require every application +database connection to close at the existing default 60-second idle timeout. 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 +elapsed time from simulated elapsed time. PostgreSQL's server clock is not +advanced; this window does not replace owner-specific durable deadline tests. + +The final enabled-mode claim still requires a separate real 90-second quiet +window after interval removal. A fake clock alone is insufficient. + +## Evidence and remaining gates + +Before removal, empty enabled boot issued 325 startup SQL statements and another +26 executions in the next real 31-second window: 24 extended executions and two +simple queries, covering residual heartbeat/native/resource/wake recovery. +This establishes that the observer sees the legacy interval's actual effect. +The obsolete-positive assertion will be replaced when the interval is removed; +the raw/prepared real-PostgreSQL calibration remains. + +Current assembled cases cover: + +- A fixture-only goal-control INSERT error that previously aborted actual boot. + With owner isolation, the goal row stays pending, a new real residual lease + obligation is discovered within one second of child time and is cleaned by its + separate owner, and ordinary attention ingestion still archives its source. + Native classification remains fail-closed. +- Actual readiness while a fixture-owned PostgreSQL lock holds the real startup + retention INSERT. Health becomes ready with active work and pending SQL; + releasing the lock produces the archive and zero active work. Readiness alone + therefore does not stand in for idle eligibility. +- Disabled-heartbeat empty boot across two simulated hourly windows. +- Actual worktree suppression with a committed residual obligation and no + recurring SQL over two minutes of child time. A normal operator settings + activation reconstructs that obligation and produces a durable cleanup result. + Activation uses the stored instance identity and timestamp; no direct worker + wake or safety-gate bypass is used. +- Ordinary HTTP company/settings/completed-task creation, independent durable + inspection, drain/resume and two additional hourly windows without owner work. +- An ordinary HTTP issue insertion held by an isolated PostgreSQL lock while + idle drain starts, followed by its actual commit and external-object source + ingestion. Provider work waits during drain, then resume fetches the simulated + merged PR, persists terminal state and settles for two hourly windows. +- Actual shutdown during an in-flight provider HTTP read, cancellation and + durable claim release without a false terminal result, followed by a fresh + server reconstructing and completing the retained object. +- 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 +spool cleanup. The Cloud coordinator deliberately disables both dispatch and +reconciliation before claim, so the hold remains even with zero transfer rows. +The test qualifies SQL/network quiet with that explicit deferred intent, then +requires the ordinary signed claim to release it to zero active work at the +existing bounded 30-second retry deadline. It does not call standby sleep-eligible +while this hold exists. Temporary diagnostic begin/end tracing used to identify +this owner is not part of the production change. + +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. + +The recorded acceptance source is `6ccd1ed3e661e086b793de32b809ff8d3799eb69`, +with C2 predecessor `b29334feafd5bf1740aa74e2ff0649c6b4cefc20`. It passes all +41 cases with no skips: the original 34 (16 actual boot, three passive observer, +seven runtime-status and eight single-flight cases), plus seven focused harness +support and artifact-ownership cases. All 26 PostgreSQL decision-target controls, +121 startup contracts and the scoped test TypeScript graph also pass. These +receipts qualify that recorded executable source; later parent changes need +their affected-source verification. +Startup contracts retain independent registration, failure isolation and shutdown +tracking. Required CI and exact-head review remain PR gates; these measurements +do not qualify deployment or host suspension. + +The assembled routine proof also uses ordinary subtree cancel and restore APIs. +It requires the same persisted routine run to return to `issue_created`, clears +its terminal projection, and retains zero agent executions while suppression +remains active. The passive PostgreSQL observer records connection opens, closes +and byte counts separately from SQL frames so final quiet receipts can expose +connection housekeeping without hiding it in a zero-query total. + +The saved-execution receipt case uses ordinary task Retry and subtree release +HTTP endpoints. A pause retains the accepted receipt without a run; the saved +wait reason receives its existing one-time 30-second grace, then remains quiet +for 30 real seconds. Releasing the hold with `wakeAgents: false` launches exactly +one actual local process with the expected stdout. This phase uses real elapsed +time through PostgreSQL controller-lease ownership. Clock acceleration begins +only after the process is terminal and its task is closed. + +## Final enabled idle measurements + +Both fixtures start the actual server and use its default application database +idle timeout. The independent fixture client stays outside the passive proxy. +The real observation follows readiness and complete work settlement. + +| Fixture | Real observation | Application SQL | DB connections opened / closed at end | DB bytes during real window | Further qualified clock | +| --- | --- | --- | --- | --- | --- | +| Empty | 90,003 ms | 0 | 7 / 7 | 30 frontend, 0 backend | 7,200,000 ms | +| Configured terminal history | 90,002 ms | 0 | 7 / 7 | 30 frontend, 0 backend | 7,200,000 ms | + +Each fixture had already closed one startup connection before observation. The +remaining six connections each sent one five-byte PostgreSQL Terminate frame. +No Query, Parse, Bind, Execute or Sync frame occurred during either real window. +Neither fixture opened another database connection or changed either byte count +across the following two qualified hours. Both windows also retained zero new +outgoing HTTP requests, non-database TCP connections and non-database bytes. +The test requires zero active tracked work throughout the settled boundaries. +These are application-level observations, not a host or Neon suspension test. + +## Portable harness and receipt ownership + +When no external database is supplied, the PostgreSQL suites use the repository's +embedded support probe and skip only on an unsupported environment. An explicit +acceptance database bypasses that probe and runs every case. The passive protocol +unit cases still run without PostgreSQL. Dedicated acceptance requires all 44 +cases to execute with no skips. + +The harness owns its temporary directory, proxy, port reservation, log and child +from setup through stop. Stop is idempotent. It finishes child shutdown before +releasing the proxy and files, and it removes owned files on startup failure too. +By default, configuration, logs, session state and measurement receipts are deleted. +Set `PAPERCLIP_MAINTENANCE_KEEP_ARTIFACTS=1` for an explicit proof run. Stop then +prints `[maintenance-test] retained artifacts: ` and exposes the same +path as `retainedDirectory` on the harness result. This mode retains receipts on +failure as well. The caller owns removal of explicitly retained proof artifacts. diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index 1dedcddc243..67daeba44c8 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -157,6 +157,9 @@ function registerModuleMocks() { })); } +let agentRoutes: typeof import("../routes/agents.js")["agentRoutes"]; +let errorHandler: typeof import("../middleware/index.js")["errorHandler"]; + async function createApp( db: Record = {}, actor: Record = { @@ -168,15 +171,6 @@ async function createApp( }, options: { chatRunRetries?: typeof mockChatRunRetries } = {}, ) { - // Vitest tracks factory-mock resolution in one shared call stack. Importing - // these graphs concurrently can drop the services/index factory mock and - // accidentally run real DB-backed activity logging against this test stub. - const { agentRoutes } = await vi.importActual< - typeof import("../routes/agents.js") - >("../routes/agents.js"); - const { errorHandler } = await vi.importActual< - typeof import("../middleware/index.js") - >("../middleware/index.js"); const app = express(); app.use(express.json()); app.use((req, _res, next) => { @@ -291,7 +285,7 @@ describe("agent live run routes", () => { limit, { summary: query.includes("summary=true") }); expect(mockRunSecretRedactionRegistry.redactForRuns).toHaveBeenCalled(); }); - beforeEach(() => { + beforeEach(async () => { vi.resetModules(); vi.doUnmock("../services/agents.js"); vi.doUnmock("../services/heartbeat.js"); @@ -424,6 +418,17 @@ describe("agent live run routes", () => { skipped: 0, skipReasons: [], }); + + // Module loading is fixture setup; each behavioral case keeps its own timeout. + // Vitest tracks factory-mock resolution in one shared call stack. Importing + // these graphs concurrently can drop the services/index factory mock and + // accidentally run real DB-backed activity logging against this test stub. + ({ agentRoutes } = await vi.importActual< + typeof import("../routes/agents.js") + >("../routes/agents.js")); + ({ errorHandler } = await vi.importActual< + typeof import("../middleware/index.js") + >("../middleware/index.js")); }); describe("heartbeat run ID validation", () => { diff --git a/server/src/__tests__/decision-target-work.test.ts b/server/src/__tests__/decision-target-work.test.ts index f9ec92d1068..626af25b8ee 100644 --- a/server/src/__tests__/decision-target-work.test.ts +++ b/server/src/__tests__/decision-target-work.test.ts @@ -271,8 +271,33 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes it("retains a first target change through drain without passive SQL", async () => { const row = await create(); await start(); await quiet(); draining = true; await change("cancelled"); - const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); - expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(recovery.hasDecisionTargetWork()).toBe(true); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("keeps a real uncommitted target write held during drain and releases only after commit", async () => { + const row = await create(); await start(); await quiet(); draining = true; + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const writing = db.transaction(async tx => { + await issueService(db).update(targetId, { status: "cancelled", actorUserId: userId }, tx); + entered(); await gate; + }); + try { + await ready; await vi.advanceTimersByTimeAsync(120_000); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + expect((await service().get(row.id))?.status).toBe("open"); expect(wake).not.toHaveBeenCalled(); + } finally { release(); await writing; } + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(recovery.hasDecisionTargetWork()).toBe(true); draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); }); @@ -286,8 +311,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes const row = await create(); await start(); await quiet(); await failExpiry(); await change("cancelled"); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); }); draining = true; - const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); - expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(recovery.hasDecisionTargetWork()).toBe(true); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); await db.execute(sql`drop trigger decision_target_expire_fail on decisions`); draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); for (const error of errors) expect(error).toMatchObject({ cause: { message: "expiry unavailable" } }); errors.length = 0; diff --git a/server/src/__tests__/helpers/maintenance/environment.ts b/server/src/__tests__/helpers/maintenance/environment.ts new file mode 100644 index 00000000000..226d9f65915 --- /dev/null +++ b/server/src/__tests__/helpers/maintenance/environment.ts @@ -0,0 +1,24 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { getEmbeddedPostgresTestSupport, type EmbeddedPostgresTestSupport } from "../embedded-postgres.js"; + +export async function maintenanceDatabaseSupport(databaseUrl?: string): Promise { + return databaseUrl ? { supported: true } : getEmbeddedPostgresTestSupport(); +} + +export async function createMaintenanceArtifacts(root = tmpdir()) { + const directory = await mkdtemp(path.join(root, "paperclip-maintenance-")); + const retainedDirectory = process.env.PAPERCLIP_MAINTENANCE_KEEP_ARTIFACTS === "1" ? directory : undefined; + let disposal: Promise | undefined; + return { + directory, + retainedDirectory, + dispose() { + return disposal ??= (async () => { + if (retainedDirectory) process.stdout.write(`[maintenance-test] retained artifacts: ${retainedDirectory}\n`); + else await rm(directory, { recursive: true, force: true }); + })(); + }, + }; +} diff --git a/server/src/__tests__/helpers/maintenance/postgres-proxy.ts b/server/src/__tests__/helpers/maintenance/postgres-proxy.ts new file mode 100644 index 00000000000..d80aa2fc689 --- /dev/null +++ b/server/src/__tests__/helpers/maintenance/postgres-proxy.ts @@ -0,0 +1,109 @@ +import { createHash } from "node:crypto"; +import { createConnection, createServer, type Socket } from "node:net"; + +export type ObservedQuery = { protocol: "simple" | "extended"; sql: string; fingerprint: string }; + +export function createPostgresProtocolObserver(onQuery: (query: ObservedQuery) => void, onFrame: (type: string) => void = () => {}) { + let frontend = Buffer.alloc(0), backend = Buffer.alloc(0), startup = true, ready = false, pending = 1; + const statements = new Map(), portals = new Map(); + function cstring(body: Buffer, offset = 0): [string, number] { + const end = body.indexOf(0, offset); + if (end < 0) throw new Error("PostgreSQL protocol string is incomplete"); + return [body.toString("utf8", offset, end), end + 1]; + } + function observe(protocol: ObservedQuery["protocol"], sql: string) { + onQuery({ protocol, sql, fingerprint: createHash("sha256").update(sql).digest("hex").slice(0, 16) }); + } + return { + get pending() { return pending; }, + get ready() { return ready; }, + frontend(chunk: Buffer) { + frontend = Buffer.concat([frontend, chunk]); + while (frontend.length >= (startup ? 8 : 5)) { + const length = frontend.readInt32BE(startup ? 0 : 1); + if (length < (startup ? 8 : 4) || length > 64 * 1024 * 1024) throw new Error(`Invalid PostgreSQL frame length: ${JSON.stringify({ length, startup, bufferedBytes: frontend.length, header: frontend.subarray(0, startup ? 8 : 5).toString("hex") })}`); + const total = length + (startup ? 0 : 1); + if (frontend.length < total) return; + const frame = frontend.subarray(0, total); + frontend = frontend.subarray(total); + if (startup) { + const protocol = frame.readInt32BE(4); + if (protocol === 196608) startup = false; + else if (protocol !== 80877103 && protocol !== 80877104 && protocol !== 80877102) throw new Error("Unsupported PostgreSQL startup protocol"); + continue; + } + const type = String.fromCharCode(frame[0]!), body = frame.subarray(5); + onFrame(type); + if (type === "Q") { pending++; observe("simple", cstring(body)[0]); } + else if (type === "P") { + const [name, offset] = cstring(body); + statements.set(name, cstring(body, offset)[0]); + } else if (type === "B") { + const [portal, offset] = cstring(body); + portals.set(portal, cstring(body, offset)[0]); + } else if (type === "E") { + const portal = cstring(body)[0], statement = portals.get(portal); + if (statement === undefined || !statements.has(statement)) throw new Error("Executed PostgreSQL statement was not observed"); + observe("extended", statements.get(statement)!); + } else if (type === "S") pending++; + else if (type === "C") { + const name = cstring(body, 1)[0]; + if (body[0] === 83) statements.delete(name); + else portals.delete(name); + } + } + }, + backend(chunk: Buffer) { + backend = Buffer.concat([backend, chunk]); + if (startup && backend.length === 1 && backend[0] === 78) { backend = Buffer.alloc(0); return; } + while (backend.length >= 5) { + const length = backend.readInt32BE(1); + if (length < 4 || length > 64 * 1024 * 1024) throw new Error("Unsupported encrypted or invalid PostgreSQL response"); + if (backend.length < length + 1) return; + if (backend[0] === 90) { ready = true; pending = Math.max(0, pending - 1); } + backend = backend.subarray(length + 1); + } + }, + }; +} + +export async function startMaintenancePostgresProxy(databaseUrl: string) { + const upstream = new URL(databaseUrl), queries: ObservedQuery[] = [], errors: Error[] = []; + const frames = { Query: 0, Parse: 0, Bind: 0, Execute: 0, Sync: 0, Terminate: 0 }; + const transport = { connectionsOpened: 0, connectionsClosed: 0, frontendBytes: 0, backendBytes: 0 }; + const sockets = new Set(), observers = new Set>(); + const server = createServer(client => { + transport.connectionsOpened++; + client.once("close", () => { transport.connectionsClosed++; }); + const target = createConnection({ host: upstream.hostname, port: Number(upstream.port) || 5432 }); + const observer = createPostgresProtocolObserver(query => queries.push(query), type => { + if (type === "Q") frames.Query++; else if (type === "P") frames.Parse++; else if (type === "B") frames.Bind++; + else if (type === "E") frames.Execute++; else if (type === "S") frames.Sync++; else if (type === "X") frames.Terminate++; + }); + observers.add(observer); + for (const socket of [client, target]) { + sockets.add(socket); + socket.on("error", error => { errors.push(error); client.destroy(); target.destroy(); }); + socket.on("close", () => { sockets.delete(socket); if (client.destroyed && target.destroyed) observers.delete(observer); }); + } + client.on("data", chunk => { transport.frontendBytes += chunk.length; try { observer.frontend(chunk); } catch (error) { errors.push(error as Error); client.destroy(); target.destroy(); } }); + target.on("data", chunk => { transport.backendBytes += chunk.length; try { observer.backend(chunk); } catch (error) { errors.push(error as Error); client.destroy(); target.destroy(); } }); + client.pipe(target); target.pipe(client); + client.on("end", () => target.end()); target.on("end", () => client.end()); + }); + await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("PostgreSQL observer failed to bind"); + const url = new URL(databaseUrl); url.hostname = "127.0.0.1"; url.port = String(address.port); url.searchParams.set("sslmode", "disable"); + return { + url: url.toString(), queries, errors, + get frames() { return { ...frames }; }, + get transport() { return { ...transport }; }, + get pending() { return [...observers].reduce((sum, observer) => sum + observer.pending, 0); }, + get starting() { return [...observers].filter(observer => !observer.ready).length; }, + async stop() { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + }, + }; +} diff --git a/server/src/__tests__/helpers/maintenance/runtime-build.ts b/server/src/__tests__/helpers/maintenance/runtime-build.ts new file mode 100644 index 00000000000..3b43d7c205c --- /dev/null +++ b/server/src/__tests__/helpers/maintenance/runtime-build.ts @@ -0,0 +1,21 @@ +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import { fileURLToPath } from "node:url"; + +let ready: Promise | undefined; +export function prepareMaintenanceRuntime() { + ready ??= (async () => { + const script = process.env.npm_execpath; + const javascript = script && /\.[cm]?js$/i.test(script); + const command = javascript ? process.execPath : script ?? "pnpm"; + const prefix = javascript ? [script!] : []; + for (const args of [ + ["--filter", "@paperclipai/plugin-sdk", "ensure-build-deps"], + ["--filter", "@paperclipai/paperclip-runner", "build:typescript"], + ]) await promisify(execFile)(command, [...prefix, ...args], { + cwd: fileURLToPath(new URL("../../../../..", import.meta.url)), env: process.env, + timeout: 90_000, maxBuffer: 8 * 1024 * 1024, + }); + })(); + return ready; +} diff --git a/server/src/__tests__/helpers/maintenance/server-child.ts b/server/src/__tests__/helpers/maintenance/server-child.ts new file mode 100644 index 00000000000..5569b29d2e9 --- /dev/null +++ b/server/src/__tests__/helpers/maintenance/server-child.ts @@ -0,0 +1,99 @@ +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; +const fakeClock = process.env.PAPERCLIP_MAINTENANCE_FAKE_CLOCK === "true"; +let bootClock: ReturnType | undefined; +function runClock() { + if (bootClock) clearRealInterval(bootClock); + let previous = realNow(); + bootClock = realInterval(() => { const now = realNow(); mock.timers.tick(Math.max(0, now - previous)); previous = now; }, 10); +} +if (fakeClock) { + mock.timers.enable({ apis: ["Date", "setTimeout", "setInterval"], now: realNow() }); + runClock(); +} + +if (process.env.PAPERCLIP_MAINTENANCE_GITHUB_ORIGIN) { + const fetch = globalThis.fetch; + globalThis.fetch = (input, init) => { + const url = new URL(input instanceof Request ? input.url : input.toString()); + return fetch(url.origin === "https://api.github.com" + ? new URL(url.pathname + url.search, process.env.PAPERCLIP_MAINTENANCE_GITHUB_ORIGIN) : input, init); + }; +} + +let outgoingHttpRequests = 0; +const sockets: Array<{ socket: Socket; database: boolean }> = []; +channel("net.client.socket").subscribe(message => { + const socket = (message as { socket: Socket }).socket; + socket.once("connect", () => sockets.push({ socket, database: socket.remotePort === Number(new URL(process.env.DATABASE_URL!).port) })); +}); +function network() { + return { nonDatabaseConnections: sockets.filter(entry => !entry.database).length, + databaseConnections: sockets.filter(entry => entry.database).length, + nonDatabaseBytesWritten: sockets.filter(entry => !entry.database).reduce((sum, entry) => sum + entry.socket.bytesWritten, 0) }; +} +const observeRequest = () => { outgoingHttpRequests++; }; +channel("undici:request:create").subscribe(observeRequest); +channel("http.client.request.start").subscribe(observeRequest); + +try { + const { startServer } = await import("../../../index.js"); + 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; issueUpdate?: { id: string; status: "done" }; orphanLease?: { companyId: string }; 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); } + if (message.command === "realtime" && fakeClock) runClock(); + if (message.command === "advance") { + if (!fakeClock || typeof message.ms !== "number" || !Number.isFinite(message.ms) || message.ms < 0 || message.ms > 30_000) throw new Error("Invalid maintenance clock advance"); + mock.timers.tick(message.ms); + } + if (message.command === "drain") startTaskDrain({ purpose: "idle" }); + 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 === "createOrphanLease") { + if (!message.orphanLease) throw new Error("Missing orphan lease fixture"); + const { environmentLeases } = await import("@paperclipai/db"); + const { writeEnvironmentCleanup } = await import("../../../services/environment-cleanup-notifications.js"); + const old = new Date(Date.now() - 6 * 60_000); + const companyId = message.orphanLease.companyId; + const [lease] = await writeEnvironmentCleanup(app.locals.paperclipDb, tx => tx.insert(environmentLeases).values({ + companyId, provider: "local", providerLeaseId: null, status: "active", leasePolicy: "ephemeral", metadata: { driver: "local" }, + acquiredAt: old, lastUsedAt: old, createdAt: old, updatedAt: old, + }).returning(), { companyId }); + process.send?.({ id: message.id, kind: "leaseCreated", leaseId: lease!.id }); + } 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 (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)); + process.exit(0); + } else throw new Error(`Unknown maintenance harness command: ${message.command}`); + } catch (error) { process.send?.({ id: message.id, kind: "error", error: error instanceof Error ? error.stack : String(error) }); } + }); +} catch (error) { + if (bootClock) clearRealInterval(bootClock); + process.send?.({ kind: "error", error: error instanceof Error ? error.stack : String(error) }); + process.exitCode = 1; +} diff --git a/server/src/__tests__/helpers/maintenance/server-harness.ts b/server/src/__tests__/helpers/maintenance/server-harness.ts new file mode 100644 index 00000000000..922f90292a8 --- /dev/null +++ b/server/src/__tests__/helpers/maintenance/server-harness.ts @@ -0,0 +1,218 @@ +import { fork } from "node:child_process"; +import type { Server as HttpServer } from "node:http"; +import { randomBytes, randomUUID } from "node:crypto"; +import { createWriteStream } from "node:fs"; +import { writeFile } from "node:fs/promises"; +import { createServer } from "node:net"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { setTimeout as delay } from "node:timers/promises"; +import { startMaintenancePostgresProxy } from "./postgres-proxy.js"; +import { createMaintenanceArtifacts } from "./environment.js"; + +async function withDeadline(promise: Promise, ms: number, message: () => string): Promise { + const controller = new AbortController(); + try { + return await Promise.race([promise, delay(ms, undefined, { signal: controller.signal }).then(() => { throw new Error(message()); })]); + } finally { controller.abort(); } +} + +export async function withMaintenanceProvider(provider: HttpServer, run: (origin: string) => Promise): Promise { + let failure: { error: unknown } | undefined; + try { + await new Promise((resolve, reject) => { + provider.once("error", reject); + provider.listen(0, "127.0.0.1", () => { provider.off("error", reject); resolve(); }); + }); + const address = provider.address(); + if (!address || typeof address === "string") throw new Error("Maintenance provider failed to bind"); + return await run(`http://127.0.0.1:${address.port}`); + } catch (error) { + failure = { error }; + throw error; + } finally { + try { + const closing = provider.listening ? new Promise((resolve, reject) => provider.close(error => error ? reject(error) : resolve())) : Promise.resolve(); + provider.closeAllConnections(); + await closing; + } catch (error) { + if (failure) throw new AggregateError([failure.error, error], "Maintenance provider work and cleanup failed"); + throw error; + } + } +} + +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: { artifactRoot?: string; enabled?: boolean; clock?: boolean; suppressed?: boolean; githubOrigin?: string; connector?: { origin: string; signPrivateKey: string; sealPrivateKey: string }; cloud?: { jwks: string; s3Origin: string; standby: boolean } } = {}) { + const artifacts = await createMaintenanceArtifacts(options.artifactRoot); + const { directory, retainedDirectory } = artifacts; + const reservation = createServer(); + let stopProxy = async () => {}; + let stopChild = async () => {}; + let stopLog = async () => {}; + let stopping: Promise | undefined; + function stop() { + return stopping ??= (async () => { + const errors: unknown[] = []; + for (const cleanup of [stopChild, stopProxy, + async () => { if (reservation.listening) await new Promise(resolve => reservation.close(() => resolve())); }, + stopLog, artifacts.dispose]) { + try { await cleanup(); } catch (error) { errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Maintenance cleanup failed"); + })(); + } + try { + const proxy = await startMaintenancePostgresProxy(databaseUrl); + stopProxy = proxy.stop; + await new Promise((resolve, reject) => { reservation.once("error", reject); reservation.listen(0, "127.0.0.1", resolve); }); + const address = reservation.address(); + if (!address || typeof address === "string") throw new Error("Maintenance HTTP reservation failed"); + const configPath = path.join(directory, "config.json"); + await writeFile(configPath, JSON.stringify({ + $meta: { version: 1, updatedAt: new Date().toISOString(), source: "onboard" }, + database: { mode: "postgres", connectionString: proxy.url, backup: { enabled: false, dir: path.join(directory, "backups") } }, + logging: { mode: "file", logDir: path.join(directory, "logs") }, + server: { deploymentMode: "local_trusted", exposure: "private", bind: "loopback", host: "127.0.0.1", port: address.port, serveUi: false }, + storage: options.cloud ? { provider: "s3", s3: { bucket: "maintenance-test", region: "us-east-1", endpoint: options.cloud.s3Origin, forcePathStyle: true } } + : { provider: "local_disk", localDisk: { baseDir: path.join(directory, "storage") } }, + secrets: { provider: "local_encrypted", strictMode: false, localEncrypted: { keyFilePath: path.join(directory, "master.key") } }, + telemetry: { enabled: false }, updates: { checkEnabled: false }, + }), { mode: 0o600 }); + await new Promise(resolve => reservation.close(() => resolve())); + const child = fork(fileURLToPath(new URL("./server-child.ts", import.meta.url)), [], { + cwd: fileURLToPath(new URL("../../../../..", import.meta.url)), + execArgv: ["--import", fileURLToPath(new URL("../../../../node_modules/tsx/dist/loader.mjs", import.meta.url))], + 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(), + CODEX_HOME: path.join(directory, "codex"), PAPERCLIP_ANNOUNCEMENTS_ENABLED: "false", + PAPERCLIP_DB_BACKUP_ENABLED: "false", PAPERCLIP_UI_DEV_MIDDLEWARE: "false", + HEARTBEAT_SCHEDULER_ENABLED: String(options.enabled ?? true), + 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"), + PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "maintenance-tenant-token", PAPERCLIP_CLOUD_STACK_ID: "maintenance-stack", + PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: options.cloud.jwks, PAPERCLIP_CLOUD_WARM_STANDBY: options.cloud.standby ? "1" : "0", + PAPERCLIP_PUBLIC_URL: "https://pool-maintenance.staging.paperclip.app", PAPERCLIP_PRIMARY_HOST: "pool-maintenance.staging.paperclip.app", + AWS_ACCESS_KEY_ID: "maintenance-fixture", AWS_SECRET_ACCESS_KEY: "maintenance-fixture", AWS_EC2_METADATA_DISABLED: "true", + } : {}), + PAPERCLIP_MIGRATION_PROMPT: "never", PAPERCLIP_MIGRATION_AUTO_APPLY: "true", PAPERCLIP_MANAGED_RUNTIME_HTTPS: "off", + }, + stdio: ["ignore", "pipe", "pipe", "ipc"], + }); + let output = "", requestId = 0; + const log = createWriteStream(path.join(directory, "server.log")); + stopLog = () => new Promise(resolve => log.end(resolve)); + let serverReady = false; + child.stdout?.on("data", chunk => { output += String(chunk); log.write(chunk); }); + child.stderr?.on("data", chunk => { output += String(chunk); log.write(chunk); }); + const requests = new Map void; reject: (error: Error) => void }>(); + let readyResolve: (port: number) => void, readyReject: (error: Error) => void; + const ready = new Promise((resolve, reject) => { readyResolve = resolve; readyReject = reject; }); + const exited = new Promise(resolve => child.once("exit", () => resolve())); + child.on("message", (message: any) => { + if (message.kind === "ready") { serverReady = true; readyResolve(message.port); } + else if (message.id) { + const request = requests.get(message.id); requests.delete(message.id); + if (message.kind === "error") request?.reject(new Error(message.error)); else request?.resolve(message); + } else if (message.kind === "error") readyReject(new Error(message.error)); + }); + child.once("error", error => readyReject(error)); + child.once("exit", (code, signal) => { + const error = new Error(`Maintenance child exited (${code}/${signal}): ${output.slice(-12_000)}`); + readyReject(error); + for (const request of requests.values()) request.reject(error); + requests.clear(); + }); + async function request(command: string, payload: { ms?: number; issueUpdate?: { id: string; status: "done" }; orphanLease?: { companyId: string }; 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 }); + try { return await withDeadline(result, 30_000, () => `Maintenance child ${command} timed out: ${output.slice(-8_000)}`); } + finally { requests.delete(id); } + } + stopChild = async () => { + try { + if (!serverReady && child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + else if (child.exitCode === null && child.signalCode === null && child.connected) await request("shutdown"); + await withDeadline(exited, 30_000, () => "Maintenance child did not exit after shutdown"); + } finally { + if (child.exitCode === null && child.signalCode === null) { + child.kill("SIGKILL"); + await withDeadline(exited, 5_000, () => "Maintenance child did not exit after cancellation"); + } + } + }; + const port = await withDeadline(ready, 90_000, () => `Maintenance boot timed out: ${output.slice(-12_000)}`); + return { + proxy, directory, retainedDirectory, port, stop, + 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", ...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 createOrphanLease(companyId: string) { + if (options.clock) await request("realtime"); + try { return await request<{ leaseId: string }>("createOrphanLease", { orphanLease: { companyId } }); } + 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"), + realtime: () => request("realtime"), + async advance(ms: number) { + const deadline = performance.now() + 30_000; + while (proxy.starting > 0) { + if (proxy.errors.length) throw new AggregateError(proxy.errors, "PostgreSQL observer failed"); + if (performance.now() >= deadline) throw new Error("Maintenance database startup did not finish before clock advance"); + await delay(5); + } + if (proxy.errors.length) throw new AggregateError(proxy.errors, "PostgreSQL observer failed"); + return request("advance", { ms }); + }, + freeze: () => request("freeze"), + get output() { return output; }, + async settled(quietMs = 500, deferredStartupHolds = 0) { + let previous = -1, quietSince = Date.now(); + for (const deadline = Date.now() + 30_000; Date.now() < deadline;) { + const snapshot = await request("snapshot"); + if (proxy.errors.length) throw new AggregateError(proxy.errors, "PostgreSQL observer failed"); + if (snapshot.idle.active !== deferredStartupHolds || proxy.pending !== 0 || proxy.queries.length !== previous) quietSince = Date.now(); + previous = proxy.queries.length; + if (snapshot.idle.active === deferredStartupHolds && proxy.pending === 0 && Date.now() - quietSince >= quietMs) return snapshot; + await delay(25); + } + throw new Error(`Maintenance did not settle: ${JSON.stringify({ queries: proxy.queries.length, pending: proxy.pending, snapshot: await request("snapshot") })}\n${output.slice(-12_000)}`); + }, + }; + } catch (error) { + try { await stop(); } + catch (cleanupError) { throw new AggregateError([error, cleanupError], "Maintenance startup and cleanup failed"); } + throw error; + } +} diff --git a/server/src/__tests__/helpers/maintenance/tsconfig.json b/server/src/__tests__/helpers/maintenance/tsconfig.json new file mode 100644 index 00000000000..864a6b58de6 --- /dev/null +++ b/server/src/__tests__/helpers/maintenance/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../../../../tsconfig.base.json", + "compilerOptions": { + "noEmit": true, + "rootDir": "../../../../.." + }, + "include": ["*.ts", "../../maintenance-*.test.ts", "../../decision-target-work.test.ts", "../../../types/express.d.ts"] +} diff --git a/server/src/__tests__/maintenance-harness-resources.test.ts b/server/src/__tests__/maintenance-harness-resources.test.ts new file mode 100644 index 00000000000..179bb9e20eb --- /dev/null +++ b/server/src/__tests__/maintenance-harness-resources.test.ts @@ -0,0 +1,127 @@ +import { createServer, request } from "node:http"; +import { access, mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createMaintenanceArtifacts, maintenanceDatabaseSupport } from "./helpers/maintenance/environment.js"; +import { startMaintenanceServer, withMaintenanceProvider } from "./helpers/maintenance/server-harness.js"; + +const embeddedSupport = vi.hoisted(() => vi.fn()); +vi.mock("./helpers/embedded-postgres.js", () => ({ getEmbeddedPostgresTestSupport: embeddedSupport })); + +let root: string | undefined; +beforeEach(() => { vi.clearAllMocks(); vi.stubEnv("PAPERCLIP_MAINTENANCE_KEEP_ARTIFACTS", ""); }); +afterEach(async () => { + vi.unstubAllEnvs(); vi.restoreAllMocks(); + if (root) await rm(root, { recursive: true, force: true }); + root = undefined; +}); +async function ownedRoot() { + return root = await mkdtemp(path.join(tmpdir(), "maintenance-environment-test-")); +} + +describe("maintenance database support", () => { + it("does not probe or skip an explicitly configured acceptance database", async () => { + embeddedSupport.mockRejectedValue(new Error("Embedded PostgreSQL must not be launched")); + expect(await maintenanceDatabaseSupport("postgresql://fixture/owned_test")).toEqual({ supported: true }); + expect(embeddedSupport).not.toHaveBeenCalled(); + }); + it("allows a supported default embedded environment", async () => { + embeddedSupport.mockResolvedValue({ supported: true }); + expect(await maintenanceDatabaseSupport()).toEqual({ supported: true }); + expect(embeddedSupport).toHaveBeenCalledOnce(); + }); + it("returns the standard unsupported reason to the suite gate", async () => { + embeddedSupport.mockResolvedValue({ supported: false, reason: "No usable initdb" }); + expect(await maintenanceDatabaseSupport()).toEqual({ supported: false, reason: "No usable initdb" }); + expect(embeddedSupport).toHaveBeenCalledOnce(); + }); +}); + +describe("owned maintenance artifacts", () => { + it("removes owned config and receipt files by default, including repeated disposal", async () => { + const artifacts = await createMaintenanceArtifacts(await ownedRoot()); + await writeFile(path.join(artifacts.directory, "measurement.json"), "{}"); + expect(artifacts.retainedDirectory).toBeUndefined(); + await artifacts.dispose(); await artifacts.dispose(); + await expect(access(artifacts.directory)).rejects.toMatchObject({ code: "ENOENT" }); + expect(await readdir(root!)).toEqual([]); + }); + it("retains receipts only with explicit opt-in and exposes their path once", async () => { + vi.stubEnv("PAPERCLIP_MAINTENANCE_KEEP_ARTIFACTS", "1"); + const print = vi.spyOn(process.stdout, "write").mockImplementation(() => true); + const artifacts = await createMaintenanceArtifacts(await ownedRoot()); + const receipt = path.join(artifacts.directory, "measurement.json"); + await writeFile(receipt, "{}"); + await artifacts.dispose(); await artifacts.dispose(); + await access(receipt); + expect(artifacts.retainedDirectory).toBe(artifacts.directory); + expect(print).toHaveBeenCalledExactlyOnceWith(`[maintenance-test] retained artifacts: ${artifacts.directory}\n`); + }); + it("cleans the owned directory when setup fails before the child starts", async () => { + const parent = await ownedRoot(); + await expect(startMaintenanceServer("not-a-database-url", { artifactRoot: parent })).rejects.toThrow("Invalid URL"); + expect(await readdir(parent)).toEqual([]); + }); + it("prints retained diagnostics when failed startup was explicitly opted in", async () => { + vi.stubEnv("PAPERCLIP_MAINTENANCE_KEEP_ARTIFACTS", "1"); + const print = vi.spyOn(process.stdout, "write").mockImplementation(() => true); + const parent = await ownedRoot(); + await expect(startMaintenanceServer("not-a-database-url", { artifactRoot: parent })).rejects.toThrow("Invalid URL"); + const [name] = await readdir(parent); + expect(name).toMatch(/^paperclip-maintenance-/); + expect(print).toHaveBeenCalledExactlyOnceWith(`[maintenance-test] retained artifacts: ${path.join(parent, name!)}\n`); + }); +}); + + +describe("owned maintenance providers", () => { + it("closes all six real listeners when actual server setup rejects", async () => { + const parent = await ownedRoot(); + const providers = Array.from({ length: 6 }, () => createServer((_req, res) => res.end())); + let bound = 0; + try { + const results = await Promise.allSettled(providers.map(provider => withMaintenanceProvider(provider, async () => { + expect(provider.listening).toBe(true); bound++; + return startMaintenanceServer("not-a-database-url", { artifactRoot: parent }); + }))); + expect(bound).toBe(6); + for (const result of results) expect(result).toMatchObject({ status: "rejected", reason: { message: "Invalid URL" } }); + expect(providers.map(provider => provider.listening)).toEqual(Array(6).fill(false)); + expect(providers.map(provider => provider.address())).toEqual(Array(6).fill(null)); + expect(await readdir(parent)).toEqual([]); + } finally { + await Promise.all(providers.map(async provider => { + provider.closeAllConnections(); + if (provider.listening) await new Promise(resolve => provider.close(() => resolve())); + })); + } + }); + + it("closes an active provider connection and preserves an inner cleanup rejection", async () => { + let arrived!: () => void, activeClosed = false; + const active = new Promise(resolve => { arrived = resolve; }); + const provider = createServer((_req, res) => { res.once("close", () => { activeClosed = true; }); arrived(); }); + const failure = new Error("Controlled runtime cleanup failure"); + let client: ReturnType | undefined, closed: Promise | undefined; + try { + await expect(withMaintenanceProvider(provider, async origin => { + try { + closed = new Promise(resolve => { + client = request(origin, { agent: false }); + client.once("error", () => {}); client.once("close", () => resolve()); client.end(); + }); + await active; + } finally { throw failure; } + })).rejects.toBe(failure); + await closed; + expect(activeClosed).toBe(true); + expect(provider.listening).toBe(false); + expect(provider.address()).toBeNull(); + } finally { + client?.destroy(); + const closing = provider.listening ? new Promise(resolve => provider.close(() => resolve())) : Promise.resolve(); + provider.closeAllConnections(); await closing; + } + }); +}); diff --git a/server/src/__tests__/maintenance-postgres-proxy.test.ts b/server/src/__tests__/maintenance-postgres-proxy.test.ts new file mode 100644 index 00000000000..c09e99b0a5c --- /dev/null +++ b/server/src/__tests__/maintenance-postgres-proxy.test.ts @@ -0,0 +1,115 @@ +import { createDb } from "@paperclipai/db"; +import { createConnection } from "node:net"; +import { once } from "node:events"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { maintenanceDatabaseSupport } from "./helpers/maintenance/environment.js"; +import { createPostgresProtocolObserver, startMaintenancePostgresProxy, type ObservedQuery } from "./helpers/maintenance/postgres-proxy.js"; + +const startup = Buffer.from([0, 0, 0, 8, 0, 3, 0, 0]); +function frame(type: string, body: Buffer | string = "") { + const payload = Buffer.from(body), header = Buffer.alloc(5); + header[0] = type.charCodeAt(0); header.writeInt32BE(payload.length + 4, 1); + return Buffer.concat([header, payload]); +} + +describe("passive maintenance PostgreSQL observer", () => { + it("counts fragmented simple and reused prepared executions without counting parses or parameter values", () => { + const queries: ObservedQuery[] = []; + const observer = createPostgresProtocolObserver(query => queries.push(query)); + expect(observer.pending).toBe(1); + expect(observer.ready).toBe(false); + for (const byte of startup) observer.frontend(Buffer.from([byte])); + expect(observer.pending).toBe(1); + const initialReady = frame("Z", "I"); + observer.backend(initialReady.subarray(0, 3)); + expect(observer.pending).toBe(1); + expect(observer.ready).toBe(false); + observer.backend(initialReady.subarray(3)); + expect(observer.pending).toBe(0); + expect(observer.ready).toBe(true); + const bytes = Buffer.concat([ + frame("Q", "select 1\0"), + frame("P", "statement\0select $1::text\0"), + frame("B", "portal\0statement\0parameter-value-sentinel"), + frame("E", Buffer.concat([Buffer.from("portal\0"), Buffer.alloc(4)])), frame("S"), + frame("B", "portal2\0statement\0other-value-sentinel"), + frame("E", Buffer.concat([Buffer.from("portal2\0"), Buffer.alloc(4)])), frame("S"), + ]); + for (let index = 0; index < bytes.length; index += 3) observer.frontend(bytes.subarray(index, index + 3)); + expect(queries.map(query => [query.protocol, query.sql])).toEqual([ + ["simple", "select 1"], ["extended", "select $1::text"], ["extended", "select $1::text"], + ]); + expect(queries[1]!.fingerprint).toBe(queries[2]!.fingerprint); + expect(JSON.stringify(queries)).not.toContain("value-sentinel"); + expect(observer.pending).toBe(3); + const ready = frame("Z", "I"); + for (const byte of Buffer.concat([ready, ready, ready])) observer.backend(Buffer.from([byte])); + expect(observer.pending).toBe(0); + }); + + it("rejects unobserved prepared executions instead of silently claiming quiet", () => { + const observer = createPostgresProtocolObserver(() => {}); + observer.frontend(startup); + expect(() => observer.frontend(frame("E", Buffer.from("missing\0\0\0\0\0")))).toThrow("not observed"); + }); +}); + +let databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MAINTENANCE_TEST_DATABASE_URL"); +let temporary: Awaited> | undefined; +const databaseSupport = await maintenanceDatabaseSupport(databaseUrl); +const describeDatabase = databaseSupport.supported ? describe : describe.skip; +if (!databaseSupport.supported) console.warn(`Skipping maintenance PostgreSQL tests: ${databaseSupport.reason ?? "unsupported environment"}`); +describeDatabase("maintenance observer against PostgreSQL", () => { + beforeAll(async () => { + if (!databaseUrl) { temporary = await startEmbeddedPostgresTestDatabase("maintenance-observer-"); databaseUrl = temporary.connectionString; } + }, 90_000); + afterAll(async () => { await temporary?.cleanup(); }); + it("observes raw tagged SQL, simple SQL, rollback and reused prepared executions with real sockets", async () => { + const proxy = await startMaintenancePostgresProxy(databaseUrl!); + const db = createDb(proxy.url, { prepare: true, maxConnections: 1 }); + try { + const tagged = (value: number) => db.$client`select ${value}::integer as value`; + expect((await tagged(1))[0]!.value).toBe(1); + const first = proxy.queries.length, frames = proxy.frames; + expect((await tagged(2))[0]!.value).toBe(2); + expect(proxy.queries.length).toBe(first + 1); + expect(proxy.frames.Parse).toBe(frames.Parse); + expect(proxy.frames.Execute).toBe(frames.Execute + 1); + expect(proxy.queries.at(-1)!.fingerprint).toBe(proxy.queries.at(-2)!.fingerprint); + await db.$client.unsafe("select 3 as value").simple(); + await expect(db.$client.begin(async tx => { await tx`select 4`; throw new Error("rollback control"); })).rejects.toThrow("rollback control"); + expect(proxy.queries.some(query => query.protocol === "simple" && query.sql.includes("select 3"))).toBe(true); + expect(proxy.queries.some(query => /rollback/i.test(query.sql))).toBe(true); + expect(proxy.pending).toBe(0); + expect(proxy.errors).toEqual([]); + expect(proxy.transport).toMatchObject({ connectionsOpened: 1, connectionsClosed: 0 }); + expect(proxy.transport.frontendBytes).toBeGreaterThan(0); + expect(proxy.transport.backendBytes).toBeGreaterThan(0); + const before = proxy.queries.length, transport = proxy.transport; + await new Promise(resolve => setTimeout(resolve, 100)); + expect(proxy.queries.length).toBe(before); + expect(proxy.transport).toEqual(transport); + expect(proxy.starting).toBe(0); + const address = new URL(proxy.url); + for (const cancel of [false, true]) { + const socket = createConnection({ host: address.hostname, port: Number(address.port) }); + try { + await once(socket, "connect"); + await expect.poll(() => proxy.starting).toBe(1); + const closed = once(socket, "close"); + if (cancel) { + const request = Buffer.alloc(16); + request.writeInt32BE(16); request.writeInt32BE(80877102, 4); + socket.write(request); + } else socket.end(); + await closed; + await expect.poll(() => proxy.starting).toBe(0); + expect(proxy.pending).toBe(0); + expect(proxy.errors).toEqual([]); + } finally { socket.destroy(); } + } + } finally { await db.$client.end(); await proxy.stop(); } + }); +}); diff --git a/server/src/__tests__/maintenance-server-routes.test.ts b/server/src/__tests__/maintenance-server-routes.test.ts new file mode 100644 index 00000000000..3ea2ac56c5c --- /dev/null +++ b/server/src/__tests__/maintenance-server-routes.test.ts @@ -0,0 +1,926 @@ +import { createCipheriv, createHash, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, sign, verify } from "node:crypto"; +import { createServer } from "node:http"; +import { Socket } from "node:net"; +import { once } from "node:events"; +import { agents, agentApiKeys, agentSessionGoalActions, agentTaskSessions, authUsers, companies, companyMemberships, connectionGrants, createDb, heartbeatRuns, issues, issueThreadInteractions, 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"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { maintenanceDatabaseSupport } from "./helpers/maintenance/environment.js"; +import { CLOUD_RUNTIME_IDENTITY_AUDIENCE, CLOUD_RUNTIME_IDENTITY_ISSUER, CLOUD_RUNTIME_IDENTITY_JWS_TYPE } from "../services/cloud-runtime-identity.js"; +import { prepareMaintenanceRuntime } from "./helpers/maintenance/runtime-build.js"; +import { startMaintenanceServer, withMaintenanceProvider } from "./helpers/maintenance/server-harness.js"; + +let databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MAINTENANCE_BOOT_TEST_DATABASE_URL"); +let temporary: Awaited> | undefined; +const databaseSupport = await maintenanceDatabaseSupport(databaseUrl); +const describeDatabase = databaseSupport.supported ? describe : describe.skip; +if (!databaseSupport.supported) console.warn(`Skipping maintenance PostgreSQL tests: ${databaseSupport.reason ?? "unsupported environment"}`); +async function clearDatabase() { + if (!temporary && !new URL(databaseUrl!).pathname.endsWith("_test")) throw new Error("Maintenance boot requires an owned disposable _test database"); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + const tables = await fixture.$client<{ tablename: string }[]>`select tablename from pg_tables where schemaname='public'`; + 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) { + const key = `${query.protocol}:${query.fingerprint}`; + const entry = groups.get(key) ?? { fingerprint: query.fingerprint, protocol: query.protocol, calls: 0, + tables: [...new Set([...query.sql.matchAll(/(?:from|join|update|into)\s+"?([a-z_][a-z_0-9]*)"?/gi)].map(match => match[1]!))] }; + entry.calls++; groups.set(key, entry); + } + return [...groups.values()]; +} +describeDatabase("assembled maintenance server production wiring", () => { + beforeAll(async () => { + await prepareMaintenanceRuntime(); + if (!databaseUrl) { temporary = await startEmbeddedPostgresTestDatabase("maintenance-server-"); databaseUrl = temporary.connectionString; } + }, 150_000); + afterAll(async () => { await temporary?.cleanup(); }); + it.each(["empty", "configured terminal history"] as const)("keeps enabled %s quiet for 90 real seconds and two qualified hourly windows", async state => { + await clearDatabase(); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + if (state !== "empty") { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await fixture.insert(companies).values({ id: companyId, name: "Retained idle history", issuePrefix: "MIH" }); + await fixture.insert(agents).values({ id: agentId, companyId, name: "Idle retained agent", status: "idle", adapterType: "process", + adapterConfig: { command: process.execPath, args: ["-e", "process.exit(0)"] }, runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } } }); + await fixture.insert(issues).values({ id: issueId, companyId, title: "Retained terminal task", status: "done", assigneeAgentId: agentId }); + await fixture.insert(heartbeatRuns).values({ companyId, agentId, issueId, status: "succeeded", finishedAt: new Date(), contextSnapshot: { issueId } }); + } + const server = await startMaintenanceServer(databaseUrl!, { clock: true }); + try { + if (state !== "empty") await server.api("PATCH", "/instance/settings/experimental", { enableExternalObjects: true, enableStatusCards: true }); + await settleClock(server); await server.freeze(); + const before = await server.snapshot(), queryCount = server.proxy.queries.length, frames = server.proxy.frames, transport = server.proxy.transport; + expect(before.startup.phase).toBe("ready"); expect(before.idle.active).toBe(0); + await server.realtime(); + const wallStart = performance.now(); + for (let remaining = 90_000; remaining > 0; remaining = 90_000 - (performance.now() - wallStart)) { + await delay(Math.ceil(remaining)); + } + const wallElapsedMs = performance.now() - wallStart; + await server.freeze(); + const real = await server.settled(100); + expect(wallElapsedMs).toBeGreaterThanOrEqual(90_000); + expect(real.now - before.now).toBeGreaterThanOrEqual(89_000); + expect(real.idle.active).toBe(0); + expect(server.proxy.queries.length).toBe(queryCount); + expect(real.outgoingHttpRequests).toBe(before.outgoingHttpRequests); + expect(real.network.nonDatabaseConnections).toBe(before.network.nonDatabaseConnections); + expect(real.network.nonDatabaseBytesWritten).toBe(before.network.nonDatabaseBytesWritten); + const realFrames = server.proxy.frames, realTransport = server.proxy.transport; + expect(realTransport.connectionsOpened).toBeGreaterThan(0); + expect(realTransport.connectionsOpened).toBe(transport.connectionsOpened); + expect(realTransport.connectionsClosed).toBe(realTransport.connectionsOpened); + expect({ ...realFrames, Terminate: frames.Terminate }).toEqual(frames); + expect(realTransport.frontendBytes - transport.frontendBytes).toBe((realFrames.Terminate - frames.Terminate) * 5); + expect(realTransport.backendBytes).toBe(transport.backendBytes); + for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10); } + const after = await server.snapshot(); + expect(after.now - real.now).toBe(2 * 60 * 60_000); + expect(after.idle.active).toBe(0); + expect(server.proxy.queries.length).toBe(queryCount); + expect(after.outgoingHttpRequests).toBe(before.outgoingHttpRequests); + expect(after.network.nonDatabaseConnections).toBe(before.network.nonDatabaseConnections); + expect(after.network.nonDatabaseBytesWritten).toBe(before.network.nonDatabaseBytesWritten); + expect(server.proxy.errors).toEqual([]); + expect(server.proxy.transport).toEqual(realTransport); + await writeFile(path.join(server.directory, "enabled-quiet-measurement.json"), JSON.stringify({ phase: "enabled-quiet", state, + wallElapsedMs, realClockElapsedMs: real.now - before.now, qualifiedClockMs: after.now - real.now, stepMs: 30_000, ioSettledBetweenSteps: true, + measuredQueries: 0, startupQueries: queryCount, startup: receipt(server.proxy.queries), before, real, after, + frames: { before: frames, real: realFrames, after: server.proxy.frames }, transport: { before: transport, real: realTransport, after: server.proxy.transport }, + boundary: "actual startServer, real elapsed 90s plus qualified child clock; pool connection housekeeping reported separately; no active provider/native work, Cloud enrollment, installed plugins, telemetry exporter, announcements or database backup; no host or database suspension claim", + }, null, 2)); + } finally { await server.stop(); await fixture.$client.end(); } + }, 240_000); + + it("releases an ordinary held task Retry into one real process execution", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { clock: true }); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + await settleClock(server); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Receipt ingress" }); + await fixture.insert(authUsers).values({ id: "local-board", name: "Local operator", email: "maintenance-local@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await fixture.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: "local-board", status: "active", membershipRole: "owner" }).onConflictDoNothing(); + const [agent] = await fixture.insert(agents).values({ companyId: company.id, name: "Receipt process", adapterType: "process", status: "idle", lifecycleState: "ready", + adapterConfig: { command: process.execPath, args: ["-e", "console.log('assembled-receipt-proof')"], cwd: process.cwd() }, + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } } }).returning(); + const [issue] = await fixture.insert(issues).values({ companyId: company.id, title: "Resume saved task", status: "todo", assigneeAgentId: agent!.id, responsibleUserId: "local-board" }).returning(); + const [source] = await fixture.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent!.id, issueId: issue!.id, status: "failed", finishedAt: new Date(), responsibleUserId: "local-board", errorCode: "process_failed", + contextSnapshot: { issueId: issue!.id }, resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }).returning(); + const hold = await server.api<{ hold: { id: string } }>("POST", `/issues/${issue!.id}/tree-holds`, { mode: "pause", reason: "Retain ordinary Retry" }); + const retry = await server.api<{ accepted: boolean; requestId: string }>("POST", `/issues/${issue!.id}/retry`, { observedRunId: source!.id, idempotencyKey: randomUUID() }); + expect(retry.accepted).toBe(true); + await settleClock(server); + const [waiting] = await fixture.$client`select status, run_id, payload->'executionWait'->>'reason' as reason from agent_wakeup_requests where id=${retry.requestId}`; + expect(waiting).toMatchObject({ status: "deferred_issue_execution", run_id: null, reason: "issue_tree_hold_active" }); + const beforeGrace = server.proxy.queries.length; + await server.realtime(); await delay(31_000); await server.freeze(); await settleClock(server); + const grace = receipt(server.proxy.queries.slice(beforeGrace)); + const [parked] = await fixture.$client`select status, run_id, payload->'executionWait'->>'reason' as reason from agent_wakeup_requests where id=${retry.requestId}`; + expect(parked).toEqual(waiting); + const baseline = server.proxy.queries.length; + await server.realtime(); await delay(30_000); await server.freeze(); await settleClock(server); + expect(server.proxy.queries.length).toBe(baseline); + await server.api("POST", `/issues/${issue!.id}/tree-holds/${hold.hold.id}/release`, { reason: "Resume accepted Retry", metadata: { wakeAgents: false } }); + await server.realtime(); + let executed: Record | undefined; + for (let step = 0; step < 300; step++) { + await delay(100); + [executed] = await fixture.$client`select status, stdout_excerpt, error, error_code from heartbeat_runs where context_snapshot->>'taskRetryRequestId'=${retry.requestId}`; + if (executed?.status === "succeeded") break; + } + await server.freeze(); + expect(executed, JSON.stringify(executed)).toMatchObject({ status: "succeeded", stdout_excerpt: expect.stringContaining("assembled-receipt-proof") }); + const [count] = await fixture.$client`select count(*)::integer as count from heartbeat_runs where context_snapshot->>'taskRetryRequestId'=${retry.requestId}`; + expect(count!.count).toBe(1); + await server.updateIssue({ id: issue!.id, status: "done" }); await settleClock(server); + const quiet = server.proxy.queries.length; + for (let step = 0; step < 4; step++) { await server.advance(30_000); await settleClock(server); } + expect(server.proxy.queries.length).toBe(quiet); + await writeFile(path.join(server.directory, "receipt-measurement.json"), JSON.stringify({ phase: "ordinary-held-task-retry", companyId: company.id, issueId: issue!.id, + requestId: retry.requestId, waiting, grace, parked, executed, count: count!.count, parkedRealMs: 30_000, parkedQueries: 0, settledClockMs: 120_000, settledQueries: 0, + boundary: "actual Retry and subtree release HTTP, wakeAgents false, real clock through process/controller authority, real local process, no outcome seeded", frames: server.proxy.frames }, null, 2)); + } finally { await server.stop(); await fixture.$client.end(); } + }, 150_000); + + it("reconstructs residual work from a normal scheduler-admission change after suppression", 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: "Admission reconstruction" }); + const { leaseId } = await server.createOrphanLease(company.id); + await server.advance(1_000); await settleClock(server); + const [suppressed] = await fixture.$client`select status, cleanup_status from environment_leases where id=${leaseId}`; + expect(suppressed).toMatchObject({ status: "active", cleanup_status: null }); + const baseline = server.proxy.queries.length, before = await server.snapshot(); + for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(baseline); + expect((await server.snapshot()).outgoingHttpRequests).toBe(before.outgoingHttpRequests); + await server.api("PATCH", "/instance/settings/experimental", { enableWorktreeRunExecution: true }); + await server.advance(1_000); await settleClock(server); + const [admitted] = await fixture.$client`select experimental from instance_settings where singleton_key='default'`; + expect(admitted!.experimental).toMatchObject({ enableWorktreeRunExecution: true, + worktreeRunExecutionActivatedAt: expect.any(String), worktreeRunExecutionActivationInstanceId: "maintenance-test" }); + const [reconstructed] = await fixture.$client`select status, failure_reason, cleanup_status from environment_leases where id=${leaseId}`; + expect(reconstructed).toMatchObject({ status: "expired", failure_reason: "pending_cleanup_retry", cleanup_status: "success" }); + await server.advance(30_000); await settleClock(server); + const [cleaned] = await fixture.$client`select status, cleanup_status from environment_leases where id=${leaseId}`; + expect(cleaned).toMatchObject({ status: "expired", cleanup_status: "success" }); + await writeFile(path.join(server.directory, "admission-measurement.json"), JSON.stringify({ phase: "ordinary-settings-admission-reconstruction", + companyId: company.id, leaseId, suppressed, suppressedClockMs: 120_000, suppressedQueries: 0, + admitted: admitted!.experimental, reconstructed, cleaned, frames: server.proxy.frames, + boundary: "actual suppressed runtime, committed environment-cleanup ingress, normal operator activation; no direct worker wake or safety-gate bypass", + }, null, 2)); + } finally { await server.stop(); await fixture.$client.end(); } + }, 150_000); + + it("isolates a real goal-startup write failure and still registers residual and attention owners", async () => { + await clearDatabase(); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + const companyId = randomUUID(), agentId = randomUUID(), userId = randomUUID(), issueId = randomUUID(), requestId = randomUUID(); + await fixture.insert(authUsers).values({ id: userId, name: "Goal control owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await fixture.insert(companies).values({ id: companyId, name: "Startup failure isolation", issuePrefix: "MFI", defaultResponsibleUserId: userId }); + await fixture.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" }); + await fixture.insert(agents).values({ id: agentId, companyId, name: "Goal owner", adapterType: "codex_local", status: "idle", lifecycleState: "ready" }); + await fixture.insert(issues).values({ id: issueId, companyId, title: "Retained terminal goal task", status: "done", assigneeAgentId: agentId, responsibleUserId: userId }); + const [session] = await fixture.insert(agentTaskSessions).values({ companyId, agentId, taskKey: issueId, adapterType: "codex_local" }).returning(); + const [action] = await fixture.insert(agentSessionGoalActions).values({ companyId, sessionId: session!.id, requestId, + action: "clear", payloadJson: { requestId, action: "clear" }, status: "pending" }).returning(); + await fixture.$client.unsafe(`create or replace function maintenance_fail_goal_startup() returns trigger language plpgsql as $$begin + if new.reason = 'goal_control' then raise exception 'maintenance_goal_startup_injected'; end if; return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_fail_goal_startup before insert on agent_wakeup_requests for each row execute function maintenance_fail_goal_startup()`); + let server: Awaited> | undefined; + try { + server = await startMaintenanceServer(databaseUrl!, { clock: true }); + await settleClock(server); + expect((await server.snapshot()).startup.phase).toBe("ready"); + expect(server.output).toContain("maintenance_goal_startup_injected"); + const [retained] = await fixture.$client`select status from agent_session_goal_actions where id=${action!.id}`; + expect(retained!.status).toBe("pending"); + const [failedWake] = await fixture.$client`select count(*)::integer as count from agent_wakeup_requests where reason='goal_control'`; + expect(failedWake!.count).toBe(0); + const start = (await server.snapshot()).now; + const { leaseId } = await server.createOrphanLease(companyId); + let discovered: Record | undefined; + for (let step = 0; step < 10; step++) { + await server.advance(1_000); await delay(50); + [discovered] = await fixture.$client`select status, failure_reason, provider, provider_lease_id from environment_leases where id=${leaseId}`; + if (discovered?.status === "pending_cleanup") break; + } + expect(discovered).toMatchObject({ status: "pending_cleanup", failure_reason: "orphaned_active_lease_recovered", provider: "local", provider_lease_id: null }); + const elapsedClockMs = (await server.snapshot()).now - start; + expect(elapsedClockMs).toBeLessThan(30_000); + await server.advance(30_000); await settleClock(server); + const [lease] = await fixture.$client`select status, cleanup_status, provider, provider_lease_id from environment_leases where id=${leaseId}`; + expect(lease).toMatchObject({ status: "expired", cleanup_status: "success", provider: "local", provider_lease_id: null }); + await fixture.$client.unsafe(`create or replace function maintenance_age_isolation_attention() returns trigger language plpgsql as $$begin + if new.title = 'Attention after goal failure' then new.updated_at = now() - interval '91 days'; end if; return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_age_isolation_attention before insert on issue_thread_interactions for each row execute function maintenance_age_isolation_attention()`); + const issue = await server.api<{ id: string }>("POST", `/companies/${companyId}/issues`, { title: "Independent attention owner", status: "todo" }); + const card = await server.api<{ id: string }>("POST", `/issues/${issue.id}/interactions`, { kind: "request_confirmation", title: "Attention after goal failure", + resolverPolicy: "human_only", continuationPolicy: "none", payload: { version: 1, prompt: "Choose a color?" } }); + await server.advance(1_000); await settleClock(server); + const [archived] = await fixture.$client`select archived_reason, archived_at is not null as archived from decision_retention where source_id=${card.id}`; + expect(archived).toMatchObject({ archived_reason: "idle_ttl", archived: true }); + await writeFile(path.join(server.directory, "startup-failure-measurement.json"), JSON.stringify({ phase: "actual-goal-startup-failure-isolation", + companyId, actionId: action!.id, retained, failedWake, leaseId, discovered, lease, elapsedClockMs, archived, frames: server.proxy.frames, + boundary: "fixture-only conditional goal_control INSERT failure; actual native classification untouched; actual environment cleanup ingress and ordinary attention HTTP producer", + }, null, 2)); + } finally { + await fixture.$client.unsafe("drop trigger if exists maintenance_fail_goal_startup on agent_wakeup_requests").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_fail_goal_startup()").catch(() => {}); + await fixture.$client.unsafe("drop trigger if exists maintenance_age_isolation_attention on issue_thread_interactions").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_age_isolation_attention()").catch(() => {}); + await server?.stop(); await fixture.$client.end(); + } + if (server!.retainedDirectory) await access(server!.retainedDirectory); + else await expect(access(server!.directory)).rejects.toMatchObject({ code: "ENOENT" }); + }, 150_000); + + it("exposes readiness while real deferred housekeeping is held and settles its work after release", async () => { + await clearDatabase(); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + const companyId = randomUUID(), issueId = randomUUID(), interactionId = randomUUID(); + await fixture.insert(companies).values({ id: companyId, name: "Readiness fixture", issuePrefix: "MRD" }); + await fixture.insert(issues).values({ id: issueId, companyId, title: "Historical human decision", status: "todo" }); + await fixture.insert(issueThreadInteractions).values({ id: interactionId, companyId, issueId, kind: "request_confirmation", + title: "Historical human choice", continuationPolicy: "none", payload: { version: 1, prompt: "Choose a color?" }, + updatedAt: new Date(Date.now() - 91 * 86_400_000) }); + await fixture.$client.unsafe(`create or replace function maintenance_hold_startup_archive() returns trigger language plpgsql as $$begin + perform pg_advisory_xact_lock(913703); return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_hold_startup_archive before insert on decision_retention for each row execute function maintenance_hold_startup_archive()`); + let reserved: Awaited> | undefined = await fixture.$client.reserve(); + await reserved`select pg_advisory_lock(913703)`; + let server: Awaited> | undefined; + try { + server = await startMaintenanceServer(databaseUrl!, { suppressed: true }); + const current = server; + await expect.poll(() => current.proxy.queries.some(query => /insert into "decision_retention"/.test(query.sql)), { timeout: 10_000 }).toBe(true); + const snapshot = await server.snapshot(); + expect(snapshot.startup.phase).toBe("ready"); expect(snapshot.idle.active).toBeGreaterThan(0); + expect(server.proxy.pending).toBeGreaterThan(0); + const health = await server.api<{ status: string; startupRecovery: { phase: string } }>("GET", "/health"); + expect(health).toMatchObject({ status: "ok", startupRecovery: { phase: "ready" } }); + await reserved`select pg_advisory_unlock(913703)`; reserved.release(); reserved = undefined; + await server.settled(); + const [archived] = await fixture.$client`select archived_reason, archived_at is not null as archived from decision_retention where source_id=${interactionId}`; + expect(archived).toMatchObject({ archived_reason: "idle_ttl", archived: true }); + await writeFile(path.join(server.directory, "readiness-measurement.json"), JSON.stringify({ phase: "real-housekeeping-held-after-readiness", + companyId, issueId, interactionId, beforeRelease: { startup: snapshot.startup, idle: snapshot.idle }, afterRelease: await server.snapshot(), + archived, boundary: "fixture owns advisory lock on actual startup archive INSERT; no startup/DB/service mocks", frames: server.proxy.frames, + }, null, 2)); + } finally { + if (reserved) { await reserved`select pg_advisory_unlock(913703)`.catch(() => {}); reserved.release(); } + await fixture.$client.unsafe("drop trigger if exists maintenance_hold_startup_archive on decision_retention").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_hold_startup_archive()").catch(() => {}); + await fixture.$client.end(); await server?.stop(); + } + }, 150_000); + + it("does not advance the child clock over an actual driver socket whose startup write is held", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true }); + const client = createDb(server.proxy.url, { maxConnections: 1, idleTimeoutSeconds: 0, connectTimeoutSeconds: 30 }); + const address = new URL(server.proxy.url); + const socket = new Socket(); + let release = () => {}; + let query: Promise | undefined; + let advancing: ReturnType | undefined; + try { + await settleClock(server); await server.freeze(); + socket.connect({ host: address.hostname, port: Number(address.port) }); + await once(socket, "connect"); + const write = socket.write; + let entered!: () => void; + const startupHeld = new Promise(resolve => { entered = resolve; }); + socket.write = ((...args: Parameters) => { + socket.write = write; + expect(Buffer.isBuffer(args[0]) && args[0].readInt32BE(4)).toBe(196608); + let released = false; + release = () => { if (!released) { released = true; Reflect.apply(write, socket, args); } }; + entered(); + return true; + }) as typeof socket.write; + Object.assign(client.$client.options, { socket: () => socket }); + query = client.$client`select 1 as startup_control`.then(rows => rows); + await startupHeld; + const before = await server.snapshot(); + advancing = server.advance(30_000); + const during = await server.snapshot(); + expect(during.now).toBe(before.now); + release(); + expect(await query).toMatchObject([{ startup_control: 1 }]); + expect((await advancing).now).toBe(before.now + 30_000); + await server.settled(); + await writeFile(path.join(server.directory, "startup-clock-measurement.json"), JSON.stringify({ phase: "held-real-driver-startup-clock-barrier", + before, during, after: await server.snapshot(), boundary: "actual postgres.js StartupMessage write held before wire flush; child clock IPC stays behind startup, independent of the unexplained invalid-frame failure", + }, null, 2)); + } finally { + release(); + await Promise.allSettled([query, advancing]); + await client.$client.end(); socket.destroy(); await server.stop(); + } + }, 150_000); + + it("keeps the disabled-heartbeat boot quiet across two former hourly windows with a qualified child clock", async () => { + await clearDatabase(); + const server = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true }); + try { + await server.settled(); + await server.freeze(); + const baseline = server.proxy.queries.length, start = (await server.snapshot()).now; + for (let step = 0; step < 240; step++) { + await server.advance(30_000); + await server.settled(10); + } + const after = await server.snapshot(); + expect(after.now - start).toBe(2 * 60 * 60_000); + expect(server.proxy.queries.length).toBe(baseline); + expect(after.outgoingHttpRequests).toBe(0); + expect(after.network.nonDatabaseConnections).toBe(0); + expect(server.proxy.errors).toEqual([]); + await writeFile(path.join(server.directory, "measurement.json"), JSON.stringify({ phase: "disabled-empty-boot-clock", + clockAdvanceMs: after.now - start, stepMs: 30_000, ioSettledBetweenSteps: true, + startupQueries: baseline, startup: receipt(server.proxy.queries), frames: server.proxy.frames, network: after.network, measuredQueries: 0, outgoingHttpRequests: 0 }, null, 2)); + + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Configured idle maintenance fixture" }); + await server.api("PATCH", "/instance/settings/experimental", { enableExternalObjects: true, enableStatusCards: true }); + const history = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { title: "Retained completed history", status: "done" }); + await server.advance(30_000); + await server.settled(); + const persisted = createDb(databaseUrl!, { maxConnections: 1 }); + try { + const [saved] = await persisted.$client`select company_id, status from issues where id=${history.id}`; + expect(saved).toMatchObject({ company_id: company.id, status: "done" }); + } finally { await persisted.$client.end(); } + const historyBaseline = server.proxy.queries.length; + await server.drain(); + for (let step = 0; step < 3; step++) { await server.advance(30_000); await server.settled(10); } + await server.resume(); await server.settled(); + const resumed = server.proxy.queries.length; + for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(resumed); + expect((await server.snapshot()).outgoingHttpRequests).toBe(0); + await writeFile(path.join(server.directory, "history-measurement.json"), JSON.stringify({ phase: "disabled-configured-history-drain", + companyId: company.id, issueId: history.id, historyBaseline, resumedQueries: resumed, + postResumeClockMs: 2 * 60 * 60_000, measuredQueries: server.proxy.queries.length - resumed, + frames: server.proxy.frames, outgoingHttpRequests: 0 }, null, 2)); + } 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 withMaintenanceProvider(provider, async origin => { + const options = { enabled: false, clock: true, githubOrigin: origin }; + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + let server: Awaited> | undefined; + try { + server = await startMaintenanceServer(databaseUrl!, options); + 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 { + try { await server?.stop(); } finally { await fixture.$client.end(); } + } + }); + }, 150_000); + + it("processes the first ordinary external-object producer committed during idle drain and becomes quiet again", async () => { + await clearDatabase(); + const providerRequests: string[] = []; + const provider = createServer((req, res) => { + providerRequests.push(req.url!); + res.setHeader("Content-Type", "application/json"); + if (!/^\/repos\/maintenance\/example\/pulls\/[12]$/.test(req.url!)) { res.statusCode = 500; res.end(JSON.stringify({ error: "Unexpected simulated GitHub request" })); return; } + if (req.url!.endsWith("/2") && providerRequests.filter(url => url.endsWith("/2")).length === 1) { + res.statusCode = 503; res.setHeader("Retry-After", "2"); res.end(JSON.stringify({ message: "Controlled provider outage" })); return; + } + const number = Number(req.url!.split("/").at(-1)); + res.end(JSON.stringify({ id: number, number, state: "closed", merged: true, merged_at: new Date().toISOString(), + title: "Controlled merged pull request", html_url: `https://github.com/maintenance/example/pull/${number}`, user: { login: "fixture" } })); + }); + await withMaintenanceProvider(provider, async origin => { + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + let reserved: Awaited> | undefined; + let pendingIssue: Promise<{ id: string }> | undefined; + let runtime: Awaited> | undefined; + try { + const server = runtime = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true, + githubOrigin: origin }); + await server.settled(); + const company = await server.api<{ id: string }>("POST", "/companies", { name: "Drain activation" }); + await server.api("PATCH", "/instance/settings/experimental", { enableExternalObjects: true }); + await server.advance(30_000); await server.settled(); + await fixture.$client.unsafe(`create or replace function maintenance_hold_issue() returns trigger language plpgsql as $$begin + if new.title = 'First write during idle drain' then perform pg_advisory_xact_lock(913701); end if; + return new; end$$`); + await fixture.$client.unsafe(`create trigger maintenance_hold_issue before insert on issues for each row execute function maintenance_hold_issue()`); + reserved = await fixture.$client.reserve(); + await reserved`select pg_advisory_lock(913701)`; + const before = server.proxy.queries.length; + pendingIssue = server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { + title: "First write during idle drain", description: "https://github.com/maintenance/example/pull/1", status: "done", + }); + void pendingIssue.catch(() => {}); + await expect.poll(() => server.proxy.queries.slice(before).some(query => /insert into "issues"/.test(query.sql)), { timeout: 10_000 }).toBe(true); + await server.drain(); + await reserved`select pg_advisory_unlock(913701)`; reserved.release(); reserved = undefined; + const issue = await pendingIssue; + await server.settled(); + expect(providerRequests).toEqual([]); + const [waiting] = await fixture.$client`select eo.company_id, eo.is_terminal, eo.last_resolved_at from external_objects eo + join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${issue.id}`; + expect(waiting).toMatchObject({ company_id: company.id, is_terminal: false, last_resolved_at: null }); + const draining = server.proxy.queries.length; + for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(draining); + await server.resume(); await server.advance(30_000); await server.settled(); + const [resolved] = await fixture.$client`select eo.company_id, eo.is_terminal, eo.status_key, floor(extract(epoch from eo.last_resolved_at) * 1000)::float8 as resolved_ms from external_objects eo + join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${issue.id}`; + expect(resolved).toMatchObject({ company_id: company.id, is_terminal: true, status_key: "merged", resolved_ms: expect.any(Number) }); + expect(resolved!.resolved_ms).toBeGreaterThan(0); + expect(resolved!.resolved_ms).toBeLessThanOrEqual((await server.snapshot()).now); + expect(providerRequests).toEqual(["/repos/maintenance/example/pulls/1"]); + const settledQueries = server.proxy.queries.length, network = await server.snapshot(); + for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(settledQueries); + expect(providerRequests).toHaveLength(1); + expect((await server.snapshot()).outgoingHttpRequests).toBe(network.outgoingHttpRequests); + await writeFile(path.join(server.directory, "external-object-measurement.json"), JSON.stringify({ + phase: "ordinary-issue-producer-commit-during-drain", providerBoundary: "simulated GitHub HTTP; real application/database/transactions", + companyId: company.id, issueId: issue.id, resolved, drainedClockMs: 120_000, drainedQueries: 0, + postResolveClockMs: 7_200_000, postResolveQueries: 0, providerRequests, frames: server.proxy.frames, network, + }, null, 2)); + + const retryIssue = await server.api<{ id: string }>("POST", `/companies/${company.id}/issues`, { + title: "Retry after idle drain", description: "https://github.com/maintenance/example/pull/2", status: "done", + }); + await server.advance(1_000); await server.settled(100); + const [failed] = await fixture.$client`select eo.last_error_code, floor(extract(epoch from eo.next_refresh_at) * 1000)::float8 as next_ms + from external_objects eo join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${retryIssue.id}`; + expect(failed!.last_error_code).toBe("github_unreachable"); + expect(failed!.next_ms).toBeGreaterThan((await server.snapshot()).now); + await server.drain(); await server.settled(); + const retryDrainQueries = server.proxy.queries.length; + for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(retryDrainQueries); + expect(providerRequests.filter(url => url.endsWith("/2"))).toHaveLength(1); + await server.resume(); await server.advance(30_000); await server.settled(); + const [retried] = await fixture.$client`select eo.is_terminal, eo.status_key, eo.last_error_code from external_objects eo + join external_object_mentions em on em.object_id=eo.id where em.source_issue_id=${retryIssue.id}`; + expect(retried).toMatchObject({ is_terminal: true, status_key: "merged", last_error_code: null }); + expect(providerRequests.filter(url => url.endsWith("/2"))).toHaveLength(2); + const retrySettled = server.proxy.queries.length; + for (let step = 0; step < 20; step++) { await server.advance(30_000); await server.settled(10); } + expect(server.proxy.queries.length).toBe(retrySettled); + await writeFile(path.join(server.directory, "retry-measurement.json"), JSON.stringify({ phase: "provider-failure-retry-across-drain", + providerBoundary: "simulated GitHub HTTP 503 then merged", retryIssueId: retryIssue.id, failed, retried, + drainedClockMs: 120_000, drainedQueries: 0, quietAfterRetryMs: 600_000, quietAfterRetryQueries: 0, + providerRequests, frames: server.proxy.frames }, null, 2)); + } finally { + if (reserved) { await reserved`select pg_advisory_unlock(913701)`.catch(() => {}); reserved.release(); } + await pendingIssue?.catch(() => {}); + await fixture.$client.unsafe("drop trigger if exists maintenance_hold_issue on issues").catch(() => {}); + await fixture.$client.unsafe("drop function if exists maintenance_hold_issue()").catch(() => {}); + try { await fixture.$client.end(); } finally { await runtime?.stop(); } + } + }); + }, 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 withMaintenanceProvider(broker, async origin => { + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + let runtime: Awaited> | undefined; + try { + const server = runtime = await startMaintenanceServer(databaseUrl!, { enabled: false, clock: true, + githubOrigin: origin, connector: { + origin: origin, signPrivateKey: signing.privateKey.export({ format: "jwk" }).d!, + sealPrivateKey: sealing.privateKey.export({ format: "jwk" }).d!, + } }); + 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 { + try { await fixture.$client.end(); } finally { await runtime?.stop(); } + } + }); + }, 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 withMaintenanceProvider(provider, async origin => { + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + let runtime: Awaited> | undefined; + try { + const server = runtime = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true, + githubOrigin: origin }); + 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 { + try { await fixture.$client.end(); } finally { await runtime?.stop(); } + } + }); + }, 150_000); + + it("projects the first ordinary routine completion in the actual suppressed Cloud runtime", async () => { + await clearDatabase(); + const companyId = randomUUID(), agentId = randomUUID(); + 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 withMaintenanceProvider(storage, async origin => { + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + let runtime: Awaited> | undefined; + try { + 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 server = runtime = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true, + cloud: { jwks: JSON.stringify({ keys: [publicJwk] }), s3Origin: origin, standby: false } }); + 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 restoredRun = await server.api<{ id: string; linkedIssueId: string }>("POST", `/routines/${routine.id}/run`, { + source: "manual", idempotencyKey: randomUUID(), + }); + const cancel = await server.api<{ hold: { id: string } }>("POST", `/issues/${restoredRun.linkedIssueId}/tree-holds`, { mode: "cancel", reason: "Ordinary cancel before restore" }); + await server.advance(1_000); await settleClock(server); + const [cancelled] = await fixture.$client`select status from routine_runs where id=${restoredRun.id}`; + expect(cancelled!.status).toBe("failed"); + await server.api("POST", `/issues/${restoredRun.linkedIssueId}/tree-holds`, { mode: "restore", reason: "Ordinary restore" }); + await server.advance(1_000); await settleClock(server); + const [restored] = await fixture.$client`select rr.status, rr.completed_at, rr.failure_reason, i.status as issue_status + from routine_runs rr join issues i on i.id=rr.linked_issue_id where rr.id=${restoredRun.id}`; + expect(restored).toMatchObject({ status: "issue_created", completed_at: null, failure_reason: null, issue_status: "todo" }); + const [cancelHold] = await fixture.$client`select status from issue_tree_holds where id=${cancel.hold.id}`; + expect(cancelHold!.status).toBe("released"); + 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, restoredRun, cancelled, restored, cancelHold, 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(() => {}); + try { await runtime?.stop(); } finally { await fixture.$client.end(); } + } + }); + }, 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"); + const publicJwk = { ...pair.publicKey.export({ format: "jwk" }), kid: "maintenance-test-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 empty standby fixture"); }); + await withMaintenanceProvider(storage, async origin => { + let runtime: Awaited> | undefined; + try { + const server = runtime = await startMaintenanceServer(databaseUrl!, { clock: true, + cloud: { jwks: JSON.stringify({ keys: [publicJwk] }), s3Origin: origin, standby: true } }); + const standby = await server.settled(500, 1); await server.freeze(); + expect(standby.idle.active).toBe(1); + const before = server.proxy.queries.length; + for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10, 1); } + const health = await fetch(`http://127.0.0.1:${server.port}/api/health`, { signal: AbortSignal.timeout(10_000) }); + expect(health.status).toBe(200); expect(await health.json()).toMatchObject({ warmStandby: true }); + expect(server.proxy.queries.length).toBe(before); + expect(storageRequests).toEqual([]); + expect((await server.snapshot()).network.nonDatabaseConnections).toBe(0); + const now = Math.floor((await server.snapshot()).now / 1000); + const header = Buffer.from(JSON.stringify({ alg: "EdDSA", typ: CLOUD_RUNTIME_IDENTITY_JWS_TYPE, kid: publicJwk.kid })).toString("base64url"); + const payload = Buffer.from(JSON.stringify({ v: 1, iss: CLOUD_RUNTIME_IDENTITY_ISSUER, aud: CLOUD_RUNTIME_IDENTITY_AUDIENCE, + sub: "maintenance-stack", claimId: "maintenance-claim", previousOrigin: "https://pool-maintenance.staging.paperclip.app", + canonicalOrigin: "https://claimed-maintenance.staging.paperclip.app", stackSlug: "claimed-maintenance", iat: now, exp: now + 300 })).toString("base64url"); + const assertion = `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`, "ascii"), pair.privateKey).toString("base64url")}`; + const claimed = await fetch(`http://127.0.0.1:${server.port}/api/health`, { + headers: { "x-paperclip-cloud-runtime-identity": assertion }, signal: AbortSignal.timeout(10_000), + }); + expect(claimed.status).toBe(200); expect((await claimed.json()).warmStandby).toBeUndefined(); + await server.advance(30_000); const afterClaim = await server.settled(); + expect(afterClaim.idle.active).toBe(0); + expect(server.proxy.queries.length).toBeGreaterThan(before); + const fixture = createDb(databaseUrl!, { maxConnections: 1 }); + try { + const [transfers] = await fixture.$client`select count(*)::integer as count from company_transfer_runs`; + expect(transfers!.count).toBe(0); + const [identity] = await fixture.$client`select general from instance_settings where singleton_key='cloud-runtime-identity/v1'`; + expect(identity!.general).toMatchObject({ stackId: "maintenance-stack", claimId: "maintenance-claim", canonicalOrigin: "https://claimed-maintenance.staging.paperclip.app" }); + } finally { await fixture.$client.end(); } + await writeFile(path.join(server.directory, "standby-measurement.json"), JSON.stringify({ phase: "real-cloud-standby-signed-claim", + cloudBoundary: "actual managed configuration, local S3 HTTP guard, generated Ed25519 trust root; no production Cloud claim", + clockMs: 7_200_000, standbyQueries: 0, + deferredStartupIntent: { count: 1, queue: "import-spool-cleanup", producer: "companyTransferRunService.recoverStrandedApplyingRuns", + reason: "Cloud standby disables both reconciliation and dispatch", releasedAfterSignedClaim: true, boundedResumeMs: 30_000, sleepEligibleWhileHeld: false }, storageRequests, beforeClaimQueries: before, afterClaimQueries: server.proxy.queries.length, + frames: server.proxy.frames, network: await server.snapshot() }, null, 2)); + } finally { + await runtime?.stop(); + } + }); + }, 150_000); + +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 2ada07645ec..dabee1c2f03 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -186,7 +186,6 @@ const { scanSilentActiveRuns: vi.fn(async () => ({ created: 0, escalated: 0 })), sweepStaleIssueLocks: vi.fn(async () => ({ cleared: 0 })), sweepPendingCleanupLeases: vi.fn(async () => ({ swept: 0, destroyed: 0, capped: 0 })), - sweepExpiredRuntimeStatuses: vi.fn(() => 0), tickAgentTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })), nextAgentTimerAt: vi.fn(async () => null), tickIssueMonitors: vi.fn(async () => ({ checked: 0, triggered: 0, skipped: 0 })), @@ -836,7 +835,7 @@ describe("startServer feedback export wiring", () => { expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.skillVersionRetention)).toEqual([]); }); - it.each([false, true])("registers status-card evaluation only with maintenance and never ticks it from the heartbeat loop (%s)", async enabled => { + it.each([false, true])("registers status-card evaluation only with maintenance and creates no aggregate interval (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 1000 })); let tick: (() => void) | undefined; const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, interval: number) => { @@ -850,8 +849,8 @@ describe("startServer feedback export wiring", () => { if (enabled) { const service = vi.mocked(statusCardEvaluationService).mock.results.at(-1)!.value; expect(service.run).toHaveBeenCalledOnce(); - expect(tick).toBeDefined(); - for (let i = 0; i < 3; i++) { tick!(); await settleStartupHousekeeping(); } + expect(tick).toBeUndefined(); + await settleStartupHousekeeping(); expect(service.run).toHaveBeenCalledOnce(); } expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.statusCardEvaluation)).toEqual([]); @@ -1073,7 +1072,7 @@ describe("startServer feedback export wiring", () => { expect(registration).toBeUndefined(); expect(attentionRetentionService).not.toHaveBeenCalled(); expect(decisionRetentionService).not.toHaveBeenCalled(); - expect(tick).toBeDefined(); tick!(); await settleStartupHousekeeping(); + expect(tick).toBeUndefined(); await settleStartupHousekeeping(); expect(decisionRetentionService).not.toHaveBeenCalled(); return; } @@ -1089,8 +1088,8 @@ describe("startServer feedback export wiring", () => { const companyReads = () => vi.mocked(db.select).mock.results.filter(({ value }) => vi.mocked(value.from).mock.calls.some(([table]) => table === companies)).length; const before = companyReads(); - expect(before).toBe(0); expect(tick).toBeDefined(); - tick!(); await settleStartupHousekeeping(); + expect(before).toBe(0); expect(tick).toBeUndefined(); + await settleStartupHousekeeping(); expect(companyReads()).toBe(before); expect(source.load).toHaveBeenCalledOnce(); expect(retention.deliverNotifications).toHaveBeenCalledOnce(); @@ -1397,7 +1396,7 @@ describe("startServer feedback export wiring", () => { } finally { intervals.mockRestore(); } }); - it("does not rescan idle tool health from the aggregate maintenance tick", async () => { + it("registers idle tool health without an aggregate maintenance interval", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, heartbeatSchedulerIntervalMs: 30_000 })); let tick: (() => void) | undefined; const intervals = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, ms: number) => { @@ -1414,9 +1413,8 @@ describe("startServer feedback export wiring", () => { const loads = vi.mocked(toolAccessService).mock.results.map(result => result.value.scheduledHealth.load); const loadCount = () => loads.reduce((total, load) => total + load.mock.calls.length, 0); expect(loadCount()).toBe(1); - expect(tick).toEqual(expect.any(Function)); - tick!(); await new Promise(resolve => setImmediate(resolve)); - tick!(); await new Promise(resolve => setImmediate(resolve)); + expect(tick).toBeUndefined(); + await new Promise(resolve => setImmediate(resolve)); expect(count()).toBe(0); expect(loadCount()).toBe(1); } finally { intervals.mockRestore(); } @@ -1439,8 +1437,7 @@ describe("startServer feedback export wiring", () => { await settleStartupHousekeeping(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled(); - expect(intervalCallback).not.toBeNull(); - intervalCallback?.(); + expect(intervalCallback).toBeNull(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled(); @@ -1533,8 +1530,7 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); - expect(intervalCallback).not.toBeNull(); - intervalCallback?.(); + expect(intervalCallback).toBeNull(); await Promise.resolve(); await Promise.resolve(); @@ -1591,14 +1587,11 @@ describe("startServer feedback export wiring", () => { try { await startServer(); - // The disabled path still creates one heartbeat runtime. This runtime owns - // the orphan-sandbox cleanup sweep, so a leaked provider sandbox is still - // reaped at startup and on the interval. + // The disabled path retains startup recovery and cleanup deadlines. expect(heartbeatServiceFactoryMock).toHaveBeenCalledTimes(1); expect(heartbeatServiceMock.sweepPendingCleanupLeases).toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); - expect(intervalCallback).not.toBeNull(); - intervalCallback?.(); + expect(intervalCallback).toBeNull(); await Promise.resolve(); await Promise.resolve(); @@ -1629,8 +1622,7 @@ describe("startServer feedback export wiring", () => { expect((await startServer()).server).toBe(fakeServer); await settleStartupHousekeeping(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); - expect(tick).toBeDefined(); - tick?.(); + expect(tick).toBeUndefined(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.accounting); @@ -1724,10 +1716,12 @@ describe("startServer feedback export wiring", () => { expect(getStartupRecoveryState().phase).toBe("ready"); await settleStartupHousekeeping(); expect(heartbeatIntervals).toHaveLength(0); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.routineSchedules)).toBe(false); releaseCleanup?.(); await settleStartupHousekeeping(); - expect(heartbeatIntervals).toHaveLength(1); + expect(heartbeatIntervals).toHaveLength(0); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.routineSchedules)).toBe(true); } finally { setIntervalSpy.mockRestore(); } @@ -1755,7 +1749,8 @@ describe("startServer feedback export wiring", () => { expect(await registration![1].hasPending!()).toBe(true); expect(environmentCustomImagesServiceMock.nextSetupCleanupAt).not.toHaveBeenCalled(); // The scheduler starts only after every housekeeping step settled. - expect(heartbeatIntervals).toHaveLength(1); + expect(heartbeatIntervals).toHaveLength(0); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.routineSchedules)).toBe(true); } finally { setIntervalSpy.mockRestore(); } diff --git a/server/src/index.ts b/server/src/index.ts index c0e6a7ffc9c..51af366e482 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1251,7 +1251,6 @@ async function startServerWithDatabaseTeardown( drainRunIds?: string[]; }>) | null = null; let heartbeatSchedulerStopped = false; - let heartbeatSchedulerInterval: ReturnType | null = null; const heartbeatSchedulerInFlight = new Set>(); const trackHeartbeatSchedulerWork = (work: Promise) => { const finishIdleWork = beginIdleTrackedWork(); @@ -1269,12 +1268,6 @@ async function startServerWithDatabaseTeardown( await Promise.allSettled([...heartbeatSchedulerInFlight]); } }; - const startHeartbeatSchedulerInterval = (callback: () => void) => { - heartbeatSchedulerInterval = setInterval(() => { - if (!isWarmStandby() && !isIdleTaskDrainActive()) callback(); - }, config.heartbeatSchedulerIntervalMs); - heartbeatSchedulerInterval?.unref?.(); - }; const skillSourceSync = skillSourceSyncService(db as any); const nativeHarnessBackupExpiration = nativeHarnessBackupExpirationService(db as any, { duty: "backups", @@ -1799,12 +1792,6 @@ async function startServerWithDatabaseTeardown( if (config.heartbeatSchedulerEnabled) await heartbeat.registerStoppedInstructionCollectionWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; - // Startup housekeeping: provider and connection sweeps that every timer tick - // repeats and that neither serving requests nor run recovery depends on. - // Readiness does not wait for it. It is tracked as scheduler work, so idle - // sleep and shutdown wait for it. Each step logs its own failure and the - // next step still runs. The scheduler interval starts after it, so the login - // reapers and the lease cleanup still finish before the first timer tick. const runStartupHousekeeping = async () => { const step = (name: string, work: () => Promise) => work().catch((err: unknown) => { @@ -1847,31 +1834,6 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); }; - const createHeartbeatSchedulerTick = (scheduled: ReturnType) => () => { - // Track the outer async callback as well as the work it starts. Shutdown - // can then wait through an already-running suppression check before it - // captures the authoritative set of running heartbeat rows. - trackHeartbeatSchedulerWork((async () => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses(); - if (sweptRuntimeStatuses > 0) { - logger.info( - { swept: sweptRuntimeStatuses }, - "heartbeat runtime-status sweeper cleared expired entries", - ); - } - - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - - - })().catch((err) => { - logger.error({ err }, "heartbeat scheduler tick failed"); - })); - }; const startHeartbeatScheduler = () => { if (isCloudManagedInstance()) { const status = registerRoutineExecutionStatusWork(deliveryWork, routineExecutionStatusService(db as any), @@ -1897,7 +1859,6 @@ async function startServerWithDatabaseTeardown( }), }); trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.monitors.ready, scheduled.routines.ready])); - startHeartbeatSchedulerInterval(createHeartbeatSchedulerTick(scheduled)); }; trackHeartbeatSchedulerWork(runStartupHousekeeping().then(() => { if (!heartbeatSchedulerStopped) startHeartbeatScheduler(); @@ -1908,8 +1869,6 @@ async function startServerWithDatabaseTeardown( // this path registers the same startup recovery and pending-work deadlines // as the enabled path. await startEnvironmentCleanup(); - startHeartbeatSchedulerInterval(() => { - }); } if (config.databaseBackupEnabled) { @@ -2026,10 +1985,6 @@ async function startServerWithDatabaseTeardown( await workspaceRecoveryWork.stop(); unregisterTerminalWorkspaceProbe?.(); unsubscribeNativeWorkspaceSyncRetention?.(); - if (heartbeatSchedulerInterval) { - clearInterval(heartbeatSchedulerInterval); - heartbeatSchedulerInterval = null; - } const heartbeatShutdown = await coordinateHeartbeatSchedulerShutdown({ signal, diff --git a/server/src/services/decision-target-work.ts b/server/src/services/decision-target-work.ts index 219c9ad6f2b..b26e658a3b3 100644 --- a/server/src/services/decision-target-work.ts +++ b/server/src/services/decision-target-work.ts @@ -7,11 +7,13 @@ export function registerDecisionTargetWork( service: Pick, "sweepInvalidTargets" | "decisionTargetsChanged" | "hasDecisionTargetWork">, canRun: () => boolean, ) { + const retryMs = 30_000; return coordinator.register(DELIVERY_QUEUES.decisionTargets, { - retryMs: 30_000, + retryMs, pausePassiveMaintenance: true, onSettled: service.decisionTargetsChanged, run: signal => canRun() ? service.sweepInvalidTargets(new Date(), signal) : Promise.resolve(), - hasPending: async () => service.hasDecisionTargetWork(), + hasPending: async () => canRun() && service.hasDecisionTargetWork(), + nextRunAt: () => service.hasDecisionTargetWork() ? Date.now() + retryMs : null, }); } diff --git a/server/src/services/heartbeat-run-runtime-status.test.ts b/server/src/services/heartbeat-run-runtime-status.test.ts index 80984155eea..f75eb616444 100644 --- a/server/src/services/heartbeat-run-runtime-status.test.ts +++ b/server/src/services/heartbeat-run-runtime-status.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { clearAllHeartbeatRunRuntimeStatuses, clearHeartbeatRunRuntimeStatus, @@ -12,6 +12,7 @@ import { describe("heartbeat run runtime status store", () => { afterEach(() => { clearAllHeartbeatRunRuntimeStatuses(); + vi.useRealTimers(); }); it("stores scoped ephemeral status and expires stale entries", () => { @@ -166,6 +167,43 @@ describe("heartbeat run runtime status store", () => { }); }); + it("cleans unread abandoned statuses on later output without polling or expiring fresh context", () => { + vi.useFakeTimers({ toFake: ["Date"] }); + const start = new Date("2026-10-30T00:00:00Z"); + vi.setSystemTime(start); + const write = (runId: string) => setHeartbeatRunRuntimeStatus({ + companyId: "company-1", issueId: null, agentId: "agent-1", runId, + phase: "run_activity", message: "Using Bash", + }); + write("abandoned"); + write("active"); + vi.setSystemTime(start.getTime() + 60_000); + touchHeartbeatRunRuntimeStatus({ + companyId: "company-1", issueId: null, agentId: "agent-1", runId: "active", + }); + vi.setSystemTime(start.getTime() + 90_000); + write("boundary"); + expect(getHeartbeatRunRuntimeStatus("abandoned")).toMatchObject({ message: "Using Bash" }); + expect(clearHeartbeatRunRuntimeStatus("boundary")).toBe(true); + vi.setSystemTime(start.getTime() + 90_001); + touchHeartbeatRunRuntimeStatus({ + companyId: "company-1", issueId: null, agentId: "agent-1", runId: "active", + }); + expect(clearHeartbeatRunRuntimeStatus("abandoned")).toBe(false); + expect(getHeartbeatRunRuntimeStatus("active")).toMatchObject({ message: "Using Bash" }); + setHeartbeatRunRuntimeStatus({ + companyId: "company-1", issueId: null, agentId: "agent-1", runId: "old-output", + phase: "run_activity", message: "Delayed output", updatedAt: start, + }); + vi.setSystemTime(start.getTime() + 90_002); + write("next-output"); + expect(clearHeartbeatRunRuntimeStatus("old-output")).toBe(true); + vi.setSystemTime(start.getTime() + 180_002); + write("new"); + expect(clearHeartbeatRunRuntimeStatus("active")).toBe(false); + expect(getHeartbeatRunRuntimeStatus("new")).toMatchObject({ message: "Using Bash" }); + }); + it("sweeps expired statuses without touching fresh entries", () => { setHeartbeatRunRuntimeStatus({ companyId: "company-1", diff --git a/server/src/services/heartbeat-run-runtime-status.ts b/server/src/services/heartbeat-run-runtime-status.ts index d2995337e23..1d78b1ffb0a 100644 --- a/server/src/services/heartbeat-run-runtime-status.ts +++ b/server/src/services/heartbeat-run-runtime-status.ts @@ -20,6 +20,18 @@ export interface HeartbeatRunRuntimeStatus { } const runtimeStatusesByRunId = new Map(); +let nextWriteCleanupAt: number | null = null; + +function cleanupExpiredStatusesOnWrite() { + const now = Date.now(); + if (nextWriteCleanupAt === null) { + nextWriteCleanupAt = now + HEARTBEAT_RUN_RUNTIME_STATUS_TTL_MS; + return; + } + if (now <= nextWriteCleanupAt) return; + sweepExpiredHeartbeatRunRuntimeStatuses(new Date(now)); + nextWriteCleanupAt = now + HEARTBEAT_RUN_RUNTIME_STATUS_TTL_MS; +} function cloneStatus(status: HeartbeatRunRuntimeStatus): HeartbeatRunRuntimeStatus { return { @@ -64,6 +76,7 @@ export function setHeartbeatRunRuntimeStatus( lastEventAt?: Date | null; }, ): HeartbeatRunRuntimeStatus | null { + cleanupExpiredStatusesOnWrite(); const message = sanitizeHeartbeatRunRuntimeStatusMessage(input.message); if (!message) { clearHeartbeatRunRuntimeStatus(input.runId); @@ -106,6 +119,7 @@ export function touchHeartbeatRunRuntimeStatus(input: { fallbackPhase?: HeartbeatRunStatusPhase; fallbackMessage?: string; }): HeartbeatRunRuntimeStatus | null { + cleanupExpiredStatusesOnWrite(); const at = input.at ?? new Date(); const existing = runtimeStatusesByRunId.get(input.runId); if ( @@ -167,6 +181,7 @@ export function clearHeartbeatRunRuntimeStatus(runId: string): boolean { export function clearAllHeartbeatRunRuntimeStatuses(): void { runtimeStatusesByRunId.clear(); + nextWriteCleanupAt = null; } export function sweepExpiredHeartbeatRunRuntimeStatuses( diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 28fcf43e644..77e28c77660 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -525,7 +525,6 @@ import { environmentRunOrchestrator } from "./environment-run-orchestrator.js"; import { clearHeartbeatRunRuntimeStatus, - sweepExpiredHeartbeatRunRuntimeStatuses, } from "./heartbeat-run-runtime-status.js"; import { resolveAndRetainRunTrustPreset } from "./run-trust-preset.js"; @@ -4026,7 +4025,6 @@ export function heartbeatService( decorateActiveRunStatus: decorateHeartbeatRunRuntimeStatus, recordRuntimeProgress: recordCurrentHeartbeatRunRuntimeProgress, - sweepExpiredRuntimeStatuses: sweepExpiredHeartbeatRunRuntimeStatuses, getRunReadAccess, getRunLogAccess, diff --git a/server/src/services/native-runtime/runner-api.integration.test.ts b/server/src/services/native-runtime/runner-api.integration.test.ts index fdb5b7f70a2..15d9b687c8c 100644 --- a/server/src/services/native-runtime/runner-api.integration.test.ts +++ b/server/src/services/native-runtime/runner-api.integration.test.ts @@ -6,6 +6,7 @@ import { and, eq } from "drizzle-orm"; import { agents, agentWakeupRequests, assets, documents, heartbeatRuns, issues, projects, routineDocuments, routines, runnerApiResponseReservations, companySkills, companySkillVersions, companySkillTestRuns, companyMemberships, principalPermissionGrants } from "@paperclipai/db"; import { initializeRunIdentity } from "../run-identity.js"; import { beforeAll, afterAll, beforeEach, afterEach, describe, expect, it, vi } from "vitest"; +import * as nativeSetupReadiness from "./setup-readiness.js"; import { secretService } from "../secrets.js"; import { authorizationService } from "../authorization.js"; import { startRunnerApiTestServer } from "../../__tests__/helpers/runner-api-server.js"; @@ -23,6 +24,11 @@ import { RUNNER_TOOL_RESULT_MAX_BYTES } from "./runner-tool-results.js"; describe("runner API against real HTTP routes", () => { let server: Awaited>; + const nativeAuthProbe = vi.fn(async () => ({ + adapterType: "paperclip_runner", status: "fail", testedAt: new Date().toISOString(), + checks: [{ code: "codex_hello_probe_auth_required", level: "error", message: "The API authority fixture has no provider credentials." }], + })); + let restoreNativeAuthProbe: (() => void) | undefined; const oldSecret = process.env.PAPERCLIP_AGENT_JWT_SECRET; const oldMasterKeyFile = process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; beforeEach(() => { @@ -31,16 +37,22 @@ describe("runner API against real HTTP routes", () => { }); afterEach(() => vi.unstubAllEnvs()); beforeAll(async () => { + // API authorization and replay use the real lifecycle; external model login is unavailable here. + const probe = vi.spyOn(nativeSetupReadiness, "testNativeRunnerAuthentication").mockImplementation(nativeAuthProbe); + restoreNativeAuthProbe = () => probe.mockRestore(); process.env.PAPERCLIP_AGENT_JWT_SECRET = randomUUID(); server = await startRunnerApiTestServer(); process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = join(server.root, "master.key"); }, 60_000); afterAll(async () => { - await server?.close(); - if (oldMasterKeyFile === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; - else process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = oldMasterKeyFile; - if (oldSecret === undefined) delete process.env.PAPERCLIP_AGENT_JWT_SECRET; - else process.env.PAPERCLIP_AGENT_JWT_SECRET = oldSecret; + try { await server?.close(); } + finally { + restoreNativeAuthProbe?.(); + if (oldMasterKeyFile === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; + else process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = oldMasterKeyFile; + if (oldSecret === undefined) delete process.env.PAPERCLIP_AGENT_JWT_SECRET; + else process.env.PAPERCLIP_AGENT_JWT_SECRET = oldSecret; + } }); it.each(["http_response", "runner_response"])("counts committed effects independently after losing the %s", async boundary => { @@ -408,6 +420,12 @@ else if(m.id!==undefined) send({id:m.id,result:{}}); const call = { tool, callId: "completed-bridge-write", arguments: args }; const result = await fixture.authority.execute(call); expect(result).toMatchObject({ ok: true, status: expect.any(Number), apiOperationId: expect.any(String) }); + if (tool === "hire_agent") { + await expect.poll(() => nativeAuthProbe.mock.calls + .filter(([context]) => context.companyId === fixture.companyId) + .map(([context, provider]) => ({ companyId: context.companyId, provider, configProvider: context.config.provider })), + { timeout: 5_000 }).toEqual([{ companyId: fixture.companyId, provider: "codex", configProvider: "codex" }]); + } const reopened = new PaperclipRunnerToolAuthority(server.db, fixture.authority.binding); const effects = async () => { const snapshot = await fixture.snapshot();