From 82c998233c10ca23f88e0b4ed7dbceb74feb2684 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 25 Oct 2026 11:06:22 -0700 Subject: [PATCH] perf(recovery): omit unused launch payloads from reconciliation reads (#17645) Project unused launch context and native execution input out of two preliminary recovery reads. Preserve full results, unknown controls, complete legacy profiles, current filters, native rejection and full locked terminalization. Apex 5/5 with no unresolved comments and all CI checks green. Ten combined local suites pass (370 tests), plus workspace typecheck and server build. Hosted traffic effects require staging verification. Co-Authored-By: Paperclip --- doc/DATABASE.md | 6 + .../src/services/legacy-execution-recovery.ts | 20 +++- .../recovery/reconciliation-run-read.test.ts | 107 ++++++++++++++++++ server/src/services/recovery/service.ts | 5 +- 4 files changed, 135 insertions(+), 3 deletions(-) create mode 100644 server/src/services/recovery/reconciliation-run-read.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 51f565e8bed..961d532aef9 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -66,6 +66,12 @@ the PostgreSQL row version. Native blocked-wait repair compares that version against a fresh full source read before applying changes. Any intervening row update, including an edit outside the projection, defers repair to a fresh sweep. +The two preliminary reconciliation reads omit named launch renderings and +preparation payloads. Native profiles omit unused native execution input. +They retain all other context, full results, legacy profiles, task identity, +retry accounting, and unknown controls. Native runs still reach their native +recovery guard. Terminalization keeps its full locked source and stored input. + Execution-blocker reads select process, lifecycle, and task-binding fields rather than full execution rows. Context retains typed task and pre-dispatch evidence; Runner profiles retain the adapter type, native input schema, binding, provider, diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 5bbea0e61a5..3561bff8768 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -8,7 +8,7 @@ import { normalizeMaxTurnStopReason } from "./heartbeat-stop-metadata.js"; import { claimedAdapterType, hasConversationContinuationPolicy } from "./conversation-continuation.js"; import { canonicalExecutionUuid } from "./execution-identity.js"; import { randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, isNull, or, sql, type SQL } from "drizzle-orm"; +import { and, desc, eq, getTableColumns, inArray, isNull, or, sql, type SQL } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; @@ -20,6 +20,24 @@ import { issueService } from "./issues.js"; type Run = typeof heartbeatRuns.$inferSelect; export const LEGACY_RECOVERY_CAUSE = "legacy_execution_requires_reconciliation"; +const unusedReconciliationContextKeys = [ + "executionContinuation", "paperclipWake", "paperclipWakeComment", + "paperclipTaskMarkdown", "paperclipTaskMarkdownCompact", "paperclipTaskAssignment", + "paperclipTaskAssignmentCompact", "paperclipWorkspace", "paperclipEnvironment", "paperclipContinuationSummary", +] as const; + +export const legacyReconciliationRunFields = { + ...getTableColumns(heartbeatRuns), + contextSnapshot: sql`case + when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' + then ${heartbeatRuns.contextSnapshot} - ARRAY[${sql.join(unusedReconciliationContextKeys.map(key => sql`${key}`), sql`, `)}]::text[] + else ${heartbeatRuns.contextSnapshot} end`, + runnerProfileJson: sql`case + when ${heartbeatRuns.runtimeMode} = 'native' and jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' + then ${heartbeatRuns.runnerProfileJson} - 'nativeExecutionInput' + else ${heartbeatRuns.runnerProfileJson} end`, +}; + /** Error families describe availability, not whether earlier actions happened. */ export function legacyExecutionNeedsReconciliation( run: Pick & Partial>, diff --git a/server/src/services/recovery/reconciliation-run-read.test.ts b/server/src/services/recovery/reconciliation-run-read.test.ts new file mode 100644 index 00000000000..ff42eb35aac --- /dev/null +++ b/server/src/services/recovery/reconciliation-run-read.test.ts @@ -0,0 +1,107 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueRecoveryActions, issues } from "@paperclipai/db"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { executionFailureRetryCount } from "../execution-recovery-attempt.js"; +import { legacyExecutionNeedsReconciliationWithEvidence, legacyReconciliationRunFields, terminalizeLegacyExecution } from "../legacy-execution-recovery.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("reconciliation source PostgreSQL reads", () => { + let temporary: Awaited>; + let db: ReturnType; + let companyId: string, agentId: string; + const launchKeys = ["executionContinuation", "paperclipWake", "paperclipWakeComment", "paperclipTaskMarkdown", + "paperclipTaskMarkdownCompact", "paperclipTaskAssignment", "paperclipTaskAssignmentCompact", "paperclipWorkspace", + "paperclipEnvironment", "paperclipContinuationSummary"]; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("reconciliation-source-"); + db = createDb(temporary.connectionString); + companyId = randomUUID(); agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Recovery reads", issuePrefix: "RCR" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Recovery worker", adapterType: "process", status: "active" }); + }); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + + async function fixture(patch: Partial = {}) { + const payload = randomBytes(64 * 1024).toString("base64"); + const contextSnapshot = { ...Object.fromEntries(launchKeys.map(key => [key, { prompt: payload }])), + issueId: randomUUID(), taskId: randomUUID(), executionWorkspaceId: randomUUID(), + executionRetryAccounting: { version: 1, failureRetries: 2, maxTurnContinuations: 3 }, + failureRetriesBeforeAiConnectionWait: 1, failureRetriesBeforeWorkspaceWait: 1, + conversationSessionGeneration: 7, connectionIntentResolution: "existing_connection", source: "connection_intent.resolved", + interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", interactionId: randomUUID(), + futureControl: { nonce: "retained", nested: null }, cancellation: { state: "pending" }, + }; + const [full] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", runtimeMode: "legacy", + errorCode: "process_lost", contextSnapshot, resultJson: { futureOutput: { retained: true } }, + runnerProfileJson: { adapterDispatch: { adapterType: "process" }, futureControl: { retained: true } }, + ...patch, + }).returning(); + const [selected] = await db.select(legacyReconciliationRunFields).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, full!.id))); + return { full: full!, selected: selected! }; + } + + it.each(["native", "legacy"])("omits unused launch text and preserves current control fields (%s)", async runtimeMode => { + const profile = { nativeExecutionInput: { prompt: randomBytes(128 * 1024).toString("base64") }, futureControl: { retained: true } }; + const f = await fixture({ runtimeMode, ...(runtimeMode === "native" ? { runnerProfileJson: profile } : {}) }); + const expected = { ...f.full.contextSnapshot }; + for (const key of launchKeys) delete expected[key]; + expect(f.selected).toEqual({ ...f.full, contextSnapshot: expected, + runnerProfileJson: runtimeMode === "native" ? { futureControl: profile.futureControl } : f.full.runnerProfileJson }); + expect(Buffer.byteLength(JSON.stringify(f.full))).toBeGreaterThan(800_000); + expect(Buffer.byteLength(JSON.stringify(f.selected))).toBeLessThan(8192); + expect(executionFailureRetryCount(f.selected)).toBe(executionFailureRetryCount(f.full)); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, f.selected)) + .toBe(await legacyExecutionNeedsReconciliationWithEvidence(db, f.full)); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.full.id)))[0]).toEqual(f.full); + }); + + it.each([null, [], ["unexpected"], "scalar", 42].map(value => ({ value })))("preserves non-object JSON $value", async ({ value }) => { + for (const runtimeMode of ["native", "legacy"]) { + const f = await fixture({ runtimeMode, + contextSnapshot: value as typeof heartbeatRuns.$inferInsert.contextSnapshot, + runnerProfileJson: value as typeof heartbeatRuns.$inferInsert.runnerProfileJson, + }); + expect(f.selected).toEqual(f.full); + } + }); + + it.each([ + { status: "failed", errorCode: "setup_failed", resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }, + { status: "cancelled", errorCode: "cancelled", resultJson: { executionCancellation: { state: "acknowledged" }, + executionRecovery: { kind: "interrupted", providerStopped: true, sessionPreserved: true, actionOutcomes: "settled" } } }, + { status: "cancelled", errorCode: "cancelled", resultJson: { executionCancellation: { state: "requested" }, + executionRecovery: { kind: "interrupted", providerStopped: false, sessionPreserved: true, actionOutcomes: "unknown" } } }, + { status: "cancelled", errorCode: "ai_connection_busy", resultJson: { executionRecovery: { kind: "ai_connection_wait", providerWorkStarted: false } } }, + { status: "cancelled", errorCode: "workspace_busy", resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false } } }, + { status: "failed", errorCode: "workspace_restore_failed", resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } }, + { status: "failed", errorCode: "process_lost", resultJson: { conversationContinuation: "continue_conversation_v1" } }, + { status: "cancelled", errorCode: "issue_waiting_for_response", resultJson: { stopReason: "issue_waiting_for_response", timeoutSource: "stale_queued_run_gate" } }, + ])("keeps reconciliation, cancellation, and retry decisions ($errorCode)", async patch => { + const f = await fixture(patch); + expect(f.selected.resultJson).toEqual(f.full.resultJson); + expect(executionFailureRetryCount(f.selected)).toBe(executionFailureRetryCount(f.full)); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, f.selected)) + .toBe(await legacyExecutionNeedsReconciliationWithEvidence(db, f.full)); + }); + + it("keeps full stored input and newer ownership when the projected source is terminalized", async () => { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Recover the stopped source", status: "in_progress", assigneeAgentId: agentId }); + const f = await fixture({ issueId, contextSnapshot: { issueId, paperclipTaskMarkdown: randomBytes(128 * 1024).toString("base64"), + executionRetryAccounting: { version: 1, failureRetries: 2, maxTurnContinuations: 0 } } }); + const [successor] = await db.insert(heartbeatRuns).values({ companyId, agentId, issueId, status: "running" }).returning(); + await db.update(issues).set({ executionRunId: successor!.id, checkoutRunId: successor!.id }).where(eq(issues.id, issueId)); + await terminalizeLegacyExecution({ db, run: f.selected, status: "failed", fromStatuses: ["failed"] }); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.full.id)); + expect(stored).toMatchObject({ contextSnapshot: f.full.contextSnapshot, runnerProfileJson: f.full.runnerProfileJson, resultJson: f.full.resultJson }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]) + .toMatchObject({ executionRunId: successor!.id, checkoutRunId: successor!.id }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))) + .toEqual([expect.objectContaining({ cause: "legacy_execution_requires_reconciliation", ownerType: "board", + evidence: expect.objectContaining({ runId: f.full.id, attempt: 3 }) })]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).toEqual([]); + }); +}); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 5701571e864..488ef813306 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -106,6 +106,7 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { hasUnverifiedNativeWorkspaceFinalizationOwner } from "../native-runtime/native-workspace-finalization-recovery-state.js"; import { legacyExecutionNeedsReconciliationWithEvidence, + legacyReconciliationRunFields, terminalizeLegacyExecution, } from "../legacy-execution-recovery.js"; import { issueTreeControlService } from "../issue-tree-control.js"; @@ -2010,7 +2011,7 @@ export function recoveryService( }) { if (input.retryOfRunId) { const [predecessor] = await db - .select() + .select(legacyReconciliationRunFields) .from(heartbeatRuns) .where( and( @@ -4809,7 +4810,7 @@ export function recoveryService( ) ) { const [source] = await db - .select() + .select(legacyReconciliationRunFields) .from(heartbeatRuns) .where( and(