From cb6c878c38274b13d7d8ae1013569d48223e6724 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 24 Oct 2026 14:14:46 -0700 Subject: [PATCH] Read execution blocker evidence without full execution payloads (#17491) Read the lifecycle, process, and typed context evidence consumed by execution blockers. Preserve full result evidence, native workspace receipts, current state checks, and stored history. Validated with 458 real PostgreSQL and recovery tests, full build, recursive typecheck, all55 CI checks, and Greptile 5/5. Co-Authored-By: Paperclip --- doc/DATABASE.md | 6 + .../src/services/cancelled-native-startup.ts | 7 +- .../services/execution-blocker-read.test.ts | 115 ++++++++++++++++++ server/src/services/execution-blocker.ts | 48 +++++++- 4 files changed, 174 insertions(+), 2 deletions(-) create mode 100644 server/src/services/execution-blocker-read.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 28950218145..1ff88e3368d 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -62,6 +62,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. +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 and complete native execution input and +workspace-sync receipts. Full result evidence stays available to the existing +retry and workspace checks. Stored context, profiles, and output are unchanged. + 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/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 53a17527f48..810c4d5939d 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -7,7 +7,12 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canContinueCancelledRun } from "./run-cancellation.js"; import { legacyControllerBootId } from "./legacy-controller-lease.js"; -type Run = typeof heartbeatRuns.$inferSelect; +type Run = Pick; type Coordinator = typeof nativeRunFinalizations.$inferSelect; /** Candidate for explicit user Retry only, never automatic replay after Stop. */ diff --git a/server/src/services/execution-blocker-read.test.ts b/server/src/services/execution-blocker-read.test.ts new file mode 100644 index 00000000000..309385a25cb --- /dev/null +++ b/server/src/services/execution-blocker-read.test.ts @@ -0,0 +1,115 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, companies, createDb, heartbeatRuns, issueRecoveryActions, issues, type Db } 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 { getExecutionBlocker } from "./execution-blocker.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("execution blocker PostgreSQL read projection", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("execution-blocker-read-"); + db = createDb(database.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function fixture(patch: Partial = {}, extraEvidence: Record = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Blocker fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Blocker agent", adapterType: "process", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Blocker task", status: "blocked", assigneeAgentId: agentId }); + const payload = randomBytes(128 * 1024).toString("base64"); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", runtimeMode: "legacy", + startedAt: new Date("2020-01-01T00:00:00Z"), finishedAt: new Date("2020-01-01T00:01:00Z"), + errorCode: "adapter_failed", error: "Fixture failure", contextSnapshot: { issueId, prompt: payload }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner", adapterConfig: { instructions: payload } }, instructions: payload }, + stdoutExcerpt: payload, ...patch, + }).returning(); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", + cause: "legacy_execution_requires_reconciliation", fingerprint: run.id, status: "active", nextAction: "Inspect the source run.", + evidence: { runId: run.id, ...extraEvidence } }); + return { companyId, agentId, issueId, run }; + } + + function recordRunReads(fullRows = false) { + const reads: unknown[][] = []; + const recorded = new Proxy(db, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = args[0] === heartbeatRuns && fullRows ? target.select().from(heartbeatRuns) : from(...args); + if (args[0] === heartbeatRuns) { + const execute = query.execute.bind(query); + query.execute = async (...executeArgs: Parameters) => { + const rows = await execute(...executeArgs); reads.push(rows); return rows; + }; + } + return query; + }) as typeof builder.from; + return builder; + }; + }}); + return { recorded, reads }; + } + + async function compare(f: Awaited>) { + const baseline = recordRunReads(true), projected = recordRunReads(); + const expected = await getExecutionBlocker(baseline.recorded, f.companyId, f.issueId); + const result = await getExecutionBlocker(projected.recorded, f.companyId, f.issueId); + expect(expected).not.toBeNull(); + expect(result).toEqual(expected); + expect(projected.reads).toHaveLength(baseline.reads.length); + expect(projected.reads.flat()).toHaveLength(1); + expect(Buffer.byteLength(JSON.stringify(projected.reads))).toBeLessThan(4096); + expect(Buffer.byteLength(JSON.stringify(baseline.reads))).toBeGreaterThan(128 * 1024); + return result; + } + + it("preserves blocker and retry decisions without returning prompts, profiles, or output", async () => { + const f = await fixture(); + expect(await compare(f)).toMatchObject({ runId: f.run.id, agentId: f.agentId, canRetry: true, canContinue: false }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(f.run); + expect(await getExecutionBlocker(db, randomUUID(), f.issueId)).toBeNull(); + }); + + it("retains the claimed adapter and startup receipts for a cancelled pre-provider run", async () => { + const f = await fixture({ status: "cancelled", startedAt: null, errorCode: "operator_interrupted", + resultJson: { startupCancellation: { beforeNativeSelection: true }, startupPreparationSettledAt: "2020-01-01T00:01:00Z" } }); + expect(await compare(f)).toMatchObject({ canRetry: true, canContinue: false }); + await db.update(heartbeatRuns).set({ processPid: 424242 }).where(eq(heartbeatRuns.id, f.run.id)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ canRetry: false }); + }); + + it.each(["valid", "number", "missing"])("preserves typed pre-dispatch interaction evidence (%s)", async kind => { + const context = { source: "connection_intent.resolved", connectionIntentResolution: "existing_connection", + interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", + ...(kind === "missing" ? {} : { interactionId: kind === "valid" ? randomUUID() : 42 }) }; + const f = await fixture({ status: "cancelled", errorCode: "issue_waiting_for_response", contextSnapshot: context, + resultJson: { stopReason: "issue_waiting_for_response", timeoutSource: "stale_queued_run_gate", + preDispatchResponseWait: { version: 1, providerWorkStarted: false } } }); + expect(await compare(f)).toMatchObject({ canRetry: kind === "valid" }); + }); + + it.each([true, false])("preserves native no-workspace evidence and cwd presence (%s)", async hasCwd => { + const f = await fixture({ runtimeMode: "native", driverKind: "grok_bot_mcp" }, { + workspaceRestoreFailure: "restore_unsafe_archive", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [] }, + }); + const input = { schema: "paperclip.native-execution-input.v6", + binding: { runId: f.run.id, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId }, + provider: { kind: "grok_bot" }, workspace: { access: "none", ...(hasCwd ? { cwd: null } : {}) } }; + await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, + runnerProfileJson: { ...f.run.runnerProfileJson, nativeExecutionInput: input } }).where(eq(heartbeatRuns.id, f.run.id)); + expect(await compare(f)).toMatchObject({ workspaceRepairRequired: !hasCwd, canRetry: false }); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + expect(stored.runnerProfileJson?.nativeExecutionInput).toEqual(input); + }); + + it.each([null, [], "scalar", 42, { issueId: 42, taskId: false }].map(context => ({ context })))("preserves decisions for malformed context $context", async ({ context }) => { + const f = await fixture({ contextSnapshot: context as Record | null }); + expect(await compare(f)).toMatchObject({ canRetry: true }); + }); +}); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index b1277e53f1e..4777d5947b3 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -14,6 +14,52 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; +const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", + "interactionKind", "interactionStatus", "mutation", "interactionId"] as const; +const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync"] as const; +const blockerContextColumn = sql | null>`case + when ${heartbeatRuns.contextSnapshot} is null then null else ( + select coalesce(jsonb_object_agg(field.key, field.value), '{}'::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(blockerContextKeys.map(key => sql`${key}`), sql`, `)}) + ) end`; +const blockerProfileColumn = sql | null>`case + when ${heartbeatRuns.runnerProfileJson} is null then null else ( + select coalesce(jsonb_object_agg(field.key, case field.key + when 'adapterDispatch' then jsonb_build_object('adapterType', field.value->'adapterType') + else field.value end), '{}'::jsonb) + from jsonb_each(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' + then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) as field(key, value) + where field.key in (${sql.join(blockerProfileKeys.map(key => sql`${key}`), sql`, `)}) + ) end`; +const blockerRunFields = { + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + startedAt: heartbeatRuns.startedAt, + finishedAt: heartbeatRuns.finishedAt, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + issueId: heartbeatRuns.issueId, + nativeIssueId: heartbeatRuns.nativeIssueId, + nativeSessionId: heartbeatRuns.nativeSessionId, + runtimeMode: heartbeatRuns.runtimeMode, + runtimeModeResolvedAt: heartbeatRuns.runtimeModeResolvedAt, + driverKind: heartbeatRuns.driverKind, + processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, + sessionIdAfter: heartbeatRuns.sessionIdAfter, + controllerBootId: heartbeatRuns.controllerBootId, + controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + executionStage: heartbeatRuns.executionStage, + resultJson: heartbeatRuns.resultJson, + contextSnapshot: blockerContextColumn, + runnerProfileJson: blockerProfileColumn, +}; + export async function hasPendingRunCleanup(db: Db, companyId: string, runId: string): Promise { const [state] = await db.execute<{ pending: boolean }>(sql`SELECT ${heartbeatRuns.runnerProfileJson}->>'nativeWorkspaceFinalizationOwner' IS NOT NULL @@ -99,7 +145,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st if (!action) return null; const parsedRunId = z.string().guid().safeParse(action.evidence.runId ?? action.evidence.sourceRunId); const runId = parsedRunId.success ? parsedRunId.data : null; - const [run] = runId ? await db.select().from(heartbeatRuns).where(and( + const [run] = runId ? await db.select(blockerRunFields).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId), )).limit(1) : []; const withoutWorkspace = Boolean(run?.driverKind === "grok_bot_mcp" && nativeRunHasNoWorkspaceAccess(run));