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,