diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 1e9be2453b8..f58ef608b00 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1358,8 +1358,16 @@ Native selection retains the row lock that serializes it with cancellation and controller lease renewal. The executor receives an explicit selection result; lifecycle and ownership callbacks update its teardown state immediately, including when preparation later throws. Public sandbox lifecycle and model fallback helpers -remain re-exported by `heartbeat.ts`. Provider dispatch and final cleanup stay in -the executor. +remain re-exported by `heartbeat.ts`. + +Runtime execution is in `server/src/services/heartbeat/runtime-execution.ts`. +`executeHeartbeatRuntime` dispatches a prepared native session or legacy adapter, +captures usage, finalizes workspace operations, and releases gateway tokens and +instruction copies. The executor supplies the atomic dispatch gate and retains +admission, terminal run status, retry handling, and outer lease cleanup. An explicit +dispatch result preserves early cancellation returns; ownership callbacks update +teardown state before errors propagate. Instruction getters retain updates made +during provider execution. Keep policy changes separate from this extraction. ## Wake Context Delivery diff --git a/doc/composer-stop.md b/doc/composer-stop.md index 5579d55f10e..bd0165a19ca 100644 --- a/doc/composer-stop.md +++ b/doc/composer-stop.md @@ -159,7 +159,9 @@ pnpm exec playwright test --config tests/e2e/playwright-composer-stop.config.ts The suite boots a disposable local-trusted instance on port 3199 (override with `PAPERCLIP_E2E_PORT`). It never attaches to an existing server. Native coverage is explicitly skipped without the fixture; it must not use a logged-in provider -as a fallback. Test companies are archived during cleanup. +as a fallback. The acceptance server pins its Codex command resolver to the fixture +and uses an empty credential home, because an installed Codex dependency takes +precedence over `PATH`. Test companies are archived during cleanup. For each runner, the journey starts a parent, child, and unrelated task, plus a terminal child. It sends while running and verifies the durable queue, clicks diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 10cbb2b81fe..ae86b02f878 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { executeHeartbeatRuntime, NativeSessionResumeScheduledError, NativeWorkspaceFinalizeScheduledError } from "./heartbeat/runtime-execution.js"; import { selectHeartbeatRuntime } from "./heartbeat/runtime-selection.js"; import { createHeartbeatRunCompletion } from "./heartbeat/run-completion.js"; export { @@ -9,7 +10,6 @@ export { import { createHeartbeatScheduling, formatIssueIdentifierLink, - parseNativeSessionGoalControl, } from "./heartbeat/scheduling.js"; import { agentExecutionsHaveStopped } from "./agent-execution-stop.js"; import { @@ -120,12 +120,6 @@ import { applyRunScopedMentionedSkillKeys, MANAGED_GITHUB_TOKEN_KEYS, configuredPaperclipApiBaseUrl, - buildPaperclipRuntimeMcpServers, - createAdapterRuntimeToolAccess, - paperclipApiBaseUrl, - createAdapterRuntimeMcpAccess, - createManagedMcpRunConfig, - revokeHeartbeatRunGatewayTokens, createHeartbeatRunPreparation, } from "./heartbeat/run-preparation.js"; export { @@ -168,9 +162,6 @@ import { buildRunWorkspaceHints, buildReferencedProjectRunObservability, assertGitSensitiveAdapterWorkspaceValid, - isWorkspaceValidationFailure, - fingerprintFinalizeWorkspaceBranchValidation, - attachPaperclipSessionMetadataToSessionParams, type EffectiveRunSessionConfigMetadata, createHeartbeatWorkspaceResolver, } from "./heartbeat/workspaces.js"; @@ -245,19 +236,14 @@ export { import { buildPaperclipTaskMarkdown } from "./heartbeat/task-markdown.js"; export { buildPaperclipTaskMarkdown } from "./heartbeat/task-markdown.js"; -import { recordLegacyWorkspaceRestoreFailure } from "./legacy-execution-recovery.js"; -import { - configuredEnvironmentProjection, -} from "../vendor/paperclip-runner/index.js"; import { decisionModelService } from "./decision-models.js"; import { TASK_QUESTION_GUIDANCE } from "./issue-question-context.js"; import { createAgentIdentityRedactor } from "./agent-identity-redaction.js"; import { agentIdentityService, supportsManagedAgentIdentity } from "./agent-identity.js"; -import { buildAgentIdentityEnv } from "@paperclipai/adapter-utils/server-utils"; import { prepareConnectionInstructionDelivery } from "./connection-instructions.js"; import { resolveAssignedConnectionInstructionsForRun } from "./native-runtime/assigned-mcp-tools.js"; -import { externalObjectService } from "./external-objects.js"; + import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { @@ -270,22 +256,14 @@ import { import { AgentDirectoryReuseInvalidatedError, isAgentDirectoryCopy } from "./agent-directory-working-copies.js"; import type { PaperclipTurnContext } from "@paperclipai/adapter-utils/server-utils"; -import { restoreNativeWorkspaceBestEffort } from "./native-runtime/native-workspace-best-effort.js"; import { - withNativeWorkspaceFinalizationOwnership, NativeWorkspaceFinalizationBusyError, NativeWorkspaceFinalizationOwnershipLostError, - type NativeWorkspaceFinalizationOwnership, } from "./native-runtime/native-workspace-finalization-ownership.js"; import { reserveRunBudget } from "./budget-reservations.js"; import { accountRunCost, createCostAccountingReconciler } from "./run-cost-accounting.js"; -import { createRunUsageRecorder } from "./usage-receipts.js"; -import { applyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-result"; -import { - hasWorkspaceRestoreFailure, -} from "@paperclipai/shared"; import { toolActionDeliveryService } from "./tool-action-delivery.js"; import { liveVoiceExecutionGuidance } from "./voice/voice-execution-guidance.js"; @@ -307,9 +285,6 @@ import { watchLegacyControllerLease, } from "./legacy-controller-lease.js"; -import { - remoteExecutionHasStopped, -} from "./remote-execution-termination.js"; import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js"; import { connectionIntentService } from "./connection-intents.js"; @@ -336,7 +311,6 @@ import { import { adapterExecutionControls, createAdapterExecutionControl, - registerAdapterExecutionControl, } from "./adapter-execution-control.js"; import { executionFailureRetryCount } from "./execution-recovery-attempt.js"; import { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js"; @@ -357,9 +331,7 @@ import { startAdapterExecutionTargetPaperclipBridge, } from "@paperclipai/adapter-utils/execution-target"; import { agentInstructionWorkingCopyService, collectStoppedInstructionCopyWithRetries, instructionWorkingCopyGuidance } from "./agent-instruction-working-copies.js"; -import { - resolveManagedOpenAiBilling, -} from "@paperclipai/adapter-utils"; + import fs from "node:fs/promises"; import path from "node:path"; import { execFile as execFileCallback } from "node:child_process"; @@ -398,11 +370,9 @@ import { nativeRunFinalizations, projects, projectWorkspaces, - workspaceOperations, } from "@paperclipai/db"; -import { conflict, HttpError, notFound } from "../errors.js"; +import { HttpError, notFound } from "../errors.js"; import { - getStartupTraceContext, getStartupTracer, } from "../instrumentation.js"; import { createHostDuplexObservabilityRecorder } from "./duplex-observability-recorder.js"; @@ -425,27 +395,19 @@ import { import { getTaskPlanContext } from "./task-plan-context.js"; import { - buildNativeProviderEnvironment, claimWarmNativeInstructionCopy, nativeSessionWorkspaceScope, reserveWarmNativeInstructionDirectory, dispatchNativeSessionResumptions, - executePaperclipNativeSession, - finalizeNativeRun, - isRunnerIngressAuthorized, - materializeLegacyQuestionResponseWakeProjection, NativeCancellationPendingRecoveryError, NativeControllerDetachedForRestartError, prepareNativeWorkspaceSync, readNativeWorkspaceSyncReference, - recordNativeFinalizationFailure, type NativeRestartRecoveryClaim, resolveHeartbeatNativeRuntimeMode, } from "./native-runtime/index.js"; import { - buildNativeHeartbeatPreparationSpans, - buildNativeWakeIngressSpan, recordFailedSkillPreparation, type NativeRunHistoricalSpan, } from "./native-runtime/native-run-trace.js"; @@ -459,19 +421,16 @@ import { createNativeSessionHandoffLoader } from "./native-runtime/native-sessio import { getRunLogStore, type RunLogHandle } from "./run-log-store.js"; import { providerTraceStore, - PROVIDER_TRACE_MAX_BYTES, } from "./provider-trace-store.js"; import { getServerAdapter, runningProcesses } from "../adapters/index.js"; import type { - AdapterExecutionResult, AdapterInvocationMeta, AdapterRuntimeEvent, } from "../adapters/index.js"; -import { createLocalAgentJwt } from "../agent-auth-jwt.js"; + import { createRuntimeToolsToken } from "../runtime-tools-token.js"; import { parseObject, - asNumber, } from "../adapters/utils.js"; import { @@ -523,11 +482,8 @@ import { buildWorkspaceReadyMetadata, buildWorkspaceReadyPresentation, cleanupExecutionWorkspaceArtifacts, - ensureGitWorktreeBranchCoherent, ensurePersistedExecutionWorkspaceAvailable, ensureRuntimeServicesForRun, - formatManagedGitWorktreeBranchInspection, - inspectManagedGitWorktreeBranch, persistAdapterManagedRuntimeServices, realizeExecutionWorkspace, releaseRuntimeServicesForRun, @@ -542,10 +498,7 @@ import { } from "./workspace-instance-cleanup.js"; import { issueService } from "./issues.js"; import { - blockRunnerGoalRecovery, failRunnerGoalAction, - isRunnerGoalActionCompleted, - settleLiveRunnerGoalBeforeInterrupt, } from "./runner-goals.js"; import { @@ -961,25 +914,6 @@ export interface HeartbeatServiceOptions { }) => Promise; } -class NativeSessionResumeScheduledError extends Error { - constructor(readonly original: unknown) { - super("Native session recovery has been scheduled for the same run."); - this.name = "NativeSessionResumeScheduledError"; - } -} - -class NativeWorkspaceFinalizeScheduledError extends Error { - constructor( - readonly original: unknown, - readonly terminalFailure: boolean, - readonly reasonCode: - "workspace_sync_out_failed" | "workspace_sync_out_unrecoverable", - ) { - super("Native workspace finalization recovery has been scheduled."); - this.name = "NativeWorkspaceFinalizeScheduledError"; - } -} - type WorkspaceReadyCommentWriter = { addComment: ( issueId: string, @@ -5656,918 +5590,106 @@ export function heartbeatService( if (!runtimeSelection.selected) return; const { nativeExecution, nativeRunnerInstanceId, getNativeFreshSessionHandoff } = runtimeSelection; nativeWorkspaceSync = runtimeSelection.nativeWorkspaceSync; - const localAgentJwtScope = - issueRef?.workMode === "skill_test" - ? { kind: "skill_test" as const, issueId: issueRef.id } - : { kind: "standard" as const }; - const authToken = - nativeRuntimeResolution.kind === "legacy" && - adapter.supportsLocalAgentJwt - ? createLocalAgentJwt( - agent.id, - agent.companyId, - agent.adapterType, - run.id, - run.responsibleUserId, - localAgentJwtScope, - ) - : null; - if ( - nativeRuntimeResolution.kind === "legacy" && - adapter.supportsLocalAgentJwt && - !authToken - ) { - logger.warn( - { - companyId: agent.companyId, - agentId: agent.id, - runId: run.id, - adapterType: agent.adapterType, - }, - "local agent jwt secret missing or invalid; running without injected PAPERCLIP_API_KEY", - ); - } - let adapterFinalizeOutcome: "succeeded" | "failed" | null = null; - const inspectFinalizeWorkspaceBranch = async () => { - const workspaceRecord = persistedExecutionWorkspace?.id - ? await executionWorkspacesSvc.getById( - persistedExecutionWorkspace.id, - ) - : persistedExecutionWorkspace; - if (workspaceRecord?.strategyType !== "git_worktree") return null; - - const worktreePath = - readNonEmptyString(workspaceRecord.providerRef) ?? - readNonEmptyString(workspaceRecord.cwd) ?? - readNonEmptyString(executionWorkspace.worktreePath) ?? - readNonEmptyString(executionWorkspace.cwd); - const expectedBranchName = - readNonEmptyString(workspaceRecord.branchName) ?? - readNonEmptyString(executionWorkspace.branchName); - if (!worktreePath || !expectedBranchName) return null; - - const inspection = await inspectManagedGitWorktreeBranch({ - worktreePath, - expectedBranchName, - }); - return { workspaceRecord, inspection }; - }; - const recordWorkspaceFinalize = async ( - status: "succeeded" | "failed", - metadata?: Record, - ) => { - if (adapterFinalizeOutcome) return; - let finalizeBranchMetadata: Record | null = null; - let finalizeBranchRepairMetadata: Record | null = - null; - if (status === "succeeded") { - const branchInspection = await inspectFinalizeWorkspaceBranch(); - if (branchInspection) { - let inspection = branchInspection.inspection; - const initialManagedGitWorktreeBranch = - formatManagedGitWorktreeBranchInspection(inspection); - if ( - !inspection.valid && - inspection.reasonCode === "branch_mismatch" && - inspection.repoRoot - ) { - let repairedExpectedBranchName = inspection.expectedBranchName; - try { - const coherence = await ensureGitWorktreeBranchCoherent({ - db, - repoRoot: inspection.repoRoot, - worktreePath: inspection.worktreePath, - expectedBranchName: inspection.expectedBranchName, - actualBranchName: inspection.actualBranchName, - sourceIssue: issueRef - ? { - id: issueRef.id, - identifier: issueRef.identifier, - title: issueRef.title, - workMode: issueRef.workMode, - } - : null, - executionWorkspaceId: branchInspection.workspaceRecord.id, - heartbeatRunId: run.id, - enableWorkspaceBranchReconcileForward: - resolvedInstanceSettings.experimental - .enableWorkspaceBranchReconcileForward, - enableWorkspaceDirtyQuarantineRepair: - resolvedInstanceSettings.experimental - .enableWorkspaceDirtyQuarantineRepair, - persistForwardReconcile: false, - reconcileOperationPhase: "workspace_finalize", - recorder: workspaceOperationRecorder, - }); - if ( - coherence.branchName && - coherence.branchName !== - branchInspection.workspaceRecord.branchName - ) { - repairedExpectedBranchName = coherence.branchName; - executionWorkspace.branchName = coherence.branchName; - executionWorkspace.warnings.push(...coherence.warnings); - } - } catch (repairErr) { - const workspaceValidationFailure = - isWorkspaceValidationFailure(repairErr) ? repairErr : null; - finalizeBranchMetadata = { - executionWorkspaceId: branchInspection.workspaceRecord.id, - ...initialManagedGitWorktreeBranch, - }; - finalizeBranchRepairMetadata = { - attempted: true, - succeeded: false, - initial: initialManagedGitWorktreeBranch, - reason: - repairErr instanceof Error - ? repairErr.message - : String(repairErr), - }; - await workspaceOperationRecorder.recordOperation({ - phase: "workspace_finalize", - cwd: executionWorkspace.cwd, - metadata: { - adapterType: agent.adapterType, - executionTargetKind: executionTarget?.kind ?? "local", - ...metadata, - managedGitWorktreeBranch: finalizeBranchMetadata, - managedGitWorktreeBranchRepair: - finalizeBranchRepairMetadata, - ...(workspaceValidationFailure?.resultJson - ? { - workspaceValidation: - workspaceValidationFailure.resultJson - .workspaceValidation ?? - workspaceValidationFailure.resultJson, - } - : {}), - }, - run: async () => ({ - status: "failed", - stderr: `Managed git worktree branch check failed: ${repairErr instanceof Error ? repairErr.message : String(repairErr)}\n`, - }), - }); - adapterFinalizeOutcome = "failed"; - throw repairErr; - } - - const repairedInspection = - await inspectManagedGitWorktreeBranch({ - worktreePath: inspection.worktreePath, - expectedBranchName: repairedExpectedBranchName, - repoRoot: inspection.repoRoot, - }); - finalizeBranchRepairMetadata = { - attempted: true, - succeeded: repairedInspection.valid, - initial: initialManagedGitWorktreeBranch, - repaired: - formatManagedGitWorktreeBranchInspection( - repairedInspection, - ), - }; - inspection = repairedInspection; - } - - const managedGitWorktreeBranch = - formatManagedGitWorktreeBranchInspection(inspection); - finalizeBranchMetadata = { - executionWorkspaceId: branchInspection.workspaceRecord.id, - ...managedGitWorktreeBranch, - }; - if (!inspection.valid) { - const workspaceValidationFingerprint = - fingerprintFinalizeWorkspaceBranchValidation({ - issueId: issueRef?.id ?? null, - executionWorkspaceId: branchInspection.workspaceRecord.id, - inspection: managedGitWorktreeBranch, - }); - await workspaceOperationRecorder.recordOperation({ - phase: "workspace_finalize", - cwd: executionWorkspace.cwd, - metadata: { - adapterType: agent.adapterType, - executionTargetKind: executionTarget?.kind ?? "local", - ...metadata, - managedGitWorktreeBranch: finalizeBranchMetadata, - ...(finalizeBranchRepairMetadata - ? { - managedGitWorktreeBranchRepair: - finalizeBranchRepairMetadata, - } - : {}), - }, - run: async () => ({ - status: "failed", - stderr: `Managed git worktree branch check failed: ${inspection.reason ?? "unknown branch mismatch"}\n`, - }), - }); - adapterFinalizeOutcome = "failed"; - throw new WorkspaceValidationFailure( - `Execution workspace ${branchInspection.workspaceRecord.id} expected git worktree branch "${inspection.expectedBranchName}" at "${inspection.worktreePath}", but ${inspection.reason ?? "the checked-out branch could not be verified"}. Record a sanctioned execution-workspace branch transition or restore the workspace branch before completing the run.`, - { - workspaceValidation: { - reason: "git_worktree_branch_incoherence", - fingerprint: workspaceValidationFingerprint, - adapterType: agent.adapterType, - issueId: issueRef?.id ?? null, - issueIdentifier: issueRef?.identifier ?? null, - persistedExecutionWorkspaceId: - branchInspection.workspaceRecord.id, - executionWorkspaceCwd: executionWorkspace.cwd, - managedGitWorktreeBranch: finalizeBranchMetadata, - }, - }, - ); - } - } - } - await workspaceOperationRecorder.recordOperation({ - phase: "workspace_finalize", - cwd: executionWorkspace.cwd, - metadata: { - adapterType: agent.adapterType, - executionTargetKind: executionTarget?.kind ?? "local", - ...metadata, - ...(finalizeBranchMetadata - ? { managedGitWorktreeBranch: finalizeBranchMetadata } - : {}), - ...(finalizeBranchRepairMetadata - ? { - managedGitWorktreeBranchRepair: - finalizeBranchRepairMetadata, - } - : {}), - }, - run: async () => ({ status }), - }); - // Only mark the outcome after the row landed, so a transient write - // failure on the succeeded path can still be recovered by recording - // finalize=failed from the catch path below. - adapterFinalizeOutcome = status; - }; - - const usageRecorder = await createRunUsageRecorder(db, { companyId: run.companyId, runId: run.id, adapterType: agent.adapterType }); - persistUsageCaptureFailure = usageRecorder.persistFailure; - let adapterResult: AdapterExecutionResult; - const runGoalControlRequestId = readNonEmptyString( - context.goalControlRequestId, - ); - try { - if (nativeRuntimeResolution.kind === "native") { - if (!nativeExecution || !nativeRunnerInstanceId) - throw new Error("native_runtime_selection_not_persisted"); - const expectedNativeMcpDigest = - "runtimeContext" in nativeExecution && - nativeExecution.runtimeContext.mcp.bindingId - ? nativeExecution.runtimeContext.mcp.digest - : null; - const nativeMcpServers = await buildPaperclipRuntimeMcpServers({ - db, - agent, - runId: run.id, - expectedAssignmentDigest: expectedNativeMcpDigest, - }); - if ("runtimeContext" in nativeExecution) { - if (nativeMcpServers.length > 1) - throw new Error( - "native MCP realization must produce one aggregate gateway", - ); - const server = nativeMcpServers[0] ?? null; - const digest = server?.connectionId.startsWith("assignment:") - ? server.connectionId.slice("assignment:".length) - : null; - if (digest && digest !== expectedNativeMcpDigest) { - throw new Error("native MCP assignment digest mismatch"); - } - } - const nativeMcpServer = nativeMcpServers[0] ?? null; - let sessionGoalControl = parseNativeSessionGoalControl( - context.runnerGoalControl, - ); - if (runGoalControlRequestId && !sessionGoalControl) { - throw new Error("session_goal_control_payload_invalid"); - } - // A hard restart replays the heartbeat context, not a new user - // action. Do not repeat a completed create/replace/edit (which - // could reactivate or clear a goal that finished while detached). - const completedGoalControl = - sessionGoalControl !== null && - taskKey !== null && - (await isRunnerGoalActionCompleted( - db, - { - companyId: agent.companyId, - agentId: agent.id, - issueId: taskKey, - }, - sessionGoalControl.requestId, - )); - if (completedGoalControl) sessionGoalControl = null; - const nativeDispatchAtMs = Date.now(); - const runCreatedAtMs = run.createdAt.getTime(); - const runStartedAtMs = (run.startedAt ?? run.createdAt).getTime(); - const wakeComments = Array.isArray( - parseObject(context.paperclipWake).comments, - ) - ? (parseObject(context.paperclipWake).comments as unknown[]) - : []; - const wakeIngressSpan = buildNativeWakeIngressSpan({ - runCreatedAtMs, - wakeComments, - attestedQuestionResponseAtMs, - }); - if (wakeIngressSpan) - nativeRunnerPreparationSpans.unshift(wakeIngressSpan); - nativeRunnerPreparationSpans.push( - ...buildNativeHeartbeatPreparationSpans({ - runCreatedAtMs, - runStartedAtMs, - attemptStartedAtMs, - environmentAcquireStartedAtMs, - environmentRealizeEndedAtMs, - nativeDispatchAtMs, - }), - ); - const guardedDispatch = - await dispatchResolvedInteractionContinuationWithAtomicGate( - (markDispatchStarted) => { - return executePaperclipNativeSession({ - db, - execution: nativeExecution, - getFreshSessionHandoff: getNativeFreshSessionHandoff, - refreshTools: context.refreshTools === true, - conversationMode: isConversation(issueContext), - turnTimeoutMs: Math.max(0, asNumber(runtimeConfig.timeoutSec, 0)) * 1_000, - runnerInstanceId: nativeRunnerInstanceId, - leaseOwner: runOptions.nativeLeaseOwner, - restartRecovery: runOptions.nativeRestartRecovery, - backend: - options.nativeSessionBackendFactory?.(nativeExecution), - useRunnerd: agent.adapterType === "paperclip_runner", - adapterType: agent.adapterType, - sessionGoalControl, - resumeSessionGoalHeartbeat: - context.resumeSessionGoalHeartbeat === true || - completedGoalControl, - onGoalCheckpoint: async (snapshot) => { - if (!taskKey) return; - const params = - attachPaperclipSessionMetadataToSessionParams( - { - ...runtimeSessionParamsForAdapter, - sessionId: snapshot.identity.sessionId, - cwd: executionWorkspace.cwd, - }, - configuredModel, - sessionConfigMetadata, - )!; - const displayId = - snapshot.providerSessionId ?? snapshot.sessionId; - await upsertTaskSession({ - companyId: agent.companyId, - agentId: agent.id, - adapterType: agent.adapterType, - taskKey, - sessionParamsJson: params, - sessionDisplayId: displayId, - lastRunId: run.id, - lastError: null, - }); - goalCheckpointSession.current = { params, displayId }; - }, - onLog, - onEvent: onAdapterEvent, - instructionWorkingCopy: nativeInstructionWorkingCopy(), - - onUsage: async receipt => { await usageRecorder.capture(receipt); }, - preparationSpans: nativeRunnerPreparationSpans, - // Bootstrap with executable/home discovery while keeping - // configured provider values and the server-selected - // workspace boundary authoritative. - managedGitHub: !useHostGitHub && githubSelection.configured, - billingIdentity: managedAiRuntime ? { provider: managedAiRuntime.attribution.provider, biller: managedAiRuntime.attribution.provider === "openai" ? resolveManagedOpenAiBilling(managedAiRuntime.config.managedAiRouting)?.biller ?? managedAiRuntime.attribution.provider : managedAiRuntime.attribution.provider, billingType: managedAiRuntime.attribution.method === "subscription" ? "subscription_included" : "metered_api" } : undefined, - managedAiCredentialIdentity: managedAiRuntime?.identity, - managedAiCredentialHome: managedAiRuntime ? String((managedAiRuntime.config.env as Record).CODEX_HOME) : undefined, - dotWorkspaceRoot: nativeExecution.provider.kind === "openai_dot" && resolvedConfig.dotWorkspaceAccess === true ? executionWorkspace.cwd : undefined, - runnerEnvironment: { - ...configuredEnvironmentProjection(configuredTaskEnvironment), - ...buildNativeProviderEnvironment( - adapterEnv, - process.env, - executionWorkspace.cwd, - ), - ...buildAgentIdentityEnv(agentIdentity), - ...(instructionCopy && isAgentDirectoryCopy(instructionCopy) ? { AGENT_HOME: instructionCopy.executionRoot } : {}), - ...(nativeMcpServer - ? { - PAPERCLIP_NATIVE_MCP_NAME: nativeMcpServer.name, - PAPERCLIP_NATIVE_MCP_URL: nativeMcpServer.url, - PAPERCLIP_NATIVE_MCP_TOKEN: nativeMcpServer.token, - } - : {}), - ...(providerTraceCapture - ? { - PAPERCLIP_PROVIDER_TRACE_PATH: - providerTraceCapture.path, - PAPERCLIP_PROVIDER_TRACE_MAX_BYTES: String( - PROVIDER_TRACE_MAX_BYTES, - ), - } - : {}), - }, - runnerExecutionTarget: executionTarget, - runnerIngressAuthorized: isRunnerIngressAuthorized( - nativeRuntimeResolution, - ), - runnerPublicUrl: - runtimeEnv.PAPERCLIP_RUNNER_PUBLIC_URL?.trim() || null, - runnerCaBundlePath: - runtimeEnv.PAPERCLIP_RUNNER_CA_BUNDLE_PATH?.trim() || - null, - runnerRemoteBinaryPath: - runtimeEnv.PAPERCLIP_RUNNER_REMOTE_BINARY_PATH?.trim() || - null, - runnerRemoteCodexPath: - runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH?.trim() || - null, - runnerRemoteCodexNpmSpec: - runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC?.trim() || - null, - runnerRemoteProviderPackPath: - runtimeEnv.PAPERCLIP_RUNNER_REMOTE_PROVIDER_PACK_PATH?.trim() || - null, - stopTaskForReassignment: async (target) => { - await settleLiveRunnerGoalBeforeInterrupt(db, target); - if (!target.runId) return; - const prior = await getRun(target.runId); - if (!prior || prior.companyId !== target.companyId || prior.agentId !== target.agentId) { - throw conflict("Reassignment run binding changed"); - } - const stopped = await cancelRunInternal(target.runId, "Cancelled for task reassignment", { - errorCode: "issue_reassigned", suppressImmediateRecovery: true, - resultJson: { reassignmentStopConfirmed: true }, - }); - if (stopped && ["running", "queued", "scheduled_retry"].includes(stopped.status)) { - throw conflict("The previous run did not stop; reassignment was not applied"); - } - }, - enqueueWakeup, - syncIssueExternalObjects: externalObjectService(db, { - pluginWorkerManager: options.pluginWorkerManager, - enabled: async () => (await instanceSettings.getExperimental()).enableExternalObjects === true, - }).syncIssueSafely, - onSpawn: async (meta) => { - markDispatchStarted(); - await persistRunProcessMetadata(run.id, { ...meta, targetKind: executionTarget?.kind ?? "local" }); - }, - }); - }, - ); - if (!guardedDispatch.dispatched) return; - nativeDispatchStarted = true; - adapterResult = await guardedDispatch.resultPromise; - } else { - const interactionId = readNonEmptyString(context.interactionId); - const legacyQuestionResponse = - issueRef && - interactionId && - readNonEmptyString(context.interactionKind) === - "ask_user_questions" && - readNonEmptyString(context.interactionStatus) === "answered" - ? await materializeLegacyQuestionResponseWakeProjection({ - db, - companyId: agent.companyId, - issueId: issueRef.id, - runId: run.id, - agentId: agent.id, - interactionId, - }) - : null; - // Do not write the answer projection back to `context`: legacy - // adapters need it in their prompt, but the authoritative answers - // remain on the interaction instead of being duplicated in the - // heartbeat run snapshot. - const adapterContext: Record = { - ...context, - ...(legacyQuestionResponse - ? { - [PAPERCLIP_WAKE_PAYLOAD_KEY]: { - ...parseObject(context[PAPERCLIP_WAKE_PAYLOAD_KEY]), - questionResponse: legacyQuestionResponse, - }, - } - : {}), - }; - const runtimeTools = createAdapterRuntimeToolAccess({ - agentId: agent.id, - companyId: agent.companyId, - runId: run.id, - responsibleUserId: run.responsibleUserId, - }); - if (!runtimeTools) { - logger.warn( - { - companyId: agent.companyId, - agentId: agent.id, - runId: run.id, - }, - "runtime connection tools could not be delivered", - ); - } - const runtimeMcpServers = await buildPaperclipRuntimeMcpServers({ - db, - agent, - runId: run.id, - }); - const runtimeToolDelivery = - adapter.runtimeToolDelivery ?? "invocation_context"; - if (runtimeTools && runtimeToolDelivery === "native_mcp") { - runtimeMcpServers.unshift({ - name: "Paperclip connections", - url: runtimeTools.mcpEndpoint, - token: runtimeTools.bearerToken, - connectionId: "paperclip-runtime-tools", - }); - } - if (authToken && configuredPaperclipApiBaseUrl() && issueRef) { - runtimeMcpServers.unshift({ name: "Paperclip projects", url: `${paperclipApiBaseUrl()}/api/mcp/project-tools`, - token: authToken, connectionId: "paperclip-project-tools" }); - } - const runtimeMcp = createAdapterRuntimeMcpAccess(runtimeMcpServers); - if (runtimeTools && runtimeToolDelivery === "invocation_context") { - adapterContext.paperclipRuntimeTools = runtimeTools; - } - const managedMcpConfig = await createManagedMcpRunConfig({ - db, - agent, - runId: run.id, - config: runtimeConfig, - projectId: issueRef?.projectId ?? null, - issueId: issueRef?.id ?? null, - }); - if (managedMcpConfig) { - adapterContext.paperclipManagedMcp = managedMcpConfig; - } - const guardedDispatch = - await dispatchResolvedInteractionContinuationWithAtomicGate( - (markDispatchStarted) => { - legacyAdapterEntered = true; - return withAdapterExecutionPhase(executionPhaseContext, "adapter_execution", () => adapter.execute({ - getFreshSessionHandoff, - agentIdentity, - runId: run.id, - agent, - runtime: runtimeForAdapter, - config: runtimeConfig, - context: adapterContext, - executionContinuation: executionContinuation ?? null, - runtimeCommandSpec: - adapter.getRuntimeCommandSpec?.(runtimeConfig) ?? null, - executionTarget, - executionTransport: remoteExecution - ? { - remoteExecution: remoteExecution as unknown as Record< - string, - unknown - >, - } - : undefined, - runtimeMcp, - runtimeTools, - onLog, - onMeta: onAdapterMeta, - onEvent: onAdapterEvent, - onUsage: async receipt => { await usageRecorder.capture(receipt); }, - onExecutionPhase: executionControl.phases.enter, - startupTraceContext: getStartupTraceContext(), - onRuntimeProgress: async (progress) => { - await recordCurrentHeartbeatRunRuntimeProgress( - run, - progress, - issueId, - ); - }, - onProviderStopped: collectStoppedInstructions, - onDispatch: markDispatchStarted, - signal: executionControl.controller.signal, - ...(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox" ? { - stopRemoteStartup: async () => { - // Scope comes from the running host invocation, never agent - // config. Keep adapter ownership until setup has unwound. - if (!executionControl.controller.signal.aborted) { - throw new Error("Remote startup stop requires a cancelled run"); - } - const release = await envOrchestrator.releaseForRun({ - heartbeatRunId: run.id, - companyId: agent.companyId, - agentId: agent.id, - status: "released", - providerResourceDisposition: "stop_and_retain", - cancelActiveWork: true, - }); - if (release.errors.length || !await remoteExecutionHasStopped(db, agent.companyId, run.id)) { - throw new Error("Could not verify remote startup stopped"); - } - }, - } : {}), - onCancellationReady: async () => { - await registerAdapterExecutionControl(run.id, executionControl); - const current = await getRun(run.id); - if (!current || isHeartbeatRunTerminalStatus(current.status)) { - executionControl.controller.abort(new Error("Run stopped before provider startup")); - } - }, - onSpawn: async (meta) => { - markDispatchStarted(); - await persistRunProcessMetadata(run.id, { - pid: meta.pid, - processGroupId: - "processGroupId" in meta && - typeof meta.processGroupId === "number" - ? meta.processGroupId - : null, - startedAt: meta.startedAt, - }); - }, - authToken: authToken ?? undefined, - })); - }, - ); - if (!guardedDispatch.dispatched) return; - adapterResult = await guardedDispatch.resultPromise; - } - adapterResult = identityRedactor.redact(adapterResult); - if (run.runtimeMode === "legacy" && hasWorkspaceRestoreFailure(adapterResult.resultJson) - && executionTarget?.kind === "remote" && executionTarget.transport === "sandbox") { - requiredWorkspaceRestoreEvidence = { - workspaceRestoreFailure: adapterResult.resultJson!.workspaceRestoreFailure, - ...(adapterResult.resultJson?.workspaceRestoreDiagnostic ? { workspaceRestoreDiagnostic: adapterResult.resultJson.workspaceRestoreDiagnostic } : {}), - }; - // Retention is the fallback even if recording this receipt fails. - providerResourceDispositionForRun = "stop_and_retain"; - await recordLegacyWorkspaceRestoreFailure(db, run, requiredWorkspaceRestoreEvidence, workspaceRestoreSource); - } - for (const stream of ["stdout", "stderr"] as const) { - const tail = identityRedactor.finish(stream); - if (tail) await appendIdentityRedactedLog(stream, tail); - } - if (instructionSave) adapterResult.resultJson = { ...adapterResult.resultJson, instructionSave }; - - if (parseObject(adapterResult.executionRecovery).providerWorkStarted !== false) { - const captured = await usageRecorder.complete(adapterResult); - adapterResult = { ...adapterResult, ...captured, usageComplete: captured.complete }; - } else { - // Stop may already own the terminal result. Preserve its metadata - // while durably recording the proof needed to release admission. - await db.update(heartbeatRuns).set({ costAccountingPending: true, - usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || '{"accountingProviderWorkStarted":false}'::jsonb`, - }).where(and(eq(heartbeatRuns.id, run.id), isNull(heartbeatRuns.costAccountedAt))); - } - adapterResult = applyWorkspaceRestoreFailure(adapterResult); - // A returned result can include a failed restore. Keep the workspace - // barrier closed until required files have been restored. - // If recording the barrier itself fails, propagate as a run failure - // rather than silently leaving dependents stranded behind a missing - // finalize row. - const completeWorkspace = async (ownership?: NativeWorkspaceFinalizationOwnership) => { - try { - if (nativeWorkspaceSync) { - const exported = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( - eq(workspaceOperations.companyId, run.companyId), - eq(workspaceOperations.heartbeatRunId, run.id), - eq(workspaceOperations.phase, "workspace_finalize"), - eq(workspaceOperations.status, "succeeded"), - )).limit(1); - if (exported.length) adapterFinalizeOutcome = "succeeded"; - else await restoreNativeWorkspaceBestEffort({ - db, runId: run.id, assertOwnership: ownership?.assertHeld, - restore: () => nativeWorkspaceSync!.restoreWorkspace(ownership?.assertHeld), - }); - } - await ownership?.assertHeld(); - await db - .update(heartbeatRuns) - .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) - .where( - and( - eq(heartbeatRuns.id, run.id), - eq(heartbeatRuns.status, "running"), - ), - ); - const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded"; - await recordWorkspaceFinalize(workspaceFinalizeStatus); - if (adapterResult.nativeFinalization) { - adapterResult.nativeFinalization.workspaceFinalizeStatus = - workspaceFinalizeStatus; - try { - const finalized = await finalizeNativeRun({ - db, - runId: run.id, - workspaceFinalizeStatus, - preserveProviderAttempt: Boolean(nativeWorkspaceSync), - }); - await dispatchPendingNativeStatusWakeups({ - companyId: run.companyId, - }); - if (finalized.phase === "committed") { - await nativeWorkspaceSync?.cleanup(); - } - } catch (finalizeErr) { - logger.warn( - { err: finalizeErr, runId: run.id }, - "native result persisted but finalization did not apply; the reconciliation loop will retry", - ); - } - } - } catch (error) { - if (ownership) { - await ownership.assertHeld(); - await recordWorkspaceFinalize("failed"); - } - throw error; - } - }; - if (nativeWorkspaceSync) { - const owned = await withNativeWorkspaceFinalizationOwnership({ - db, companyId: run.companyId, runId: run.id, - }, completeWorkspace); - if (!owned.acquired) throw new NativeWorkspaceFinalizationBusyError(); - } else { - await completeWorkspace(); - } - } catch (adapterErr) { - if (adapterErr instanceof NativeCancellationPendingRecoveryError) { - // Durable cancellation is settled by the outer recovery handler; - // it does not imply a failed workspace or a persisted run result. - throw adapterErr; - } - if (adapterErr instanceof NativeWorkspaceFinalizationBusyError - || adapterErr instanceof NativeWorkspaceFinalizationOwnershipLostError) { - nativeWorkspaceFinalizeScheduled = true; - throw adapterErr; - } - if (adapterErr instanceof NativeControllerDetachedForRestartError) { - // Preserve the provider and its run for the new controller. This - // also keeps generic teardown from terminalizing/releasing its lease. - nativeSessionResumeScheduled = true; - throw adapterErr; - } - if (adapterErr instanceof NativeRunnerOwnershipUnverifiedError) { - nativeOwnershipHeld = true; - throw adapterErr; - } - await db - .update(heartbeatRuns) - .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) - .where( - and( - eq(heartbeatRuns.id, run.id), - eq(heartbeatRuns.status, "running"), - ), - ); - if ( - issueRef && - context.resumeSessionGoalHeartbeat === true && - !runGoalControlRequestId - ) { - await blockRunnerGoalRecovery( - db, - { - companyId: run.companyId, - issueId: issueRef.id, - agentId: agent.id, - adapterType: agent.adapterType, - }, - "provider_session_goal_recovery_failed", - ).catch(() => undefined); - } - if (issueRef && runGoalControlRequestId) { - await failRunnerGoalAction( - db, - { - companyId: run.companyId, - issueId: issueRef.id, - agentId: agent.id, - adapterType: agent.adapterType, - }, - runGoalControlRequestId, - adapterErr instanceof Error - ? adapterErr.message - : "session_goal_control_failed", - ).catch(() => undefined); - } - const nativeResumeScheduled = - nativeRuntimeResolution.kind === "native" - ? await db - .select({ - phase: nativeRunFinalizations.phase, - resultId: nativeRunFinalizations.resultId, - }) - .from(nativeRunFinalizations) - .where(eq(nativeRunFinalizations.runId, run.id)) - .limit(1) - .then( - (rows) => - rows[0]?.phase === "retryable_failure" && - rows[0]?.resultId === null, - ) - : false; - if (nativeResumeScheduled) { - nativeSessionResumeScheduled = true; - throw new NativeSessionResumeScheduledError(adapterErr); - } - // Adapter (or its restore finally) threw — or the finalize record - // write itself threw. Either way the workspace may be in a partial - // state. Best-effort record finalize=failed so the dependent readiness - // check keeps the gate closed instead of waking on stale local state, - // and surface the original error to the caller. - try { - await recordWorkspaceFinalize("failed", { - errorMessage: - adapterErr instanceof Error - ? adapterErr.message - : String(adapterErr), - }); - } catch (recordErr) { - logger.warn( - { - err: recordErr, - runId: run.id, - executionWorkspaceId: persistedExecutionWorkspace?.id ?? null, - }, - "failed to record workspace_finalize=failed operation; dependents may remain gated", - ); - } - if (nativeRuntimeResolution.kind === "native") { - const proposedResult = await db - .select({ resultId: nativeRunFinalizations.resultId }) - .from(nativeRunFinalizations) - .where(eq(nativeRunFinalizations.runId, run.id)) - .limit(1) - .then((rows) => rows[0]?.resultId ?? null); - if (proposedResult && nativeWorkspaceSync) { - const workspaceFailureMessage = - adapterErr instanceof Error ? adapterErr.message : ""; - const unrecoverable = - workspaceFailureMessage === - "workspace_sync_out_unrecoverable" || - workspaceFailureMessage.includes("daytona_sandbox_not_found"); - const failure = await recordNativeFinalizationFailure({ - db, - runId: run.id, - error: new Error( - unrecoverable - ? "native_workspace_sync_out_unrecoverable" - : "native_workspace_sync_out_failed", - ), - projectRunStatus: true, - failureScope: "workspace", - permanent: unrecoverable, - }); - nativeWorkspaceFinalizeScheduled = true; - throw new NativeWorkspaceFinalizeScheduledError( - adapterErr, - failure.phase === "terminal_failure", - unrecoverable - ? "workspace_sync_out_unrecoverable" - : "workspace_sync_out_failed", - ); - } - try { - await finalizeNativeRun({ - db, - runId: run.id, - workspaceFinalizeStatus: "failed", - }); - await dispatchPendingNativeStatusWakeups({ - companyId: run.companyId, - }); - } catch (finalizeErr) { - logger.warn( - { err: finalizeErr, runId: run.id }, - "native result could not be marked workspace_failed; the reconciliation loop will retry persisted results", - ); - } - } - throw adapterErr; - } finally { - try { - await revokeHeartbeatRunGatewayTokens({ - db, - companyId: agent.companyId, - runId: run.id, - }); - } catch (revokeErr) { - logger.warn( - { err: revokeErr, runId: run.id, companyId: agent.companyId }, - "failed to revoke heartbeat-run MCP gateway tokens", - ); - } - await nativeInstructionReservation?.release(); - await withAdapterExecutionPhase(executionPhaseContext, "instruction_cleanup", releaseInstructionCopy); - } + const execution = await executeHeartbeatRuntime(db, { + run, agent, options, + task: { + issueRef, + issueContext, + context, + issueId, + taskKey, + executionContinuation, + }, + runtime: { + nativeRuntimeResolution, + nativeExecution, + nativeRunnerInstanceId, + adapter, + runtimeForAdapter, + getNativeFreshSessionHandoff, + getFreshSessionHandoff, + runOptions, + }, + workspace: { + persistedExecutionWorkspace, + executionWorkspace, + executionTarget, + workspaceOperationRecorder, + nativeWorkspaceSync, + workspaceRestoreSource, + remoteExecution, + }, + config: { + runtimeConfig, + resolvedConfig, + resolvedInstanceSettings, + configuredTaskEnvironment, + adapterEnv, + agentIdentity, + managedAiRuntime, + runtimeEnv, + useHostGitHub, + githubSelection, + }, + session: { + runtimeSessionParamsForAdapter, + configuredModel, + sessionConfigMetadata, + goalCheckpointSession, + }, + instructions: { + nativeInstructionWorkingCopy, + collectStoppedInstructions, + nativeInstructionReservation, + releaseInstructionCopy, + get instructionCopy() { return instructionCopy; }, + getInstructionSave: () => instructionSave, + }, + trace: { + providerTraceCapture, + nativeRunnerPreparationSpans, + attestedQuestionResponseAtMs, + attemptStartedAtMs, + environmentAcquireStartedAtMs, + environmentRealizeEndedAtMs, + }, + control: { + executionControl, + executionPhaseContext, + dispatchResolvedInteractionContinuationWithAtomicGate, + }, + output: { + identityRedactor, + appendIdentityRedactedLog, + onLog, + onAdapterEvent, + onAdapterMeta, + }, + services: { + executionWorkspacesSvc, + instanceSettings, + envOrchestrator, + upsertTaskSession, + getRun, + cancelRunInternal, + enqueueWakeup, + persistRunProcessMetadata, + recordCurrentHeartbeatRunRuntimeProgress, + dispatchPendingNativeStatusWakeups, + }, + effects: { + onUsageCaptureReady: (value) => { persistUsageCaptureFailure = value; }, + onNativeDispatchStarted: () => { nativeDispatchStarted = true; }, + onLegacyAdapterEntered: () => { legacyAdapterEntered = true; }, + onNativeWorkspaceFinalizeScheduled: () => { nativeWorkspaceFinalizeScheduled = true; }, + onNativeSessionResumeScheduled: () => { nativeSessionResumeScheduled = true; }, + onNativeOwnershipHeld: () => { nativeOwnershipHeld = true; }, + onProviderResourceDisposition: (value) => { providerResourceDispositionForRun = value; }, + onWorkspaceRestoreFailure: (evidence) => { requiredWorkspaceRestoreEvidence = evidence; }, + }, + }); + if (!execution.dispatched) return; + const { adapterResult } = execution; // Reconcile the referenced-project set against the real remote staging outcome. A referenced // project can pass authorization and clone locally at run prep, then fail to stage into the // sandbox during execution. The run-prep observability above counts such a project as synced, diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts new file mode 100644 index 00000000000..6fdf8fa0e9b --- /dev/null +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -0,0 +1,252 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations, runUsageReceipts, workspaceOperations, type Db } 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"; +import * as legacyRecovery from "../legacy-execution-recovery.js"; +import { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; +import { createAdapterExecutionControl } from "../adapter-execution-control.js"; +import { executionWorkspaceService } from "../execution-workspaces.js"; +import { workspaceOperationService } from "../workspace-operations.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { executePaperclipNativeSession, NativeCancellationPendingRecoveryError, NativeControllerDetachedForRestartError } from "../native-runtime/native-session-executor.js"; +import { NativeRunnerOwnershipUnverifiedError } from "../native-runtime/native-runner-ownership.js"; +import { NativeWorkspaceFinalizationBusyError, NativeWorkspaceFinalizationOwnershipLostError } from "../native-runtime/native-workspace-finalization-ownership.js"; +import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mode.js"; +import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js"; +import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js"; +import { createHeartbeatRunState } from "./run-state.js"; +import { buildEffectiveRunSessionConfigMetadata } from "./workspaces.js"; +import { revokeHeartbeatRunGatewayTokens } from "./run-preparation.js"; +import { executeHeartbeatRuntime, type HeartbeatRuntimeExecutionInput } from "./runtime-execution.js"; + +// Keep real accounting, workspace barriers, and task-session persistence. Providers +// and gateway delivery are external boundaries; browser acceptance covers their wiring. +vi.mock("../native-runtime/native-session-executor.js", async original => ({ + ...await original(), + executePaperclipNativeSession: vi.fn(), +})); +vi.mock("./run-preparation.js", async original => ({ + ...await original(), + buildPaperclipRuntimeMcpServers: vi.fn(async () => []), + createAdapterRuntimeToolAccess: vi.fn(() => null), + createManagedMcpRunConfig: vi.fn(async () => null), + revokeHeartbeatRunGatewayTokens: vi.fn(async () => {}), +})); + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () => { + let database: Awaited>; + let db: Db; + let home: string; + beforeAll(async () => { + home = await mkdtemp(join(tmpdir(), "paperclip-runtime-execution-")); + vi.stubEnv("PAPERCLIP_HOME", home); + database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-execution"); + db = createDb(database.connectionString); + await db.execute(sql`set client_min_messages = warning`); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); }); + beforeEach(async () => { + await db.execute(sql`truncate table companies restart identity cascade`); + vi.clearAllMocks(); + vi.mocked(executePaperclipNativeSession).mockResolvedValue({ exitCode: 0, signal: null, timedOut: false }); + }); + + async function fixture(native = false) { + const [company] = await db.insert(companies).values({ name: "Runtime execution", issuePrefix: "EXE" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Executing agent", status: "running", adapterType: native ? "paperclip_runner" : "process", adapterConfig: { provider: "codex" } }).returning(); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Complete this task", assigneeAgentId: agent.id }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: native ? "native" : "legacy" }).returning(); + const state = createHeartbeatRunState(db); + const executionControl = createAdapterExecutionControl(); + const adapter = { type: "process", label: "Test process", testEnvironment: vi.fn(), execute: vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false })) }; + let instructionSave: Record | null = null; + const nativeExecution = native ? buildNativeExecutionInput({ + companyId: company.id, agentId: agent.id, runId: run.id, issue, + taskPrompt: issue.title, workspace: { id: run.id, cwd: home, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: null, provider: "codex", runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", contract: { revision: "1", objective: issue.title, criteria: [{ id: "output", requirement: issue.title }] } }, + }) : null; + const input: HeartbeatRuntimeExecutionInput = { + run, agent, options: {}, + task: { issueRef: issue, issueContext: null, context: {}, issueId: issue.id, taskKey: issue.id, executionContinuation: null }, + runtime: { nativeExecution, nativeRunnerInstanceId: native ? randomUUID() : null, getNativeFreshSessionHandoff: undefined, + nativeRuntimeResolution: resolveHeartbeatNativeRuntimeMode({ persisted: run, enabled: native, runtimeConfig: {}, adapterConfig: agent.adapterConfig, agent, issue, target: null, workspaceId: null }), + adapter, runtimeForAdapter: { taskKey: issue.id, sessionId: null, sessionParams: null, sessionDisplayId: null }, getFreshSessionHandoff: undefined, runOptions: {}, + }, + workspace: { persistedExecutionWorkspace: null, executionWorkspace: { baseCwd: home, cwd: home, source: "agent_home", projectId: null, workspaceId: null, repoUrl: null, repoRef: null, strategy: "project_primary", branchName: null, worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false }, + executionTarget: null, nativeWorkspaceSync: null, workspaceRestoreSource: undefined, remoteExecution: null, + workspaceOperationRecorder: workspaceOperationService(db).createRecorder({ companyId: company.id, heartbeatRunId: run.id, issueId: issue.id }), + }, + config: { runtimeConfig: {}, resolvedConfig: {}, managedAiRuntime: undefined, + resolvedInstanceSettings: { experimental: { enableWorkspaceBranchReconcileForward: false, enableWorkspaceDirtyQuarantineRepair: false } }, + configuredTaskEnvironment: {}, adapterEnv: {}, agentIdentity: undefined, runtimeEnv: {}, useHostGitHub: false, githubSelection: { configured: false }, + }, + session: { runtimeSessionParamsForAdapter: null, configuredModel: null, + sessionConfigMetadata: await buildEffectiveRunSessionConfigMetadata({ adapterType: agent.adapterType, effectiveAdapterConfig: {}, agentRuntimeConfig: {}, issueOverrides: null, workspaceConfig: null, environment: null, environmentEnv: null, projectEnv: null, routineEnv: null, runtimeSkills: [] }), goalCheckpointSession: { current: null }, + }, + instructions: { instructionCopy: null, getInstructionSave: () => instructionSave, nativeInstructionWorkingCopy: () => undefined, + collectStoppedInstructions: async () => { instructionSave = { state: "saved", revisionId: "during-execution" }; }, + nativeInstructionReservation: null, releaseInstructionCopy: vi.fn(async () => {}), + }, + trace: { providerTraceCapture: null, nativeRunnerPreparationSpans: [], attestedQuestionResponseAtMs: null, attemptStartedAtMs: Date.now(), environmentAcquireStartedAtMs: Date.now(), environmentRealizeEndedAtMs: Date.now() }, + control: { executionControl, executionPhaseContext: { onExecutionPhase: executionControl.phases.enter }, + dispatchResolvedInteractionContinuationWithAtomicGate: async dispatch => ({ dispatched: true, resultPromise: dispatch(vi.fn()) }), + }, + output: { identityRedactor: createAgentIdentityRedactor(), appendIdentityRedactedLog: vi.fn(async () => {}), onLog: vi.fn(async () => {}), onAdapterEvent: vi.fn(async () => {}), onAdapterMeta: vi.fn(async () => {}) }, + services: { ...state, executionWorkspacesSvc: executionWorkspaceService(db), instanceSettings: instanceSettingsService(db), + envOrchestrator: { releaseForRun: vi.fn() }, cancelRunInternal: vi.fn(), enqueueWakeup: vi.fn(), + persistRunProcessMetadata: vi.fn(), recordCurrentHeartbeatRunRuntimeProgress: vi.fn(), dispatchPendingNativeStatusWakeups: vi.fn(), + }, + effects: { onUsageCaptureReady: vi.fn(), onNativeDispatchStarted: vi.fn(), onLegacyAdapterEntered: vi.fn(), onWorkspaceRestoreFailure: vi.fn(), onProviderResourceDisposition: vi.fn(), onNativeWorkspaceFinalizeScheduled: vi.fn(), onNativeSessionResumeScheduled: vi.fn(), onNativeOwnershipHeld: vi.fn() }, + }; + return { input, adapter, run }; + } + const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); + const finalizations = () => db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "workspace_finalize")); + function expectCleanup(input: HeartbeatRuntimeExecutionInput) { + expect(revokeHeartbeatRunGatewayTokens).toHaveBeenCalledWith({ db, companyId: input.agent.companyId, runId: input.run.id }); + expect(input.instructions.releaseInstructionCopy).toHaveBeenCalledOnce(); + } + + it.each([false, true])("preserves the dispatch gate's early return and cleanup (native: %s)", async native => { + const { input, adapter } = await fixture(native); + input.control.dispatchResolvedInteractionContinuationWithAtomicGate = async () => ({ dispatched: false }); + expect(await executeHeartbeatRuntime(db, input)).toEqual({ dispatched: false }); + expect(adapter.execute).not.toHaveBeenCalled(); + expect(executePaperclipNativeSession).not.toHaveBeenCalled(); + expect(input.effects.onNativeDispatchStarted).not.toHaveBeenCalled(); + expect(input.effects.onLegacyAdapterEntered).not.toHaveBeenCalled(); + expect(await finalizations()).toHaveLength(0); + expectCleanup(input); + }); + + it("captures usage and live instruction saves before finalizing a legacy workspace", async () => { + const { input, adapter, run } = await fixture(); + adapter.execute.mockImplementationOnce(async ctx => { + expect(input.effects.onLegacyAdapterEntered).toHaveBeenCalledOnce(); + expect(input.effects.onUsageCaptureReady).toHaveBeenCalledOnce(); + await ctx.onProviderStopped?.(); + return { exitCode: 0, signal: null, timedOut: false, usage: { inputTokens: 10, outputTokens: 4 }, costUsd: 0.01, provider: "test" }; + }); + expect(await executeHeartbeatRuntime(db, input)).toMatchObject({ dispatched: true, adapterResult: { usageComplete: true, resultJson: { instructionSave: { state: "saved", revisionId: "during-execution" } } } }); + expect(await read(run.id)).toMatchObject({ status: "running", costAccountingPending: true, usageJson: { inputTokens: 10, outputTokens: 4, accountingReceiptReady: true } }); + expect(await db.select().from(runUsageReceipts)).toHaveLength(1); + expect(await finalizations()).toMatchObject([{ status: "succeeded" }]); + expectCleanup(input); + }); + + it("preserves Stop metadata and marks no-work accounting without manufacturing a receipt", async () => { + const { input, adapter, run } = await fixture(); + await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: { stoppedByUser: true } }).where(eq(heartbeatRuns.id, run.id)); + adapter.execute.mockResolvedValueOnce({ exitCode: 0, signal: null, timedOut: false, executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }); + await executeHeartbeatRuntime(db, input); + expect(await read(run.id)).toMatchObject({ status: "cancelled", resultJson: { stoppedByUser: true }, usageJson: { accountingProviderWorkStarted: false } }); + expect(await db.select().from(runUsageReceipts)).toHaveLength(0); + expectCleanup(input); + }); + + it("records the failed workspace and keeps the original provider error", async () => { + const { input, adapter } = await fixture(); + const error = new Error("provider failed"); + adapter.execute.mockRejectedValueOnce(error); + await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error); + expect(await finalizations()).toMatchObject([{ status: "failed", metadata: { errorMessage: "provider failed" } }]); + expect(input.effects.onLegacyAdapterEntered).toHaveBeenCalledOnce(); + expectCleanup(input); + }); + + it.each([ + [new NativeCancellationPendingRecoveryError(), null], + [new NativeControllerDetachedForRestartError(), "onNativeSessionResumeScheduled"], + [new NativeRunnerOwnershipUnverifiedError(), "onNativeOwnershipHeld"], + [new NativeWorkspaceFinalizationBusyError(), "onNativeWorkspaceFinalizeScheduled"], + [new NativeWorkspaceFinalizationOwnershipLostError(), "onNativeWorkspaceFinalizeScheduled"], + ] as const)("transfers recovery ownership for %s before cleanup", async (error, effect) => { + const { input } = await fixture(true); + vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error); + input.instructions.releaseInstructionCopy = vi.fn(async () => { + if (effect) expect(input.effects[effect]).toHaveBeenCalledOnce(); + }); + await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error); + expect(input.effects.onNativeDispatchStarted).toHaveBeenCalledOnce(); + expect(input.effects.onLegacyAdapterEntered).not.toHaveBeenCalled(); + expect(await finalizations()).toHaveLength(0); + expectCleanup(input); + }); + + it("dispatches native execution with the selected session and persists its usage", async () => { + const { input, adapter, run } = await fixture(true); + vi.mocked(executePaperclipNativeSession).mockImplementationOnce(async native => { + expect(native.execution).toBe(input.runtime.nativeExecution); + expect(native.runnerInstanceId).toBe(input.runtime.nativeRunnerInstanceId); + await native.onUsage?.({ usage: { inputTokens: 7, outputTokens: 3 }, complete: true }); + return { exitCode: 0, signal: null, timedOut: false, usageComplete: false }; + }); + expect(await executeHeartbeatRuntime(db, input)).toMatchObject({ dispatched: true, adapterResult: { usage: { inputTokens: 7, outputTokens: 3 }, usageComplete: true } }); + expect(adapter.execute).not.toHaveBeenCalled(); + expect(input.effects.onNativeDispatchStarted).toHaveBeenCalledOnce(); + expect(await read(run.id)).toMatchObject({ status: "running", usageJson: { inputTokens: 7, outputTokens: 3, accountingReceiptReady: true } }); + expect(await finalizations()).toMatchObject([{ status: "succeeded" }]); + expectCleanup(input); + }); + + it("retains the same-run retry signal without recording a failed workspace", async () => { + const { input, run } = await fixture(true); + await db.update(heartbeatRuns).set({ nativeIssueId: input.task.issueId }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, issueId: input.task.issueId!, runId: run.id, phase: "retryable_failure" }); + const error = new Error("session connection lost"); + vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error); + await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({ + name: "NativeSessionResumeScheduledError", original: error, + }); + expect(input.effects.onNativeSessionResumeScheduled).toHaveBeenCalledOnce(); + expect(await finalizations()).toHaveLength(0); + expectCleanup(input); + }); + + it("records a failed barrier if the successful finalization write fails", async () => { + const { input } = await fixture(); + const error = new Error("finalization write failed"); + vi.spyOn(input.workspace.workspaceOperationRecorder, "recordOperation").mockRejectedValueOnce(error); + await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error); + expect(await finalizations()).toMatchObject([{ status: "failed" }]); + expectCleanup(input); + }); + + it("retains a failed sandbox restore even when persisting its receipt throws", async () => { + const { input, adapter, run } = await fixture(); + input.workspace.executionTarget = { kind: "remote", transport: "sandbox", remoteCwd: home }; + const evidence = { workspaceRestoreFailure: "restore_failed", workspaceRestoreDiagnostic: { message: "copy-back failed" } }; + adapter.execute.mockResolvedValueOnce({ exitCode: 0, signal: null, timedOut: false, resultJson: evidence }); + const order: string[] = []; + input.effects.onWorkspaceRestoreFailure = vi.fn(() => { order.push("restore evidence"); }); + input.effects.onProviderResourceDisposition = vi.fn(() => { order.push("retain sandbox"); }); + input.instructions.releaseInstructionCopy = vi.fn(async () => { order.push("cleanup"); }); + const error = new Error("restore receipt write failed"); + const record = vi.spyOn(legacyRecovery, "recordLegacyWorkspaceRestoreFailure").mockImplementationOnce(async () => { + order.push("receipt write"); + throw error; + }); + try { + await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error); + expect(record).toHaveBeenCalledWith(db, run, evidence, input.workspace.workspaceRestoreSource); + expect(input.effects.onWorkspaceRestoreFailure).toHaveBeenCalledWith(evidence); + expect(input.effects.onProviderResourceDisposition).toHaveBeenCalledWith("stop_and_retain"); + expect(order).toEqual(["restore evidence", "retain sandbox", "receipt write", "cleanup"]); + expect(await finalizations()).toMatchObject([{ status: "failed" }]); + expectCleanup(input); + } finally { record.mockRestore(); } + }); + + it("runs instruction cleanup even when gateway revocation fails", async () => { + const { input } = await fixture(); + vi.mocked(revokeHeartbeatRunGatewayTokens).mockRejectedValueOnce(new Error("gateway unavailable")); + expect(await executeHeartbeatRuntime(db, input)).toMatchObject({ dispatched: true }); + expectCleanup(input); + }); +}); diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts new file mode 100644 index 00000000000..814ffa2e709 --- /dev/null +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -0,0 +1,1150 @@ +import { parseNativeSessionGoalControl } from "./scheduling.js"; +import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js"; +import { + PAPERCLIP_WAKE_PAYLOAD_KEY, + configuredPaperclipApiBaseUrl, + buildPaperclipRuntimeMcpServers, + createAdapterRuntimeToolAccess, + paperclipApiBaseUrl, + createAdapterRuntimeMcpAccess, + createManagedMcpRunConfig, + revokeHeartbeatRunGatewayTokens, +} from "./run-preparation.js"; +import { + WorkspaceValidationFailure, + isWorkspaceValidationFailure, + fingerprintFinalizeWorkspaceBranchValidation, + attachPaperclipSessionMetadataToSessionParams, +} from "./workspaces.js"; +import { recordLegacyWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; +import { configuredEnvironmentProjection } from "../../vendor/paperclip-runner/index.js"; +import { buildAgentIdentityEnv } from "@paperclipai/adapter-utils/server-utils"; +import { externalObjectService } from "../external-objects.js"; +import { isAgentDirectoryCopy } from "../agent-directory-working-copies.js"; +import { restoreNativeWorkspaceBestEffort } from "../native-runtime/native-workspace-best-effort.js"; +import { + withNativeWorkspaceFinalizationOwnership, + NativeWorkspaceFinalizationBusyError, + NativeWorkspaceFinalizationOwnershipLostError, + type NativeWorkspaceFinalizationOwnership, +} from "../native-runtime/native-workspace-finalization-ownership.js"; +import { createRunUsageRecorder } from "../usage-receipts.js"; +import { applyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-result"; +import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; +import { isConversation } from "../agent-conversations.js"; +import { withAdapterExecutionPhase } from "@paperclipai/adapter-utils/execution-phase"; +import { remoteExecutionHasStopped } from "../remote-execution-termination.js"; +import { registerAdapterExecutionControl } from "../adapter-execution-control.js"; +import { resolveManagedOpenAiBilling } from "@paperclipai/adapter-utils"; +import { and, eq, isNull, sql } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db"; +import { conflict } from "../../errors.js"; +import { getStartupTraceContext } from "../../instrumentation.js"; +import { logger } from "../../middleware/logger.js"; +import { + buildNativeProviderEnvironment, + executePaperclipNativeSession, + finalizeNativeRun, + isRunnerIngressAuthorized, + materializeLegacyQuestionResponseWakeProjection, + NativeCancellationPendingRecoveryError, + NativeControllerDetachedForRestartError, + recordNativeFinalizationFailure, +} from "../native-runtime/index.js"; +import { + buildNativeHeartbeatPreparationSpans, + buildNativeWakeIngressSpan, +} from "../native-runtime/native-run-trace.js"; +import { PROVIDER_TRACE_MAX_BYTES } from "../provider-trace-store.js"; +import type { AdapterExecutionResult } from "../../adapters/index.js"; +import { createLocalAgentJwt } from "../../agent-auth-jwt.js"; +import { parseObject, asNumber } from "../../adapters/utils.js"; +import { NativeRunnerOwnershipUnverifiedError } from "../native-runtime/native-runner-ownership.js"; +import { + ensureGitWorktreeBranchCoherent, + formatManagedGitWorktreeBranchInspection, + inspectManagedGitWorktreeBranch, +} from "../workspace-runtime.js"; +import { + blockRunnerGoalRecovery, + failRunnerGoalAction, + isRunnerGoalActionCompleted, + settleLiveRunnerGoalBeforeInterrupt, +} from "../runner-goals.js"; +import type { Db, agents, executionWorkspaces } from "@paperclipai/db"; +import type { AdapterExecutionContext, ServerAdapterModule } from "../../adapters/index.js"; +import type { HeartbeatRuntimeSelectionInput, HeartbeatRuntimeSelectionResult } from "./runtime-selection.js"; +import type { createHeartbeatRunState } from "./run-state.js"; +import type { createHeartbeatLifecycle } from "./run-lifecycle.js"; +import type { createHeartbeatRunControl } from "./run-control.js"; +import type { createHeartbeatQueue } from "./queue.js"; +import type { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; +import type { createAdapterExecutionControl } from "../adapter-execution-control.js"; +import type { executionWorkspaceService } from "../execution-workspaces.js"; +import type { workspaceOperationService } from "../workspace-operations.js"; +import type { environmentRunOrchestrator } from "../environment-run-orchestrator.js"; +import type { instanceSettingsService } from "../instance-settings.js"; +import type { EffectiveRunSessionConfigMetadata } from "./workspaces.js"; +import type { ProviderResourceDisposition } from "../environment-runtime.js"; +import type { NativeExecutionInput, NativeSessionBackend } from "../../vendor/paperclip-runner/index.js"; +import type { PluginWorkerManager } from "../plugin-worker-manager.js"; +import type { RealizedExecutionWorkspace } from "../workspace-runtime.js"; +import type { reserveWarmNativeInstructionDirectory } from "../native-runtime/index.js"; + +type Selection = HeartbeatRuntimeSelectionInput; +type Selected = Extract; +type NativeSessionInput = Parameters[0]; +type RunState = ReturnType; +type Lifecycle = ReturnType; + +/** Dispatches a prepared run; admission and terminal run status remain owned by the executor. */ +export interface HeartbeatRuntimeExecutionInput { + run: typeof heartbeatRuns.$inferSelect; + agent: typeof agents.$inferSelect; + options: { + nativeSessionBackendFactory?: (execution: NativeExecutionInput) => NativeSessionBackend; + pluginWorkerManager?: PluginWorkerManager; + }; + task: Pick & { + issueRef: (NonNullable & { projectId: string | null }) | null; + issueId: string | null; + taskKey: string | null; + }; + runtime: Pick & { + nativeRuntimeResolution: Selection["nativeRuntimeResolution"]; + adapter: ServerAdapterModule; + runtimeForAdapter: AdapterExecutionContext["runtime"]; + getFreshSessionHandoff: AdapterExecutionContext["getFreshSessionHandoff"]; + runOptions: Selection["recovery"]; + }; + workspace: { + persistedExecutionWorkspace: typeof executionWorkspaces.$inferSelect | null; + executionWorkspace: RealizedExecutionWorkspace; + executionTarget: Selection["workspace"]["executionTarget"]; + workspaceOperationRecorder: ReturnType["createRecorder"]>; + nativeWorkspaceSync: Selected["nativeWorkspaceSync"]; + workspaceRestoreSource: Parameters[3]; + remoteExecution: unknown; + }; + config: Pick & { + resolvedInstanceSettings: { experimental: Pick["getExperimental"]>>, "enableWorkspaceBranchReconcileForward" | "enableWorkspaceDirtyQuarantineRepair"> }; + configuredTaskEnvironment: Parameters[0]; + adapterEnv: Parameters[0]; + agentIdentity: AdapterExecutionContext["agentIdentity"]; + runtimeEnv: Selection["runtimeEnv"]; + useHostGitHub: boolean; + githubSelection: { configured: boolean }; + }; + session: { + runtimeSessionParamsForAdapter: Record | null; + configuredModel: string | null; + sessionConfigMetadata: EffectiveRunSessionConfigMetadata; + goalCheckpointSession: { current: { params: Record; displayId: string } | null }; + }; + instructions: { + // Instruction callbacks can replace the copy and receipt while a provider is running. + readonly instructionCopy: Selection["config"]["instructionCopy"]; + getInstructionSave: () => Record | null; + nativeInstructionWorkingCopy: () => NativeSessionInput["instructionWorkingCopy"]; + collectStoppedInstructions: AdapterExecutionContext["onProviderStopped"]; + nativeInstructionReservation: Awaited>; + releaseInstructionCopy: () => Promise; + }; + trace: { + providerTraceCapture: { path: string } | null; + nativeRunnerPreparationSpans: NonNullable; + attestedQuestionResponseAtMs: Parameters[0]["attestedQuestionResponseAtMs"]; + attemptStartedAtMs: number; + environmentAcquireStartedAtMs: number; + environmentRealizeEndedAtMs: number; + }; + control: { + executionControl: ReturnType; + executionPhaseContext: Parameters[0]; + dispatchResolvedInteractionContinuationWithAtomicGate: (dispatch: (markDispatchStarted: () => void) => Promise) => Promise< + { dispatched: true; resultPromise: Promise } | { dispatched: false } + >; + }; + output: Pick & { + onAdapterEvent: NonNullable; + onAdapterMeta: AdapterExecutionContext["onMeta"]; + identityRedactor: ReturnType; + appendIdentityRedactedLog: AdapterExecutionContext["onLog"]; + }; + services: Pick & + Pick & + Pick, "cancelRunInternal"> & + Pick, "enqueueWakeup" | "dispatchPendingNativeStatusWakeups"> & { + executionWorkspacesSvc: Pick, "getById">; + instanceSettings: Pick, "getExperimental">; + envOrchestrator: Pick, "releaseForRun">; + }; + // Teardown must observe these updates immediately, including when execution later throws. + effects: { + onUsageCaptureReady: (persistFailure: () => Promise) => void; + onNativeDispatchStarted: () => void; + onLegacyAdapterEntered: () => void; + onWorkspaceRestoreFailure: (evidence: Record) => void; + onProviderResourceDisposition: (disposition: ProviderResourceDisposition) => void; + onNativeWorkspaceFinalizeScheduled: () => void; + onNativeSessionResumeScheduled: () => void; + onNativeOwnershipHeld: () => void; + }; +} + +export type HeartbeatRuntimeExecutionResult = { dispatched: false } | { + dispatched: true; + adapterResult: AdapterExecutionResult; +}; + +export class NativeSessionResumeScheduledError extends Error { + constructor(readonly original: unknown) { + super("Native session recovery has been scheduled for the same run."); + this.name = "NativeSessionResumeScheduledError"; + } +} + +export class NativeWorkspaceFinalizeScheduledError extends Error { + constructor( + readonly original: unknown, + readonly terminalFailure: boolean, + readonly reasonCode: + "workspace_sync_out_failed" | "workspace_sync_out_unrecoverable", + ) { + super("Native workspace finalization recovery has been scheduled."); + this.name = "NativeWorkspaceFinalizeScheduledError"; + } +} + +function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExecutionInput): Promise { + const { run, agent, options } = input; + const { issueRef, issueContext, context, issueId, taskKey, executionContinuation } = input.task; + const { nativeRuntimeResolution, nativeExecution, nativeRunnerInstanceId, adapter, runtimeForAdapter, getNativeFreshSessionHandoff, getFreshSessionHandoff, runOptions } = input.runtime; + const { persistedExecutionWorkspace, executionWorkspace, executionTarget, workspaceOperationRecorder, nativeWorkspaceSync, workspaceRestoreSource, remoteExecution } = input.workspace; + const { runtimeConfig, resolvedConfig, resolvedInstanceSettings, configuredTaskEnvironment, adapterEnv, agentIdentity, managedAiRuntime, runtimeEnv, useHostGitHub, githubSelection } = input.config; + const { runtimeSessionParamsForAdapter, configuredModel, sessionConfigMetadata, goalCheckpointSession } = input.session; + const { nativeInstructionWorkingCopy, collectStoppedInstructions, nativeInstructionReservation, releaseInstructionCopy } = input.instructions; + const { providerTraceCapture, nativeRunnerPreparationSpans, attestedQuestionResponseAtMs, attemptStartedAtMs, environmentAcquireStartedAtMs, environmentRealizeEndedAtMs } = input.trace; + const { executionControl, executionPhaseContext, dispatchResolvedInteractionContinuationWithAtomicGate } = input.control; + const { identityRedactor, appendIdentityRedactedLog, onLog, onAdapterEvent, onAdapterMeta } = input.output; + const { executionWorkspacesSvc, instanceSettings, envOrchestrator, upsertTaskSession, getRun, cancelRunInternal, enqueueWakeup, persistRunProcessMetadata, recordCurrentHeartbeatRunRuntimeProgress, dispatchPendingNativeStatusWakeups } = input.services; + const localAgentJwtScope = + issueRef?.workMode === "skill_test" + ? { kind: "skill_test" as const, issueId: issueRef.id } + : { kind: "standard" as const }; + const authToken = + nativeRuntimeResolution.kind === "legacy" && + adapter.supportsLocalAgentJwt + ? createLocalAgentJwt( + agent.id, + agent.companyId, + agent.adapterType, + run.id, + run.responsibleUserId, + localAgentJwtScope, + ) + : null; + if ( + nativeRuntimeResolution.kind === "legacy" && + adapter.supportsLocalAgentJwt && + !authToken + ) { + logger.warn( + { + companyId: agent.companyId, + agentId: agent.id, + runId: run.id, + adapterType: agent.adapterType, + }, + "local agent jwt secret missing or invalid; running without injected PAPERCLIP_API_KEY", + ); + } + let adapterFinalizeOutcome: "succeeded" | "failed" | null = null; + const inspectFinalizeWorkspaceBranch = async () => { + const workspaceRecord = persistedExecutionWorkspace?.id + ? await executionWorkspacesSvc.getById( + persistedExecutionWorkspace.id, + ) + : persistedExecutionWorkspace; + if (workspaceRecord?.strategyType !== "git_worktree") return null; + + const worktreePath = + readNonEmptyString(workspaceRecord.providerRef) ?? + readNonEmptyString(workspaceRecord.cwd) ?? + readNonEmptyString(executionWorkspace.worktreePath) ?? + readNonEmptyString(executionWorkspace.cwd); + const expectedBranchName = + readNonEmptyString(workspaceRecord.branchName) ?? + readNonEmptyString(executionWorkspace.branchName); + if (!worktreePath || !expectedBranchName) return null; + + const inspection = await inspectManagedGitWorktreeBranch({ + worktreePath, + expectedBranchName, + }); + return { workspaceRecord, inspection }; + }; + const recordWorkspaceFinalize = async ( + status: "succeeded" | "failed", + metadata?: Record, + ) => { + if (adapterFinalizeOutcome) return; + let finalizeBranchMetadata: Record | null = null; + let finalizeBranchRepairMetadata: Record | null = + null; + if (status === "succeeded") { + const branchInspection = await inspectFinalizeWorkspaceBranch(); + if (branchInspection) { + let inspection = branchInspection.inspection; + const initialManagedGitWorktreeBranch = + formatManagedGitWorktreeBranchInspection(inspection); + if ( + !inspection.valid && + inspection.reasonCode === "branch_mismatch" && + inspection.repoRoot + ) { + let repairedExpectedBranchName = inspection.expectedBranchName; + try { + const coherence = await ensureGitWorktreeBranchCoherent({ + db, + repoRoot: inspection.repoRoot, + worktreePath: inspection.worktreePath, + expectedBranchName: inspection.expectedBranchName, + actualBranchName: inspection.actualBranchName, + sourceIssue: issueRef + ? { + id: issueRef.id, + identifier: issueRef.identifier, + title: issueRef.title, + workMode: issueRef.workMode, + } + : null, + executionWorkspaceId: branchInspection.workspaceRecord.id, + heartbeatRunId: run.id, + enableWorkspaceBranchReconcileForward: + resolvedInstanceSettings.experimental + .enableWorkspaceBranchReconcileForward, + enableWorkspaceDirtyQuarantineRepair: + resolvedInstanceSettings.experimental + .enableWorkspaceDirtyQuarantineRepair, + persistForwardReconcile: false, + reconcileOperationPhase: "workspace_finalize", + recorder: workspaceOperationRecorder, + }); + if ( + coherence.branchName && + coherence.branchName !== + branchInspection.workspaceRecord.branchName + ) { + repairedExpectedBranchName = coherence.branchName; + executionWorkspace.branchName = coherence.branchName; + executionWorkspace.warnings.push(...coherence.warnings); + } + } catch (repairErr) { + const workspaceValidationFailure = + isWorkspaceValidationFailure(repairErr) ? repairErr : null; + finalizeBranchMetadata = { + executionWorkspaceId: branchInspection.workspaceRecord.id, + ...initialManagedGitWorktreeBranch, + }; + finalizeBranchRepairMetadata = { + attempted: true, + succeeded: false, + initial: initialManagedGitWorktreeBranch, + reason: + repairErr instanceof Error + ? repairErr.message + : String(repairErr), + }; + await workspaceOperationRecorder.recordOperation({ + phase: "workspace_finalize", + cwd: executionWorkspace.cwd, + metadata: { + adapterType: agent.adapterType, + executionTargetKind: executionTarget?.kind ?? "local", + ...metadata, + managedGitWorktreeBranch: finalizeBranchMetadata, + managedGitWorktreeBranchRepair: + finalizeBranchRepairMetadata, + ...(workspaceValidationFailure?.resultJson + ? { + workspaceValidation: + workspaceValidationFailure.resultJson + .workspaceValidation ?? + workspaceValidationFailure.resultJson, + } + : {}), + }, + run: async () => ({ + status: "failed", + stderr: `Managed git worktree branch check failed: ${repairErr instanceof Error ? repairErr.message : String(repairErr)}\n`, + }), + }); + adapterFinalizeOutcome = "failed"; + throw repairErr; + } + + const repairedInspection = + await inspectManagedGitWorktreeBranch({ + worktreePath: inspection.worktreePath, + expectedBranchName: repairedExpectedBranchName, + repoRoot: inspection.repoRoot, + }); + finalizeBranchRepairMetadata = { + attempted: true, + succeeded: repairedInspection.valid, + initial: initialManagedGitWorktreeBranch, + repaired: + formatManagedGitWorktreeBranchInspection( + repairedInspection, + ), + }; + inspection = repairedInspection; + } + + const managedGitWorktreeBranch = + formatManagedGitWorktreeBranchInspection(inspection); + finalizeBranchMetadata = { + executionWorkspaceId: branchInspection.workspaceRecord.id, + ...managedGitWorktreeBranch, + }; + if (!inspection.valid) { + const workspaceValidationFingerprint = + fingerprintFinalizeWorkspaceBranchValidation({ + issueId: issueRef?.id ?? null, + executionWorkspaceId: branchInspection.workspaceRecord.id, + inspection: managedGitWorktreeBranch, + }); + await workspaceOperationRecorder.recordOperation({ + phase: "workspace_finalize", + cwd: executionWorkspace.cwd, + metadata: { + adapterType: agent.adapterType, + executionTargetKind: executionTarget?.kind ?? "local", + ...metadata, + managedGitWorktreeBranch: finalizeBranchMetadata, + ...(finalizeBranchRepairMetadata + ? { + managedGitWorktreeBranchRepair: + finalizeBranchRepairMetadata, + } + : {}), + }, + run: async () => ({ + status: "failed", + stderr: `Managed git worktree branch check failed: ${inspection.reason ?? "unknown branch mismatch"}\n`, + }), + }); + adapterFinalizeOutcome = "failed"; + throw new WorkspaceValidationFailure( + `Execution workspace ${branchInspection.workspaceRecord.id} expected git worktree branch "${inspection.expectedBranchName}" at "${inspection.worktreePath}", but ${inspection.reason ?? "the checked-out branch could not be verified"}. Record a sanctioned execution-workspace branch transition or restore the workspace branch before completing the run.`, + { + workspaceValidation: { + reason: "git_worktree_branch_incoherence", + fingerprint: workspaceValidationFingerprint, + adapterType: agent.adapterType, + issueId: issueRef?.id ?? null, + issueIdentifier: issueRef?.identifier ?? null, + persistedExecutionWorkspaceId: + branchInspection.workspaceRecord.id, + executionWorkspaceCwd: executionWorkspace.cwd, + managedGitWorktreeBranch: finalizeBranchMetadata, + }, + }, + ); + } + } + } + await workspaceOperationRecorder.recordOperation({ + phase: "workspace_finalize", + cwd: executionWorkspace.cwd, + metadata: { + adapterType: agent.adapterType, + executionTargetKind: executionTarget?.kind ?? "local", + ...metadata, + ...(finalizeBranchMetadata + ? { managedGitWorktreeBranch: finalizeBranchMetadata } + : {}), + ...(finalizeBranchRepairMetadata + ? { + managedGitWorktreeBranchRepair: + finalizeBranchRepairMetadata, + } + : {}), + }, + run: async () => ({ status }), + }); + // Only mark the outcome after the row landed, so a transient write + // failure on the succeeded path can still be recovered by recording + // finalize=failed from the catch path below. + adapterFinalizeOutcome = status; + }; + + const usageRecorder = await createRunUsageRecorder(db, { companyId: run.companyId, runId: run.id, adapterType: agent.adapterType }); + input.effects.onUsageCaptureReady(usageRecorder.persistFailure); + let adapterResult: AdapterExecutionResult; + const runGoalControlRequestId = readNonEmptyString( + context.goalControlRequestId, + ); + try { + if (nativeRuntimeResolution.kind === "native") { + if (!nativeExecution || !nativeRunnerInstanceId) + throw new Error("native_runtime_selection_not_persisted"); + const expectedNativeMcpDigest = + "runtimeContext" in nativeExecution && + nativeExecution.runtimeContext.mcp.bindingId + ? nativeExecution.runtimeContext.mcp.digest + : null; + const nativeMcpServers = await buildPaperclipRuntimeMcpServers({ + db, + agent, + runId: run.id, + expectedAssignmentDigest: expectedNativeMcpDigest, + }); + if ("runtimeContext" in nativeExecution) { + if (nativeMcpServers.length > 1) + throw new Error( + "native MCP realization must produce one aggregate gateway", + ); + const server = nativeMcpServers[0] ?? null; + const digest = server?.connectionId.startsWith("assignment:") + ? server.connectionId.slice("assignment:".length) + : null; + if (digest && digest !== expectedNativeMcpDigest) { + throw new Error("native MCP assignment digest mismatch"); + } + } + const nativeMcpServer = nativeMcpServers[0] ?? null; + let sessionGoalControl = parseNativeSessionGoalControl( + context.runnerGoalControl, + ); + if (runGoalControlRequestId && !sessionGoalControl) { + throw new Error("session_goal_control_payload_invalid"); + } + // A hard restart replays the heartbeat context, not a new user + // action. Do not repeat a completed create/replace/edit (which + // could reactivate or clear a goal that finished while detached). + const completedGoalControl = + sessionGoalControl !== null && + taskKey !== null && + (await isRunnerGoalActionCompleted( + db, + { + companyId: agent.companyId, + agentId: agent.id, + issueId: taskKey, + }, + sessionGoalControl.requestId, + )); + if (completedGoalControl) sessionGoalControl = null; + const nativeDispatchAtMs = Date.now(); + const runCreatedAtMs = run.createdAt.getTime(); + const runStartedAtMs = (run.startedAt ?? run.createdAt).getTime(); + const wakeComments = Array.isArray( + parseObject(context.paperclipWake).comments, + ) + ? (parseObject(context.paperclipWake).comments as unknown[]) + : []; + const wakeIngressSpan = buildNativeWakeIngressSpan({ + runCreatedAtMs, + wakeComments, + attestedQuestionResponseAtMs, + }); + if (wakeIngressSpan) + nativeRunnerPreparationSpans.unshift(wakeIngressSpan); + nativeRunnerPreparationSpans.push( + ...buildNativeHeartbeatPreparationSpans({ + runCreatedAtMs, + runStartedAtMs, + attemptStartedAtMs, + environmentAcquireStartedAtMs, + environmentRealizeEndedAtMs, + nativeDispatchAtMs, + }), + ); + const guardedDispatch = + await dispatchResolvedInteractionContinuationWithAtomicGate( + (markDispatchStarted) => { + return executePaperclipNativeSession({ + db, + execution: nativeExecution, + getFreshSessionHandoff: getNativeFreshSessionHandoff, + refreshTools: context.refreshTools === true, + conversationMode: isConversation(issueContext), + turnTimeoutMs: Math.max(0, asNumber(runtimeConfig.timeoutSec, 0)) * 1_000, + runnerInstanceId: nativeRunnerInstanceId, + leaseOwner: runOptions.nativeLeaseOwner, + restartRecovery: runOptions.nativeRestartRecovery, + backend: + options.nativeSessionBackendFactory?.(nativeExecution), + useRunnerd: agent.adapterType === "paperclip_runner", + adapterType: agent.adapterType, + sessionGoalControl, + resumeSessionGoalHeartbeat: + context.resumeSessionGoalHeartbeat === true || + completedGoalControl, + onGoalCheckpoint: async (snapshot) => { + if (!taskKey) return; + const params = + attachPaperclipSessionMetadataToSessionParams( + { + ...runtimeSessionParamsForAdapter, + sessionId: snapshot.identity.sessionId, + cwd: executionWorkspace.cwd, + }, + configuredModel, + sessionConfigMetadata, + )!; + const displayId = + snapshot.providerSessionId ?? snapshot.sessionId; + await upsertTaskSession({ + companyId: agent.companyId, + agentId: agent.id, + adapterType: agent.adapterType, + taskKey, + sessionParamsJson: params, + sessionDisplayId: displayId, + lastRunId: run.id, + lastError: null, + }); + goalCheckpointSession.current = { params, displayId }; + }, + onLog, + onEvent: onAdapterEvent, + instructionWorkingCopy: nativeInstructionWorkingCopy(), + + onUsage: async receipt => { await usageRecorder.capture(receipt); }, + preparationSpans: nativeRunnerPreparationSpans, + // Bootstrap with executable/home discovery while keeping + // configured provider values and the server-selected + // workspace boundary authoritative. + managedGitHub: !useHostGitHub && githubSelection.configured, + billingIdentity: managedAiRuntime ? { provider: managedAiRuntime.attribution.provider, biller: managedAiRuntime.attribution.provider === "openai" ? resolveManagedOpenAiBilling(managedAiRuntime.config.managedAiRouting)?.biller ?? managedAiRuntime.attribution.provider : managedAiRuntime.attribution.provider, billingType: managedAiRuntime.attribution.method === "subscription" ? "subscription_included" : "metered_api" } : undefined, + managedAiCredentialIdentity: managedAiRuntime?.identity, + managedAiCredentialHome: managedAiRuntime ? String((managedAiRuntime.config.env as Record).CODEX_HOME) : undefined, + dotWorkspaceRoot: nativeExecution.provider.kind === "openai_dot" && resolvedConfig.dotWorkspaceAccess === true ? executionWorkspace.cwd : undefined, + runnerEnvironment: { + ...configuredEnvironmentProjection(configuredTaskEnvironment), + ...buildNativeProviderEnvironment( + adapterEnv, + process.env, + executionWorkspace.cwd, + ), + ...buildAgentIdentityEnv(agentIdentity), + ...(input.instructions.instructionCopy && isAgentDirectoryCopy(input.instructions.instructionCopy) ? { AGENT_HOME: input.instructions.instructionCopy.executionRoot } : {}), + ...(nativeMcpServer + ? { + PAPERCLIP_NATIVE_MCP_NAME: nativeMcpServer.name, + PAPERCLIP_NATIVE_MCP_URL: nativeMcpServer.url, + PAPERCLIP_NATIVE_MCP_TOKEN: nativeMcpServer.token, + } + : {}), + ...(providerTraceCapture + ? { + PAPERCLIP_PROVIDER_TRACE_PATH: + providerTraceCapture.path, + PAPERCLIP_PROVIDER_TRACE_MAX_BYTES: String( + PROVIDER_TRACE_MAX_BYTES, + ), + } + : {}), + }, + runnerExecutionTarget: executionTarget, + runnerIngressAuthorized: isRunnerIngressAuthorized( + nativeRuntimeResolution, + ), + runnerPublicUrl: + runtimeEnv.PAPERCLIP_RUNNER_PUBLIC_URL?.trim() || null, + runnerCaBundlePath: + runtimeEnv.PAPERCLIP_RUNNER_CA_BUNDLE_PATH?.trim() || + null, + runnerRemoteBinaryPath: + runtimeEnv.PAPERCLIP_RUNNER_REMOTE_BINARY_PATH?.trim() || + null, + runnerRemoteCodexPath: + runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH?.trim() || + null, + runnerRemoteCodexNpmSpec: + runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC?.trim() || + null, + runnerRemoteProviderPackPath: + runtimeEnv.PAPERCLIP_RUNNER_REMOTE_PROVIDER_PACK_PATH?.trim() || + null, + stopTaskForReassignment: async (target) => { + await settleLiveRunnerGoalBeforeInterrupt(db, target); + if (!target.runId) return; + const prior = await getRun(target.runId); + if (!prior || prior.companyId !== target.companyId || prior.agentId !== target.agentId) { + throw conflict("Reassignment run binding changed"); + } + const stopped = await cancelRunInternal(target.runId, "Cancelled for task reassignment", { + errorCode: "issue_reassigned", suppressImmediateRecovery: true, + resultJson: { reassignmentStopConfirmed: true }, + }); + if (stopped && ["running", "queued", "scheduled_retry"].includes(stopped.status)) { + throw conflict("The previous run did not stop; reassignment was not applied"); + } + }, + enqueueWakeup, + syncIssueExternalObjects: externalObjectService(db, { + pluginWorkerManager: options.pluginWorkerManager, + enabled: async () => (await instanceSettings.getExperimental()).enableExternalObjects === true, + }).syncIssueSafely, + onSpawn: async (meta) => { + markDispatchStarted(); + await persistRunProcessMetadata(run.id, { ...meta, targetKind: executionTarget?.kind ?? "local" }); + }, + }); + }, + ); + if (!guardedDispatch.dispatched) return { dispatched: false }; + input.effects.onNativeDispatchStarted(); + adapterResult = await guardedDispatch.resultPromise; + } else { + const interactionId = readNonEmptyString(context.interactionId); + const legacyQuestionResponse = + issueRef && + interactionId && + readNonEmptyString(context.interactionKind) === + "ask_user_questions" && + readNonEmptyString(context.interactionStatus) === "answered" + ? await materializeLegacyQuestionResponseWakeProjection({ + db, + companyId: agent.companyId, + issueId: issueRef.id, + runId: run.id, + agentId: agent.id, + interactionId, + }) + : null; + // Do not write the answer projection back to `context`: legacy + // adapters need it in their prompt, but the authoritative answers + // remain on the interaction instead of being duplicated in the + // heartbeat run snapshot. + const adapterContext: Record = { + ...context, + ...(legacyQuestionResponse + ? { + [PAPERCLIP_WAKE_PAYLOAD_KEY]: { + ...parseObject(context[PAPERCLIP_WAKE_PAYLOAD_KEY]), + questionResponse: legacyQuestionResponse, + }, + } + : {}), + }; + const runtimeTools = createAdapterRuntimeToolAccess({ + agentId: agent.id, + companyId: agent.companyId, + runId: run.id, + responsibleUserId: run.responsibleUserId, + }); + if (!runtimeTools) { + logger.warn( + { + companyId: agent.companyId, + agentId: agent.id, + runId: run.id, + }, + "runtime connection tools could not be delivered", + ); + } + const runtimeMcpServers = await buildPaperclipRuntimeMcpServers({ + db, + agent, + runId: run.id, + }); + const runtimeToolDelivery = + adapter.runtimeToolDelivery ?? "invocation_context"; + if (runtimeTools && runtimeToolDelivery === "native_mcp") { + runtimeMcpServers.unshift({ + name: "Paperclip connections", + url: runtimeTools.mcpEndpoint, + token: runtimeTools.bearerToken, + connectionId: "paperclip-runtime-tools", + }); + } + if (authToken && configuredPaperclipApiBaseUrl() && issueRef) { + runtimeMcpServers.unshift({ name: "Paperclip projects", url: `${paperclipApiBaseUrl()}/api/mcp/project-tools`, + token: authToken, connectionId: "paperclip-project-tools" }); + } + const runtimeMcp = createAdapterRuntimeMcpAccess(runtimeMcpServers); + if (runtimeTools && runtimeToolDelivery === "invocation_context") { + adapterContext.paperclipRuntimeTools = runtimeTools; + } + const managedMcpConfig = await createManagedMcpRunConfig({ + db, + agent, + runId: run.id, + config: runtimeConfig, + projectId: issueRef?.projectId ?? null, + issueId: issueRef?.id ?? null, + }); + if (managedMcpConfig) { + adapterContext.paperclipManagedMcp = managedMcpConfig; + } + const guardedDispatch = + await dispatchResolvedInteractionContinuationWithAtomicGate( + (markDispatchStarted) => { + input.effects.onLegacyAdapterEntered(); + return withAdapterExecutionPhase(executionPhaseContext, "adapter_execution", () => adapter.execute({ + getFreshSessionHandoff, + agentIdentity, + runId: run.id, + agent, + runtime: runtimeForAdapter, + config: runtimeConfig, + context: adapterContext, + executionContinuation: executionContinuation ?? null, + runtimeCommandSpec: + adapter.getRuntimeCommandSpec?.(runtimeConfig) ?? null, + executionTarget, + executionTransport: remoteExecution + ? { + remoteExecution: remoteExecution as unknown as Record< + string, + unknown + >, + } + : undefined, + runtimeMcp, + runtimeTools, + onLog, + onMeta: onAdapterMeta, + onEvent: onAdapterEvent, + onUsage: async receipt => { await usageRecorder.capture(receipt); }, + onExecutionPhase: executionControl.phases.enter, + startupTraceContext: getStartupTraceContext(), + onRuntimeProgress: async (progress) => { + await recordCurrentHeartbeatRunRuntimeProgress( + run, + progress, + issueId, + ); + }, + onProviderStopped: collectStoppedInstructions, + onDispatch: markDispatchStarted, + signal: executionControl.controller.signal, + ...(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox" ? { + stopRemoteStartup: async () => { + // Scope comes from the running host invocation, never agent + // config. Keep adapter ownership until setup has unwound. + if (!executionControl.controller.signal.aborted) { + throw new Error("Remote startup stop requires a cancelled run"); + } + const release = await envOrchestrator.releaseForRun({ + heartbeatRunId: run.id, + companyId: agent.companyId, + agentId: agent.id, + status: "released", + providerResourceDisposition: "stop_and_retain", + cancelActiveWork: true, + }); + if (release.errors.length || !await remoteExecutionHasStopped(db, agent.companyId, run.id)) { + throw new Error("Could not verify remote startup stopped"); + } + }, + } : {}), + onCancellationReady: async () => { + await registerAdapterExecutionControl(run.id, executionControl); + const current = await getRun(run.id); + if (!current || isHeartbeatRunTerminalStatus(current.status)) { + executionControl.controller.abort(new Error("Run stopped before provider startup")); + } + }, + onSpawn: async (meta) => { + markDispatchStarted(); + await persistRunProcessMetadata(run.id, { + pid: meta.pid, + processGroupId: + "processGroupId" in meta && + typeof meta.processGroupId === "number" + ? meta.processGroupId + : null, + startedAt: meta.startedAt, + }); + }, + authToken: authToken ?? undefined, + })); + }, + ); + if (!guardedDispatch.dispatched) return { dispatched: false }; + adapterResult = await guardedDispatch.resultPromise; + } + adapterResult = identityRedactor.redact(adapterResult); + if (run.runtimeMode === "legacy" && hasWorkspaceRestoreFailure(adapterResult.resultJson) + && executionTarget?.kind === "remote" && executionTarget.transport === "sandbox") { + const requiredWorkspaceRestoreEvidence = { + workspaceRestoreFailure: adapterResult.resultJson!.workspaceRestoreFailure, + ...(adapterResult.resultJson?.workspaceRestoreDiagnostic ? { workspaceRestoreDiagnostic: adapterResult.resultJson.workspaceRestoreDiagnostic } : {}), + }; + input.effects.onWorkspaceRestoreFailure(requiredWorkspaceRestoreEvidence); + // Retention is the fallback even if recording this receipt fails. + input.effects.onProviderResourceDisposition("stop_and_retain"); + await recordLegacyWorkspaceRestoreFailure(db, run, requiredWorkspaceRestoreEvidence, workspaceRestoreSource); + } + for (const stream of ["stdout", "stderr"] as const) { + const tail = identityRedactor.finish(stream); + if (tail) await appendIdentityRedactedLog(stream, tail); + } + const instructionSave = input.instructions.getInstructionSave(); + if (instructionSave) adapterResult.resultJson = { ...adapterResult.resultJson, instructionSave }; + + if (parseObject(adapterResult.executionRecovery).providerWorkStarted !== false) { + const captured = await usageRecorder.complete(adapterResult); + adapterResult = { ...adapterResult, ...captured, usageComplete: captured.complete }; + } else { + // Stop may already own the terminal result. Preserve its metadata + // while durably recording the proof needed to release admission. + await db.update(heartbeatRuns).set({ costAccountingPending: true, + usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || '{"accountingProviderWorkStarted":false}'::jsonb`, + }).where(and(eq(heartbeatRuns.id, run.id), isNull(heartbeatRuns.costAccountedAt))); + } + adapterResult = applyWorkspaceRestoreFailure(adapterResult); + // A returned result can include a failed restore. Keep the workspace + // barrier closed until required files have been restored. + // If recording the barrier itself fails, propagate as a run failure + // rather than silently leaving dependents stranded behind a missing + // finalize row. + const completeWorkspace = async (ownership?: NativeWorkspaceFinalizationOwnership) => { + try { + if (nativeWorkspaceSync) { + const exported = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, run.companyId), + eq(workspaceOperations.heartbeatRunId, run.id), + eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "succeeded"), + )).limit(1); + if (exported.length) adapterFinalizeOutcome = "succeeded"; + else await restoreNativeWorkspaceBestEffort({ + db, runId: run.id, assertOwnership: ownership?.assertHeld, + restore: () => nativeWorkspaceSync!.restoreWorkspace(ownership?.assertHeld), + }); + } + await ownership?.assertHeld(); + await db + .update(heartbeatRuns) + .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) + .where( + and( + eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.status, "running"), + ), + ); + const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded"; + await recordWorkspaceFinalize(workspaceFinalizeStatus); + if (adapterResult.nativeFinalization) { + adapterResult.nativeFinalization.workspaceFinalizeStatus = + workspaceFinalizeStatus; + try { + const finalized = await finalizeNativeRun({ + db, + runId: run.id, + workspaceFinalizeStatus, + preserveProviderAttempt: Boolean(nativeWorkspaceSync), + }); + await dispatchPendingNativeStatusWakeups({ + companyId: run.companyId, + }); + if (finalized.phase === "committed") { + await nativeWorkspaceSync?.cleanup(); + } + } catch (finalizeErr) { + logger.warn( + { err: finalizeErr, runId: run.id }, + "native result persisted but finalization did not apply; the reconciliation loop will retry", + ); + } + } + } catch (error) { + if (ownership) { + await ownership.assertHeld(); + await recordWorkspaceFinalize("failed"); + } + throw error; + } + }; + if (nativeWorkspaceSync) { + const owned = await withNativeWorkspaceFinalizationOwnership({ + db, companyId: run.companyId, runId: run.id, + }, completeWorkspace); + if (!owned.acquired) throw new NativeWorkspaceFinalizationBusyError(); + } else { + await completeWorkspace(); + } + } catch (adapterErr) { + if (adapterErr instanceof NativeCancellationPendingRecoveryError) { + // Durable cancellation is settled by the outer recovery handler; + // it does not imply a failed workspace or a persisted run result. + throw adapterErr; + } + if (adapterErr instanceof NativeWorkspaceFinalizationBusyError + || adapterErr instanceof NativeWorkspaceFinalizationOwnershipLostError) { + input.effects.onNativeWorkspaceFinalizeScheduled(); + throw adapterErr; + } + if (adapterErr instanceof NativeControllerDetachedForRestartError) { + // Preserve the provider and its run for the new controller. This + // also keeps generic teardown from terminalizing/releasing its lease. + input.effects.onNativeSessionResumeScheduled(); + throw adapterErr; + } + if (adapterErr instanceof NativeRunnerOwnershipUnverifiedError) { + input.effects.onNativeOwnershipHeld(); + throw adapterErr; + } + await db + .update(heartbeatRuns) + .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) + .where( + and( + eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.status, "running"), + ), + ); + if ( + issueRef && + context.resumeSessionGoalHeartbeat === true && + !runGoalControlRequestId + ) { + await blockRunnerGoalRecovery( + db, + { + companyId: run.companyId, + issueId: issueRef.id, + agentId: agent.id, + adapterType: agent.adapterType, + }, + "provider_session_goal_recovery_failed", + ).catch(() => undefined); + } + if (issueRef && runGoalControlRequestId) { + await failRunnerGoalAction( + db, + { + companyId: run.companyId, + issueId: issueRef.id, + agentId: agent.id, + adapterType: agent.adapterType, + }, + runGoalControlRequestId, + adapterErr instanceof Error + ? adapterErr.message + : "session_goal_control_failed", + ).catch(() => undefined); + } + const nativeResumeScheduled = + nativeRuntimeResolution.kind === "native" + ? await db + .select({ + phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, + }) + .from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)) + .limit(1) + .then( + (rows) => + rows[0]?.phase === "retryable_failure" && + rows[0]?.resultId === null, + ) + : false; + if (nativeResumeScheduled) { + input.effects.onNativeSessionResumeScheduled(); + throw new NativeSessionResumeScheduledError(adapterErr); + } + // Adapter (or its restore finally) threw — or the finalize record + // write itself threw. Either way the workspace may be in a partial + // state. Best-effort record finalize=failed so the dependent readiness + // check keeps the gate closed instead of waking on stale local state, + // and surface the original error to the caller. + try { + await recordWorkspaceFinalize("failed", { + errorMessage: + adapterErr instanceof Error + ? adapterErr.message + : String(adapterErr), + }); + } catch (recordErr) { + logger.warn( + { + err: recordErr, + runId: run.id, + executionWorkspaceId: persistedExecutionWorkspace?.id ?? null, + }, + "failed to record workspace_finalize=failed operation; dependents may remain gated", + ); + } + if (nativeRuntimeResolution.kind === "native") { + const proposedResult = await db + .select({ resultId: nativeRunFinalizations.resultId }) + .from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)) + .limit(1) + .then((rows) => rows[0]?.resultId ?? null); + if (proposedResult && nativeWorkspaceSync) { + const workspaceFailureMessage = + adapterErr instanceof Error ? adapterErr.message : ""; + const unrecoverable = + workspaceFailureMessage === + "workspace_sync_out_unrecoverable" || + workspaceFailureMessage.includes("daytona_sandbox_not_found"); + const failure = await recordNativeFinalizationFailure({ + db, + runId: run.id, + error: new Error( + unrecoverable + ? "native_workspace_sync_out_unrecoverable" + : "native_workspace_sync_out_failed", + ), + projectRunStatus: true, + failureScope: "workspace", + permanent: unrecoverable, + }); + input.effects.onNativeWorkspaceFinalizeScheduled(); + throw new NativeWorkspaceFinalizeScheduledError( + adapterErr, + failure.phase === "terminal_failure", + unrecoverable + ? "workspace_sync_out_unrecoverable" + : "workspace_sync_out_failed", + ); + } + try { + await finalizeNativeRun({ + db, + runId: run.id, + workspaceFinalizeStatus: "failed", + }); + await dispatchPendingNativeStatusWakeups({ + companyId: run.companyId, + }); + } catch (finalizeErr) { + logger.warn( + { err: finalizeErr, runId: run.id }, + "native result could not be marked workspace_failed; the reconciliation loop will retry persisted results", + ); + } + } + throw adapterErr; + } finally { + try { + await revokeHeartbeatRunGatewayTokens({ + db, + companyId: agent.companyId, + runId: run.id, + }); + } catch (revokeErr) { + logger.warn( + { err: revokeErr, runId: run.id, companyId: agent.companyId }, + "failed to revoke heartbeat-run MCP gateway tokens", + ); + } + await nativeInstructionReservation?.release(); + await withAdapterExecutionPhase(executionPhaseContext, "instruction_cleanup", releaseInstructionCopy); + } + return { dispatched: true, adapterResult }; +} diff --git a/tests/e2e/fixtures/composer-stop-codex-loader.mjs b/tests/e2e/fixtures/composer-stop-codex-loader.mjs new file mode 100644 index 00000000000..0815a0c62bf --- /dev/null +++ b/tests/e2e/fixtures/composer-stop-codex-loader.mjs @@ -0,0 +1,23 @@ +// The installed Codex dependency takes precedence over PATH. Pin only this +// acceptance server's command resolver to its deterministic protocol fixture. +import { accessSync, constants } from "node:fs"; +import { registerHooks } from "node:module"; +import { isAbsolute } from "node:path"; + +const command = process.env.PAPERCLIP_STOP_CODEX_COMMAND; +if (process.env.NODE_ENV !== "test" || !command || !isAbsolute(command)) { + throw new Error("Composer Stop requires an absolute fixture command in NODE_ENV=test"); +} +accessSync(command, constants.X_OK); +registerHooks({ + load(url, context, nextLoad) { + if (/\/packages\/paperclip-runner\/(src|dist)\/drivers\/codex\/codex-command\.(js|ts)$/.test(url)) { + return { + format: "module", + shortCircuit: true, + source: `export const resolveCodexCommand = () => ${JSON.stringify(command)}; export const resolvePinnedCodexCommand = resolveCodexCommand;`, + }; + } + return nextLoad(url, context); + }, +}); diff --git a/tests/e2e/fixtures/composer-stop-codex-loader.test.mjs b/tests/e2e/fixtures/composer-stop-codex-loader.test.mjs new file mode 100644 index 00000000000..35a0de116ee --- /dev/null +++ b/tests/e2e/fixtures/composer-stop-codex-loader.test.mjs @@ -0,0 +1,41 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { test } from "node:test"; + +const loader = fileURLToPath(new URL("./composer-stop-codex-loader.mjs", import.meta.url)); +const resolver = new URL("../../../packages/paperclip-runner/src/drivers/codex/codex-command.ts", import.meta.url).href; +function launch(env, source = "") { + return spawnSync(process.execPath, ["--import", loader, "--input-type=module", "-e", source], { + encoding: "utf8", env: { ...process.env, NODE_ENV: "test", PAPERCLIP_STOP_CODEX_COMMAND: "", ...env }, + }); +} + +test("pins the fixture even when a bundled Codex dependency is installed", () => { + const dir = mkdtempSync(join(tmpdir(), "composer-stop-loader-")); + const command = join(dir, "codex"); + try { + writeFileSync(command, "#!/bin/sh\nprintf 'composer-stop-fixture'\n", { mode: 0o755 }); + const result = launch({ PAPERCLIP_STOP_CODEX_COMMAND: command }, ` + import { resolveCodexCommand, resolvePinnedCodexCommand } from ${JSON.stringify(resolver)}; + import { execFileSync } from "node:child_process"; + import assert from "node:assert/strict"; + assert.equal(resolveCodexCommand(), ${JSON.stringify(command)}); + assert.equal(resolvePinnedCodexCommand(), ${JSON.stringify(command)}); + process.stdout.write(execFileSync(resolveCodexCommand(), ["--version"])); + `); + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout, "composer-stop-fixture"); + } finally { rmSync(dir, { recursive: true, force: true }); } +}); + +for (const env of [{}, { PAPERCLIP_STOP_CODEX_COMMAND: "codex" }, { NODE_ENV: "production", PAPERCLIP_STOP_CODEX_COMMAND: process.execPath }]) { + test(`rejects a missing or unsafe fixture configuration: ${JSON.stringify(env)}`, () => { + const result = launch(env); + assert.notEqual(result.status, 0); + assert.match(result.stderr, /Composer Stop requires an absolute fixture command/); + }); +} diff --git a/tests/e2e/playwright-composer-stop.config.ts b/tests/e2e/playwright-composer-stop.config.ts index 41d03194680..a58f8507034 100644 --- a/tests/e2e/playwright-composer-stop.config.ts +++ b/tests/e2e/playwright-composer-stop.config.ts @@ -19,6 +19,7 @@ if (fixture) { throw new Error( "PAPERCLIP_STOP_FAKE_CODEX must name the built fake-codex-app-server binary", ); + fs.mkdirSync(path.join(fixtureDir, "codex-home")); const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`; fs.writeFileSync( path.join(fixtureDir, "codex"), @@ -37,6 +38,14 @@ export default defineConfig({ ...server?.env, HEARTBEAT_SCHEDULER_INTERVAL_MS: "10000", PATH: `${fixtureDir}${path.delimiter}${process.env.PATH ?? ""}`, + ...(fixture ? { + // Provider selection prefers the installed dependency over PATH. + // Scope this override and its empty credential home to the test server. + PAPERCLIP_STOP_CODEX_COMMAND: path.join(fixtureDir, "codex"), + CODEX_HOME: path.join(fixtureDir, "codex-home"), + OPENAI_API_KEY: "composer-stop-fixture-key", + NODE_OPTIONS: `${server?.env?.NODE_OPTIONS ?? ""} --import=${path.resolve(import.meta.dirname, "fixtures/composer-stop-codex-loader.mjs")}`, + } : {}), }, }, });