perf(server): retire aggregate maintenance polling (#18425)

Reviewed and verified at 901c4c4bdc.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-31 17:22:31 -05:00
1 parent 9ca9af544a
commit 681f7b7726
19 files changed
+2022 -93

No files matched your search

@@ -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: <directory>` 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.
@@ -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<string, unknown> = {},
actor: Record<string, unknown> = {
@@ -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", () => {
@@ -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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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;
@@ -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<EmbeddedPostgresTestSupport> {
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<void> | 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 });
})();
},
};
}
@@ -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<string, string>(), portals = new Map<string, string>();
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<Socket>(), observers = new Set<ReturnType<typeof createPostgresProtocolObserver>>();
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<void>((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<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
},
};
}
@@ -0,0 +1,21 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { fileURLToPath } from "node:url";
let ready: Promise<void> | 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;
}
@@ -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<typeof setInterval> | 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;
}
@@ -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<T>(promise: Promise<T>, ms: number, message: () => string): Promise<T> {
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<T>(provider: HttpServer, run: (origin: string) => Promise<T>): Promise<T> {
let failure: { error: unknown } | undefined;
try {
await new Promise<void>((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<void>((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<void> | undefined;
function stop() {
return stopping ??= (async () => {
const errors: unknown[] = [];
for (const cleanup of [stopChild, stopProxy,
async () => { if (reservation.listening) await new Promise<void>(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<void>((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<void>(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<void>(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<number, { resolve: (value: any) => void; reject: (error: Error) => void }>();
let readyResolve: (port: number) => void, readyReject: (error: Error) => void;
const ready = new Promise<number>((resolve, reject) => { readyResolve = resolve; readyReject = reject; });
const exited = new Promise<void>(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<T>(command: string, payload: { ms?: number; issueUpdate?: { id: string; status: "done" }; orphanLease?: { companyId: string }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } } = {}): Promise<T> {
const id = ++requestId;
const result = new Promise<T>((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<T>(method: string, route: string, body?: unknown, headers: Record<string, string> = {}): Promise<T> {
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>("snapshot"),
drain: () => request<Snapshot>("drain"), resume: () => request<Snapshot>("resume"),
realtime: () => request<Snapshot>("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<Snapshot>("advance", { ms });
},
freeze: () => request<Snapshot>("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>("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;
}
}
@@ -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"]
}
@@ -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<void>(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<void>(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<typeof request> | undefined, closed: Promise<void> | undefined;
try {
await expect(withMaintenanceProvider(provider, async origin => {
try {
closed = new Promise<void>(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<void>(resolve => provider.close(() => resolve())) : Promise.resolve();
provider.closeAllConnections(); await closing;
}
});
});
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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(); }
});
});
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<ReturnType<typeof startMaintenanceServer>>) {
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<string, { fingerprint: string; protocol: string; calls: number; tables: string[] }>();
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<string, unknown> | 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<ReturnType<typeof startMaintenanceServer>> | 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<string, unknown> | 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<ReturnType<typeof fixture.$client.reserve>> | undefined = await fixture.$client.reserve();
await reserved`select pg_advisory_lock(913703)`;
let server: Awaited<ReturnType<typeof startMaintenanceServer>> | 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<unknown> | undefined;
let advancing: ReturnType<typeof server.advance> | 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<void>(resolve => { entered = resolve; });
socket.write = ((...args: Parameters<typeof socket.write>) => {
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<void>(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<ReturnType<typeof startMaintenanceServer>> | 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<ReturnType<typeof fixture.$client.reserve>> | undefined;
let pendingIssue: Promise<{ id: string }> | undefined;
let runtime: Awaited<ReturnType<typeof startMaintenanceServer>> | 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<string, unknown> | 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<ReturnType<typeof startMaintenanceServer>> | 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<ReturnType<typeof startMaintenanceServer>> | 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<ReturnType<typeof startMaintenanceServer>> | 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<ReturnType<typeof startMaintenanceServer>> | 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);
});
@@ -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<void>(resolve => setImmediate(resolve));
tick!(); await new Promise<void>(resolve => setImmediate(resolve));
expect(tick).toBeUndefined();
await new Promise<void>(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<void>((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<void>((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();
}
-45
View File
@@ -1251,7 +1251,6 @@ async function startServerWithDatabaseTeardown(
drainRunIds?: string[];
}>) | null = null;
let heartbeatSchedulerStopped = false;
let heartbeatSchedulerInterval: ReturnType<typeof setInterval> | null = null;
const heartbeatSchedulerInFlight = new Set<Promise<void>>();
const trackHeartbeatSchedulerWork = (work: Promise<unknown>) => {
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<unknown>) =>
work().catch((err: unknown) => {
@@ -1847,31 +1834,6 @@ async function startServerWithDatabaseTeardown(
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
};
const createHeartbeatSchedulerTick = (scheduled: ReturnType<typeof registerScheduledWork>) => () => {
// 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,
+4 -2
View File
@@ -7,11 +7,13 @@ export function registerDecisionTargetWork(
service: Pick<ReturnType<typeof decisionService>, "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,
});
}
@@ -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",
@@ -20,6 +20,18 @@ export interface HeartbeatRunRuntimeStatus {
}
const runtimeStatusesByRunId = new Map<string, HeartbeatRunRuntimeStatus>();
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(
-2
View File
@@ -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,
@@ -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<ReturnType<typeof startRunnerApiTestServer>>;
const nativeAuthProbe = vi.fn<typeof nativeSetupReadiness.testNativeRunnerAuthentication>(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();