From 7bd449dbb66473ccb3fc9965186eed495d7cc623 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 24 Oct 2026 12:14:41 -0700 Subject: [PATCH] fix(recovery): project context evidence for latest-run checks (#17463) ## Thinking Path > - Paperclip manages AI agents and their tasks. > - Recovery sweeps decide whether tasks have a valid next step. > - Those decisions inspect saved run metadata. > - Latest-run queries also return full prompts and execution setup data. > - This repeats large responses when a sweep leaves a task unchanged. > - This change projects the context evidence recovery actually reads. > - A row-version check keeps stale evidence from changing a task. ## Linked Issues or Issue Description **What happened?** The latest-run recovery queries retrieve large stored contexts for each candidate. The decision usually consumes small recovery, message, or workspace-mode fields. **Expected behavior** Keep all decision evidence while selecting a small response. Preserve complete stored contexts and full execution reads. Detect concurrent source changes. **Steps to reproduce** 1. Store a terminal run with large task Markdown, wake, workspace, and continuation bodies. 2. Call `reconcileStrandedAssignedIssues` for its assigned task. 3. Observe the full context returned by the latest-run query on each sweep. **Paperclip version or commit** Baseline: `410e5c6eedba98588a87e4104153898fb0437532`. Related read optimization: Refs #17405. Complementary fix: Refs #17460. These PRs have no code dependency. Searched existing PRs and issues for run projections, recovery reads, and database transfer. No duplicate was found. **Deployment mode** External PostgreSQL. Verification uses disposable real PostgreSQL and synthetic data. ## What Changed - Project typed context evidence in all three latest-run recovery queries. - Keep only the used workspace mode and external-interaction flag from nested setup objects. - Preserve original JSON types, message references, retry evidence, and existing query scopes and order. - Read PostgreSQL `xmin` with the run. Require the same row version during native blocked-wait repair. - Retain the original full-context comparison for callers that supply a full run without a row version. - Test large payloads, malformed JSON shapes, review-message recovery, and concurrent source edits. ## Verification - Final-source integration checks: **450 passed** across context projections, native finalization recovery, and the assembled heartbeat recovery service. - Focused issue-lock race check: **9 passed** (401 outside the filter), including omitted-context and late-output edits and successful repair on a fresh sweep. - Final-source recursive typecheck and full build pass. The full local stable suite is still running; current-head CI must also pass before merge. - Real-service SQL capture against the baseline: one returned row falls from 699,724 to 573 serialized JSON bytes (99.918% smaller). - The fixture uses four 174,764-byte incompressible execution fields. All unaffected returned columns compare equal. Regression tests check the retained decision evidence and unchanged stored records. - Five alternating warm samples after two warmups: baseline median 1.479 ms, treatment 1.084 ms. Ranges overlap; no request-latency speedup is claimed. Plans were captured outside the timing interval. - An initial full-context hash projection cut bytes but raised the local median to 7.457 ms. The final row-version check removes that cost. - One integration run overlapped the hash-export rename and failed seven tests on the removed export. A fresh final-source run passed all 450 tests. The earlier mixed-source failure remains recorded and is not counted as a pass. - All current-head CI checks pass on `17add20e391c30ed0496f4b39eb9c59779efa6b4`. Greptile: **5/5, medium impact**, no findings or unresolved threads on that head. - Hosted transfer and product behavior remain to be verified on staging after merge. ## Risks The context selection must track its recovery consumers. Tests preserve typed flags, retry budgets, external-chat references, and malformed-input behavior. The row-version guard is conservative: any intervening run update defers native blocker repair to a fresh sweep. Concurrent omitted-context and output edits are covered through the real service and issue-lock wait; a fresh sweep must still repair. There is no migration, prompt truncation, history deletion, or authorization cache. ## Model Used OpenAI GPT-6 through Codex, with source inspection, code execution, and GitHub tools. The exact served model ID and context-window size are not exposed in this session. ## 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 | 7 ++ .../heartbeat-process-recovery.test.ts | 12 +++ .../recovery/run-context-read.test.ts | 84 +++++++++++++++++++ .../src/services/recovery/run-context-read.ts | 28 +++++++ server/src/services/recovery/service.ts | 22 +++-- 5 files changed, 147 insertions(+), 6 deletions(-) create mode 100644 server/src/services/recovery/run-context-read.test.ts create mode 100644 server/src/services/recovery/run-context-read.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 2698c899e51..3ab9490ee1f 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -49,6 +49,13 @@ and leaves all existing cases passive. Source identities are not grants. ### Task query performance +Recovery's latest-run queries return the context fields used by recovery +decisions. Prompts, execution continuation bodies, and workspace setup payloads +remain in the stored run and its full detail reads. These queries also return +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. + Run-context lookups use expression indexes for both `context_snapshot.issueId` and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own company-scoped expression index that follows the query's native issue ID and diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 3af86f73e77..ca2576a5b59 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -15676,6 +15676,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { "comment", "wake", "run", + "run_context", + "run_metadata", "attention", ] as const)( "does not clobber a newer %s while native blocker repair waits on the issue lock", @@ -15750,6 +15752,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { continuationPolicy: "wake_assignee", payload: { version: 1, questions: [] }, }); + if (change === "run_context") + await tx.update(heartbeatRuns).set({ + contextSnapshot: sql`jsonb_set(${heartbeatRuns.contextSnapshot}, '{paperclipTaskMarkdown}', '"Changed execution instructions"'::jsonb)`, + }).where(eq(heartbeatRuns.id, runId)); + if (change === "run_metadata") + await tx.update(heartbeatRuns).set({ stdoutExcerpt: "Late execution output" }).where(eq(heartbeatRuns.id, runId)); [snapshot] = await tx .select() .from(issues) @@ -15793,6 +15801,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { ), ).toHaveLength(0); expect(mockAdapterExecute).not.toHaveBeenCalled(); + if (change === "run_context" || change === "run_metadata") { + expect((await heartbeat.reconcileStrandedAssignedIssues()).escalated).toBe(1); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + } }, ); diff --git a/server/src/services/recovery/run-context-read.test.ts b/server/src/services/recovery/run-context-read.test.ts new file mode 100644 index 00000000000..682aea7ab96 --- /dev/null +++ b/server/src/services/recovery/run-context-read.test.ts @@ -0,0 +1,84 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { isExternalChatPresentationContext } from "../heartbeat-run-summary.js"; +import { legacyDispositionEpisode } from "./legacy-continuation.js"; +import { decideIssueReviewPathRecovery } from "./review-path-recovery.js"; +import { recoveryRunContextColumn, recoveryRunVersionColumn } from "./run-context-read.js"; + +const support = await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("recovery context PostgreSQL projection", () => { + let database: Awaited>; + let db: ReturnType; + let companyId: string; + let agentId: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("recovery-context-read-"); + db = createDb(database.connectionString); + const [company] = await db.insert(companies).values({ name: "Recovery fixture", issuePrefix: "RC" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Recovery agent", adapterType: "process" }).returning(); + companyId = company.id; + agentId = agent.id; + }, 60_000); + afterAll(async () => { + await db?.$client.end(); + await database?.cleanup(); + }); + + async function read(contextSnapshot: unknown) { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, contextSnapshot: contextSnapshot as Record | null }).returning(); + const [selected] = await db.select({ context: recoveryRunContextColumn, version: recoveryRunVersionColumn }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + return { run, selected }; + } + + it("retains typed recovery evidence while excluding large execution data", async () => { + const issueId = randomUUID(); + const payload = randomBytes(128 * 1024).toString("base64"); + const evidence = { + issueId, taskId: issueId, retryReason: "issue_continuation_needed", source: "chat:slack", + wakeReason: "assignment", handoffReason: "successful_run_missing_state", handoffRequired: true, + handoffAttempt: 2, maxHandoffAttempts: 3, sourceRunId: randomUUID(), resumeFromRunId: randomUUID(), + missingDisposition: "clear_next_step", providerQuotaRetryNotBefore: "2099-01-01T00:00:00Z", + transientRetryNotBefore: "2099-01-02T00:00:00Z", dispositionRepairAttempt: 2, + dispositionRepairFingerprint: "fingerprint", legacyDispositionEpisode: { id: "episode", attempt: 1, maxAttempts: 2 }, + reviewPathRecoveryAttempt: 1, reviewPathConsumedRef: "reference", interactionId: randomUUID(), + approvalId: randomUUID(), clearReason: "resolved", monitorAttemptCount: 4, + wakeCommentIds: [randomUUID(), randomUUID()], wakeCommentId: randomUUID(), commentId: randomUUID(), + externalChatContinuation: true, + }; + const context = { ...evidence, paperclipWake: { externalInteractionContinuation: true, prompt: payload }, + paperclipWorkspace: { mode: "shared_workspace", instructions: payload }, executionContinuation: { prompt: payload }, paperclipTaskMarkdown: payload }; + const { run, selected } = await read(context); + expect(selected.context).toEqual({ ...evidence, paperclipWake: { externalInteractionContinuation: true }, paperclipWorkspace: { mode: "shared_workspace" } }); + expect(Buffer.byteLength(JSON.stringify(selected))).toBeLessThan(4096); + expect(isExternalChatPresentationContext(selected.context)).toBe(isExternalChatPresentationContext(context)); + expect(legacyDispositionEpisode({ ...run, contextSnapshot: selected.context })).toEqual(legacyDispositionEpisode(run)); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(stored.contextSnapshot).toEqual(context); + const [serialized] = await db.$client.unsafe("SELECT xmin::text AS version FROM heartbeat_runs WHERE id = $1", [run.id]); + expect(selected.version).toBe(serialized.version); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...context, paperclipTaskMarkdown: "changed" } }).where(eq(heartbeatRuns.id, run.id)); + const [changed] = await db.select({ context: recoveryRunContextColumn, version: recoveryRunVersionColumn }).from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(changed.context).toEqual(selected.context); + expect(changed.version).not.toBe(selected.version); + }); + + it.each([null, [], ["unexpected"], "scalar", 42, { issueId: 42, taskId: false, paperclipWake: [], paperclipWorkspace: null }].map(context => ({ context })))("preserves decisions for malformed context $context", async ({ context }) => { + const { selected } = await read(context); + expect(isExternalChatPresentationContext(selected.context)).toBe(isExternalChatPresentationContext(context)); + expect(legacyDispositionEpisode({ id: "run", contextSnapshot: selected.context })).toEqual(legacyDispositionEpisode({ id: "run", contextSnapshot: context })); + }); + + it("retains external-chat review recovery message references and monitor identity", async () => { + const context = { source: "chat:slack", wakeReason: "issue_monitor_fired", clearReason: "fired", monitorAttemptCount: 3, + wakeCommentIds: [randomUUID(), randomUUID()], prompt: randomBytes(64 * 1024).toString("base64") }; + const { selected } = await read(context); + const input = { issueId: randomUUID(), sourceRunId: randomUUID(), assigneeAgentId: agentId, reviewAttention: { state: "stalled" } as Parameters[0]["reviewAttention"], existingWake: false }; + expect(decideIssueReviewPathRecovery({ ...input, contextSnapshot: selected.context })) + .toEqual(decideIssueReviewPathRecovery({ ...input, contextSnapshot: context })); + }); +}); diff --git a/server/src/services/recovery/run-context-read.ts b/server/src/services/recovery/run-context-read.ts new file mode 100644 index 00000000000..b453aa79d09 --- /dev/null +++ b/server/src/services/recovery/run-context-read.ts @@ -0,0 +1,28 @@ +import { heartbeatRuns } from "@paperclipai/db"; +import { sql } from "drizzle-orm"; + +const recoveryContextKeys = [ + "issueId", "taskId", "retryReason", "source", "wakeReason", + "handoffReason", "handoffRequired", "handoffAttempt", "maxHandoffAttempts", + "sourceRunId", "resumeFromRunId", "missingDisposition", + "providerQuotaRetryNotBefore", "transientRetryNotBefore", + "dispositionRepairAttempt", "dispositionRepairFingerprint", "legacyDispositionEpisode", + "reviewPathRecoveryAttempt", "reviewPathConsumedRef", "interactionId", "approvalId", + "clearReason", "monitorAttemptCount", "wakeCommentIds", "wakeCommentId", "commentId", + "externalChatContinuation", "paperclipWake", "paperclipWorkspace", +] as const; + +export const recoveryRunContextColumn = sql | null>` + case when ${heartbeatRuns.contextSnapshot} is null then null else ( + select coalesce(jsonb_object_agg(field.key, case field.key + when 'paperclipWake' then jsonb_build_object( + 'externalInteractionContinuation', field.value->'externalInteractionContinuation') + when 'paperclipWorkspace' then jsonb_build_object('mode', field.value->'mode') + else field.value end), '{}'::jsonb) + from jsonb_each(case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' + then ${heartbeatRuns.contextSnapshot} else '{}'::jsonb end) as field(key, value) + where field.key in (${sql.join(recoveryContextKeys.map(key => sql`${key}`), sql`, `)}) + ) end +`; + +export const recoveryRunVersionColumn = sql`${heartbeatRuns}.xmin::text`; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 42765bb3606..e517b69ec69 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { recoveryRunContextColumn, recoveryRunVersionColumn } from "./run-context-read.js"; import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js"; import { onboardingFirstTaskAwaitingUserSql } from "./onboarding-first-task-wait.js"; import { isAiAuthenticationBlocked } from "../ai-auth-failure.js"; @@ -25,6 +26,7 @@ import { desc, eq, gt, + getTableColumns, gte, inArray, isNotNull, @@ -247,6 +249,7 @@ type LatestIssueRun = | "createdAt" > & { resultJson?: unknown; + rowVersion?: string; }) | null; type SuccessfulLatestIssueRun = NonNullable & { @@ -977,7 +980,8 @@ export function recoveryService( status: heartbeatRuns.status, error: heartbeatRuns.error, errorCode: heartbeatRuns.errorCode, - contextSnapshot: heartbeatRuns.contextSnapshot, + contextSnapshot: recoveryRunContextColumn, + rowVersion: recoveryRunVersionColumn, livenessState: heartbeatRuns.livenessState, resultJson: heartbeatRuns.resultJson, startedAt: heartbeatRuns.startedAt, @@ -1007,7 +1011,8 @@ export function recoveryService( status: heartbeatRuns.status, error: heartbeatRuns.error, errorCode: heartbeatRuns.errorCode, - contextSnapshot: heartbeatRuns.contextSnapshot, + contextSnapshot: recoveryRunContextColumn, + rowVersion: recoveryRunVersionColumn, livenessState: heartbeatRuns.livenessState, resultJson: heartbeatRuns.resultJson, startedAt: heartbeatRuns.startedAt, @@ -1618,7 +1623,10 @@ export function recoveryService( return false; const source = await tx - .select() + .select({ + ...getTableColumns(heartbeatRuns), + rowVersion: recoveryRunVersionColumn, + }) .from(heartbeatRuns) .where( and( @@ -1634,8 +1642,9 @@ export function recoveryService( source.id !== latestRun.id || source.agentId !== current.assigneeAgentId || source.status !== "succeeded" || - JSON.stringify(source.contextSnapshot) !== - JSON.stringify(latestRun.contextSnapshot) + (latestRun.rowVersion !== undefined + ? source.rowVersion !== latestRun.rowVersion + : JSON.stringify(source.contextSnapshot) !== JSON.stringify(latestRun.contextSnapshot)) ) return false; const proof = await nativeBlockedUnblockAction( @@ -1903,7 +1912,8 @@ export function recoveryService( status: heartbeatRuns.status, error: heartbeatRuns.error, errorCode: heartbeatRuns.errorCode, - contextSnapshot: heartbeatRuns.contextSnapshot, + contextSnapshot: recoveryRunContextColumn, + rowVersion: recoveryRunVersionColumn, livenessState: heartbeatRuns.livenessState, resultJson: heartbeatRuns.resultJson, startedAt: heartbeatRuns.startedAt,