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));