diff --git a/doc/plans/2026-10-30-native-controller-checkpoints.md b/doc/plans/2026-10-30-native-controller-checkpoints.md index 81223312d9d..049f83f1855 100644 --- a/doc/plans/2026-10-30-native-controller-checkpoints.md +++ b/doc/plans/2026-10-30-native-controller-checkpoints.md @@ -391,7 +391,7 @@ pilot is underway. Database cost acceptance, distinct-release application startu compatible rollback, and native Daytona qualification remain open. The published Daytona qualification image was independently resolved from successful workflow 40709784515 to `sha256:b38ed051d00fed409092d354eea9ae1b01fb232bcb7d5a187ed763382dda2d0b` (source `8fe5d5defcdfb55a1b9049a801cfa5498629cebc`). The exact Linux Runner was -extracted from that image. A single five-minute outage Product E2E attempt is running. +extracted from that image. The local attempt failed during the environment probe before agent creation. A read-only lookup independently returned `DaytonaAuthenticationError: Invalid credentials`. The private database and cleanup obligation remain preserved. The protected single-cell workflow [41093316372](https://github.com/paperclipai/paperclip/actions/runs/41093316372), targeting `512cc6d6dc`, was cancelled during image/artifact builds before provider allocation so acceptance can use the rebased candidate. Its protected secret is separate from the invalid local fixture. No production deployment or complete zero-downtime acceptance is claimed. Failures retained in the implementation record: sandbox listener denial and host @@ -428,3 +428,50 @@ columns it actually needs. The initial paired 300-controller ABBA comparison completed every offered operation with no ownership losses, but candidate latency and CPU increased above the investigation threshold. It is not performance acceptance. A new paired run will measure the narrower snapshot projection. + + +Qualification update at `512cc6d6dc`: the snapshot port now combines coordinator +and run locks into one statement, with a database-clock check after both waits. +All 59 real-Postgres port tests pass. Snapshot and PRP transaction counts remain +unchanged; the first completed 300-controller baseline/candidate pair each issued +74,790 total statements and 7,800 control transactions across all pools. The +candidate adds the metadata bytes but no SQL per streamed chunk. + +The dedicated Daytona suite now fixes the active outage at 300 seconds and adds +normal UI follow-up after a second replacement following completion. Its oracle +requires a new successful native run, retained native/provider session identity, +a completed tool receipt and correct saved-project output. Focused E2E support +checks passed (56 tests, one optional external-database case skipped), plus all +137 Node support tests and E2E typecheck. The explicit external-database fixture +was separately exercised against owned Postgres earlier. The broad E2E unit +attempt lacked local listener permissions and was stopped; that failure remains +in `/tmp/native-controller-e2e-unit.log`, and broad qualification is still open. + +Database load remains **unqualified**. The narrowed-read and combined-query ABBA +runs became heavily contended as other local checks ran, including our E2E support +suite during the final pair. Both baseline and candidate lost absolute gates; +dropped operations invalidate relative comparisons. The combined-query first +pair completed all work, but its candidate p99 was 303 ms versus 131 ms and PG CPU +35.69 s versus 25.89 s. Unchanged business queries also slowed. These observations +require a quiet-host paired rerun; they are neither a performance pass nor proof +that the implementation caused all observed regression. The connection limit +remains unchanged. Summarized counters are retained in the adjacent +`2026-10-30-native-controller-load-evidence.json`; raw measurements remain under +`/tmp/native-controller-load/`. The reproducible script records failed absolute +and relative gates and marks unequal-operation comparisons ineligible. + +Remaining critical path: finish the protected Daytona attempt and cleanup; +complete full repository checks on the final rebased revision; finish the quiet +steady/burst/replacement load gates and explicit lock-wait latency measurement; +qualify distinct installed A/B/A artifacts; then resolve review and prepare the +PR. Current server startup smoke and three-controller component tests are useful +layer-specific evidence, not substitutes for those remaining gates. + + +The full local `pnpm test:run` attempt was stopped after roughly 45 minutes while +still progressing through the general chat suite. Its output is retained at +`/tmp/native-controller-full-tests.log`; it is incomplete, not a pass. Broad +checks will run on hosted Linux for the rebased candidate. The broad E2E support +rerun with two workers was also stopped before rebasing; focused support checks +remain the completed evidence. A separate `server-config.test.ts` diagnostic +confirmed the original restricted run's `listen EPERM` cause. diff --git a/doc/plans/2026-10-30-native-controller-load-evidence.json b/doc/plans/2026-10-30-native-controller-load-evidence.json new file mode 100644 index 00000000000..1fadc46a050 --- /dev/null +++ b/doc/plans/2026-10-30-native-controller-load-evidence.json @@ -0,0 +1,390 @@ +{ + "schema": "paperclip.native-controller-load-summary/v1", + "baselineRevision": "59e191c368", + "candidateRevision": "4f710fd7e7", + "qualification": "failed; host contention and unequal completed work invalidate the ABBA comparison", + "sourceReport": "/tmp/native-controller-load/combined-steady/report.json", + "fixture": { + "mode": "steady", + "normalReadsPerSecond": 500, + "normalWritesPerSecond": 100, + "databaseAdmissionEverySeconds": 5, + "checkpointControlEverySeconds": 5, + "controlPayloadBytes": 4096, + "phaseSeconds": 65, + "snapshotTransactionsPerControl": 1, + "baselineCheckpointPool": 2, + "baselineRenewalIntervalSeconds": 20, + "candidateRenewalIntervalSeconds": 20, + "historicalIssues": 10000 + }, + "limits": [ + "Database request latency is not HTTP latency", + "Pool queue includes client encoding; BEGIN response includes network time", + "pg_stat_statements execution includes lock waits; samples report concurrent lock waiters, not a lock latency distribution", + "Peak memory is sampled once a second", + "Use an otherwise idle cluster and host; WAL and cgroup CPU are cluster-wide" + ], + "results": [ + { + "label": "01_baseline_300", + "controls": 3900, + "normalOffered": 32500, + "normalCompleted": 32500, + "normalWritesCompleted": 6500, + "dropped": 0, + "renewalLosses": 0, + "checkpointTransactions": 3900, + "snapshotTransactions": 3900, + "sqlCount": 54687, + "livenessSqlCount": 4502, + "checkpointSqlCount": 15601, + "changedRows": 7800, + "changedBytes": 17371500, + "walBytes": 63819704, + "postgresCpuUsec": 25887716, + "postgresPeakSampledMemoryBytes": 697917440, + "normalReadSummary": { + "n": 32500, + "p50": 53.82264099999884, + "p95": 83.97040299999935, + "p99": 131.4543599999888, + "max": 249.75419500000135 + }, + "checkpointLatencySummary": { + "n": 3900, + "p50": 3.5057919999962905, + "p95": 14.36212499999965, + "p99": 79.32949999999255, + "max": 220.70087599998806 + }, + "poolQueueByPool": { + "request": { + "n": 3913, + "p50": 0.04241599999659229, + "p95": 4.89183399999456, + "p99": 25.90625, + "max": 210.7756250000093 + }, + "liveness": { + "n": 900, + "p50": 401.07175099999586, + "p95": 1987.666584000006, + "p99": 2269.4225429999933, + "max": 2296.889626000004 + }, + "checkpoint": { + "n": 3900, + "p50": 0.05258299999695737, + "p95": 0.14449999999487773, + "p99": 41.99183400000038, + "max": 188.36633400000574 + } + }, + "transactionLatencyByPool": { + "request": { + "n": 3913, + "p50": 3.6501250000001164, + "p95": 14.756584000002476, + "p99": 62.41062499998952, + "max": 240.88450000000012 + }, + "liveness": { + "n": 900, + "p50": 406.2205419999955, + "p95": 2044.2198760000028, + "p99": 2277.324250999998, + "max": 2308.3716669999994 + }, + "checkpoint": { + "n": 3900, + "p50": 3.297791999997571, + "p95": 14.160208000001148, + "p99": 79.12295900000026, + "max": 220.50066700000025 + } + }, + "drainMs": 26.716459000002942, + "absoluteGates": { + "requestP99": true, + "drain": true, + "throughput": true, + "postgresCpu": true, + "postgresMemory": true + }, + "passed": true, + "startedAt": "2026-10-30T16:40:56.503Z", + "endedAt": "2026-10-30T16:42:01.534Z" + }, + { + "label": "02_candidate_300", + "controls": 3900, + "normalOffered": 32500, + "normalCompleted": 32500, + "normalWritesCompleted": 6500, + "dropped": 0, + "renewalLosses": 0, + "checkpointTransactions": 3900, + "snapshotTransactions": 3900, + "sqlCount": 54687, + "livenessSqlCount": 4502, + "checkpointSqlCount": 15601, + "changedRows": 7800, + "changedBytes": 17371500, + "walBytes": 66746424, + "postgresCpuUsec": 35685605, + "postgresPeakSampledMemoryBytes": 666578944, + "normalReadSummary": { + "n": 32500, + "p50": 68.02364699999453, + "p95": 127.37661100001424, + "p99": 303.326441000012, + "max": 474.2829830000119 + }, + "checkpointLatencySummary": { + "n": 3900, + "p50": 5.670542000007117, + "p95": 241.97500100001344, + "p99": 479.3440830000036, + "max": 553.2187080000003 + }, + "poolQueueByPool": { + "request": { + "n": 3913, + "p50": 0.05162500002188608, + "p95": 37.2935839999991, + "p99": 136.02920900000026, + "max": 420.89487499999814 + }, + "liveness": { + "n": 900, + "p50": 536.7470420000027, + "p95": 3620.454543, + "p99": 4146.325211000018, + "max": 4238.34758500001 + }, + "checkpoint": { + "n": 3900, + "p50": 0.057792000006884336, + "p95": 209.87512499999139, + "p99": 443.83324999999604, + "max": 528.2805419999931 + } + }, + "transactionLatencyByPool": { + "request": { + "n": 3913, + "p50": 6.640833000012208, + "p95": 66.06445900001563, + "p99": 184.52537499999744, + "max": 449.3999579999945 + }, + "liveness": { + "n": 900, + "p50": 545.1022919999959, + "p95": 3665.246376999974, + "p99": 4172.158336000022, + "max": 4248.078752000001 + }, + "checkpoint": { + "n": 3900, + "p50": 5.484875000009197, + "p95": 241.83212600002298, + "p99": 478.8787080000038, + "max": 553.1449999999895 + } + }, + "drainMs": 76.8760000000184, + "absoluteGates": { + "requestP99": false, + "drain": true, + "throughput": true, + "postgresCpu": true, + "postgresMemory": true + }, + "passed": false, + "startedAt": "2026-10-30T16:42:16.915Z", + "endedAt": "2026-10-30T16:43:21.995Z" + }, + { + "label": "03_candidate_300", + "controls": 3887, + "normalOffered": 32500, + "normalCompleted": 32379, + "normalWritesCompleted": 6470, + "dropped": 164, + "renewalLosses": 0, + "checkpointTransactions": 3887, + "snapshotTransactions": 3887, + "sqlCount": 54484, + "livenessSqlCount": 4502, + "checkpointSqlCount": 15549, + "changedRows": 7774, + "changedBytes": 17313559, + "walBytes": 66470080, + "postgresCpuUsec": 53715752, + "postgresPeakSampledMemoryBytes": 670437376, + "normalReadSummary": { + "n": 32379, + "p50": 87.29533699998865, + "p95": 407.7868620000081, + "p99": 715.8863589999964, + "max": 1413.3376139999891 + }, + "checkpointLatencySummary": { + "n": 3887, + "p50": 35.722666999994544, + "p95": 2254.7568759999995, + "p99": 4195.890377000003, + "max": 4784.590544000006 + }, + "poolQueueByPool": { + "request": { + "n": 3900, + "p50": 0.12583400000585243, + "p95": 218.51912499999162, + "p99": 498.72595799999544, + "max": 1368.8183760000102 + }, + "liveness": { + "n": 900, + "p50": 2733.2611259999976, + "p95": 7990.602337999997, + "p99": 8707.207421, + "max": 8915.786796 + }, + "checkpoint": { + "n": 3887, + "p50": 3.155832999997074, + "p95": 2207.667333999998, + "p99": 4154.626294000016, + "max": 4726.676418999996 + } + }, + "transactionLatencyByPool": { + "request": { + "n": 3900, + "p50": 27.709541999996873, + "p95": 276.61000000001513, + "p99": 625.9165419999918, + "max": 1627.7409179999959 + }, + "liveness": { + "n": 900, + "p50": 2782.7440009999846, + "p95": 8019.347628999996, + "p99": 8734.34838000001, + "max": 8964.291171000019 + }, + "checkpoint": { + "n": 3887, + "p50": 35.49954200000502, + "p95": 2254.554459999985, + "p99": 4195.5352520000015, + "max": 4784.371294000011 + } + }, + "drainMs": 4928.940960000007, + "absoluteGates": { + "requestP99": false, + "drain": false, + "throughput": true, + "postgresCpu": true, + "postgresMemory": true + }, + "passed": false, + "startedAt": "2026-10-30T16:43:41.479Z", + "endedAt": "2026-10-30T16:44:51.414Z" + }, + { + "label": "04_baseline_300", + "controls": 2390, + "normalOffered": 32500, + "normalCompleted": 10713, + "normalWritesCompleted": 1503, + "dropped": 28305, + "renewalLosses": 0, + "checkpointTransactions": 2390, + "snapshotTransactions": 2390, + "sqlCount": 21797, + "livenessSqlCount": 4502, + "checkpointSqlCount": 9561, + "changedRows": 4780, + "changedBytes": 10643952, + "walBytes": 31871472, + "postgresCpuUsec": 55047197, + "postgresPeakSampledMemoryBytes": 643657728, + "normalReadSummary": { + "n": 10713, + "p50": 291.001007999992, + "p95": 841.957837999973, + "p99": 2601.1680469999555, + "max": 3106.3534709999803 + }, + "checkpointLatencySummary": { + "n": 2390, + "p50": 6933.93029400002, + "p95": 16553.939049999986, + "p99": 17176.082591000013, + "max": 17434.499467000016 + }, + "poolQueueByPool": { + "request": { + "n": 2392, + "p50": 0.10450000001583248, + "p95": 520.4013340000529, + "p99": 798.97454200004, + "max": 2614.81958499999 + }, + "liveness": { + "n": 900, + "p50": 5497.6995860000025, + "p95": 16836.80413299997, + "p99": 17748.473174999992, + "max": 18895.275049999997 + }, + "checkpoint": { + "n": 2390, + "p50": 6882.121711000043, + "p95": 16492.48646600003, + "p99": 17097.382091999985, + "max": 17352.715134 + } + }, + "transactionLatencyByPool": { + "request": { + "n": 2392, + "p50": 55.64137500000652, + "p95": 638.8517920000013, + "p99": 2574.2484169999952, + "max": 2879.7436260000104 + }, + "liveness": { + "n": 900, + "p50": 5623.322503000032, + "p95": 16872.938590999984, + "p99": 18552.84471799998, + "max": 18921.25067499996 + }, + "checkpoint": { + "n": 2390, + "p50": 6933.636502000038, + "p95": 16553.763133, + "p99": 17175.909424000012, + "max": 17433.771509000042 + } + }, + "drainMs": 12508.142965000006, + "absoluteGates": { + "requestP99": false, + "drain": false, + "throughput": false, + "postgresCpu": true, + "postgresMemory": true + }, + "passed": false, + "startedAt": "2026-10-30T16:45:39.710Z", + "endedAt": "2026-10-30T16:46:57.219Z" + } + ] +} diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 056421adcd3..cf8600ca826 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -349,7 +349,7 @@ separate. Postgres execution time includes lock waits. This does **not** yet provide a lock-wait latency distribution or HTTP latency. Report those limits; do not call this whole-application load qualification. -Correctness and absolute gates must pass before interpreting relative costs. +Correctness and absolute gates must pass for acceptance. Dropped operations invalidate paired comparisons. The report retains failures and refuses an overall pass if p99, drain, throughput, memory, CPU or the 10% relative regression gate fails. Do not change pool limits to conceal a failure. diff --git a/scripts/benchmark-native-controller-recovery.ts b/scripts/benchmark-native-controller-recovery.ts new file mode 100644 index 00000000000..8f8aad4dbca --- /dev/null +++ b/scripts/benchmark-native-controller-recovery.ts @@ -0,0 +1,899 @@ +import fs from "node:fs"; +import { syncBuiltinESMExports } from "node:module"; +import assert from "node:assert/strict"; +import { randomUUID, randomBytes } from "node:crypto"; +import { createRequire } from "node:module"; +import { mkdir, writeFile, readFile } from "node:fs/promises"; +import { monitorEventLoopDelay } from "node:perf_hooks"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { execFileSync } from "node:child_process"; + +// Run only against an explicitly disposable, otherwise idle Postgres cluster. +// pg_stat_statements must be preloaded; CPU/memory measurements use its cgroup. +const repoRoot = fileURLToPath(new URL("../", import.meta.url)); +function required(name: string) { + const value = process.env[name]; + assert(value, `Set ${name}`); + return value; +} +const baselineRoot = path.resolve(required("BENCH_BASELINE_ROOT")); +const root = path.resolve(required("BENCH_REPORT_ROOT")); +const url = required("BENCH_DATABASE_URL"); +const parsedUrl = new URL(url); +assert( + ["postgres:", "postgresql:"].includes(parsedUrl.protocol) && + ["127.0.0.1", "localhost", "[::1]"].includes(parsedUrl.hostname), + "Use an explicit loopback Postgres URL", +); +assert.equal(process.env.BENCH_DISPOSABLE_POSTGRES, "1", "The whole cluster must be owned by this benchmark"); +const mode = process.argv[2] ?? "steady"; +assert(["pilot", "steady", "burst"].includes(mode)); +const pgCgroup = required("BENCH_POSTGRES_CGROUP"); +const [cpuQuota, cpuPeriod] = fs.readFileSync(path.join(pgCgroup, "cpu.max"), "utf8").trim().split(/\s+/).map(Number); +assert.equal(cpuQuota / cpuPeriod, 2, "The reference workload requires two Postgres CPUs"); +assert.equal( + Number(fs.readFileSync(path.join(pgCgroup, "memory.max"), "utf8")), + 1536 * 1024 * 1024, + "The reference workload requires a 1536 MiB Postgres memory limit", +); +const load = (base: string, file: string) => import(pathToFileURL(path.join(base, file)).href); +const { + completionContracts, + createDb, + livenessDb, + checkpointDb, + closeRegisteredClients, + migratePostgresIfEmpty, + companies, + agents, + issues, + heartbeatRuns, + nativeRunFinalizations, +} = await load(repoRoot, "packages/db/src/index.ts"); +const { nativeRunnerCheckpointWriter: CandidateWriter } = await load( + repoRoot, + "server/src/services/native-runtime/native-runner-checkpoint.ts", +); +const { nativeRunnerCheckpointWriter: BaselineWriter } = await load( + baselineRoot, + "server/src/services/native-runtime/native-runner-checkpoint.ts", +); +const { startNativeSessionExecutionLeaseRenewal } = await load( + repoRoot, + "server/src/services/native-runtime/native-execution-lease.ts", +); +const { loadAgentRunAuthentication } = await load(repoRoot, "server/src/middleware/agent-run-authentication.ts"); +const { DurablePrpControlPlane: CandidateCore } = await load( + repoRoot, + "packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts", +); +const { DurablePrpControlPlane: BaselineCore } = await load( + baselineRoot, + "packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts", +); +const { PaperclipControlPlanePort: CandidatePort } = await load( + repoRoot, + "server/src/services/native-runtime/paperclip-control-plane-port.ts", +); +const { PaperclipControlPlanePort: BaselinePort } = await load( + baselineRoot, + "server/src/services/native-runtime/paperclip-control-plane-port.ts", +); +const require = createRequire(path.join(repoRoot, "server/package.json")); +const postgres = createRequire(path.join(repoRoot, "packages/db/package.json"))("postgres"); +const { and, eq, desc } = require("drizzle-orm"); +const pgResources = () => { + const raw = fs.readFileSync(path.join(pgCgroup, "cpu.stat"), "utf8"); + return { + cpuUsec: Number(raw.match(/usage_usec (\d+)/)![1]), + memoryBytes: Number(fs.readFileSync(path.join(pgCgroup, "memory.current"), "utf8")), + }; +}; +const revision = (directory: string) => + execFileSync("git", ["rev-parse", "HEAD"], { + cwd: directory, + encoding: "utf8", + }).trim(); +const provenance = { + baselineRevision: process.env.BENCH_BASELINE_REVISION ?? revision(baselineRoot), + candidateRevision: process.env.BENCH_CANDIDATE_REVISION ?? revision(repoRoot), + candidateDiff: + process.env.BENCH_CANDIDATE_DIFF ?? + execFileSync("git", ["diff", "HEAD", "--", "server/src/services/native-runtime", "packages/db/src"], { + cwd: repoRoot, + encoding: "utf8", + }), +}; +await mkdir(root); // Refuse to overwrite earlier measurements. +await writeFile(path.join(root, "provenance.json"), JSON.stringify(provenance, null, 2)); +const admin = postgres(url, { + max: 1, + onnotice: () => {}, + connection: { application_name: "benchmark-observer" }, +}); +const ownedPrefix = "native_recovery_bench_" + randomUUID().replaceAll("-", ""); +const templateName = ownedPrefix + "_template"; +const databaseUrl = (name: string) => { + const value = new URL(url); + value.pathname = "/" + name; + return value.toString(); +}; +process.env.PAPERCLIP_HOME = root + "/home"; +process.env.PAPERCLIP_SECRETS_MASTER_KEY = randomBytes(32).toString("hex"); +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const summary = (a: number[]) => { + const b = [...a].sort((x, y) => x - y); + return { + n: b.length, + p50: b[Math.floor(b.length * 0.5)] ?? 0, + p95: b[Math.floor(b.length * 0.95)] ?? 0, + p99: b[Math.floor(b.length * 0.99)] ?? 0, + max: b.at(-1) ?? 0, + }; +}; +async function setup() { + await admin.unsafe(`create database ${templateName}`); + const u = databaseUrl(templateName); + await migratePostgresIfEmpty(u); + const db = createDb(u); + await db.$client.unsafe("create extension if not exists pg_stat_statements"); + const companyId = randomUUID(), + agentId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Synthetic normal-operation benchmark", + issuePrefix: "BENCH", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Benchmark fixture", + adapterType: "paperclip_runner", + adapterConfig: { provider: "codex" }, + status: "active", + }); + for (let batch = 0; batch < 20; batch++) + await db.insert(issues).values( + Array.from({ length: 500 }, (_, j) => ({ + companyId, + title: `Historical task ${batch * 500 + j}`, + status: j % 3 ? "done" : "backlog", + identifier: `BENCH-${batch * 500 + j + 1}`, + issueNumber: batch * 500 + j + 1, + description: "Synthetic context ".repeat(80), + })), + ); + await db.$client.unsafe("analyze"); + await writeFile(root + "/seed.json", JSON.stringify({ companyId, agentId })); + await closeRegisteredClients(u); +} +async function sampleStats(dbName: string) { + const [stat] = + await admin`select xact_commit,xact_rollback,deadlocks,blks_read,blks_hit,tup_inserted,tup_updated,tup_deleted,temp_bytes from pg_stat_database where datname=${dbName}`; + const [wal] = await admin`select pg_current_wal_insert_lsn()::text as lsn`; + const statements = + await admin`select query,calls,total_exec_time,rows,shared_blks_hit,shared_blks_read,wal_bytes from pg_stat_statements where dbid=(select oid from pg_database where datname=${dbName}) order by calls desc`; + return { stat, wal: wal.lsn, statements }; +} +async function run(variant: string, n: number, trial: number, durationMs: number) { + const label = `${String(trial).padStart(2, "0")}_${variant}_${n}`, + dbName = ownedPrefix + "_" + trial; + await admin.unsafe(`create database ${dbName} template ${templateName}`); + const u = databaseUrl(dbName), + db = createDb(u, { + maxConnections: 10, + applicationName: "benchmark-application", + }); + const seed = JSON.parse(await readFile(root + "/seed.json", "utf8")); + const owner = { + bootId: randomUUID(), + pid: process.pid, + processStartedAt: new Date(), + }; + const items: any[] = [], + metrics: any = { + label, + variant, + n, + durationMs, + setupMs: 0, + errors: [], + dropped: 0, + renewalLosses: 0, + controls: 0, + changedRows: 0, + changedBytes: 0, + checkpointTransactions: 0, + sqlCount: 0, + sqlCategories: {}, + normalRead: [], + normalWrite: [], + databaseAdmission: [], + controlLatency: [], + checkpointLatency: [], + normalSchedulingLag: [], + controlSchedulingLag: [], + normalOffered: 0, + normalCompleted: 0, + normalWritesOffered: 0, + normalWritesCompleted: 0, + admissionsOffered: 0, + admissionsCompleted: 0, + peakPending: 0, + samples: [], + }; + // postgres.js calls onexecute after selecting a connection and encoding BEGIN. + // Preserve that callback. Queue/admission and BEGIN response time are separate; + // SQL execution/lock waits are reported independently by Postgres statistics. + metrics.poolAdmission = []; + metrics.beginResponse = []; + metrics.poolQueueByPool = {}; + metrics.transactionLatencyByPool = {}; + for (const [pool, client] of [ + ["request", db.$client], + ["liveness", livenessDb(db).$client], + ["checkpoint", checkpointDb(db).$client], + ] as const) { + metrics.poolQueueByPool[pool] = []; + metrics.transactionLatencyByPool[pool] = []; + const originalBegin = client.begin, + originalUnsafe = client.unsafe; + let entering: { start: number; sent?: number } | null = null; + client.unsafe = (query, parameters, options) => { + const admission = entering; + if (!admission || !/^begin\b/i.test(query)) return originalUnsafe(query, parameters, options); + return originalUnsafe(query, parameters, { + ...options, + onexecute: (connection) => { + admission.sent = performance.now(); + if (measuring) metrics.poolQueueByPool[pool].push(admission.sent - admission.start); + return options.onexecute(connection); + }, + }); + }; + client.begin = (...args) => { + const admission = { + start: performance.now(), + sent: undefined as number | undefined, + }; + const index = args.length - 1, + work = args[index]; + args[index] = async (connection) => { + if (measuring) { + metrics.poolAdmission.push(performance.now() - admission.start); + if (admission.sent !== undefined) metrics.beginResponse.push(performance.now() - admission.sent); + } + return work(connection); + }; + entering = admission; + try { + return originalBegin(...args).finally(() => { + if (measuring) metrics.transactionLatencyByPool[pool].push(performance.now() - admission.start); + }); + } finally { + entering = null; + } + }; + } + metrics.snapshotLatency = []; + metrics.snapshotTransactions = 0; + let measuring = false; + const originalFsync = fs.fsyncSync; + metrics.fsyncDurations = []; + fs.fsyncSync = function (fd) { + const t = performance.now(); + try { + return originalFsync(fd); + } finally { + if (measuring) metrics.fsyncDurations.push(performance.now() - t); + } + }; + syncBuiltinESMExports(); + db.$client.options.debug = (_conn: any, query: string) => { + if (!measuring) return; + metrics.sqlCount++; + const c = /native_runner_checkpoints/.test(query) + ? "checkpoint" + : /native_run_finalizations/.test(query) + ? "ownership" + : /heartbeat_runs/.test(query) + ? "run" + : /issues/.test(query) + ? "task" + : "other"; + metrics.sqlCategories[c] = (metrics.sqlCategories[c] ?? 0) + 1; + }; + metrics.livenessSqlCount = 0; + metrics.livenessSqlKinds = {}; + { + const live = livenessDb(db); + live.$client.options.debug = (_connection, query) => { + if (!measuring) return; + metrics.livenessSqlCount++; + const kind = query.trim().split(/\s+/)[0].toLowerCase(); + metrics.livenessSqlKinds[kind] = (metrics.livenessSqlKinds[kind] ?? 0) + 1; + }; + } + metrics.checkpointSqlCount = 0; + { + checkpointDb(db).$client.options.debug = (_conn, query) => { + if (measuring) metrics.checkpointSqlCount++; + }; + } + const tSetup = performance.now(), + runNonce = randomUUID(); + for (let i = 0; i < n; i++) { + const runId = randomUUID(), + issueId = randomUUID(), + leaseOwner = randomUUID(), + runnerId = randomUUID(), + sessionId = randomUUID(), + contractId = randomUUID(); + await db.transaction(async (tx: any) => { + await tx.insert(issues).values({ + id: issueId, + companyId: seed.companyId, + title: "Active fixture " + i, + status: "in_progress", + assigneeAgentId: seed.agentId, + }); + await tx.insert(completionContracts).values({ + id: contractId, + companyId: seed.companyId, + issueId, + revision: 1, + schemaVersion: "fixture", + policyVersion: "fixture", + risk: "low", + completionAuthority: "agent", + incompleteCriteriaPolicy: "allow", + contractJson: {}, + canonicalSha256: "fixture", + createdByActorType: "system", + createdByActorId: "benchmark", + }); + await tx.insert(heartbeatRuns).values({ + id: runId, + companyId: seed.companyId, + agentId: seed.agentId, + status: "running", + runtimeMode: "native", + nativeIssueId: issueId, + nativeSessionId: sessionId, + runnerInstanceId: runnerId, + completionContractId: contractId, + completionContractSha256: "fixture", + contextSnapshot: { + issueId, + fixture: "normal-load", + prompt: "x".repeat(4096), + }, + }); + await tx.insert(nativeRunFinalizations).values({ + runId, + companyId: seed.companyId, + issueId, + phase: "observed", + attempt: 1, + leaseOwner, + controllerBootId: owner.bootId, + controllerPid: owner.pid, + controllerProcessStartedAt: owner.processStartedAt, + controllerGeneration: 1, + leaseExpiresAt: new Date(Date.now() + 1200000), + }); + await tx.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + }); + const stateDirectory = (process.env.BENCH_STATE_ROOT ?? root + "/states") + "/" + label + "/" + runNonce + "/" + i; + const Core = variant === "candidate" ? CandidateCore : BaselineCore; + const core = new Core({ + stateDirectory, + identity: { + runnerInstanceId: runnerId, + environmentLeaseId: randomUUID(), + runId, + normalizedSessionId: sessionId, + turnId: randomUUID(), + itemId: randomUUID(), + }, + expectedRunnerVersion: "fixture-v1", + expectedRunnerDigest: "sha256:" + "a".repeat(64), + }); + { + const writer = (variant === "candidate" ? CandidateWriter : BaselineWriter)(db, { + runId, + issueId, + companyId: seed.companyId, + authorityRunId: runId, + bootId: owner.bootId, + generation: 1, + leaseOwner, + }); + core.setCheckpointPersistence(async (cp: any) => { + const start = performance.now(); + await writer(cp); + if (measuring) { + metrics.checkpointTransactions++; + metrics.changedRows += cp.changedRecords.size; + metrics.changedBytes += [...cp.changedRecords.values()].reduce( + (s: number, v: any) => s + Buffer.byteLength(v), + 0, + ); + metrics.checkpointLatency.push(performance.now() - start); + } + }); + await core.flushCheckpoint(); + } + const identity = { + companyId: seed.companyId, + agentId: seed.agentId, + issueId, + runId, + sessionId, + }; + const snapshot = { + backendKind: "codex", + sessionId, + identity, + driverKind: "codex_app_server", + driverSessionId: "thread_" + runId, + providerSessionId: "thread_" + runId, + activeTurnId: null, + terminalTurns: [], + pendingRuntimeRequests: [], + }; + const Port = variant === "candidate" ? CandidatePort : BaselinePort; + const port = new Port( + db, + { + ...identity, + sourceInstanceId: runnerId, + controlPlaneSourceInstanceId: "controller", + completionContractId: contractId, + completionContractSha256: "fixture", + }, + variant === "candidate" + ? { + nativeRecovery: { + compatibility: { + configDigest: "sha256:" + "a".repeat(64), + processConfigDigest: null, + }, + owner: { + companyId: seed.companyId, + issueId, + runId, + authorityRunId: runId, + bootId: owner.bootId, + generation: 1, + leaseOwner, + }, + }, + } + : {}, + ); + await port.openRun({ + identity, + backendKind: "codex", + sourceInstanceId: runnerId, + }); + items.push({ + core, + port, + snapshot, + runId, + issueId, + leaseOwner, + companyId: seed.companyId, + agentId: seed.agentId, + }); + } + metrics.setupMs = performance.now() - tSetup; + const monitor = postgres(u, { + max: 1, + onnotice: () => {}, + connection: { application_name: "benchmark-observer" }, + }); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60000) }); + await db.$client.unsafe("analyze"); + // Warm normal queries before measuring, identical in each arm. + for (let i = 0; i < 100; i++) await loadAgentRunAuthentication(db, items[i % n]); + await admin.unsafe("select pg_stat_statements_reset()"); + const pgBefore = pgResources(); + const before = await sampleStats(dbName), + cpuBefore = process.cpuUsage(), + el = monitorEventLoopDelay({ resolution: 10 }); + el.enable(); + const pending = new Set>(), + renewals: any[] = [], + timers: any[] = []; + function offer(action: () => Promise, out: number[], scheduled = performance.now()) { + if (pending.size >= 500) { + metrics.dropped++; + return; + } + const task = Promise.resolve() + .then(action) + .then(() => out.push(performance.now() - scheduled)) + .catch((e) => metrics.errors.push({ message: e.message, code: e.code })) + .finally(() => pending.delete(task)); + pending.add(task); + metrics.peakPending = Math.max(metrics.peakPending, pending.size); + } + measuring = true; + const started = performance.now(), + startedAt = new Date().toISOString(); + await writeFile( + root + "/phase.json", + JSON.stringify({ + label, + phase: "running", + startedAt, + n, + variant, + durationMs, + }), + ); + console.log( + JSON.stringify({ + phase: "started", + label, + setupMs: metrics.setupMs, + startedAt, + }), + ); + for (const item of items) { + const input = { db, ...item, attempt: 1, controller: owner }; + renewals.push( + startNativeSessionExecutionLeaseRenewal({ + ...input, + remote: true, + onLost: async () => { + metrics.renewalLosses++; + }, + }), + ); + } + // Both arms receive the same offered workload. No await-before-next-tick pacing. + let tick = 0; + const offerNormal = () => { + const due = Math.min(Math.floor(durationMs / 100), Math.floor((performance.now() - started) / 100)); + while (tick < due) { + tick++; + metrics.normalSchedulingLag.push(Math.max(0, performance.now() - started - tick * 100)); + for (let k = 0; k < (mode === "burst" ? 10 : 50); k++) { + const idx = (tick * 50 + k) % n, + item = items[idx]; + metrics.normalOffered++; + offer( + async () => { + if (k % 10 < 7) { + const rows = await loadAgentRunAuthentication(db, item); + assert.equal(rows.length, 1); + assert.equal(rows[0].status, "running"); + } else if (k % 10 < 9) { + const rows = await db + .select({ + id: issues.id, + title: issues.title, + status: issues.status, + }) + .from(issues) + .where(and(eq(issues.companyId, seed.companyId), eq(issues.status, "backlog"))) + .orderBy(desc(issues.createdAt)) + .limit(50); + assert.equal(rows.length, 50); + } else { + const rows = await db + .select({ + runId: nativeRunFinalizations.runId, + phase: nativeRunFinalizations.phase, + }) + .from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, item.runId)); + assert.equal(rows[0]?.phase, "observed"); + } + metrics.normalCompleted++; + }, + metrics.normalRead, + started + tick * 100, + ); + } + for (let k = 0; k < (mode === "burst" ? 2 : 10); k++) { + const item = items[(tick * 10 + k) % n]; + metrics.normalWritesOffered++; + offer( + async () => { + await db.update(issues).set({ updatedAt: new Date() }).where(eq(issues.id, item.issueId)); + metrics.normalWritesCompleted++; + }, + metrics.normalWrite, + started + tick * 100, + ); + } + if (tick % 50 === 0) { + metrics.admissionsOffered++; + offer( + async () => { + const runId = randomUUID(), + issueId = randomUUID(); + await db.transaction(async (tx: any) => { + await tx.insert(issues).values({ + id: issueId, + companyId: seed.companyId, + title: "Fresh admission database probe", + status: "in_progress", + }); + await tx.insert(heartbeatRuns).values({ + id: runId, + companyId: seed.companyId, + agentId: seed.agentId, + status: "queued", + runtimeMode: "native", + nativeIssueId: issueId, + }); + await tx.insert(nativeRunFinalizations).values({ + runId, + companyId: seed.companyId, + issueId, + phase: "observed", + }); + await tx.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + }); + metrics.admissionsCompleted++; + }, + metrics.databaseAdmission, + started + tick * 100, + ); + } + } + }; + const normal = setInterval(offerNormal, 100); + timers.push(normal); + // Same aggregate control rate, spread evenly across sessions for normal operation. + let offeredControls = 0; + const offerControl = () => { + const elapsed = performance.now() - started; + const due = Math.min( + Math.floor((durationMs * n) / 5000), + mode === "burst" ? Math.floor(elapsed / 5000) * n : Math.floor((elapsed * n) / 5000), + ); + while (offeredControls < due) { + const index = offeredControls++, + item = items[index % n], + seq = Math.floor(index / n) + 1; + const scheduled = started + (mode === "burst" ? seq * 5000 : ((index + 1) * 5000) / n); + metrics.controlSchedulingLag.push(Math.max(0, performance.now() - scheduled)); + offer( + async () => { + item.core.queueCommand( + "semantic_tool.result", + { + callId: `fixture_call_${seq}`, + operationId: "fixture_read", + result: { body: "x".repeat(4096) }, + isError: false, + }, + `control_${seq}`, + ); + await item.core.flushCheckpoint(); + const snapshotStarted = performance.now(); + await item.port.checkpointSession(item.snapshot); + metrics.snapshotLatency.push(performance.now() - snapshotStarted); + metrics.snapshotTransactions++; + metrics.controls++; + }, + metrics.controlLatency, + scheduled, + ); + } + }; + const control = setInterval(offerControl, 10); + timers.push(control); + const sampler = setInterval(() => { + void (async () => { + const m = process.memoryUsage(); + const activity = + await monitor`select count(*)::int as connections,count(*) filter(where state='active')::int as active,count(*) filter(where wait_event_type='Lock')::int as locks from pg_stat_activity where datname=current_database() and application_name!='benchmark-observer'`; + metrics.samples.push({ + elapsedMs: performance.now() - started, + rss: m.rss, + heapUsed: m.heapUsed, + postgresMemoryBytes: pgResources().memoryBytes, + ...activity[0], + }); + })().catch((e) => metrics.errors.push({ observer: e.message })); + }, 1000); + timers.push(sampler); + await sleep(durationMs); + offerNormal(); + offerControl(); + for (const t of timers) clearInterval(t); + const drainStart = performance.now(); + await Promise.allSettled([...pending]); + await Promise.allSettled(renewals.map((r) => r.stop())); + metrics.drainMs = performance.now() - drainStart; + metrics.elapsedMs = performance.now() - started; + metrics.startedAt = startedAt; + metrics.endedAt = new Date().toISOString(); + measuring = false; + fs.fsyncSync = originalFsync; + syncBuiltinESMExports(); + metrics.fsyncTotalMs = metrics.fsyncDurations.reduce((a, b) => a + b, 0); + el.disable(); + metrics.nodeCpu = process.cpuUsage(cpuBefore); + metrics.eventLoopDelay = { + p95Ms: el.percentile(95) / 1e6, + p99Ms: el.percentile(99) / 1e6, + maxMs: el.max / 1e6, + }; + // Flush PG statistics after all application activity; observer SQL is tagged/excluded below. + await sleep(1100); + const after = await sampleStats(dbName); + const [wal] = await admin`select pg_wal_lsn_diff(${after.wal}::pg_lsn,${before.wal}::pg_lsn)::text as bytes`; + metrics.walBytes = Number(wal.bytes); + const pgAfter = pgResources(); + metrics.postgresCpuUsec = pgAfter.cpuUsec - pgBefore.cpuUsec; + metrics.postgresMemoryBytes = pgAfter.memoryBytes; + metrics.postgresPeakSampledMemoryBytes = Math.max( + pgBefore.memoryBytes, + pgAfter.memoryBytes, + ...metrics.samples.map((s) => s.postgresMemoryBytes), + ); + metrics.platform = { + node: process.version, + os: process.platform, + arch: process.arch, + }; + metrics.databaseStats = Object.fromEntries( + Object.keys(after.stat).map((k) => [k, Number(after.stat[k]) - Number(before.stat[k])]), + ); + metrics.statements = after.statements; + metrics.tableStats = + await monitor`select relname,n_tup_ins,n_tup_upd,n_tup_del,n_dead_tup from pg_stat_user_tables where relname in ('native_runner_checkpoints','native_run_finalizations','heartbeat_runs','issues') order by relname`; + const [storage] = + await monitor`select count(*)::int as rows,coalesce(sum(pg_column_size(material)),0)::bigint::text as encrypted_bytes,pg_total_relation_size('native_runner_checkpoints')::text as relation_bytes from native_runner_checkpoints`; + metrics.checkpointStorage = storage; + for (const name of [ + "poolAdmission", + "beginResponse", + "snapshotLatency", + "normalRead", + "normalWrite", + "databaseAdmission", + "controlLatency", + "checkpointLatency", + "normalSchedulingLag", + "controlSchedulingLag", + "fsyncDurations", + ]) { + metrics[name + "Summary"] = summary(metrics[name]); + delete metrics[name]; + } + for (const pool of Object.keys(metrics.poolQueueByPool)) { + metrics.poolQueueByPool[pool] = summary(metrics.poolQueueByPool[pool]); + metrics.transactionLatencyByPool[pool] = summary(metrics.transactionLatencyByPool[pool]); + } + metrics.nodeCpuAverageOneCorePercent = + ((metrics.nodeCpu.user + metrics.nodeCpu.system) / 1000 / metrics.elapsedMs) * 100; + metrics.correctnessPassed = + metrics.errors.length === 0 && + metrics.renewalLosses === 0 && + metrics.dropped === 0 && + metrics.normalCompleted === metrics.normalOffered && + metrics.normalWritesCompleted === metrics.normalWritesOffered && + metrics.admissionsCompleted === metrics.admissionsOffered; + metrics.absoluteGates = { + requestP99: metrics.normalReadSummary.p99 < 250, + drain: metrics.drainMs < 1000, + throughput: metrics.normalCompleted / metrics.normalOffered >= 0.99, + postgresCpu: metrics.postgresCpuUsec / 1000 / metrics.elapsedMs / 2 < 0.75, + postgresMemory: metrics.postgresPeakSampledMemoryBytes < 1382.4 * 1024 * 1024, + }; + metrics.passed = metrics.correctnessPassed && Object.values(metrics.absoluteGates).every(Boolean); + await writeFile(root + "/" + label + ".json", JSON.stringify(metrics, null, 2) + "\n"); + console.log( + JSON.stringify({ + phase: "complete", + label, + passed: metrics.passed, + reads: metrics.normalReadSummary, + writes: metrics.normalWriteSummary, + control: metrics.controlLatencySummary, + renewalLosses: metrics.renewalLosses, + dropped: metrics.dropped, + cpu: metrics.nodeCpuAverageOneCorePercent, + errors: metrics.errors.slice(0, 3), + }), + ); + await monitor.end(); + await closeRegisteredClients(u); + await admin.unsafe(`drop database ${dbName}`); + return metrics; +} +try { + await setup(); + if (mode === "pilot") { + if (!(await run("candidate", 10, 9, 25000)).passed) process.exitCode = 1; + } else { + const profile = [ + ["baseline", 300], + ["candidate", 300], + ["candidate", 300], + ["baseline", 300], + ] as const; + const result = []; + for (const [variant, n] of profile) result.push(await run(variant, n, result.length + 1, 65000)); + const report = { + schema: "paperclip.normal-recovery-load/v1", + ...provenance, + scope: + "Paired baseline/candidate, including real PaperclipControlPlanePort snapshot persistence on every control checkpoint. Database/control-plane component stress; real production checkpoint writer and natural remote lease timers. Same synthetic DB request load. Not full HTTP, harness or hosted capacity qualification.", + hardware: { + postgresCpuLimit: 2, + postgresMemoryLimitMiB: 1536, + mainPool: 10, + livenessPool: 2, + candidateCheckpointPool: 2, + }, + workload: { + mode, + normalReadsPerSecond: mode === "burst" ? 100 : 500, + normalWritesPerSecond: mode === "burst" ? 20 : 100, + databaseAdmissionEverySeconds: 5, + checkpointControlEverySeconds: 5, + controlPayloadBytes: 4096, + phaseSeconds: 65, + snapshotTransactionsPerControl: 1, + baselineCheckpointPool: 2, + baselineRenewalIntervalSeconds: 20, + candidateRenewalIntervalSeconds: 20, + historicalIssues: 10000, + }, + results: result, + }; + const perOperation = (r: any, key: string) => + r[key] / (r.normalCompleted + r.normalWritesCompleted + r.controls + r.admissionsCompleted); + const average = (values: number[]) => values.reduce((a, b) => a + b, 0) / values.length; + const comparable = + result.every((r) => r.correctnessPassed) && + result.every( + (r) => + r.controls === result[0].controls && + r.normalCompleted === result[0].normalCompleted && + r.normalWritesCompleted === result[0].normalWritesCompleted, + ); + const relative = {}; + for (const key of ["postgresCpuUsec", "walBytes", "requestP99"]) { + const value = (r: any) => (key === "requestP99" ? r.normalReadSummary.p99 : perOperation(r, key)); + const baseline = average(result.filter((r) => r.variant === "baseline").map(value)); + const candidate = average(result.filter((r) => r.variant === "candidate").map(value)); + relative[key] = { + baseline, + candidate, + increasePercent: comparable ? 100 * (candidate / baseline - 1) : null, + passed: comparable ? candidate <= baseline * 1.1 : null, + }; + } + Object.assign(report, { + relative, + comparable, + passed: result.every((r) => r.passed) && Object.values(relative).every((r: any) => r.passed), + measurementLimits: [ + "Database request latency is not HTTP latency", + "Pool queue includes client encoding; BEGIN response includes network time", + "pg_stat_statements execution includes lock waits; samples report concurrent lock waiters, not a lock latency distribution", + "Peak memory is sampled once a second", + "Use an otherwise idle cluster and host; WAL and cgroup CPU are cluster-wide", + ], + }); + await writeFile(root + "/report.json", JSON.stringify(report, null, 2) + "\n"); + if (!(report as typeof report & { passed: boolean }).passed) process.exitCode = 1; + } +} finally { + await admin.end({ timeout: 1 }); +} +// On failure, retain owned databases for diagnosis. Successful trials are dropped; +// the random template remains available for fixture inspection until cluster cleanup.