From bd64eb27b060cd54d3c3b8cbbee06aadfe0ae024 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:14:58 -0500 Subject: [PATCH] perf(recovery): use indexed execution identity lookups (#16291) ## Thinking Path > - Paperclip manages AI agents and their execution. > - Recovery must keep old execution ownership visible before new work starts. > - Ownership reads filter large run contexts through a combined text expression. > - Recovery and status delivery also cast indexed UUID columns to text. > - These reads can scan history before finding the requested execution. > - This pull request preserves identity and ownership rules while using existing indexes. > - The benefit is less database work during recovery and task polling. ## Linked Issues or Issue Description Refs #14406. Related community report: #15275. This is an independent implementation. It does not use that pull request's code or commits. It covers the reported execution joins and the additional ownership query hotspot. **What happened?** Accumulated staging query statistics showed 11,782 ownership reads at 51 ms each. A paired recovery-action read averaged 22 ms. Both queried large run JSON. Text casts also prevented indexed UUID lookups in recovery and status delivery. **What did you expect to happen?** Recovery must find the same execution evidence through existing indexes. Malformed identifiers must match nothing without a cast error. Native issue bindings must retain precedence. Missing termination evidence must continue to block new work. **Steps to reproduce** Run the disposable benchmark below. It loads the original service modules from the baseline revision. It compares their exact ownership, blocker action and status-delivery SQL with the current modules on 1,400 synthetic runs, 350 historical recovery actions and 192 KiB contexts. ## What Changed - Convert saved text to UUID only after checking its canonical lowercase form. - Keep indexed run and task UUID columns unchanged in execution joins. - Split native and legacy issue predicates so existing indexes can filter ownership candidates. - Preserve company scope, native precedence, process and lease holds, recovery eligibility and delivery receipts. - Add real PostgreSQL identity and ownership regressions, a reproducible benchmark and database documentation. ## Verification - Local service SQL benchmark: ownership 78.30 ms to 1.14 ms, with the same two complete rows. Blocker action read 76.94 ms to 22.55 ms, with the same complete row. These are synthetic SQL timings, excluding process probes, provider calls and HTTP. - Status delivery returned the same 100 rows. Timing was neutral on the small-task fixture: 7.85 ms before and 8.30 ms after. No speedup is claimed for that case. - 231 local tests passed across six server suites. They cover malformed identifiers, canonical spelling, company scope, native precedence, retained processes, workspace restore, native replacement, conversations and delivery. - `pnpm -r typecheck` and `pnpm build` passed. - `git diff --check` passed. - All current-head CI and security checks passed, including the full general and serialized suites, browser shards, Runner verification and build. Greptile reviewed the exact commit at 5/5 with no actionable findings. - The older full local sweep in the companion worktree was stopped after final-commit complete CI passed. It is not claimed as a completed local pass. The 231 focused tests above passed locally on this branch. ```sh pnpm --filter @paperclipai/server exec vitest run src/services/conversation-continuation.test.ts src/services/legacy-execution-recovery.test.ts src/__tests__/legacy-workspace-restore-recovery.test.ts src/__tests__/chat-completion-delivery.test.ts src/__tests__/agent-conversations.test.ts src/services/native-runtime/native-safe-replacement.test.ts pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-recovery-queries.ts --baseline-ref=1e731520bb ``` ## Risks - PostgreSQL accepts more UUID spellings than the old text joins. The explicit guard preserves the old lowercase spelling requirement. - Query plans depend on the data distribution. Small task sets can show neutral status-delivery timing. The benchmark is not a production latency claim. - Recovery ownership checks remain conservative. Process, lease, workspace and no-replay rules are unchanged. - No migration, new index, cache, cleanup change or lock removal is included. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository inspection, shell tools and local code execution. The runtime does not expose a more specific backend model ID or context-window size. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/DATABASE.md | 18 +++ scripts/benchmark-recovery-queries.ts | 106 ++++++++++++++++++ .../conversation-continuation.test.ts | 81 +++++++++++++ .../src/services/conversation-continuation.ts | 15 ++- server/src/services/execution-identity.ts | 6 + .../services/execution-recovery-resolution.ts | 8 +- .../src/services/execution-status-delivery.ts | 13 +-- 7 files changed, 233 insertions(+), 14 deletions(-) create mode 100644 scripts/benchmark-recovery-queries.ts create mode 100644 server/src/services/conversation-continuation.test.ts create mode 100644 server/src/services/execution-identity.ts 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))