diff --git a/doc/DATABASE.md b/doc/DATABASE.md index d0a5993faa9..e2ac1e50d99 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -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 diff --git a/scripts/benchmark-recovery-status-checks.ts b/scripts/benchmark-recovery-status-checks.ts new file mode 100644 index 00000000000..e768952807a --- /dev/null +++ b/scripts/benchmark-recovery-status-checks.ts @@ -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= + * + * Options: --fixture= (repeatable), --samples=, --skip-hash-plan, --out=. + */ +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="); +// 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; +const load = async (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(changedSources[0]!); +const baselineDiscovery = await load(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(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> | 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): 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`${discovery.retainedSourceFallbackPendingCondition()}`, + fallbackDueAt: sql`(${discovery.retainedSourceFallbackDueAt()})::text`, + fallbackDue: sql`${discovery.retainedSourceFallbackDueCondition(now)}`, + rearmPending: sql`${discovery.stopProofRearmPendingCondition()}`, + rearmDueAt: sql`(${discovery.stopProofRearmDueAt()})::text`, + rearmDue: sql`${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>; + let proof: Record; + 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 = {}; + const buffers: Record = {}; + 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 }[])[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 = {}; + 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); } +} diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 1ae2fd4dbd5..b999c1c57af 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -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 => ({ 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", diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index d9daa7185d9..c5d3d4b357f 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -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 { +export async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Promise { 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. diff --git a/server/src/services/workspace-recovery-fallback-discovery.ts b/server/src/services/workspace-recovery-fallback-discovery.ts index 5c679e584f3..6cba725d77b 100644 --- a/server/src/services/workspace-recovery-fallback-discovery.ts +++ b/server/src/services/workspace-recovery-fallback-discovery.ts @@ -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 | 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} diff --git a/server/src/services/workspace-recovery-fallback.test.ts b/server/src/services/workspace-recovery-fallback.test.ts index 82e18edbe48..2d440101d72 100644 --- a/server/src/services/workspace-recovery-fallback.test.ts +++ b/server/src/services/workspace-recovery-fallback.test.ts @@ -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);