diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 2bd6e82bc78..4ae992bc87e 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -897,3 +897,21 @@ reservation per attempt. Failed downloads and package fallback keep the full reservation. Successfully linked small imports refund unused measured work. Completed bindings do not reset company credits; waiting jobs retry on refill. The company_skill_import_budgets table stores no credentials or source contents. + +### Execution recovery query performance + +Conversation recovery keeps indexed run and task UUID columns as UUIDs. Saved +JSON identifiers pass a canonical lowercase UUID guard before conversion. +Malformed, uppercase, numeric and cross-company values retain their old behavior. +Ownership candidate reads split native and legacy issue bindings so existing +indexes can filter candidates before reading large contexts. Native bindings +continue to override legacy context IDs. No new index or migration is needed. + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-recovery-queries.ts --baseline-ref=1e731520bb +``` + +This disposable fixture compares the real ownership, recovery-action and status +delivery queries, including complete positional results and database plans. +All data is synthetic. Query timings exclude process liveness probes, HTTP and +provider work. diff --git a/scripts/benchmark-recovery-queries.ts b/scripts/benchmark-recovery-queries.ts new file mode 100644 index 00000000000..f48f8e53552 --- /dev/null +++ b/scripts/benchmark-recovery-queries.ts @@ -0,0 +1,106 @@ +import assert from "node:assert/strict"; +import { randomBytes, randomUUID } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { companies, agents, issues, heartbeatRuns, issueRecoveryActions, createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import * as conversation from "../server/src/services/conversation-continuation.js"; +import * as blocker from "../server/src/services/execution-blocker.js"; +import * as delivery from "../server/src/services/execution-status-delivery.js"; +type Query = { sql: string; params: unknown[] }; +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15) ?? "origin/master"; +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim(); +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let baselineDir: string | undefined; +let db: Db | undefined; +try { + database = await startEmbeddedPostgresTestDatabase("recovery-query-benchmark-"); + db = createDb(database.connectionString); + baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".recovery-baseline-")); + async function loadBaseline(name: string) { + const modulePath = `server/src/services/${name}.ts`; + const original = new URL(`../${modulePath}`, import.meta.url); + const source = execFileSync("git", ["show", `${baselineRevision}:${modulePath}`], { encoding: "utf8" }); + const rewritten = source.replace(/from "(\.{1,2}\/[^\"]+)"/g, + (_match, specifier: string) => `from ${JSON.stringify(name === "execution-blocker" && specifier === "./conversation-continuation.js" ? path.join(baselineDir!, "conversation-continuation.ts") : fileURLToPath(new URL(specifier, original)))}`); + const target = path.join(baselineDir!, `${name}.ts`); + await fs.writeFile(target, rewritten); + return import(pathToFileURL(target).href); + } + const oldConversation = await loadBaseline("conversation-continuation") as typeof conversation; + const oldBlocker = await loadBaseline("execution-blocker") as typeof blocker; + const oldDelivery = await loadBaseline("execution-status-delivery") as typeof delivery; + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), otherIssueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Synthetic recovery", issuePrefix: "BENCH" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "codex_local" }); + await db.insert(issues).values([{ id: issueId, companyId, title: "Target" }, { id: otherIssueId, companyId, title: "Other" }]); + const prompt = randomBytes(144 * 1024).toString("base64"); + await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, runtime_mode, native_issue_id, created_at, + context_snapshot, runner_profile_json, process_pid, error_code, execution_status_delivery_id) + SELECT $1::uuid, $2::uuid, 'failed', 'legacy', case when n = 2 then $3::uuid when n = 3 then $4::uuid end, + now() + n * interval '1 millisecond', jsonb_build_object('issueId', case when n in (1,3) then $3::text else $4::text end, 'prompt', $5::text), + '{"adapterDispatch":{"adapterType":"codex_local"}}'::jsonb, 999999999, 'process_lost', + case when n <= 100 then gen_random_uuid() end + FROM generate_series(1, 1400) n`, [companyId, agentId, issueId, otherIssueId, prompt]); + const runs = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).orderBy(heartbeatRuns.createdAt); + await db.insert(issueRecoveryActions).values(runs.slice(0, 350).map(run => ({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "board", cause: "legacy_execution_requires_reconciliation", + fingerprint: run.id, evidence: { runId: run.id, automaticRecovery: { replay: "blocked" }, workspaceRestoreFailure: "restore_unsafe_archive" }, updatedAt: new Date(Date.UTC(2026, 0, 1) + runs.indexOf(run) * 1000), nextAction: "Fixture" }))); + await db.$client.unsafe("ANALYZE"); + async function capture(invoke: (captureDb: Db) => Promise, actionQuery = false): Promise { + let query: Query | undefined; + const stop = new Error("captured"); + const captureDb = new Proxy(db!, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const selection = from(...args); + selection.execute = async () => { + const candidate = selection.toSQL(); + if (actionQuery && !candidate.sql.includes('from "issue_recovery_actions"')) return []; + query = candidate; throw stop; + }; + return selection; + }) as typeof builder.from; + return builder; + }; + }}); + try { await invoke(captureDb); } catch (error) { if (error !== stop) throw error; } + assert.ok(query); + return query; + } + async function measure(query: Query) { + const samples: number[] = []; let rows: unknown[] = []; + for (let n = 0; n < 7; n++) { + const start = performance.now(); + rows = await db!.$client.unsafe(query.sql, query.params as never[]).values(); + if (n > 1) samples.push(performance.now() - start); + } + const [plan] = await db!.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + samples.sort((a, b) => a - b); + return { medianMs: samples[2], rows, plan: plan["QUERY PLAN"] }; + } + const pairs = [ + { name: "ownership", before: (d: Db) => oldConversation.getConversationOwnershipBlocker(d, companyId, issueId), + after: (d: Db) => conversation.getConversationOwnershipBlocker(d, companyId, issueId) }, + { name: "recoveryActions", before: (d: Db) => oldBlocker.getExecutionBlocker(d, companyId, issueId), + after: (d: Db) => blocker.getExecutionBlocker(d, companyId, issueId) }, + { name: "statusDelivery", before: (d: Db) => oldDelivery.deliverExecutionStatuses(d), after: (d: Db) => delivery.deliverExecutionStatuses(d) }, + ]; + const results = []; + for (const pair of pairs) { + const before = await measure(await capture(pair.before, pair.name === "recoveryActions")), after = await measure(await capture(pair.after, pair.name === "recoveryActions")); + const normalize = (rows: unknown[]) => rows.map(row => JSON.stringify(row)).sort(); + assert.deepEqual(normalize(after.rows), normalize(before.rows)); + results.push({ name: pair.name, equalRows: after.rows.length, baselineMs: before.medianMs, treatmentMs: after.medianMs, + speedup: before.medianMs / after.medianMs, baselinePlan: before.plan, treatmentPlan: after.plan }); + } + console.log(JSON.stringify({ baselineRevision, runs: 1400, actions: 350, promptBytes: Buffer.byteLength(prompt), results }, null, 2)); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); + if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/services/conversation-continuation.test.ts b/server/src/services/conversation-continuation.test.ts new file mode 100644 index 00000000000..31166bfa59f --- /dev/null +++ b/server/src/services/conversation-continuation.test.ts @@ -0,0 +1,81 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { agents, companies, createDb, heartbeatRuns, issues, issueRecoveryActions, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js"; +import { deliverExecutionStatuses } from "./execution-status-delivery.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describePostgres = support.supported ? describe : describe.skip; +if (!support.supported) console.warn(`Skipping execution identity regressions: ${support.reason}`); + +describePostgres("execution identity query contracts", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("execution-identity-"); db = createDb(database.connectionString); }, 20_000); + afterAll(async () => { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); }); + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), otherIssueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Identity", issuePrefix: `I${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "codex_local" }); + await db.insert(issues).values([{ id: issueId, companyId, title: "Target" }, { id: otherIssueId, companyId, title: "Other" }]); + return { companyId, agentId, issueId, otherIssueId }; + } + async function run(f: Awaited>, context: Record | null, nativeIssueId: string | null = null, pid: number | null = null) { + const [row] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, nativeIssueId, + status: "interrupted", runtimeMode: "legacy", contextSnapshot: context, processPid: pid, + runnerProfileJson: { adapterDispatch: { adapterType: "codex_local" } } }).returning(); + return row; + } + it("matches only canonical evidence IDs and keeps source issue and company scope", async () => { + const f = await fixture(), foreign = await fixture(); + const ours = await run(f, { issueId: f.issueId }); + const elsewhere = await run(f, { issueId: f.otherIssueId }); + const foreignRun = await run(foreign, { issueId: f.issueId }); + const ids: unknown[] = [ours.id, ours.id.toUpperCase(), ours.id.replaceAll("-", ""), `{${ours.id}}`, "bad", "", null, 42, {}, [ours.id], elsewhere.id, foreignRun.id]; + const actions = await db.insert(issueRecoveryActions).values(ids.map((runId, index) => ({ + companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", status: "resolved", ownerType: "board", + cause: "legacy_execution_requires_reconciliation", fingerprint: `fixture-${index}`, evidence: { runId }, nextAction: "Fixture", + }))).returning(); + const matches = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions) + .where(and(eq(issueRecoveryActions.companyId, f.companyId), conversationRecoveryActionPredicate())); + expect(matches).toEqual([{ id: actions[0].id }]); + }); + it("preserves native issue precedence and does not broaden legacy spellings", async () => { + const f = await fixture(); + await run(f, { issueId: f.issueId }, f.otherIssueId, process.pid); + const uppercase = await run(f, { issueId: f.issueId.toUpperCase() }, null, process.pid); + const malformed = await run(f, { issueId: "bad" }, null, process.pid); + expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toBeNull(); + const native = await run(f, { issueId: f.otherIssueId }, f.issueId, process.pid); + expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: native.id, cause: "execution_owner_active" }); + expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId.toUpperCase())).toMatchObject({ runId: uppercase.id }); + expect(await getConversationOwnershipBlocker(db, f.companyId, "bad")).toMatchObject({ runId: malformed.id }); + }); + it("retains legacy ownership holds and releases only after ownership ends", async () => { + const f = await fixture(); + const legacy = await run(f, { issueId: f.issueId }, null, process.pid); + expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: legacy.id }); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, legacy.id)); + expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); + it("delivers canonical same-company task IDs and keeps malformed IDs private", async () => { + const f = await fixture(), foreign = await fixture(); + const contexts: unknown[] = [f.issueId, f.issueId.toUpperCase(), "bad", `{${f.issueId}}`, f.issueId.replaceAll("-", ""), null, 42, {}, [f.issueId], foreign.issueId]; + const expected = new Map(); + for (const [index, issueId] of contexts.entries()) { + const row = await run(f, { issueId }); + await db.update(heartbeatRuns).set({ executionStatusDeliveryId: randomUUID() }).where(eq(heartbeatRuns.id, row.id)); + expected.set(row.id, index === 0 ? f.issueId : null); + } + const native = await run(f, { issueId: f.otherIssueId }, f.issueId); + await db.update(heartbeatRuns).set({ executionStatusDeliveryId: randomUUID() }).where(eq(heartbeatRuns.id, native.id)); + expected.set(native.id, f.issueId); + const seen = new Map(); + const result = await deliverExecutionStatuses(db, { publish: event => { seen.set(event.payload!.runId as string, event.payload!.issueId as string | null); return { ...event, id: 0, payload: event.payload ?? {}, createdAt: new Date().toISOString() }; } }); + expect(result.delivered).toBe(expected.size); + expect(seen).toEqual(expected); + expect(await deliverExecutionStatuses(db, { publish: () => { throw new Error("duplicate delivery"); } })).toEqual({ scanned: 0, delivered: 0 }); + }); +}); diff --git a/server/src/services/conversation-continuation.ts b/server/src/services/conversation-continuation.ts index 4730357c0ff..36e8b7a4804 100644 --- a/server/src/services/conversation-continuation.ts +++ b/server/src/services/conversation-continuation.ts @@ -1,6 +1,7 @@ import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; -import { and, desc, eq, inArray, isNotNull, or, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, type Db } from "@paperclipai/db"; +import { canonicalExecutionUuid } from "./execution-identity.js"; import { readProcessStartedAt } from "./hot-restart.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; @@ -70,8 +71,11 @@ export function conversationRecoveryActionPredicate() { sql`exists ( select 1 from ${heartbeatRuns} where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId} - and ${heartbeatRuns.id}::text = ${issueRecoveryActions.evidence}->>'runId' - and coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueRecoveryActions.sourceIssueId}::text + and ${heartbeatRuns.id} = ${canonicalExecutionUuid(sql`${issueRecoveryActions.evidence}->>'runId'`)} + and ${or( + eq(heartbeatRuns.nativeIssueId, issueRecoveryActions.sourceIssueId), + and(isNull(heartbeatRuns.nativeIssueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueRecoveryActions.sourceIssueId}::text`), + )} and ${heartbeatRuns.runtimeMode} = 'legacy' and coalesce(${heartbeatRuns.resultJson}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive' and coalesce(${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema', '') <> 'paperclip.workspace-restore-recovery.v1' @@ -111,7 +115,10 @@ export async function getConversationOwnershipBlocker(db: Db, companyId: string, .where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.runtimeMode, "legacy"), conversationRunPredicate(), - sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, + or( + eq(heartbeatRuns.nativeIssueId, canonicalExecutionUuid(sql`${issueId}::text`)), + and(isNull(heartbeatRuns.nativeIssueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`), + ), inArray(heartbeatRuns.status, ["failed", "timed_out", "interrupted", "cancelled"]), or(isNotNull(heartbeatRuns.processPid), isNotNull(heartbeatRuns.processGroupId), activeLease, sql`${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'`), diff --git a/server/src/services/execution-identity.ts b/server/src/services/execution-identity.ts new file mode 100644 index 00000000000..4c45680437c --- /dev/null +++ b/server/src/services/execution-identity.ts @@ -0,0 +1,6 @@ +import { sql, type SQL } from "drizzle-orm"; + +export function canonicalExecutionUuid(text: SQL): SQL { + return sql`case when ${text} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' + then (${text})::uuid end`; +} diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index 3edd84a3ad0..f5ea0dfc87e 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -1,5 +1,6 @@ import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; import { randomUUID } from "node:crypto"; +import { canonicalExecutionUuid } from "./execution-identity.js"; import { claimedAdapterType, conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js"; import { persistActivity } from "./activity-log.js"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; @@ -394,8 +395,11 @@ export async function settleUnrecoverableExecutions( heartbeatRuns, and( eq(heartbeatRuns.companyId, issueRecoveryActions.companyId), - sql`${heartbeatRuns.id}::text = ${issueRecoveryActions.evidence}->>'runId'`, - sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueRecoveryActions.sourceIssueId}::text`, + eq(heartbeatRuns.id, canonicalExecutionUuid(sql`${issueRecoveryActions.evidence}->>'runId'`)), + or( + eq(heartbeatRuns.nativeIssueId, issueRecoveryActions.sourceIssueId), + and(isNull(heartbeatRuns.nativeIssueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueRecoveryActions.sourceIssueId}::text`), + ), ), ) .leftJoin( diff --git a/server/src/services/execution-status-delivery.ts b/server/src/services/execution-status-delivery.ts index 053eb3f3897..d98c35d996e 100644 --- a/server/src/services/execution-status-delivery.ts +++ b/server/src/services/execution-status-delivery.ts @@ -1,5 +1,6 @@ import { and, eq, isNotNull, sql } from "drizzle-orm"; import { heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { canonicalExecutionUuid } from "./execution-identity.js"; import { publishLiveEvent } from "./live-events.js"; import { logger } from "../middleware/logger.js"; @@ -23,18 +24,14 @@ export async function deliverExecutionStatuses( executionStatusDeliveryId: heartbeatRuns.executionStatusDeliveryId, }) .from(heartbeatRuns) - // Match terminalization's native-first association, but expose only an - // existing same-company canonical task ID. Compare text so malformed - // context values cannot cause a UUID cast error or escape onto the wire. .leftJoin( issues, and( eq(issues.companyId, heartbeatRuns.companyId), - sql`${issues.id}::text = coalesce( - ${heartbeatRuns.nativeIssueId}::text, - case when jsonb_typeof(${heartbeatRuns.contextSnapshot} -> 'issueId') = 'string' - then ${heartbeatRuns.contextSnapshot} ->> 'issueId' end - )`, + eq(issues.id, sql`coalesce( + ${heartbeatRuns.nativeIssueId}, + ${canonicalExecutionUuid(sql`${heartbeatRuns.contextSnapshot}->>'issueId'`)} + )`), ), ) .where(isNotNull(heartbeatRuns.executionStatusDeliveryId))