mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(recovery): probe recovery source runs and stale-continuation runs by key (#20584)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The workspace recovery worker saves the files of failed runs. After each pass, it reads the time of its next deadline (`nextWorkspaceRecoveryAt`) > - That read finds the source run, its finalizer and its leases with `id::text = evidence->>'runId'`. A text comparison cannot use a key. Each exhausted hold read every run, finalizer and lease of its company > - On a hosted staging database this statement used about 4 s per call (21 calls in the sample window). The cost grows with exhausted holds multiplied by the company's run count > - The ownership reconciler's stale-continuation check (`checkStaleDroppedComments`) has the run lookup shape that #20412 fixed in the lost-wake check: one OR across the legacy context and the native column > - This pull request compares the run id as a canonical uuid, and it gives the stale-continuation check the per-binding run probes from #20412 > - The benefit is the same recovery decisions with index-driven work: 11 s to 93 ms for the deadline read on a representative local fixture ## Linked Issues or Issue Description Refs #20412 **What happened?** The workspace recovery deadline read was one of the most expensive statements that remained on a hosted staging database after the fixes in #20393, #20405, #20386, #20412, #20391 and #20400. Query statistics showed about 4 s per call. The statement text starts with `select case when (issue_recovery_actions.…`. The statement evaluates the retained-source fallback and stop-proof conditions for each recovery action. These conditions read the source run (`heartbeat_runs`), its finalizer (`native_run_finalizations`) and its leases (`environment_leases`) with `x.id::text = evidence->>'runId'`. The cast prevents the use of the primary key. PostgreSQL reads all rows of the company for each exhausted hold. The worker's discovery (`retainedWorkspaceCandidates`) uses the same conditions. The stale-continuation check matched a task's runs with `(context_snapshot->>'issueId' = issue OR native_issue_id = issue)`. With the OR, PostgreSQL can join runs only by company. It also hashed every wake request of the company for the stale-wake test. **Expected behavior** Each condition uses a key lookup. The worker and the reconciler make the same decisions for the same saved state. **Steps to reproduce** 1. Seed a company with a deep run history and a few thousand workspace recovery actions. Let a few hundred be exhausted holds that stay active. 2. Call `nextWorkspaceRecoveryAt`. 3. The call takes seconds, and its time grows with holds multiplied by runs. The benchmark below shows this. **Paperclip version or commit** Baseline: `8d2bf78323`. **Deployment mode** Server with PostgreSQL. The local proof uses disposable PostgreSQL 18.1. Staging uses hosted PostgreSQL. ## What Changed - `workspace-recovery-fallback-discovery.ts`: the source run, finalizer and lease probes compare `id = canonicalExecutionUuid(evidence->>'runId')`. The helper returns null unless the text is the canonical lowercase uuid form. Thus the probe matches exactly the rows that `id::text = evidence->>'runId'` matched. The helper already exists in `execution-identity.ts`. - `nextWorkspaceRecoveryAt` and worker discovery use these conditions. Their SQL changes only in these probes. - `checkStaleDroppedComments` reads in two stages. The first stage keeps comments that a stale-continuation cancellation dropped and that no open wake carries. The second stage applies the run probes. The run probes use the `taskRuns` helper from #20412, which is now shared by both checks. The stale-wake lookup ends with `offset 0`, so it stays a per-comment index probe. - The final `order by c.created_at limit N`, the projection and the delivery check after the query do not change. - Add `scripts/benchmark-recovery-status-checks.ts`. It loads the changed files from an explicit baseline commit and from the working tree. Before it records time, it compares every changed condition for every recovery action, the complete service results and the positional SQL rows. - Add tests: each run binding in the stale-continuation check (native, legacy, live run on either binding, wrong agent, wrong company, failed run, run outside the comment time, closed by a person, open wake). Also a test that a hold with a non-canonical saved run id is not discovered, as before. - Document both checks and the benchmark in `doc/DATABASE.md`. - No schema change, index, migration, or new startup work. ## Verification - `pnpm exec vitest run src/__tests__/ownership-reconciliation-report.test.ts src/services/workspace-recovery-fallback.test.ts src/services/workspace-recovery-stop-proof.test.ts src/__tests__/heartbeat-held-comment-wake.test.ts --maxWorkers=1` (from `server/`). - `pnpm --filter @paperclipai/server typecheck`. - `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-recovery-status-checks.ts --baseline-ref=8d2bf78323 --samples=7` Benchmark setup: PostgreSQL 18.1 (embedded, local), `shared_buffers` 128 MB, `work_mem` 4 MB. All data is synthetic, with md5-hex filler. Each fixture has two companies (full size and a quarter). Two warmup rounds, then 7 timed samples for each arm, in alternating order. Each sample runs the real service function and asserts the same result. These are local SQL measurements on a shared machine, so the spread is wide. Workspace recovery deadline (`nextWorkspaceRecoveryAt`). About 8% of actions are exhausted holds that stay active. Some actions save a non-canonical (uppercase) run id. Before timing, the benchmark compares six conditions (fallback pending, due time, due; re-arm pending, due time, due) for every action of the fixture. | Fixture | Runs / actions / pending holds | Baseline median (min–max) | Treatment median (min–max) | Shared buffers | | --- | --- | --- | --- | --- | | small | 6,500 / 255 / 36 | 50.9 ms (47–76) | 6.8 ms (6–8) | 19,133 → 1,075 | | representative | 78,750 / 3,813 / 588 | 11,098 ms (8,277–12,790) | 93 ms (35–145) | 3,493,417 → 24,902 | | deep history | 191,250 / 3,813 / 588 | 40,146 ms (31,190–52,521) | 40 ms (36–230) | 13,491,792 → 26,911 | Stale-continuation check (`checkStaleDroppedComments`). The fixture is pessimistic: more than half of recent comments on done tasks have a stale cancelled wake. The table shows statement time from `EXPLAIN ANALYZE` with default settings, and one statement run with `enable_nestloop = off` (the company-only hash plan). The result is checked as equal in both modes. | Fixture | Runs / recent comments / reported | Statement, baseline → treatment | Hash plan, baseline → treatment | | --- | --- | --- | --- | | small | 3,792 / 76 / 24 | 3.2 → 1.5 ms | 5.5 → 3.8 ms | | representative | 75,406 / 1,125 / 200 (limit) | 34 → 26 ms | 66 → 25 ms | | busy hour | 152,031 / 5,625 / 200 (limit) | 328 → 153 ms | 2,302 → 165 ms | - Neutral path: the complete stale-continuation service call stays about the same (busy hour 750 → 655 ms median). The per-comment delivery check after the query (`deliveredQueuedCommentIds`) is most of that time. This change does not touch it. ## Risks - Low risk. Every predicate keeps the same logic. Only the probe shape and the evaluation order change. The benchmark compares each changed condition for every recovery action, including non-canonical saved run ids, and the complete results of both checks. - Previous-release story: no saved state, schema or identity changes. Holds saved by older releases are read with the same rules, so they carry forward unchanged. - Rows with the same comment `created_at` have no defined order, as before. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local PostgreSQL). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
be0d100a8b
commit
6ed6b0754b
6 files changed
+580
-43
No files matched your search
@@ -2092,6 +2092,26 @@ pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-comment-wake-che
|
||||
This disposable fixture compares the complete check result, the complete
|
||||
positional SQL rows, plans, and timings with and without nested-loop joins.
|
||||
|
||||
The stale-continuation check for dropped user comments uses the same per-binding
|
||||
run probes. It first keeps the comments that a stale-continuation cancellation
|
||||
dropped and that no open wake carries, and only those comments reach the run
|
||||
probes.
|
||||
|
||||
The workspace recovery worker's next deadline (`nextWorkspaceRecoveryAt`) and
|
||||
its discovery conditions find the source run, its finalizer and its leases by
|
||||
`id = canonical uuid of evidence->>'runId'`. The value is null unless the saved
|
||||
text is the canonical lowercase form, so the probe matches exactly the rows the
|
||||
former `id::text = evidence->>'runId'` comparison matched. The text comparison
|
||||
could not use a key, so each exhausted hold read every run, finalizer and lease
|
||||
of its company.
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-recovery-status-checks.ts --baseline-ref=8d2bf78323
|
||||
```
|
||||
|
||||
This fixture compares every changed discovery condition for every recovery
|
||||
action, the complete check results and the positional SQL rows before timing.
|
||||
|
||||
### Agent runtime synchronization journal
|
||||
|
||||
Migration 0347 adds the opt-in `agent_runtime` resource stream. Database triggers
|
||||
|
||||
@@ -0,0 +1,429 @@
|
||||
/**
|
||||
* Local red/green proof for two recovery checks:
|
||||
*
|
||||
* - `workspace-recovery`: `nextWorkspaceRecoveryAt` (the workspace recovery
|
||||
* worker's next deadline) and the fallback/stop-proof conditions it shares
|
||||
* with worker discovery (`workspace-recovery-fallback-discovery.ts`).
|
||||
* - `dropped-comments`: `checkStaleDroppedComments` in the ownership
|
||||
* invariant report.
|
||||
*
|
||||
* Loads the changed files from an explicit pre-change revision and from the
|
||||
* working tree, runs both against the same synthetic PostgreSQL fixture, and
|
||||
* asserts that the complete results are equal before timing anything. The
|
||||
* workspace recovery proof also compares every condition the change touched,
|
||||
* for every recovery action of the fixture.
|
||||
*
|
||||
* pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-recovery-status-checks.ts --baseline-ref=<commit>
|
||||
*
|
||||
* Options: --fixture=<name> (repeatable), --samples=<n>, --skip-hash-plan, --out=<file>.
|
||||
*/
|
||||
import assert from "node:assert/strict";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath, pathToFileURL } from "node:url";
|
||||
import { createRequire } from "node:module";
|
||||
import { createDb, issueRecoveryActions, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js";
|
||||
import * as schema from "../packages/db/src/schema/index.js";
|
||||
|
||||
// Resolve database drivers through the db package; the scripts folder has no own dependencies.
|
||||
const dbRequire = createRequire(new URL("../packages/db/package.json", import.meta.url));
|
||||
const postgres = (await import(pathToFileURL(dbRequire.resolve("postgres")).href)).default as typeof import("postgres").default;
|
||||
const { drizzle } = await import(pathToFileURL(dbRequire.resolve("drizzle-orm/postgres-js")).href) as typeof import("drizzle-orm/postgres-js");
|
||||
const { sql } = await import(pathToFileURL(dbRequire.resolve("drizzle-orm")).href) as typeof import("drizzle-orm");
|
||||
|
||||
const baselineRef = process.argv.find(value => value.startsWith("--baseline-ref="))?.slice(15);
|
||||
assert.ok(baselineRef, "Pass an explicit pre-change --baseline-ref=<commit>");
|
||||
// Run every git command from the repository root: the documented command runs from server/.
|
||||
const root = fileURLToPath(new URL("../", import.meta.url));
|
||||
const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8", cwd: root }).trim();
|
||||
const samplesPerArm = Number(process.argv.find(value => value.startsWith("--samples="))?.slice(10) ?? 5);
|
||||
const changedSources = [
|
||||
"server/src/services/recovery/ownership-invariant-report.ts",
|
||||
"server/src/services/workspace-recovery-fallback-discovery.ts",
|
||||
];
|
||||
// Every other helper the checks import must be unchanged, or the baseline is contaminated.
|
||||
const changedRuntime = execFileSync("git", ["diff", "--name-only", baselineRevision, "--",
|
||||
"server/src/services", "packages/db/src", "packages/shared/src", "pnpm-lock.yaml"], { encoding: "utf8", cwd: root })
|
||||
.trim().split("\n").filter(Boolean);
|
||||
assert.deepEqual(changedRuntime.filter(file => !changedSources.includes(file) && !file.endsWith(".test.ts")), [],
|
||||
"The baseline must not import changed runtime helpers");
|
||||
|
||||
// Copy the baseline version of each changed file. A relative import of another
|
||||
// changed file points at its baseline copy; every other import points at the
|
||||
// working tree, which the check above proves unchanged.
|
||||
const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.recovery-status-baseline-"));
|
||||
const copyName = (source: string) => source.replaceAll("/", "__");
|
||||
for (const source of changedSources) {
|
||||
const text = execFileSync("git", ["show", `${baselineRevision}:${source}`], { encoding: "utf8", cwd: root });
|
||||
const original = new URL(`../${source}`, import.meta.url);
|
||||
const rewritten = text.replace(/from "(\.{1,2}\/[^"]+)"/g, (_match, specifier: string) => {
|
||||
const target = fileURLToPath(new URL(specifier, original));
|
||||
const relative = path.relative(root, target).replace(/\.js$/, ".ts");
|
||||
return `from ${JSON.stringify(changedSources.includes(relative) ? path.join(baselineDir, copyName(relative)) : target)}`;
|
||||
});
|
||||
await fs.writeFile(path.join(baselineDir, copyName(source)), rewritten);
|
||||
}
|
||||
type Report = typeof import("../server/src/services/recovery/ownership-invariant-report.js");
|
||||
type Discovery = typeof import("../server/src/services/workspace-recovery-fallback-discovery.js");
|
||||
type DroppedCheck = (db: Db, now: Date, limit: number) => Promise<unknown>;
|
||||
const load = async <T>(source: string) => await import(pathToFileURL(path.join(baselineDir, copyName(source))).href) as T;
|
||||
// The baseline did not export the dropped-comment check; expose it for the benchmark only.
|
||||
const reportCopy = path.join(baselineDir, copyName(changedSources[0]!));
|
||||
const reportText = await fs.readFile(reportCopy, "utf8");
|
||||
if (!/export async function checkStaleDroppedComments/.test(reportText)) {
|
||||
await fs.writeFile(reportCopy, `${reportText}\nexport { checkStaleDroppedComments };\n`);
|
||||
}
|
||||
const baselineReport = await load<Report>(changedSources[0]!);
|
||||
const baselineDiscovery = await load<Discovery>(changedSources[1]!);
|
||||
const treatmentReport = await import("../server/src/services/recovery/ownership-invariant-report.js") as Report;
|
||||
const treatmentDiscovery = await import("../server/src/services/workspace-recovery-fallback-discovery.js") as Discovery;
|
||||
const { nextWorkspaceRecoveryAt: treatmentNext } = await import("../server/src/services/system-recovery.js");
|
||||
|
||||
// nextWorkspaceRecoveryAt builds its statement from the discovery module. The
|
||||
// baseline arm is the same statement built from the baseline discovery copy:
|
||||
// load a copy of system-recovery.ts whose discovery import is the baseline.
|
||||
const systemSource = "server/src/services/system-recovery.ts";
|
||||
assert.ok(!changedRuntime.includes(systemSource));
|
||||
{
|
||||
const text = await fs.readFile(path.join(root, systemSource), "utf8");
|
||||
const original = new URL(`../${systemSource}`, import.meta.url);
|
||||
const rewritten = text.replace(/from "(\.{1,2}\/[^"]+)"/g, (_match, specifier: string) => {
|
||||
const target = fileURLToPath(new URL(specifier, original));
|
||||
const relative = path.relative(root, target).replace(/\.js$/, ".ts");
|
||||
return `from ${JSON.stringify(changedSources.includes(relative) ? path.join(baselineDir, copyName(relative)) : target)}`;
|
||||
});
|
||||
await fs.writeFile(path.join(baselineDir, copyName(systemSource)), rewritten);
|
||||
}
|
||||
const { nextWorkspaceRecoveryAt: baselineNext } = await load<typeof import("../server/src/services/system-recovery.js")>(systemSource);
|
||||
|
||||
type Fixture = { name: string; check: "workspace-recovery" | "dropped-comments"; [key: string]: number | string };
|
||||
const fixtures: Fixture[] = [
|
||||
// Workspace recovery: actions per company (most complete), runs per company, finalizers and leases.
|
||||
{ name: "recovery-small", check: "workspace-recovery", actions: 200, runs: 5_000, contextBytes: 1_024, evidenceBytes: 512 },
|
||||
// A busy instance after restart waves: a few thousand recovery actions, most
|
||||
// complete, a few hundred exhausted holds that stay active, deep run history.
|
||||
{ name: "recovery-representative", check: "workspace-recovery", actions: 3_000, runs: 60_000, contextBytes: 2_048, evidenceBytes: 1_024 },
|
||||
// The same actions over a deeper run history. The baseline grows with holds times runs per company.
|
||||
{ name: "recovery-deep-history", check: "workspace-recovery", actions: 3_000, runs: 150_000, contextBytes: 2_048, evidenceBytes: 1_024 },
|
||||
// Dropped comments: recent user comments on done tasks over a deep run history.
|
||||
{ name: "dropped-small", check: "dropped-comments", issues: 300, runs: 3_000, hotIssues: 30, commentsPerHotIssue: 2, contextBytes: 1_024 },
|
||||
{ name: "dropped-representative", check: "dropped-comments", issues: 4_000, runs: 60_000, hotIssues: 300, commentsPerHotIssue: 3, contextBytes: 4_096 },
|
||||
{ name: "dropped-busy-hour", check: "dropped-comments", issues: 6_000, runs: 120_000, hotIssues: 1_500, commentsPerHotIssue: 3, contextBytes: 2_048 },
|
||||
];
|
||||
|
||||
const keepAlive = setInterval(() => {}, 1000);
|
||||
const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!;
|
||||
const spread = (values: number[]) => ({ min: Math.min(...values), max: Math.max(...values) });
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: Db | undefined;
|
||||
let captureClient: import("postgres").Sql | undefined;
|
||||
|
||||
/** Synthetic, incompressible text of about `bytes` characters (md5 hex chain). */
|
||||
const filler = (bytes: number, seed: string) =>
|
||||
`(select string_agg(md5(${seed}::text || ':' || g::text), '') from generate_series(1, ${Math.max(1, Math.ceil(bytes / 32))}) g)`;
|
||||
|
||||
async function seedCompany(db: Db, scale: number, prefixSeed: string, issueCount: number) {
|
||||
const run = (text: string, params: unknown[] = []) => db.$client.unsafe(text, params as never[]);
|
||||
const company = randomUUID();
|
||||
const prefix = `B${prefixSeed}${company.slice(0, 5).toUpperCase()}`;
|
||||
await run(`insert into companies (id, name, issue_prefix) values ($1::uuid, 'Synthetic', $2)`, [company, prefix]);
|
||||
await run(`insert into agents (company_id, name, adapter_type) select $1::uuid, 'agent-' || n, 'codex_local' from generate_series(1, 20) n`, [company]);
|
||||
await run(`create temp table if not exists bench_agents (company_id uuid, n int, id uuid)`);
|
||||
await run(`insert into bench_agents select company_id, row_number() over (order by id), id from agents where company_id = $1::uuid`, [company]);
|
||||
await run(`insert into issues (company_id, title, status, assignee_agent_id, identifier, issue_number)
|
||||
select $1::uuid, 'I-' || s.num, (array['done','done','todo','in_progress','done','cancelled','todo','in_progress','done','todo'])[1 + s.num % 10],
|
||||
(select a.id from bench_agents a where a.company_id = $1::uuid and a.n = 1 + s.num % 20), $2 || '-' || s.num, s.num
|
||||
from generate_series(1, $3::int) s(num)`, [company, prefix, Math.ceil(issueCount * scale)]);
|
||||
await run(`create temp table if not exists bench_issues (company_id uuid, n int, id uuid, agent_id uuid)`);
|
||||
await run(`insert into bench_issues select company_id, issue_number, id, assignee_agent_id from issues where company_id = $1::uuid`, [company]);
|
||||
return company;
|
||||
}
|
||||
|
||||
/** Run history over 14 days: legacy runs bind by context, native runs by column, some by both. */
|
||||
async function seedRunHistory(db: Db, company: string, now: Date, runCount: number, issueCount: number, contextBytes: number) {
|
||||
await db.$client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot,
|
||||
runner_profile_json, created_at, started_at, finished_at)
|
||||
select $1::uuid, i.agent_id, (array['succeeded','failed','cancelled','succeeded','timed_out'])[1 + r % 5], 'assignment',
|
||||
case when r % 3 <> 0 then i.id end,
|
||||
case when r % 3 = 1 then '{}'::jsonb else jsonb_build_object('issueId', i.id::text, 'prompt', ${filler(contextBytes, "r")}) end,
|
||||
jsonb_build_object('remoteWorkspace', jsonb_build_object('state', case when r % 4 = 0 then 'prepared' else 'released' end)),
|
||||
$2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int), $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int),
|
||||
$2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int) + interval '20 minutes'
|
||||
from generate_series(1, $3::int) r join bench_issues i on i.company_id = $1::uuid and i.n = 1 + (r * 7919) % $4::int`,
|
||||
[company, now.toISOString(), runCount, issueCount] as never[]);
|
||||
}
|
||||
|
||||
async function seedWorkspaceRecovery(db: Db, fixture: Fixture, now: Date) {
|
||||
const run = (text: string, params: unknown[] = []) => db.$client.unsafe(text, params as never[]);
|
||||
const actionsTotal = Number(fixture.actions), runsTotal = Number(fixture.runs);
|
||||
for (const [scale, tag] of [[1, "A"], [0.25, "B"]] as const) {
|
||||
const actionCount = Math.ceil(actionsTotal * scale), runCount = Math.ceil(runsTotal * scale);
|
||||
// Each recovery action has its own task; ordinary tasks carry the run history.
|
||||
const issueCount = actionCount + Math.ceil(runCount / 15);
|
||||
const company = await seedCompany(db, 1, tag, issueCount);
|
||||
await seedRunHistory(db, company, now, runCount, issueCount, Number(fixture.contextBytes));
|
||||
// Finalizers for a third of the history, some still leased.
|
||||
await run(`insert into native_run_finalizations (run_id, company_id, issue_id, phase, lease_owner, lease_expires_at)
|
||||
select r.id, r.company_id, coalesce(r.native_issue_id, (r.context_snapshot->>'issueId')::uuid), 'completed',
|
||||
case when row_number() over () % 7 = 0 then 'owner' end, $2::timestamptz + interval '10 minutes'
|
||||
from heartbeat_runs r where r.company_id = $1::uuid and r.native_issue_id is not null and r.created_at::text like '%3%'`,
|
||||
[company, now.toISOString()]);
|
||||
await run(`insert into environment_leases (company_id, heartbeat_run_id, issue_id, lease_policy, updated_at)
|
||||
select r.company_id, r.id, r.native_issue_id, 'retain_on_failure',
|
||||
r.finished_at from heartbeat_runs r where r.company_id = $1::uuid and r.native_issue_id is not null and r.created_at::text like '%7%'`,
|
||||
[company]);
|
||||
// Source runs of the recovery actions: one failed native run per action task.
|
||||
await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot,
|
||||
runner_profile_json, created_at, started_at, finished_at)
|
||||
select i.company_id, i.agent_id, 'failed', 'assignment', i.id, jsonb_build_object('issueId', i.id::text),
|
||||
jsonb_build_object('remoteWorkspace', jsonb_build_object('state', case when i.n % 3 = 0 then 'released' else 'prepared' end)),
|
||||
$2::timestamptz - (i.n * interval '1 minute'), $2::timestamptz - (i.n * interval '1 minute'), $2::timestamptz - (i.n * interval '1 minute') + interval '5 minutes'
|
||||
from bench_issues i where i.company_id = $1::uuid and i.n <= $3::int`, [company, now.toISOString(), actionCount]);
|
||||
// Stop-proof holds whose source finalizer is still leased (n % 40 = 15) or has an expired lease (n % 40 = 35).
|
||||
await run(`insert into native_run_finalizations (run_id, company_id, issue_id, phase, lease_owner, lease_expires_at)
|
||||
select r.id, r.company_id, r.native_issue_id, 'finalizing', 'owner',
|
||||
case when i.n % 40 = 15 then $2::timestamptz + interval '20 minutes' else $2::timestamptz - interval '20 minutes' end
|
||||
from heartbeat_runs r join bench_issues i on i.id = r.native_issue_id
|
||||
where r.company_id = $1::uuid and r.status = 'failed' and i.n <= $3::int and i.n % 40 in (15, 35)
|
||||
and r.created_at = $2::timestamptz - (i.n * interval '1 minute')`, [company, now.toISOString(), actionCount]);
|
||||
// A late provider stop receipt after a deferred check (n % 20 = 16 with n % 40 = 36).
|
||||
await run(`insert into environment_leases (company_id, heartbeat_run_id, issue_id, lease_policy, updated_at)
|
||||
select r.company_id, r.id, r.native_issue_id, 'retain_on_failure',
|
||||
$2::timestamptz - interval '1 minute'
|
||||
from heartbeat_runs r join bench_issues i on i.id = r.native_issue_id
|
||||
where r.company_id = $1::uuid and r.status = 'failed' and i.n <= $3::int and i.n % 40 = 36
|
||||
and r.created_at = $2::timestamptz - (i.n * interval '1 minute')`, [company, now.toISOString(), actionCount]);
|
||||
// Recovery actions by n % 20:
|
||||
// 0-11 complete and resolved (most history); 12-13 exhausted with a permanently refused source;
|
||||
// 14-15 exhausted unproven-stop holds with re-arms left; 16 the same with every re-arm spent;
|
||||
// 17 scheduled; 18 cancelled with export requests; 19 complete with a pending bootstrap continuation.
|
||||
// Exhausted holds alternate between a spent deferred check of this generation, a deferred check
|
||||
// with a future time, an older generation's check, and no check. Every due time is in the future.
|
||||
await run(`insert into issue_recovery_actions (company_id, source_issue_id, kind, status, owner_type, cause, fingerprint, next_action, evidence,
|
||||
created_at, updated_at)
|
||||
select i.company_id, i.id, 'workspace_recovery',
|
||||
case when k < 12 or k = 19 then 'resolved' when k = 18 then 'cancelled' when k in (13, 15) then 'escalated' else 'active' end,
|
||||
'system', case when i.n % 50 = 7 then 'workspace_restore_failed' else (array['native_workspace_sync_out_unrecoverable','native_workspace_sync_out_retry_exhausted'])[1 + i.n % 2] end,
|
||||
'fp-' || i.n, 'next',
|
||||
jsonb_strip_nulls(jsonb_build_object(
|
||||
'runId', case when i.n % 97 = 5 then upper(r.id::text) else r.id::text end,
|
||||
'filler', ${filler(Number(fixture.evidenceBytes), "i.n")},
|
||||
'workspaceRestoreRecovery', case when i.n % 50 = 7 then jsonb_build_object('schema', 'paperclip.workspace-restore-recovery.v1') end,
|
||||
'bootstrapPreparationSettledAt', case when k = 19 then '2026-01-01T00:00:00Z' end,
|
||||
'bootstrapPreparationContinuation', case when k = 19 then 'pending' end,
|
||||
'systemRecovery', jsonb_build_object('schema', 'paperclip.system-recovery.v1', 'policyVersion', 1, 'sourceRunId', r.id::text,
|
||||
'generationId', 'gen-' || i.n, 'revision', 2,
|
||||
'state', case when k < 12 or k = 19 then 'complete' when k = 17 then 'scheduled' else 'exhausted' end,
|
||||
'nextAttemptAt', case when k = 17 then $3::bigint + 60000 * (1 + i.n % 30) end,
|
||||
'failure', case when k in (12, 13) then jsonb_build_object('code', (array['source_unavailable','historical_source_missing','source_binding_invalid'])[1 + i.n % 3],
|
||||
'retryClass', case when i.n % 11 = 0 then 'recheck_evidence' else 'service_required' end)
|
||||
when k in (14, 15, 16, 18) then jsonb_build_object('code', 'agent_stop_unconfirmed', 'retryClass', 'recheck_evidence') end),
|
||||
'workspaceRecovery', jsonb_build_object('schema', 'paperclip.workspace-recovery.v1', 'runId', r.id::text,
|
||||
'status', case when k < 12 then 'recovered' else 'blocked' end,
|
||||
'requestIds', case when k = 18 then jsonb_build_array('req-' || i.n) else '[]'::jsonb end),
|
||||
'stopProofRecovery', case when k in (14, 15, 16) then jsonb_build_object('schema', 'paperclip.stop-proof-recovery.v1',
|
||||
'rearms', case when k = 16 then 4 else i.n % 4 end, 'nextRearmAt', $3::bigint + 60000 * (5 + i.n % 50)) end,
|
||||
'retainedSourceFallbackCheck', case when k in (12, 13, 16) then (array[
|
||||
jsonb_build_object('generationId', 'gen-' || i.n, 'checkedAt', $3::bigint - 600000),
|
||||
jsonb_build_object('generationId', 'gen-' || i.n, 'checkedAt', $3::bigint - 600000, 'nextCheckAt', $3::bigint + 60000 * (3 + i.n % 40)),
|
||||
jsonb_build_object('generationId', 'gen-old', 'checkedAt', $3::bigint - 600000, 'nextCheckAt', $3::bigint + 60000 * (3 + i.n % 40)),
|
||||
jsonb_build_object('generationId', 'gen-' || i.n, 'nextCheckAt', $3::bigint + 60000 * (7 + i.n % 40))])[1 + (i.n / 20) % 4] end)),
|
||||
$2::timestamptz, $2::timestamptz
|
||||
from (select i.*, i.n % 20 as k from bench_issues i where i.company_id = $1::uuid and i.n <= $4::int) i
|
||||
join heartbeat_runs r on r.company_id = i.company_id and r.native_issue_id = i.id and r.status = 'failed'
|
||||
and r.created_at = $2::timestamptz - (i.n * interval '1 minute')`,
|
||||
[company, now.toISOString(), now.getTime(), actionCount]);
|
||||
// Siblings: a completed recovery of the same run saved the files (n % 60 = 12).
|
||||
await run(`insert into issue_recovery_actions (company_id, source_issue_id, kind, status, owner_type, cause, fingerprint, next_action, evidence)
|
||||
select a.company_id, a.source_issue_id, 'workspace_recovery', 'resolved', 'system', a.cause, a.fingerprint || '-sibling', 'next',
|
||||
jsonb_build_object('runId', a.evidence->>'runId', 'systemRecovery', jsonb_build_object('state', 'complete'),
|
||||
'workspaceRecovery', jsonb_build_object('status', 'recovered'))
|
||||
from issue_recovery_actions a join bench_issues i on i.id = a.source_issue_id
|
||||
where a.company_id = $1::uuid and i.n % 60 = 12`, [company]);
|
||||
}
|
||||
await run("analyze");
|
||||
}
|
||||
|
||||
async function seedDroppedComments(db: Db, fixture: Fixture, now: Date) {
|
||||
const run = (text: string, params: unknown[] = []) => db.$client.unsafe(text, params as never[]);
|
||||
for (const [scale, tag] of [[1, "A"], [0.25, "B"]] as const) {
|
||||
const issueCount = Math.ceil(Number(fixture.issues) * scale), runCount = Math.ceil(Number(fixture.runs) * scale);
|
||||
const hot = Math.ceil(Number(fixture.hotIssues) * scale), perIssue = Number(fixture.commentsPerHotIssue);
|
||||
const company = await seedCompany(db, scale, tag, Number(fixture.issues));
|
||||
await seedRunHistory(db, company, now, runCount, issueCount, Number(fixture.contextBytes));
|
||||
// Hot tasks are done; the agent closed even n, a person odd n (n % 12 = 9).
|
||||
await run(`update issues set status = 'done' where company_id = $1::uuid and issue_number <= $2::int`, [company, hot]);
|
||||
await run(`insert into activity_log (company_id, actor_type, actor_id, action, entity_type, entity_id, details, created_at)
|
||||
select i.company_id, case when i.n % 12 = 9 then 'user' else 'agent' end, 'actor', 'issue.updated', 'issue', i.id::text,
|
||||
jsonb_build_object('toStatus', 'done'), $2::timestamptz - interval '50 minutes'
|
||||
from bench_issues i where i.company_id = $1::uuid and i.n <= $3::int`, [company, now.toISOString(), hot]);
|
||||
// The closing run: started before the comments and finished after them; bound by context (odd n) or column (even n).
|
||||
await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot, created_at, started_at, finished_at)
|
||||
select i.company_id, case when i.n % 12 = 11 then (select a.id from bench_agents a where a.company_id = i.company_id and a.id <> i.agent_id limit 1) else i.agent_id end,
|
||||
case when i.n % 12 = 10 then 'failed' else 'succeeded' end, 'assignment',
|
||||
case when i.n % 2 = 0 then i.id end, case when i.n % 2 = 1 then jsonb_build_object('issueId', i.id::text, 'prompt', ${filler(Number(fixture.contextBytes), "i.n")}) else '{}'::jsonb end,
|
||||
$2::timestamptz - interval '58 minutes', $2::timestamptz - interval '58 minutes',
|
||||
case when i.n % 12 = 8 then $2::timestamptz - interval '56 minutes' else $2::timestamptz - interval '1 minute' end
|
||||
from bench_issues i where i.company_id = $1::uuid and i.n <= $3::int`, [company, now.toISOString(), hot]);
|
||||
// A live run on the task (n % 12 = 6), bound by the other binding.
|
||||
await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot, created_at, started_at)
|
||||
select i.company_id, i.agent_id, (array['queued','running','scheduled_retry'])[1 + i.n % 3], 'assignment',
|
||||
case when i.n % 2 = 1 then i.id end, case when i.n % 2 = 0 then jsonb_build_object('issueId', i.id::text) else '{}'::jsonb end,
|
||||
$2::timestamptz - interval '2 minutes', $2::timestamptz - interval '2 minutes'
|
||||
from bench_issues i where i.company_id = $1::uuid and i.n <= $3::int and i.n % 12 = 6`, [company, now.toISOString(), hot]);
|
||||
await run(`insert into issue_comments (company_id, issue_id, author_user_id, body, created_at)
|
||||
select i.company_id, i.id, 'user-1', 'comment ' || k,
|
||||
$2::timestamptz - interval '55 minutes' + ((k - 1) * $3::int + i.n) * (interval '50 minutes' / ($3::int * $4::int)) + $5::int * interval '1 millisecond'
|
||||
from bench_issues i cross join generate_series(1, $4::int) k where i.company_id = $1::uuid and i.n <= $3::int`,
|
||||
[company, now.toISOString(), hot, perIssue, scale === 1 ? 0 : 500]);
|
||||
await run(`insert into issue_comments (company_id, issue_id, author_user_id, body, created_at)
|
||||
select i.company_id, i.id, 'user-1', 'old', $2::timestamptz - interval '2 hours' - (k * interval '3 hours')
|
||||
from bench_issues i cross join generate_series(1, 8) k where i.company_id = $1::uuid`, [company, now.toISOString()]);
|
||||
// Wakes by n % 12: 0-2 completed (normal delivery); 3-11 cancelled as stale continuations
|
||||
// (commentId, or a goal wake context list for n % 4 = 3); n % 12 = 7 also has an open wake.
|
||||
await run(`insert into agent_wakeup_requests (company_id, agent_id, source, status, error, payload, created_at, updated_at)
|
||||
select c.company_id, i.agent_id, 'automation', case when i.n % 12 < 3 then 'completed' else 'cancelled' end,
|
||||
case when i.n % 12 >= 3 then 'continuation_task_ownership_changed' end,
|
||||
case when i.n % 4 = 3 then jsonb_build_object('issueId', c.issue_id::text, '_paperclipWakeContext', jsonb_build_object('wakeCommentIds', jsonb_build_array(c.id::text)))
|
||||
else jsonb_build_object('issueId', c.issue_id::text, 'commentId', c.id::text) end, c.created_at, c.created_at
|
||||
from issue_comments c join bench_issues i on i.id = c.issue_id where c.company_id = $1::uuid`, [company]);
|
||||
await run(`insert into agent_wakeup_requests (company_id, agent_id, source, status, payload, created_at, updated_at)
|
||||
select c.company_id, i.agent_id, 'automation', 'deferred_issue_execution',
|
||||
jsonb_build_object('issueId', c.issue_id::text, 'commentId', c.id::text), c.created_at, c.created_at
|
||||
from issue_comments c join bench_issues i on i.id = c.issue_id
|
||||
where c.company_id = $1::uuid and i.n % 12 = 7 and c.created_at >= $2::timestamptz - interval '1 hour'`, [company, now.toISOString()]);
|
||||
}
|
||||
await run("analyze");
|
||||
}
|
||||
|
||||
type Query = { sql: string; params: unknown[] };
|
||||
let recorded: Query[] | null = null;
|
||||
let captureDb: Db | undefined;
|
||||
/** Runs `call` on a connection that records each statement it sends. */
|
||||
async function capture(call: (db: Db) => Promise<unknown>): Promise<{ queries: Query[]; result: unknown }> {
|
||||
const queries: Query[] = [];
|
||||
recorded = queries;
|
||||
try { return { queries, result: await call(captureDb!) }; } finally { recorded = null; }
|
||||
}
|
||||
|
||||
/** Every condition the change touched, for every recovery action, from one arm's discovery module. */
|
||||
async function conditionRows(db: Db, discovery: Discovery, now: number) {
|
||||
return await db.select({
|
||||
id: issueRecoveryActions.id,
|
||||
fallbackPending: sql<boolean>`${discovery.retainedSourceFallbackPendingCondition()}`,
|
||||
fallbackDueAt: sql<string>`(${discovery.retainedSourceFallbackDueAt()})::text`,
|
||||
fallbackDue: sql<boolean>`${discovery.retainedSourceFallbackDueCondition(now)}`,
|
||||
rearmPending: sql<boolean>`${discovery.stopProofRearmPendingCondition()}`,
|
||||
rearmDueAt: sql<string>`(${discovery.stopProofRearmDueAt()})::text`,
|
||||
rearmDue: sql<boolean>`${discovery.stopProofRearmDueCondition(now)}`,
|
||||
}).from(issueRecoveryActions).orderBy(issueRecoveryActions.id);
|
||||
}
|
||||
|
||||
try {
|
||||
database = await startEmbeddedPostgresTestDatabase("recovery-status-benchmark-");
|
||||
db = createDb(database.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0 });
|
||||
captureClient = postgres(database.connectionString, { max: 1, onnotice: () => {}, fetch_types: false,
|
||||
debug: (_connection, query, params) => { recorded?.push({ sql: query, params }); } });
|
||||
captureDb = drizzle(captureClient, { schema }) as unknown as Db;
|
||||
await db.$client.unsafe("set client_min_messages = warning");
|
||||
const version = await db.$client.unsafe("select version()");
|
||||
const settings = await db.$client.unsafe("select name, setting from pg_settings where name in ('shared_buffers', 'work_mem', 'jit', 'random_page_cost') order by name");
|
||||
const selectors = process.argv.filter(value => value.startsWith("--fixture=")).map(value => value.slice(10));
|
||||
const results = [];
|
||||
for (const fixture of fixtures.filter(value => !selectors.length || selectors.includes(value.name))) {
|
||||
await db.$client.unsafe("TRUNCATE companies CASCADE");
|
||||
await db.$client.unsafe("drop table if exists bench_agents, bench_issues");
|
||||
const now = new Date(), limit = 200;
|
||||
const seedStart = performance.now();
|
||||
let arms: Record<"before" | "after", (db: Db) => Promise<unknown>>;
|
||||
let proof: Record<string, unknown>;
|
||||
if (fixture.check === "workspace-recovery") {
|
||||
await seedWorkspaceRecovery(db, fixture, now);
|
||||
arms = { before: baselineNext, after: treatmentNext };
|
||||
const [expected, actual] = [await conditionRows(db, baselineDiscovery, now.getTime()), await conditionRows(db, treatmentDiscovery, now.getTime())];
|
||||
assert.deepEqual(actual, expected, `${fixture.name}: a recovery condition differs for some action`);
|
||||
const count = (key: keyof typeof expected[number]) => expected.filter(row => row[key] === true).length;
|
||||
proof = { actionsCompared: expected.length, fallbackPending: count("fallbackPending"), fallbackDue: count("fallbackDue"),
|
||||
rearmPending: count("rearmPending"), rearmDue: count("rearmDue") };
|
||||
assert.ok(Number(proof.fallbackPending) > 0 && Number(proof.rearmPending) > 0, `${fixture.name}: the fixture must have pending holds`);
|
||||
} else {
|
||||
await seedDroppedComments(db, fixture, now);
|
||||
const check = (report: Report) => (input: Db) => (report as unknown as { checkStaleDroppedComments: DroppedCheck }).checkStaleDroppedComments(input, now, limit);
|
||||
arms = { before: check(baselineReport), after: check(treatmentReport) };
|
||||
proof = {};
|
||||
}
|
||||
const seedMs = performance.now() - seedStart;
|
||||
const counts = (await db.$client.unsafe(`select (select count(*) from heartbeat_runs) runs, (select count(*) from issue_recovery_actions) recovery_actions,
|
||||
(select count(*) from issue_comments where created_at >= now() - interval '1 hour') recent_comments,
|
||||
(select count(*) from native_run_finalizations) finalizers, (select count(*) from environment_leases) leases,
|
||||
(select pg_size_pretty(pg_total_relation_size('heartbeat_runs'))) runs_size`))[0];
|
||||
const before = await capture(arms.before);
|
||||
const after = await capture(arms.after);
|
||||
assert.deepEqual(after.result, before.result, `${fixture.name}: complete service result differs`);
|
||||
if (fixture.check === "workspace-recovery") assert.ok(before.result !== null, `${fixture.name}: the fixture must have a deadline`);
|
||||
else assert.ok((before.result as unknown[]).length > 0, `${fixture.name}: the fixture must report owed comments`);
|
||||
const statement = (captured: typeof before) => captured.queries[0]!;
|
||||
const rows = async (query: Query) => db!.$client.unsafe(query.sql, query.params as never[]).values();
|
||||
const expectedRows = await rows(statement(before)), actualRows = await rows(statement(after));
|
||||
assert.ok(expectedRows.length > 0);
|
||||
// Order is part of the contract: compare the statement's rows positionally.
|
||||
assert.deepEqual(actualRows, expectedRows, `${fixture.name}: complete positional SQL output differs`);
|
||||
const samples = { before: [] as number[], after: [] as number[] };
|
||||
for (let iteration = 0; iteration < samplesPerArm + 2; iteration++) {
|
||||
for (const arm of iteration % 2 ? ["after", "before"] as const : ["before", "after"] as const) {
|
||||
const start = performance.now();
|
||||
const result = await arms[arm](db);
|
||||
const ms = performance.now() - start;
|
||||
assert.deepEqual(result, before.result);
|
||||
if (iteration >= 2) samples[arm].push(ms);
|
||||
}
|
||||
}
|
||||
const plans: Record<string, unknown> = {};
|
||||
const buffers: Record<string, number> = {};
|
||||
for (const [arm, captured] of [["before", before], ["after", after]] as const) {
|
||||
const query = statement(captured);
|
||||
const [row] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]);
|
||||
plans[arm] = row["QUERY PLAN"];
|
||||
const plan = (row["QUERY PLAN"] as { Plan: Record<string, number> }[])[0]!.Plan;
|
||||
buffers[arm] = Number(plan["Shared Hit Blocks"]) + Number(plan["Shared Read Blocks"]);
|
||||
}
|
||||
const medians = { before: median(samples.before), after: median(samples.after) };
|
||||
// Hash-join plan: without nested loops, PostgreSQL can join an OR-bound or
|
||||
// text-cast run lookup only by company. One sample each, untimed loop, same result check.
|
||||
const hashPlan: Record<string, number | string> = {};
|
||||
if (!process.argv.includes("--skip-hash-plan")) {
|
||||
await db.$client.unsafe("set enable_nestloop = off");
|
||||
await db.$client.unsafe("set statement_timeout = '300s'");
|
||||
try {
|
||||
for (const [arm, captured] of [["before", before], ["after", after]] as const) {
|
||||
const start = performance.now();
|
||||
try {
|
||||
assert.deepEqual(await rows(statement(captured)), expectedRows, `${fixture.name}: hash-plan output differs`);
|
||||
hashPlan[arm] = performance.now() - start;
|
||||
} catch (error) {
|
||||
if ((error as { code?: string }).code !== "57014") throw error;
|
||||
hashPlan[arm] = "timeout after 300 s";
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await db.$client.unsafe("reset enable_nestloop");
|
||||
await db.$client.unsafe("reset statement_timeout");
|
||||
}
|
||||
}
|
||||
results.push({ fixture, counts, seedMs, proof, reportedRows: expectedRows.length, queries: { before: statement(before).sql, after: statement(after).sql },
|
||||
samples, medians, spread: { before: spread(samples.before), after: spread(samples.after) }, buffers, hashPlan, plans });
|
||||
process.stderr.write(`${fixture.name}: ${JSON.stringify(counts)} proof=${JSON.stringify(proof)} rows=${expectedRows.length}; `
|
||||
+ `median ${medians.before.toFixed(1)} -> ${medians.after.toFixed(1)} ms (${(medians.before / medians.after).toFixed(1)}x); `
|
||||
+ `buffers ${buffers.before} -> ${buffers.after}; hash plan ${JSON.stringify(hashPlan)}; `
|
||||
+ `samples before=${samples.before.map(v => v.toFixed(0)).join(",")} after=${samples.after.map(v => v.toFixed(0)).join(",")}\n`);
|
||||
}
|
||||
const output = JSON.stringify({ baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8", cwd: root }).trim(),
|
||||
changedSources, version, settings, warmups: 2, timedSamplesPerArm: samplesPerArm, results }, null, 2);
|
||||
const out = process.argv.find(value => value.startsWith("--out="))?.slice(6);
|
||||
if (out) await fs.writeFile(out, output); else console.log(output);
|
||||
} finally {
|
||||
try { await captureClient?.end({ timeout: 0 }); await db?.$client.end({ timeout: 0 }); await database?.cleanup(); await fs.rm(baselineDir, { recursive: true, force: true }); }
|
||||
finally { clearInterval(keepAlive); }
|
||||
}
|
||||
@@ -781,6 +781,61 @@ describe("report-only ownership reconciliation", () => {
|
||||
.toEqual([lostNative.commentId, lostContext.commentId, earlyAck.commentId, logOnly.commentId, otherRunAck.commentId]);
|
||||
});
|
||||
|
||||
// The stale-continuation check reads a task's runs through the same two
|
||||
// bindings. The closing run and a live run must each count through either
|
||||
// binding alone, and only in the comment's company and for the assignee.
|
||||
it("binds each task run by its legacy context or its native column in the stale-continuation check", async () => {
|
||||
const otherCompanyId = randomUUID(), otherAgentId = randomUUID(), peerAgentId = randomUUID();
|
||||
await db.insert(companies).values({ id: otherCompanyId, name: "Other company", issuePrefix: "OTH" });
|
||||
await db.insert(agents).values([
|
||||
{ id: otherAgentId, companyId: otherCompanyId, name: "Other", role: "engineer", adapterType: "codex_local" },
|
||||
{ id: peerAgentId, companyId, name: "Peer", role: "engineer", adapterType: "codex_local" }]);
|
||||
type Run = { binding: "context" | "native"; status?: string; startedAt: Date; finishedAt?: Date | null; companyId?: string; agentId?: string };
|
||||
const task = async (title: string, runs: Run[], commentAt: Date, options: { closedBy?: string; wake?: "commentId" | "list" | "open" } = {}) => {
|
||||
const issueId = randomUUID();
|
||||
await db.insert(issues).values({ id: issueId, companyId, title, status: "done", assigneeAgentId: agentId });
|
||||
for (const run of runs) {
|
||||
const runCompanyId = run.companyId ?? companyId;
|
||||
await db.insert(heartbeatRuns).values({ companyId: runCompanyId, agentId: run.agentId ?? (runCompanyId === companyId ? agentId : otherAgentId),
|
||||
status: run.status ?? "succeeded", invocationSource: "manual",
|
||||
contextSnapshot: run.binding === "native" ? {} : { issueId }, nativeIssueId: run.binding === "context" ? null : issueId,
|
||||
startedAt: run.startedAt, createdAt: run.startedAt, finishedAt: run.finishedAt === undefined ? minutesAgo(2) : run.finishedAt });
|
||||
}
|
||||
const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", body: title,
|
||||
createdAt: commentAt }).returning();
|
||||
const commentId = comment!.id;
|
||||
const payload = options.wake === "list" ? { issueId, _paperclipWakeContext: { wakeCommentIds: [commentId] } } : { issueId, commentId };
|
||||
await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "cancelled",
|
||||
error: "continuation_task_ownership_changed", payload });
|
||||
if (options.wake === "open") await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation",
|
||||
status: "deferred_issue_execution", payload: { issueId, commentId } });
|
||||
await db.insert(activityLog).values({ companyId, actorType: options.closedBy ?? "agent", actorId: "actor", action: "issue.updated",
|
||||
entityType: "issue", entityId: issueId, details: { toStatus: "done" }, createdAt: minutesAgo(1) });
|
||||
return commentId;
|
||||
};
|
||||
const closing = (binding: Run["binding"], extra: Partial<Run> = {}): Run => ({ binding, startedAt: minutesAgo(10), ...extra });
|
||||
const nativeClosing = await task("closed by a native run", [closing("native")], minutesAgo(8));
|
||||
const contextClosing = await task("closed by a legacy run", [closing("context")], minutesAgo(7));
|
||||
const listWake = await task("carried by a wake list", [closing("native")], minutesAgo(6), { wake: "list" });
|
||||
await task("no closing run", [], minutesAgo(8));
|
||||
await task("closing run in another company", [closing("native", { companyId: otherCompanyId }), closing("context", { companyId: otherCompanyId })], minutesAgo(8));
|
||||
await task("closing run of another agent", [closing("native", { agentId: peerAgentId }), closing("context", { agentId: peerAgentId })], minutesAgo(8));
|
||||
await task("closing run failed", [closing("native", { status: "failed" })], minutesAgo(8));
|
||||
await task("closing run ended before the comment", [closing("context", { finishedAt: minutesAgo(9) })], minutesAgo(8));
|
||||
await task("closing run started after the comment", [closing("native", { startedAt: minutesAgo(7) })], minutesAgo(8));
|
||||
await task("native live run", [closing("context"), closing("native", { status: "running", startedAt: minutesAgo(3), finishedAt: null })], minutesAgo(8));
|
||||
await task("legacy live run", [closing("native"), closing("context", { status: "queued", startedAt: minutesAgo(3), finishedAt: null })], minutesAgo(8));
|
||||
const otherCompanyLive = await task("live run in another company only", [closing("native"),
|
||||
closing("context", { status: "running", companyId: otherCompanyId, finishedAt: null })], minutesAgo(5));
|
||||
await task("closed by a person", [closing("native")], minutesAgo(8), { closedBy: "user" });
|
||||
await task("an open wake carries it", [closing("context")], minutesAgo(8), { wake: "open" });
|
||||
const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
|
||||
const dropped = report.violations.filter(row => row.reason === "user_comment_dropped_by_stale_continuation");
|
||||
// Oldest first, as the check orders them. The other company's live run does not hold the last one.
|
||||
expect(dropped.map(row => row.recordId)).toEqual([nativeClosing, contextClosing, listWake, otherCompanyLive]);
|
||||
expect(dropped.every(row => row.companyId === companyId)).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => {
|
||||
const updatedAt = minutesAgo(1);
|
||||
await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution",
|
||||
|
||||
@@ -413,6 +413,20 @@ const COMMENT_WAKE_GRACE_MS = 30_000;
|
||||
|
||||
export const USER_COMMENT_DROPPED_BY_STALE_RUN = "user_comment_dropped_by_stale_continuation";
|
||||
|
||||
/** Whether a run of comment `c`'s task matches `condition`. A run belongs to
|
||||
* the task by its legacy context or its native column. Each binding is a
|
||||
* separate indexed probe per comment: one OR across both lets PostgreSQL hash
|
||||
* every run of the company and read each run's context. `offset 0` keeps each
|
||||
* probe a per-row index lookup; without it PostgreSQL can also build a hashed
|
||||
* subplan or a hash join over all runs of the company. */
|
||||
function taskRuns(alias: string, condition: SQL) {
|
||||
const run = sql.raw(alias);
|
||||
return sql`(exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
|
||||
and ${run}.context_snapshot->>'issueId' = c.issue_id::text and ${condition} offset 0)
|
||||
or exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
|
||||
and ${run}.native_issue_id = c.issue_id and ${condition} offset 0))`;
|
||||
}
|
||||
|
||||
/** Invariant 4, the durable half of the stale-continuation rule (chaos
|
||||
* s9401/t72). A user comment on a done task whose only carrying wakes were
|
||||
* cancelled as stale continuations, that a settled agent run which started
|
||||
@@ -421,32 +435,36 @@ export const USER_COMMENT_DROPPED_BY_STALE_RUN = "user_comment_dropped_by_stale_
|
||||
* also covers a stale check that ran before that run reached succeeded, or a
|
||||
* save that failed. The worker saves it so the release drain reopens the
|
||||
* task and delivers it once. */
|
||||
async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Promise<InvariantViolation[]> {
|
||||
export async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Promise<InvariantViolation[]> {
|
||||
const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString();
|
||||
// Two stages. The first keeps the comments that a stale-continuation
|
||||
// cancellation dropped and that no open wake carries. These are rare, so the
|
||||
// run probes (taskRuns) read only these few comments. `offset 0` keeps the
|
||||
// stale-wake lookup a per-comment index probe; without it PostgreSQL can
|
||||
// hash every wake request of the company and read each payload.
|
||||
const rows = await db.execute<{ id: string; company_id: string; issue_id: string; agent_id: string }>(sql`
|
||||
select c.id, c.company_id, c.issue_id, i.assignee_agent_id as agent_id from issue_comments c
|
||||
join issues i on i.id = c.issue_id and i.company_id = c.company_id
|
||||
where c.author_user_id is not null and c.deleted_at is null
|
||||
and c.author_agent_id is null and c.created_by_run_id is null and c.created_at >= ${since}::timestamptz
|
||||
and i.status = 'done' and i.assignee_agent_id is not null and i.conversation_agent_id is null
|
||||
and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}
|
||||
and exists (select 1 from agent_wakeup_requests stale where stale.company_id = c.company_id
|
||||
and stale.payload->>'issueId' = c.issue_id::text and stale.status = 'cancelled'
|
||||
and stale.error = 'continuation_task_ownership_changed'
|
||||
and (stale.payload->>'commentId' = c.id::text
|
||||
or coalesce(stale.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
|
||||
and not exists (select 1 from agent_wakeup_requests open_wake where open_wake.company_id = c.company_id
|
||||
and open_wake.payload->>'issueId' = c.issue_id::text
|
||||
and open_wake.status not in ('cancelled', 'skipped', 'failed', 'coalesced')
|
||||
and (open_wake.payload->>'commentId' = c.id::text
|
||||
or coalesce(open_wake.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
|
||||
and exists (select 1 from heartbeat_runs closing where closing.company_id = c.company_id
|
||||
and closing.agent_id = i.assignee_agent_id and closing.status = 'succeeded'
|
||||
and (closing.context_snapshot->>'issueId' = c.issue_id::text or closing.native_issue_id = c.issue_id)
|
||||
and closing.started_at < c.created_at and closing.finished_at >= c.created_at)
|
||||
and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id
|
||||
and (r.context_snapshot->>'issueId' = c.issue_id::text or r.native_issue_id = c.issue_id)
|
||||
and r.status in ('queued', 'running', 'scheduled_retry'))
|
||||
with dropped as materialized (
|
||||
select c.id, c.company_id, c.issue_id, c.created_at, i.assignee_agent_id as agent_id from issue_comments c
|
||||
join issues i on i.id = c.issue_id and i.company_id = c.company_id
|
||||
where c.author_user_id is not null and c.deleted_at is null
|
||||
and c.author_agent_id is null and c.created_by_run_id is null and c.created_at >= ${since}::timestamptz
|
||||
and i.status = 'done' and i.assignee_agent_id is not null and i.conversation_agent_id is null
|
||||
and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}
|
||||
and exists (select 1 from agent_wakeup_requests stale where stale.company_id = c.company_id
|
||||
and stale.payload->>'issueId' = c.issue_id::text and stale.status = 'cancelled'
|
||||
and stale.error = 'continuation_task_ownership_changed'
|
||||
and (stale.payload->>'commentId' = c.id::text
|
||||
or coalesce(stale.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text) offset 0)
|
||||
and not exists (select 1 from agent_wakeup_requests open_wake where open_wake.company_id = c.company_id
|
||||
and open_wake.payload->>'issueId' = c.issue_id::text
|
||||
and open_wake.status not in ('cancelled', 'skipped', 'failed', 'coalesced')
|
||||
and (open_wake.payload->>'commentId' = c.id::text
|
||||
or coalesce(open_wake.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
|
||||
)
|
||||
select c.id, c.company_id, c.issue_id, c.agent_id from dropped c
|
||||
where ${taskRuns("closing", sql`closing.agent_id = c.agent_id and closing.status = 'succeeded'
|
||||
and closing.started_at < c.created_at and closing.finished_at >= c.created_at`)}
|
||||
and not ${taskRuns("r", sql`r.status in ('queued', 'running', 'scheduled_retry')`)}
|
||||
order by c.created_at limit ${Math.min(limit, MAX_RECORDS_PER_KIND)}`);
|
||||
// Consumption is per comment: a later run counts only if it received this
|
||||
// comment (its own wake input, or a retry's task history), never ordinary
|
||||
@@ -469,19 +487,9 @@ async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Prom
|
||||
* The reconciliation worker saves it as a deferred receipt and promotes it. */
|
||||
export async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
|
||||
const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString();
|
||||
// A run belongs to the task by its legacy context or its native column. Each
|
||||
// binding is a separate indexed probe per comment: one OR across both lets
|
||||
// PostgreSQL hash every run of the company and read each run's context.
|
||||
// `offset 0` keeps each probe, and the steer check's event lookup, a
|
||||
// per-row index lookup; without it PostgreSQL can also build a hashed
|
||||
// subplan or a hash join over all runs or all events of the company.
|
||||
const taskRuns = (alias: string, condition: SQL) => {
|
||||
const run = sql.raw(alias);
|
||||
return sql`(exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
|
||||
and ${run}.context_snapshot->>'issueId' = c.issue_id::text and ${condition} offset 0)
|
||||
or exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
|
||||
and ${run}.native_issue_id = c.issue_id and ${condition} offset 0))`;
|
||||
};
|
||||
// Run lookups are taskRuns probes. `offset 0` also keeps the steer check's
|
||||
// event lookup a per-row index lookup; without it PostgreSQL can build a
|
||||
// hash join over all events of the company.
|
||||
// Three stages, cheapest first. Most user comments have a wake (stage one)
|
||||
// or a later run (stage two). Only the few comments left reach the steer
|
||||
// check, which reads run event payloads.
|
||||
|
||||
@@ -2,6 +2,7 @@ import { sql } from "drizzle-orm";
|
||||
import { issueRecoveryActions } from "@paperclipai/db";
|
||||
import { isExhaustedStopProofWait, STOP_PROOF_MAX_REARMS, STOP_PROOF_RECOVERY_KEY, STOP_PROOF_RECOVERY_SCHEMA, STOP_PROOF_WAIT_CODE,
|
||||
stopProofRearmsSpent, type SystemRecoveryState } from "./system-recovery-state.js";
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
|
||||
/** Source refusals that repeat on every retry (see workspace-recovery-fallback). */
|
||||
export const UNVERIFIABLE_SOURCE_CODES = ["allocation_ownership_ambiguous", "source_binding_invalid", "source_unavailable", "historical_source_missing"] as const;
|
||||
@@ -48,6 +49,12 @@ export function retainedHoldFallbackEligible(evidence: Record<string, unknown> |
|
||||
return isExhaustedStopProofWait(state) && stopProofRearmsSpent(evidence);
|
||||
}
|
||||
|
||||
// The source run's id as a uuid, or null when the saved text is not the
|
||||
// canonical (lowercase) form. `run.id = sourceRunId` matches exactly the rows
|
||||
// `run.id::text = evidence->>'runId'` matched, and it uses the run's primary
|
||||
// key. The text comparison cannot, so each probe read every run (or
|
||||
// finalizer, or lease) of the company.
|
||||
const sourceRunId = canonicalExecutionUuid(sql`${issueRecoveryActions.evidence}->>'runId'`);
|
||||
const stopProofMarker = sql`${issueRecoveryActions.evidence}->${STOP_PROOF_RECOVERY_KEY}`;
|
||||
const stopProofRearms = sql`case when ${stopProofMarker}->>'schema' = ${STOP_PROOF_RECOVERY_SCHEMA}
|
||||
and jsonb_typeof(${stopProofMarker}->'rearms') = 'number' then (${stopProofMarker}->>'rearms')::int else 0 end`;
|
||||
@@ -64,7 +71,7 @@ export function stopProofRearmPendingCondition() {
|
||||
return sql`(${issueRecoveryActions.status} in ('active', 'escalated')
|
||||
and ${exhaustedStopProofWait}
|
||||
and not exists (select 1 from native_run_finalizations finalizer
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id::text = ${issueRecoveryActions.evidence}->>'runId'
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id = ${sourceRunId}
|
||||
and finalizer.lease_owner is not null and coalesce(finalizer.lease_expires_at <= now(), true))
|
||||
and ${issueRecoveryActions.evidence}->'retainedSourceFallback' is null
|
||||
and ${issueRecoveryActions.evidence}->'executionReconciliation' is null
|
||||
@@ -78,9 +85,9 @@ export function stopProofRearmDueAt() {
|
||||
return sql`greatest(case when ${stopProofMarker}->>'schema' = ${STOP_PROOF_RECOVERY_SCHEMA}
|
||||
and jsonb_typeof(${stopProofMarker}->'nextRearmAt') = 'number' then (${stopProofMarker}->>'nextRearmAt')::bigint else 0 end,
|
||||
coalesce((select (extract(epoch from source.controller_lease_expires_at) * 1000)::bigint from heartbeat_runs source
|
||||
where source.company_id = ${issueRecoveryActions.companyId} and source.id::text = ${issueRecoveryActions.evidence}->>'runId'), 0),
|
||||
where source.company_id = ${issueRecoveryActions.companyId} and source.id = ${sourceRunId}), 0),
|
||||
coalesce((select max((extract(epoch from finalizer.lease_expires_at) * 1000)::bigint) from native_run_finalizations finalizer
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id::text = ${issueRecoveryActions.evidence}->>'runId'
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id = ${sourceRunId}
|
||||
and finalizer.lease_owner is not null), 0))`;
|
||||
}
|
||||
|
||||
@@ -95,7 +102,7 @@ const check = sql`${issueRecoveryActions.evidence}->${RETAINED_SOURCE_FALLBACK_C
|
||||
// hold once, even after its bounded checks are spent.
|
||||
const leaseChangedSinceCheck = sql`(jsonb_typeof(${check}->'checkedAt') = 'number' and exists (
|
||||
select 1 from environment_leases lease where lease.company_id = ${issueRecoveryActions.companyId}
|
||||
and lease.heartbeat_run_id::text = ${issueRecoveryActions.evidence}->>'runId'
|
||||
and lease.heartbeat_run_id = ${sourceRunId}
|
||||
and lease.updated_at > to_timestamp((${check}->>'checkedAt')::bigint / 1000.0)))`;
|
||||
|
||||
/**
|
||||
@@ -114,11 +121,11 @@ export function retainedSourceFallbackPendingCondition() {
|
||||
and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,code}' in (${list(UNVERIFIABLE_SOURCE_CODES)})
|
||||
and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,retryClass}' in (${list(PERMANENT_SOURCE_RETRY_CLASSES)})
|
||||
and exists (select 1 from heartbeat_runs source where source.company_id = ${issueRecoveryActions.companyId}
|
||||
and source.id::text = ${issueRecoveryActions.evidence}->>'runId'
|
||||
and source.id = ${sourceRunId}
|
||||
and source.runner_profile_json#>>'{remoteWorkspace,state}' = 'prepared'))
|
||||
or (${retainedFileHoldCondition()} and ${exhaustedStopProofWait} and ${stopProofRearms} >= ${STOP_PROOF_MAX_REARMS}
|
||||
and exists (select 1 from heartbeat_runs source where source.company_id = ${issueRecoveryActions.companyId}
|
||||
and source.id::text = ${issueRecoveryActions.evidence}->>'runId'
|
||||
and source.id = ${sourceRunId}
|
||||
and source.runner_profile_json#>>'{remoteWorkspace,state}' = 'prepared'))
|
||||
or (${issueRecoveryActions.status} in ('active', 'escalated') and exists (select 1 from issue_recovery_actions sibling
|
||||
where sibling.company_id = ${issueRecoveryActions.companyId} and sibling.source_issue_id = ${issueRecoveryActions.sourceIssueId}
|
||||
|
||||
@@ -395,6 +395,24 @@ describe("retained source fallback", () => {
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked");
|
||||
}, 60_000);
|
||||
|
||||
// The discovery probes find the source run by its primary key. They must
|
||||
// still match only the canonical id text the hold saved, as the text
|
||||
// comparison they replace did: a hold whose saved run id is not canonical
|
||||
// is not discovered through them.
|
||||
it("discovers a hold through its source run only by the canonical run id it saved", async () => {
|
||||
const f = await unverifiableRetainedSource();
|
||||
const ids = async () => (await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id);
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeLessThanOrEqual(Date.now());
|
||||
expect(await ids()).toContain(f.actionId);
|
||||
const saved = await action(f.actionId);
|
||||
await db.update(issueRecoveryActions).set({ evidence: { ...saved.evidence, runId: f.runId.toUpperCase() } })
|
||||
.where(eq(issueRecoveryActions.id, f.actionId));
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeNull();
|
||||
expect(await ids()).not.toContain(f.actionId);
|
||||
await db.update(issueRecoveryActions).set({ evidence: saved.evidence }).where(eq(issueRecoveryActions.id, f.actionId));
|
||||
expect(await ids()).toContain(f.actionId);
|
||||
}, 60_000);
|
||||
|
||||
it("waits until a deferred transient check is due before the worker visits the hold again", async () => {
|
||||
const f = await unverifiableRetainedSource();
|
||||
const source = await run(f.runId);
|
||||
|
||||
Reference in new issue
Block a user