diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 1731fc5df0e..b5b2af1a789 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -299,6 +299,20 @@ purpose alone cannot certify termination. Ordinary execution retains its existin semantic dispatcher requirement. Missing usage remains pending accounting debt; a complete receipt settles the original reservation and charge once. +A recovery setup failure does not establish that the original execution never +started. A same-run, company- and issue-bound coordinator attempt proves that +native execution or result finalization was entered; selected input alone does +not. Setup-failure writes lock that coordinator before the run, so a concurrently +committed claim cannot be missed. Current Stop metadata survives configuration +failure, including an acknowledgement committed while the terminal write waits. +Only the bounded run-Stop fields, including intentional handoff, are retained; +stale result projections are replaced. Normal Stop writes can still advance +delivery metadata. Missing usage keeps its existing hold and accounting debt. Fresh setup +that genuinely never dispatched can still settle without a provider receipt. +Account-identity checks continue to reject incompatible execution; preserving +cleanup obligations does not authorize a new provider turn or repair historical +state that was already overwritten. + ## 4. Agent-Owned vs User-Owned Execution The execution model differs depending on assignee type. diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index feafc254d80..8b8416c1891 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -2,7 +2,7 @@ import { createNativeUsageDelivery } from "../native-runtime/native-usage-delive import type { PrpEvent } from "../../vendor/paperclip-runner/index.js"; import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; -import { agents, agentWakeupRequests, caseHandoffs, cases, companies, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { activityLog, agents, agentWakeupRequests, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -18,6 +18,11 @@ import { createHeartbeatRunState, getAdapterSessionCodec } from "./run-state.js" import { buildEffectiveRunSessionConfigMetadata } from "./workspaces.js"; import { createHeartbeatRunCompletion, type CompleteHeartbeatRunInput, type FailHeartbeatRunInput, type HeartbeatRunCompletionDependencies } from "./run-completion.js"; import { createHeartbeatAgentStatus } from "./agent-status.js"; +import { prepareNativeHeartbeatRun } from "../native-runtime/prepare-native-run.js"; +import { buildNativeCompletionContract } from "../native-runtime/completion-contracts.js"; +import { readPendingNativeRunStop } from "../native-runtime/native-cancellation-request.js"; +import { reserveRunBudget } from "../budget-reservations.js"; +import { hasAcknowledgedNativeReassignmentStopIntent } from "../acknowledged-native-stop.js"; vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); @@ -122,6 +127,28 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( } const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); + async function admitNativeExecution(f: Awaited>) { + await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, f.agent.id)); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Interrupted native task" }).returning(); + await db.update(heartbeatRuns).set({ nativeIssueId: issue.id, issueId: issue.id, + contextSnapshot: { issueId: issue.id } }).where(eq(heartbeatRuns.id, f.run.id)); + Object.assign(f.run, await read(f.run.id)); + const native = await prepareNativeHeartbeatRun({ db, run: f.run, issue, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issue.id)); + const binding = { companyId: f.run.companyId, runId: f.run.id, issueId: issue.id, agentId: f.agent.id, executionWorkspaceId: f.run.id }; + const execution = { schema: "paperclip.native-execution-input.v1", binding, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: issue.id, title: issue.title, description: null, prompt: "Complete the task.", workMode: "standard" }, + workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue, { revision: contract.revision }) }, + interactionResponses: [], credentialBindings: [] }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, f.run.id)); + return { issue, native, binding, execution }; + } + it.each(["completeRun", "failRun"] as const)("settles the outgoing agent after a case handoff in %s", async operation => { const f = await fixture(); await db.update(agents).set({ status: "running", lifecycleState: "ready" }).where(eq(agents.id, f.agent.id)); @@ -434,6 +461,210 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( expect(f.deps.finalizeAgentStatus).toHaveBeenCalledOnce(); }); + it.each(["current", "stale_snapshot", "stale_profile", "prior_attempt_only", "concurrent_ack", "handoff_ack"] as const)("retains an existing native Stop and receipt debt when recovery configuration fails (%s)", async mode => { + const f = await fixture({ runtimeMode: "native", costAccountingPending: true, + usageJson: { accountingReceiptReady: false, accountingReceiptSequence: 0 } }); + const { issue, native, binding, execution } = await admitNativeExecution(f); + const intentAuditId = randomUUID(); + await db.insert(activityLog).values({ id: intentAuditId, companyId: f.run.companyId, actorType: "user", actorId: "test-board", + action: "native.cancellation_intent_recorded", entityType: "heartbeat_run", entityId: f.run.id, runId: f.run.id, agentId: f.agent.id }); + const nativeCancellation = { schema: "paperclip.native-cancellation.v1", ...binding, scope: "run", reasonCode: "cancellation_run_only", + intentId: `native-cancellation:${randomUUID()}`, intentAuditId, dispatchState: "pending", dispatched: false, effects: ["release_run_resources"] }; + await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, issueId: issue.id, phase: "observed", attempt: 1 }); + await reserveRunBudget(db, f.run.companyId, f.run.id, null); + await db.update(budgetReservations).set({ providerStartedAt: new Date() }).where(eq(budgetReservations.runId, f.run.id)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution, + ...(mode === "prior_attempt_only" ? {} : { sessionCheckpoint: { identity: { ...binding, sessionId: native.normalizedSessionId }, providerSessionId: "existing-provider-session", activeTurnId: native.turnId } }) }, + resultJson: { oldNonControlField: "replace this stale projection", nativeCancellation, startupCancellation: { requestedAt: new Date().toISOString(), beforeNativeSelection: false }, + ...(mode === "handoff_ack" ? { reassignmentStopRequested: true } : {}), + executionCancellation: { state: "requested", requestedAt: new Date().toISOString() } } }).where(eq(heartbeatRuns.id, f.run.id)); + Object.assign(f.run, await read(f.run.id)); + expect(await readPendingNativeRunStop(db, f.run)).toBe(true); + const setup = f.setup(new ConfigurationIncompleteFailure("The AI account changed while suspended.", { + configurationIncomplete: { reason: "ai_connection_changed" } })); + if (mode === "stale_snapshot") setup.run = { ...setup.run, resultJson: {} }; + if (mode === "stale_profile" || mode === "prior_attempt_only") setup.run = { ...setup.run, runnerProfileJson: {} }; + let expectedCancellation: Record = nativeCancellation; + const acknowledged = mode === "concurrent_ack" || mode === "handoff_ack"; + if (acknowledged) { + const acknowledgementAuditId = randomUUID(); + await db.insert(activityLog).values({ id: acknowledgementAuditId, companyId: f.run.companyId, actorType: "system", actorId: "native-session-cancellation", + action: "native.cancellation_dispatch_acknowledged", entityType: "heartbeat_run", entityId: f.run.id, runId: f.run.id, agentId: f.agent.id }); + expectedCancellation = { ...nativeCancellation, dispatchState: "acknowledged", dispatched: true, acknowledgementAuditId }; + let enter!: () => void, release!: () => void, blockerPid = 0; + const entered = new Promise(resolve => { enter = resolve; }); + const released = new Promise(resolve => { release = resolve; }); + const writer = db.transaction(async tx => { + const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + blockerPid = backend.pid; + const [current] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)).for("update"); + enter(); await released; + const [committed] = await tx.update(heartbeatRuns).set({ resultJson: { ...current.resultJson, nativeCancellation: expectedCancellation } }) + .where(eq(heartbeatRuns.id, f.run.id)).returning(); + expect(hasAcknowledgedNativeReassignmentStopIntent(committed)).toBe(mode === "handoff_ack"); + }); + await entered; + const failure = f.completion.failRunSetup(setup); + try { + await vi.waitFor(async () => { + const waiting = await db.execute(sql`select pid from pg_stat_activity + where ${blockerPid} = any(pg_blocking_pids(pid)) and wait_event_type = 'Lock'`); + expect(waiting.length).toBeGreaterThan(0); + }, { timeout: 5_000 }); + } finally { release(); await writer; await failure; } + } else await f.completion.failRunSetup(setup); + const failed = await read(f.run.id); + expect(failed).toMatchObject({ status: "failed", errorCode: "configuration_incomplete" }); + expect.soft(failed.resultJson?.nativeCancellation).toEqual(expectedCancellation); + expect.soft(await readPendingNativeRunStop(db, failed)).toBe(!acknowledged); + expect.soft(hasAcknowledgedNativeReassignmentStopIntent(failed)).toBe(mode === "handoff_ack"); + expect(failed.resultJson).not.toHaveProperty("oldNonControlField"); + expect.soft((failed.resultJson?.executionRecovery as Record | undefined)?.providerWorkStarted).not.toBe(false); + expect.soft(await (await import("../run-cost-accounting.js")).accountRunCost(db, f.run.id)).toBe(false); + expect.soft(await read(f.run.id)).toMatchObject({ costAccountingPending: true, costAccountedAt: null, + usageJson: { accountingReceiptReady: false, accountingReceiptSequence: 0 } }); + const [hold] = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)); + expect.soft(hold).toMatchObject({ state: "held", settledAt: null }); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, f.run.id))).toHaveLength(0); + }, 30_000); + + it.each([ + { runtimeMode: "native", providerDispatchStarted: false }, + { runtimeMode: "legacy", providerDispatchStarted: false }, + { runtimeMode: "native", providerDispatchStarted: true }, + { runtimeMode: "legacy", providerDispatchStarted: true }, + ] as const)("settles only a genuinely pre-provider setup failure ($runtimeMode, dispatched=$providerDispatchStarted)", async ({ runtimeMode, providerDispatchStarted }) => { + const f = await fixture({ runtimeMode, costAccountingPending: true, + usageJson: { accountingReceiptReady: false, accountingReceiptSequence: 0 } }); + await reserveRunBudget(db, f.run.companyId, f.run.id, null); + if (providerDispatchStarted) await db.update(budgetReservations).set({ providerStartedAt: new Date() }).where(eq(budgetReservations.runId, f.run.id)); + await f.completion.failRunSetup(f.setup(new ConfigurationIncompleteFailure("missing credential", { + configurationIncomplete: { reason: "missing_binding" } }), providerDispatchStarted)); + expect(await (await import("../run-cost-accounting.js")).accountRunCost(db, f.run.id)).toBe(!providerDispatchStarted); + expect(await read(f.run.id)).toMatchObject({ costAccountingPending: providerDispatchStarted, + costAccountedAt: providerDispatchStarted ? null : expect.any(Date) }); + const [hold] = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)); + expect(hold.state).toBe(providerDispatchStarted ? "held" : "released"); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, f.run.id))).toHaveLength(0); + }); + + it.each(["run", "company", "issue", "task_scope"] as const)("does not preserve a foreign or task-scoped cancellation projection (%s)", async mismatch => { + const f = await fixture({ runtimeMode: "native" }); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Scoped metadata" }).returning(); + const nativeCancellation = { schema: "paperclip.native-cancellation.v1", scope: mismatch === "task_scope" ? "issue" : "run", + reasonCode: "cancellation_run_only", runId: mismatch === "run" ? randomUUID() : f.run.id, + companyId: mismatch === "company" ? randomUUID() : f.run.companyId, issueId: mismatch === "issue" ? randomUUID() : issue.id, + dispatchState: "pending", dispatched: false }; + await db.update(heartbeatRuns).set({ nativeIssueId: issue.id, resultJson: { nativeCancellation, oldNonControlField: true } }) + .where(eq(heartbeatRuns.id, f.run.id)); + const resultJson = { configurationIncomplete: { reason: "missing_binding" } }; + await f.deps.setRunStatusIfRunning(f.run.id, "failed", { errorCode: "configuration_incomplete", resultJson }); + expect((await read(f.run.id)).resultJson).toEqual(resultJson); + }); + + it("keeps ordinary adapter pre-provider receipts outside setup-history checks", async () => { + const f = await fixture({ runtimeMode: "native", costAccountingPending: true, + runnerProfileJson: { nativeExecutionInput: { schema: "paperclip.native-execution-input.v1" } } }); + await reserveRunBudget(db, f.run.companyId, f.run.id, null); + await f.completion.completeRun({ ...f.input, adapterResult: { exitCode: 1, signal: null, timedOut: false, + errorCode: "provider_not_started", executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + expect(await read(f.run.id)).toMatchObject({ status: "failed", costAccountingPending: false, + costAccountedAt: expect.any(Date), resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + const [hold] = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)); + expect(hold.state).toBe("released"); + }); + + it.each(["stale", "loaded", "foreign_coordinator"] as const)("settles fresh input admitted by this attempt when no provider work started (%s)", async mode => { + const f = await fixture({ runtimeMode: "native", costAccountingPending: true, + usageJson: { accountingReceiptReady: false, accountingReceiptSequence: 0 } }); + const setup = { ...f.setup(new ConfigurationIncompleteFailure("fresh preparation failed", { + configurationIncomplete: { reason: "missing_binding" } })), run: { ...f.run } }; + const { issue } = await admitNativeExecution(f); + await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, + issueId: issue.id, phase: "observed" }); + if (mode === "foreign_coordinator") { + const [otherIssue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Other task" }).returning(); + const [otherRun] = await db.insert(heartbeatRuns).values({ companyId: f.run.companyId, agentId: f.agent.id, + nativeIssueId: otherIssue.id, status: "running", runtimeMode: "native" }).returning(); + await db.insert(nativeRunFinalizations).values({ runId: otherRun.id, companyId: otherRun.companyId, + issueId: otherIssue.id, phase: "observed", attempt: 1 }); + } + await reserveRunBudget(db, f.run.companyId, f.run.id, null); + if (mode !== "stale") setup.run = await read(f.run.id); + await f.completion.failRunSetup(setup); + expect.soft(await read(f.run.id)).toMatchObject({ status: "failed", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + expect.soft(await (await import("../run-cost-accounting.js")).accountRunCost(db, f.run.id)).toBe(true); + expect.soft(await read(f.run.id)).toMatchObject({ costAccountingPending: false, costAccountedAt: expect.any(Date) }); + const [hold] = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)); + expect.soft(hold).toMatchObject({ state: "released", providerStartedAt: expect.any(Date) }); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, f.run.id))).toHaveLength(0); + }); + + it("retains receipt debt when a first execution claim commits during setup failure", async () => { + const f = await fixture({ runtimeMode: "native", costAccountingPending: true, + usageJson: { accountingReceiptReady: false, accountingReceiptSequence: 0 } }); + const setup = f.setup(new ConfigurationIncompleteFailure("preparation failed", { + configurationIncomplete: { reason: "missing_binding" } })); + const { issue } = await admitNativeExecution(f); + await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, + issueId: issue.id, phase: "observed" }); + await reserveRunBudget(db, f.run.companyId, f.run.id, null); + let enter!: () => void, release!: () => void, blockerPid = 0; + const entered = new Promise(resolve => { enter = resolve; }); + const released = new Promise(resolve => { release = resolve; }); + const claimant = db.transaction(async tx => { + const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + blockerPid = backend.pid; + await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)).for("update"); + await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)).for("update"); + enter(); await released; + await tx.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, f.run.id)); + await tx.update(heartbeatRuns).set({ nativePhase: "observed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + }); + await entered; + const failure = f.completion.failRunSetup(setup); + try { + await vi.waitFor(async () => { + const waiting = await db.execute(sql`select pid from pg_stat_activity + where ${blockerPid} = any(pg_blocking_pids(pid)) and wait_event_type = 'Lock'`); + expect(waiting.length).toBeGreaterThan(0); + }, { timeout: 5_000 }); + } finally { release(); await claimant; await failure; } + const failed = await read(f.run.id); + expect.soft((failed.resultJson?.executionRecovery as Record | undefined)?.providerWorkStarted).not.toBe(false); + expect.soft(await (await import("../run-cost-accounting.js")).accountRunCost(db, f.run.id)).toBe(false); + expect.soft(await read(f.run.id)).toMatchObject({ status: "failed", costAccountingPending: true, costAccountedAt: null }); + const [hold] = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)); + expect(hold.state).toBe("held"); + }, 30_000); + + it("lets a completed Stop advance its budget delivery metadata", async () => { + const f = await fixture({ runtimeMode: "native" }); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Budget Stop delivery" }).returning(); + const intentAuditId = randomUUID(), acknowledgementAuditId = randomUUID(); + await db.insert(activityLog).values([ + { id: intentAuditId, companyId: f.run.companyId, actorType: "system", actorId: "native-session-cancellation", + action: "native.cancellation_intent_recorded", entityType: "heartbeat_run", entityId: f.run.id, runId: f.run.id, agentId: f.agent.id }, + { id: acknowledgementAuditId, companyId: f.run.companyId, actorType: "system", actorId: "native-session-cancellation", + action: "native.cancellation_dispatch_acknowledged", entityType: "heartbeat_run", entityId: f.run.id, runId: f.run.id, agentId: f.agent.id }, + ]); + const nativeCancellation = { schema: "paperclip.native-cancellation.v1", companyId: f.run.companyId, runId: f.run.id, issueId: issue.id, + scope: "run", reasonCode: "cancellation_run_only", intentAuditId, acknowledgementAuditId, dispatched: true, dispatchState: "acknowledged" }; + const startupCancellation = { requestedAt: new Date().toISOString(), beforeNativeSelection: false, + budgetEnforcement: { companyId: f.run.companyId, scopeType: "agent", scopeId: f.agent.id, policyId: randomUUID(), version: 1 } }; + await db.update(heartbeatRuns).set({ nativeIssueId: issue.id, resultJson: { nativeCancellation, startupCancellation } }) + .where(eq(heartbeatRuns.id, f.run.id)); + const deliveredAt = new Date().toISOString(); + const outcome = await f.deps.setRunStatusFromLive(f.run.id, "cancelled", ["running"], { + finishedAt: new Date(), resultJson: { nativeCancellation, startupCancellation: { + ...startupCancellation, budgetEnforcement: { ...startupCancellation.budgetEnforcement, deliveredAt } } }, + }); + expect(outcome.updated).toBe(true); + expect(await read(f.run.id)).toMatchObject({ status: "cancelled", resultJson: { + nativeCancellation, startupCancellation: { budgetEnforcement: { deliveredAt } } } }); + }); + it("retains the default branch needed to repair an unresolved workspace base ref", async () => { const f = await fixture(); const error = new UnresolvedWorkspaceBaseRefError({ diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index a5a31d2ccf8..17bf7a54e4f 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -1404,7 +1404,9 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC // the signal's reason when a setup step threw a different error after it. ...controllerLeaseLossEvidence(outerErr), ...controllerLeaseLossResultJson(signal), - executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + ...(!providerDispatchStarted + ? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } + : {}), }; const setupFailureWrite = await setRunStatusIfRunning(runId, "failed", { error: message, @@ -1425,7 +1427,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC : setupFailureResultJson ? { resultJson: setupFailureResultJson } : {}), - }, { error: outerErr, phase: "setup", secretValues: readFailureReportSecrets() }).catch(() => ({ run: null, updated: false as const })); + }, { error: outerErr, phase: "setup", secretValues: readFailureReportSecrets() }, { setupFailure: true }).catch(() => ({ run: null, updated: false as const })); if (!setupFailureWrite.updated) { logger.info( { diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 5cf432a9134..e343d831e3a 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -87,6 +87,7 @@ import { issueDocuments, heartbeatRunEvents, heartbeatRuns, + nativeRunFinalizations, issueApprovals, issueComments, issueRecoveryActions, @@ -645,6 +646,32 @@ function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; } +function preserveNativeExecutionObligationsSql(incoming: Record | null, setupFailure = false) { + const projected = incoming === null ? sql`null::jsonb` : sql`${JSON.stringify(incoming)}::jsonb`; + const result = setupFailure ? sql`case when ${heartbeatRuns.runtimeMode} = 'native' + and exists (select 1 from ${nativeRunFinalizations} + where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.issueId} = ${heartbeatRuns.nativeIssueId} + and ${nativeRunFinalizations.attempt} > 0) + and ${projected}->'executionRecovery'->>'kind' = 'bootstrap' + and ${projected}->'executionRecovery'->>'providerWorkStarted' = 'false' + then ${projected} - 'executionRecovery' else ${projected} end` : projected; + return sql`case when ${heartbeatRuns.runtimeMode} = 'native' + and ${heartbeatRuns.resultJson}->'nativeCancellation'->>'schema' = 'paperclip.native-cancellation.v1' + and ${heartbeatRuns.resultJson}->'nativeCancellation'->>'scope' = 'run' + and ${heartbeatRuns.resultJson}->'nativeCancellation'->>'reasonCode' = 'cancellation_run_only' + and ${heartbeatRuns.resultJson}->'nativeCancellation'->>'runId' = ${heartbeatRuns.id}::text + and ${heartbeatRuns.resultJson}->'nativeCancellation'->>'companyId' = ${heartbeatRuns.companyId}::text + and ${heartbeatRuns.resultJson}->'nativeCancellation'->>'issueId' = ${heartbeatRuns.nativeIssueId}::text + then coalesce(${result}, '{}'::jsonb) || ( + select coalesce(jsonb_object_agg(stop_field.key, stop_field.value), '{}'::jsonb) + from jsonb_each(${heartbeatRuns.resultJson}) as stop_field(key, value) + where stop_field.key in ('nativeCancellation', 'executionCancellation', 'startupCancellation', + 'cancellation', 'cancelledByActorType', 'cancelledByUserId', 'reassignmentStopRequested') + ) else ${result} end`; +} + /** Run lifecycle binds status, progress, liveness, and completion operations. */ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycleDependencies) { const { @@ -1109,6 +1136,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl .set({ status, ...patch, + ...(status === "failed" && patch?.resultJson !== undefined ? { resultJson: preserveNativeExecutionObligationsSql(patch.resultJson) } : {}), executionStatusDeliveryId: randomUUID(), updatedAt: new Date(), }) @@ -1134,8 +1162,9 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl status: string, patch?: Partial, failureReport?: RunFailureReportOptions, + options?: { setupFailure: true }, ) { - return setRunStatusFromLive(runId, status, ["running"], patch, failureReport); + return setRunStatusFromLive(runId, status, ["running"], patch, failureReport, undefined, options); } // Move a run to a new status only when its current status is one of @@ -1150,6 +1179,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl patch?: Partial, failureReport?: RunFailureReportOptions, cancellationCondition?: ReturnType, + options?: { setupFailure: true }, ) { // fromStatuses can name a terminal status as its own source (for example, // an idempotent "still failed" patch), so the write below is not always a @@ -1194,8 +1224,15 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl }, }; } - const updated = await writeRunStatus(status, previousStatus, writer => - previousStatus && previousStatus.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(status) + const updated = await writeRunStatus(status, previousStatus, async writer => { + if (options?.setupFailure && previousStatus?.runtimeMode === "native" && previousStatus.nativeIssueId) { + await writer.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.companyId, previousStatus.companyId), + eq(nativeRunFinalizations.issueId, previousStatus.nativeIssueId))) + .for("update"); + } + return previousStatus && previousStatus.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(status) ? terminalizeLegacyExecution({ db: writer, run: previousStatus, @@ -1210,6 +1247,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl .set({ status, ...patch, + ...(status === "failed" && patch?.resultJson !== undefined ? { resultJson: preserveNativeExecutionObligationsSql(patch.resultJson, options?.setupFailure) } : {}), ...(patch?.usageJson ? { usageJson: preserveAcceptedRunUsageProjectionSql(patch.usageJson) } : {}), executionStatusDeliveryId: randomUUID(), updatedAt: new Date(), @@ -1242,7 +1280,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl .catch((error: unknown) => { if (cancellationCondition) rethrowNativeCancellationLockConflict(error); throw error; - })); + }); + }); if (updated) { publishLiveEvent({