diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 3d16ff3aada..51d174bb3e7 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -24,6 +24,12 @@ the native status wake identifier from each saved run. It keeps their JSON types and reads full evidence later where finalization or recovery needs it. Saved execution input, runner profiles, and logs stay in the stored run. +Wake-queue release reads keep complete context and result receipts. Native runs +omit runner profiles and log excerpts from this read; cleanup and execution +read their own current evidence. Legacy runs retain the full release row so +workspace-restore and pre-dispatch proofs keep their existing inputs. The +selection is one scoped database query for either runtime. + The recovery card names the original run and requires evidence that its exact allocation stopped and its files were recovered. Recording that evidence does not retry the old turn, certify unknown external actions, change the current task diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 16479253a2f..6a8c360e22d 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -206,7 +206,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { return { recorded, rows }; } - it("keeps in-flight work and every recovery receipt while omitting output from release reads", async () => { + it.each(["legacy", "native"])("keeps %s in-flight work and every recovery receipt while omitting output from release reads", async runtimeMode => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); @@ -218,7 +218,9 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId, companyId, issueId }, executionCancellation: { state: "acknowledged" }, executionRecovery: { kind: "interrupted", providerStopped: true }, configurationIncomplete: { reason: "workspace_unavailable" }, unknownFutureReceipt: { retained: true } }; - await db.update(heartbeatRuns).set({ resultJson: { ...evidence, stdout: payload.repeat(3), stderr: payload } }).where(eq(heartbeatRuns.id, runId)); + await db.update(heartbeatRuns).set({ runtimeMode, + ...(runtimeMode === "native" ? { runnerProfileJson: { savedInput: payload }, stdoutExcerpt: payload, stderrExcerpt: payload } : {}), + resultJson: { ...evidence, stdout: payload.repeat(3), stderr: payload } }).where(eq(heartbeatRuns.id, runId)); await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); @@ -230,7 +232,8 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { ); expect(result.outcome.kind).toBe("released"); expect(rows).toHaveLength(1); - expect(rows[0]).toEqual([{ ...run, resultJson: evidence }]); + expect(rows[0]).toEqual([{ ...run, resultJson: evidence, + ...(runtimeMode === "native" ? { runnerProfileJson: null, stdoutExcerpt: null, stderrExcerpt: null } : {}) }]); expect(Buffer.byteLength(JSON.stringify(rows))).toBeLessThan(8 * 1024); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]).toEqual(run); expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toEqual(issue); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 6db54a82725..cd2f1aa7a48 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -5,7 +5,8 @@ import { isCompletedOnboardingHandoffWake } from "../../../services/chat-complet import { instanceSettingsService } from "../../../services/instance-settings.js"; import { currentConversationCommentCondition } from "../../../services/agent-conversations.js"; import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/execution-blocker.js"; -import { and, asc, eq, getTableColumns, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; +import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; +import { readReleaseRun, type ReleaseRun } from "./release-run-read.js"; import type { Db } from "@paperclipai/db"; import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared"; import { @@ -74,14 +75,6 @@ const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_r type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect; type IssueRow = typeof issues.$inferSelect; -const releaseRunFields = { - ...getTableColumns(heartbeatRuns), - resultJson: sql | null>`case - when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object' - then ${heartbeatRuns.resultJson} - 'stdout' - 'stderr' - else ${heartbeatRuns.resultJson} end`, -}; - function normalizeAgentNameKey(value: string | null | undefined): string | null { if (typeof value !== "string") return null; const normalized = value.trim().toLowerCase(); @@ -94,7 +87,7 @@ function toRequestedByActorType(value: string | null): "user" | "agent" | "syste return value === "user" || value === "agent" || value === "system" ? value : null; } -function toRunSnapshot(row: HeartbeatRunRow): RunSnapshot { +function toRunSnapshot(row: ReleaseRun): RunSnapshot { const configurationIncompletePayload = parseObject(parseObject(row.resultJson).configurationIncomplete); return { id: row.id, @@ -206,7 +199,7 @@ function buildHost(_tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueHost { }; } -function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, run: HeartbeatRunRow): WakeQueueTransaction { +function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, run: ReleaseRun): WakeQueueTransaction { const treeControlSvc = issueTreeControlService(tx); const issuesSvc = issueService(tx); const interruptQueueId = run.runtimeMode !== "native" && run.status === "cancelled" @@ -750,7 +743,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }; } -async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow, issue: IssueRow, now: Date): Promise { +async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: ReleaseRun, issue: IssueRow, now: Date): Promise { const applies = run.runtimeMode === "native" && ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status) && @@ -1058,11 +1051,8 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd async withIssueExecutionLock(input, fn): Promise { return db.transaction(async (rawTx) => { const tx = rawTx as unknown as Db; - const run = await tx - .select(releaseRunFields) - .from(heartbeatRuns) - .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))) - .then((rows) => rows[0] ?? null); + const evidence = await readReleaseRun(tx, input); + const run = evidence.run; if (!run) { throw new Error(`wake-queue: run ${input.runId} was not found while releasing issue execution`); } @@ -1164,7 +1154,8 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd issueStatus: issueRow?.status ?? "", hasAssigneeUser: Boolean(issueRow?.assigneeUserId), assigneeAgentMatchesRunAgent: issueRow?.assigneeAgentId === run.agentId, - legacyExecutionNeedsReconciliation: await legacyExecutionNeedsReconciliationWithEvidence(tx as unknown as Db, run), + legacyExecutionNeedsReconciliation: evidence.kind === "complete" && + await legacyExecutionNeedsReconciliationWithEvidence(tx as unknown as Db, evidence.run), // An operator stop never promotes old queued work by itself. The // next explicit wake adopts those messages atomically when it // queues a run. diff --git a/server/src/modules/wake-queue/adapters/release-run-read.test.ts b/server/src/modules/wake-queue/adapters/release-run-read.test.ts new file mode 100644 index 00000000000..e4129e6c692 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/release-run-read.test.ts @@ -0,0 +1,74 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { readReleaseRun } from "./release-run-read.js"; + +const support = await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("wake release run reads", () => { + let database: Awaited>; + let db: ReturnType; + let companyId: string; + let agentId: string; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("wake-release-read-"); + db = createDb(database.connectionString); + const [company] = await db.insert(companies).values({ name: "Wake release fixture", issuePrefix: "WRR" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Release agent", adapterType: "process" }).returning(); + companyId = company.id; + agentId = agent.id; + }, 60_000); + + afterAll(async () => { + await db?.$client.end(); + await database?.cleanup(); + }); + + it("omits native bulk at the driver boundary and preserves the complete legacy fallback", async () => { + const bulk = randomBytes(128 * 1024).toString("base64"); + const context = { issueId: randomUUID(), opaqueFutureContext: { savedInput: "keep this input" }, explicitUserContinuation: true }; + const evidence = { nativeCancellation: { schema: "paperclip.native-cancellation.v1", acknowledged: true }, unknownFutureReceipt: [true, null, 42] }; + const [before] = await db.insert(heartbeatRuns).values({ + companyId, agentId, runtimeMode: "native", status: "running", contextSnapshot: context, + runnerProfileJson: { nativeExecutionInput: { savedInput: bulk } }, stdoutExcerpt: bulk, stderrExcerpt: bulk, + resultJson: { ...evidence, stdout: bulk, stderr: bulk }, + }).returning(); + const scope = { companyId, runId: before.id }; + const events: Array> = []; + const now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + try { + const native = await db.transaction(tx => readReleaseRun(tx as unknown as typeof db, scope)); + expect(native.kind).toBe("native"); + expect(native.run?.contextSnapshot).toEqual(context); + expect(native.run?.resultJson).toEqual(evidence); + expect(native.run).not.toHaveProperty("runnerProfileJson"); + expect(native.run).not.toHaveProperty("stdoutExcerpt"); + expect(native.run).not.toHaveProperty("stderrExcerpt"); + } finally { capture.flush(); } + const summary = events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; + expect(summary.completedQueries).toBe(1); + expect(summary.responseValueBytes).toBeLessThan(8192); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, before.id)))[0]).toEqual(before); + expect(JSON.stringify(events)).not.toContain(bulk.slice(0, 64)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, before.id)); + const legacy = await readReleaseRun(db, scope); + expect(legacy.kind).toBe("complete"); + expect(legacy.run).toEqual({ ...before, runtimeMode: "legacy", resultJson: evidence }); + expect((await readReleaseRun(db, { ...scope, companyId: randomUUID() })).run).toBeUndefined(); + expect((await readReleaseRun(db, { companyId, runId: randomUUID() })).run).toBeUndefined(); + }); + + it.each([null, ["stdout", { stderr: "evidence" }], "scalar", 42].map(result => ({ result })))( + "preserves non-object native result evidence $result", async ({ result }) => { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, runtimeMode: "native", + resultJson: result as Record | null }).returning(); + expect((await readReleaseRun(db, { companyId, runId: run.id })).run?.resultJson).toEqual(result); + }, + ); +}); diff --git a/server/src/modules/wake-queue/adapters/release-run-read.ts b/server/src/modules/wake-queue/adapters/release-run-read.ts new file mode 100644 index 00000000000..cc8bcf80388 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/release-run-read.ts @@ -0,0 +1,22 @@ +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { and, eq, getTableColumns, sql } from "drizzle-orm"; + +type Run = typeof heartbeatRuns.$inferSelect; +export type ReleaseRun = Omit; + +export async function readReleaseRun(db: Db, scope: { companyId: string; runId: string }) { + const [selected] = await db.select({ + ...getTableColumns(heartbeatRuns), + runnerProfileJson: sql`case when ${heartbeatRuns.runtimeMode} = 'native' + then null else ${heartbeatRuns.runnerProfileJson} end`, + stdoutExcerpt: sql`case when ${heartbeatRuns.runtimeMode} = 'native' + then null else ${heartbeatRuns.stdoutExcerpt} end`, + stderrExcerpt: sql`case when ${heartbeatRuns.runtimeMode} = 'native' + then null else ${heartbeatRuns.stderrExcerpt} end`, + resultJson: sql`case when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object' + then ${heartbeatRuns.resultJson} - 'stdout' - 'stderr' else ${heartbeatRuns.resultJson} end`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, scope.runId), eq(heartbeatRuns.companyId, scope.companyId))); + if (!selected || selected.runtimeMode !== "native") return { kind: "complete" as const, run: selected }; + const { runnerProfileJson: _profile, stdoutExcerpt: _stdout, stderrExcerpt: _stderr, ...run } = selected; + return { kind: "native" as const, run }; +}