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 <noreply@paperclip.ing>

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,

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<Run["contextSnapshot"]>`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<Run["runnerProfileJson"]>`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<Run, "runtimeMode" | "status" | "errorCode" | "resultJson"> & Partial<Pick<Run, "scheduledRetryAttempt" | "scheduledRetryReason" | "contextSnapshot">>,

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  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<typeof heartbeatRuns.$inferInsert> = {}) {
+    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([]);
+  });
+});

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(