diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 88afc2c8072..35ae06460b8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1221,6 +1221,38 @@ Newer saved input keeps its own receipt while the existing Retry explanation sta visible. Terminal waits direct the user to send new input and hide the Retry action that cannot continue that attempt. +When a native attempt fails before launch because its predecessor's session is +quarantined, explicit task Retry verifies the retained journal's actual owner. +That owner must belong to the same company, task, agent, workspace, native session, +and runner instance. Its controller must have settled, its local leases must be +released, and its recorded worker, provider, and tool processes must be absent. +A host-bound launch receipt with the exact process birth can establish the stop +when an older server did not write a stop receipt. A later launch invalidates it. +The failed attempt must have no launch evidence and must retain that same process +identity. Missing or conflicting evidence keeps Retry deferred. + +This proof retires cleanup ownership only. It preserves the old journal, saved +result, and uncertain action outcomes. The admitted turn uses a fresh native +session under current authority. The old tool-contract fingerprint therefore +does not gate physical cleanup; existing compatibility checks still govern +session reuse and automatic replacement. +Raw runner journal events and normalized run-log events have independent sequence +numbers. Cleanup validates each normalized receipt's own sequence, event ID, and +digest, then matches the ordered provider lifecycle identities across the two +streams. It does not compare their sequence numbers to one another. + +The committed stop proof also names its successor, canonical Retry receipt, and +local execution workspace. Workspace preparation uses `resume_stopped_native_workspace` +only for that exact admitted successor and unchanged active task binding. It +rechecks the directory and refuses pending workspace operations or another live +lease. Environment selection pins that workspace again before preparation. This +keeps saved files in place while the provider session starts fresh. An unrelated +Retry or changed binding cannot borrow the proof or silently replace the files +with an empty attempt directory. +Projectless runs use their run ID as a transient workspace placeholder. That +placeholder cannot authorize managed workspace reuse; those successors keep the +existing fresh-workspace preparation path while retaining the old directory. + A continuation cancelled by the queued-run gate while waiting for review did not start a provider. Its matching dispatch-gate receipt and unclaimed execution fields keep this deliberate wait out of unknown-action reconciliation. Pending review and decision gates still apply. If an older server already created a recovery hold for this exact state, Retry or a new user message can request one fresh turn after the server verifies there is no provider, coordinator, or unfinished cleanup. Saved user input is reconsidered through the same admission path; the cancelled turn is not replayed. Other legacy queued-run gate cancellations also stay out of unknown-action reconciliation when the terminal run has no retained execution authority, session, workspace recovery, resources, or execution events. The same proof removes the effect of an old false recovery hold only for its exact company, task, and run; the audit row remains. Started or conflicting work retains its hold. Review waits retain their guarded Retry path. New work still passes the current approval, dependency, budget, pause, and ownership gates. diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 4b06ff715ee..916a3a6d0a0 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -371,6 +371,16 @@ server-authored event among these three types; provider source events cannot supply stop authority. These records stay in the local run log and do not add Telemetry or OpenTelemetry data. +Explicit continuation writes `native.stopped_conversation_verified` after it +verifies the retained worker and provider processes stopped. Its payload binds +the requesting `runId`, retained `ownerRunId`, native session, runner instance, +process IDs, retained-state fingerprint, successor run, Retry receipt, and local +workspace identity. The owner can be a preceding terminal +run when the requesting attempt failed before launch with the same process +identity. The server checks host-bound process birth and independent provider +receipts before retiring cleanup ownership. The event does not settle uncertain +actions or authorize replay of the old session. + ## Verified Local Codex Replacement Evidence The server writes `native.stopped_text_turn_verified` in the same transaction diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index eec6f8c3dde..4f0283f019c 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -496,7 +496,8 @@ export async function admitExplicitNativeContinuation(input: { companyId, runId: String(stopped.evidence.runId), agentId, eventType: "native.stopped_conversation_verified", stream: "system", level: "info", message: "The old runner and provider stopped. New user input will use a fresh session; prior action outcomes remain recorded.", - payload: stopped.evidence, + payload: { ...stopped.evidence, successorRunId: input.successorRunId, + ...(input.taskRetryRequestId ? { taskRetryRequestId: input.taskRetryRequestId } : {}) }, }); } const authorization = { ...continuation, actorId, commentId, ...(response ? { interactionId: response.source.interactionId } : {}), ...(retry ? { failedRunId: input.failedRunId } : {}), diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index 23039ff5f7f..7b376f0fa8b 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -1,5 +1,5 @@ -import { and, eq, sql } from "drizzle-orm"; -import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; +import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; import { persistActivity } from "./activity-log.js"; import { readSystemRecovery } from "./system-recovery-state.js"; @@ -16,6 +16,7 @@ import { taskRetryRunAncestry } from "./recovery-run-ancestry.js"; export type ContinuationReadiness = | { mode: "resume_preserved_workspace"; preservationReceiptId: string } | { mode: "reuse_current_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string } + | { mode: "resume_stopped_native_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string } | { mode: "fresh_isolated_workspace"; retryIntentId: string; sourceRunId: string; retainedSourceIds: string[] }; const freshReadinessSchema = z.object({ mode: z.literal("fresh_isolated_workspace"), retryIntentId: z.string().uuid(), @@ -108,6 +109,47 @@ async function reusableDirectoryAfterPreparationFailure(db: Db, source: typeof h return operation ? null : workspace; } +async function stoppedNativeContinuationWorkspace(db: Db, source: typeof heartbeatRuns.$inferSelect, + input: { runId: string; issueId: string; retryIntentId: string }) { + if (source.runtimeMode !== "native" || source.status !== "failed" || + source.errorCode !== "native_session_cleanup_quarantined" || !source.nativeSessionId || !source.runnerInstanceId) return null; + const [proof] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, source.companyId), eq(heartbeatRunEvents.runId, source.id), + eq(heartbeatRunEvents.agentId, source.agentId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified"), + isNull(heartbeatRunEvents.sourceEventId), + sql`${heartbeatRunEvents.payload}->>'successorRunId' = ${input.runId}`, + sql`${heartbeatRunEvents.payload}->>'taskRetryRequestId' = ${input.retryIntentId}`, + )).limit(1); + if (!proof) return null; + const parsed = z.object({ schema: z.literal("paperclip.stopped_native_conversation.v1"), + runId: z.literal(source.id), nativeSessionId: z.literal(source.nativeSessionId), runnerInstanceId: z.literal(source.runnerInstanceId), + executionWorkspaceId: z.string().uuid(), workspaceCwd: z.string().min(1), + }).safeParse(proof.payload); + if (!parsed.success) throw new Error("Verified native continuation workspace is unavailable"); + if (parsed.data.executionWorkspaceId === source.id) return null; + const [bound] = await db.select({ id: executionWorkspaces.id, cwd: executionWorkspaces.cwd }).from(issues).innerJoin(executionWorkspaces, and( + eq(executionWorkspaces.companyId, issues.companyId), eq(executionWorkspaces.id, issues.executionWorkspaceId), + )).where(and(eq(issues.companyId, source.companyId), eq(issues.id, input.issueId), + eq(issues.assigneeAgentId, source.agentId), eq(executionWorkspaces.id, parsed.data.executionWorkspaceId), + eq(executionWorkspaces.cwd, parsed.data.workspaceCwd), eq(executionWorkspaces.status, "active"), + eq(executionWorkspaces.providerType, "local_fs"), + or(isNull(executionWorkspaces.sourceIssueId), eq(executionWorkspaces.sourceIssueId, input.issueId)), + )).limit(1); + if (!bound) throw new Error("Verified native continuation workspace binding changed"); + const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, source.companyId), eq(workspaceOperations.executionWorkspaceId, bound.id), + eq(workspaceOperations.status, "running"), or(isNull(workspaceOperations.heartbeatRunId), ne(workspaceOperations.heartbeatRunId, input.runId)), + )).limit(1); + const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, source.companyId), eq(environmentLeases.executionWorkspaceId, bound.id), + or(isNull(environmentLeases.heartbeatRunId), ne(environmentLeases.heartbeatRunId, input.runId)), + or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "pending_cleanup"])), + )).limit(1); + if (operation || lease) throw new Error("Verified native continuation workspace cleanup is pending"); + return { mode: "resume_stopped_native_workspace" as const, retryIntentId: input.retryIntentId, + sourceRunId: source.id, workspaceId: bound.id, cwd: parsed.data.workspaceCwd }; +} + /** Preparation trusts the committed admission receipt, not a raw wake payload. */ export async function readFreshWorkspaceReadiness(db: Db, run: Pick): Promise { @@ -151,6 +193,8 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick { + const { input, issue, workspace } = await fixture(); + const saved = await workspace(); + await db.update(issues).set({ executionWorkspaceId: saved.id }).where(eq(issues.id, issue.id)); + input.task.issueRef!.executionWorkspaceId = saved.id; + input.task.issueRef!.executionWorkspacePreference = null; + input.task.freshWorkspace = { mode: "resume_stopped_native_workspace", retryIntentId: randomUUID(), + sourceRunId: randomUUID(), workspaceId: saved.id, cwd: home }; + if (mode === "changed_binding") input.task.issueRef!.executionWorkspaceId = (await workspace()).id; + if (mode === "changed_cwd") await executionWorkspaceService(db).update(saved.id, { cwd: join(home, "changed") }); + if (mode !== "preserved") { + await expect(selectHeartbeatEnvironment(db, input)).rejects.toThrow("The saved Retry workspace binding changed."); + return; + } + const result = await selectHeartbeatEnvironment(db, input); + expect(result.requestedShouldReuseExisting).toBe(true); + expect(result.reusableExistingExecutionWorkspace).toMatchObject({ id: saved.id, cwd: home }); + expect(result.nativeRecoveryExecutionWorkspaceId).toBeNull(); + }); + it("does not make an archived issue workspace available for reuse", async () => { const { input, workspace } = await fixture(); const saved = await workspace(); diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index 534459f7001..2d90ae57351 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -145,10 +145,11 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron existingExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, existingExecutionWorkspace.id); } const retainedDirectory = input.task.freshWorkspace?.mode === "reuse_current_workspace" ? input.task.freshWorkspace : null; - if (retainedDirectory && (issueRef?.executionWorkspaceId !== retainedDirectory.workspaceId || - existingExecutionWorkspace?.id !== retainedDirectory.workspaceId || existingExecutionWorkspace.cwd !== retainedDirectory.cwd)) { + const retainedWorkspace = input.task.freshWorkspace?.mode === "resume_stopped_native_workspace" ? input.task.freshWorkspace : retainedDirectory; + if (retainedWorkspace && (issueRef?.executionWorkspaceId !== retainedWorkspace.workspaceId || + existingExecutionWorkspace?.id !== retainedWorkspace.workspaceId || existingExecutionWorkspace.cwd !== retainedWorkspace.cwd)) { throw new WorkspaceValidationFailure("The saved Retry workspace binding changed.", { - workspaceValidation: { reason: "isolated_task_directory_binding_mismatch", issueId }, + workspaceValidation: { reason: retainedDirectory ? "isolated_task_directory_binding_mismatch" : "stopped_native_workspace_binding_mismatch", issueId }, }); } const nativeRecoveryExecutionWorkspaceId = @@ -159,7 +160,7 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron const workspaceReuseRequest = resolveExecutionWorkspaceReuseRequestForIssue({ issueExecutionWorkspaceId: requestedExecutionWorkspaceId, - issueExecutionWorkspacePreference: controllerOnly ? null : nativeRecoveryExecutionWorkspaceId + issueExecutionWorkspacePreference: controllerOnly ? null : nativeRecoveryExecutionWorkspaceId || retainedWorkspace ? "reuse_existing" : (issueRef?.executionWorkspacePreference ?? null), existingExecutionWorkspaceStatus: diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index a294c37d6f4..37fc1569498 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -26,12 +26,13 @@ function absent(pid: number): boolean { * the original host, process birth, and launch event before accepting it. */ export async function recordedLocalExecutionStopped(db: Db, run: Pick) { - const [identity] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + const [identity] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), - eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED), isNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED]), + isNull(heartbeatRunEvents.sourceEventId), )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); const recorded = identity?.payload; - if (recorded?.targetKind !== "local" || recorded.hostname !== os.hostname() || + if (identity?.eventType !== PROCESS_IDENTITY_RECORDED || recorded?.targetKind !== "local" || recorded.hostname !== os.hostname() || recorded.processPid !== run.processPid || recorded.processGroupId !== run.processGroupId || !run.processStartedAt || recorded.processStartedAt !== run.processStartedAt.toISOString()) return false; return Boolean(run.processPid && absent(run.processPid) && diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index e7c63c6d419..1a6bdf4d5b8 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -5972,15 +5972,20 @@ describe("stopped native conversation physical cleanup", () => { ...(mode === "startup_intent" ? { startupAttempt: { phase: "intent" } } : {}), ...(mode === "pending_identity" ? { pendingEvents: [{ eventType: "session.started" }] } : {}), }; - const normalizedEvent = mode === "normalized_session_receipt" ? { ...event, turnId: undefined, itemId: undefined } : event; + const normalizedEvent = { ...event, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:1`, + ...(mode === "normalized_session_receipt" ? { turnId: undefined, itemId: undefined } : {}) }; const receipt = { sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:1`, + sourceSeq: 1, sourceInstanceId: identity.runnerInstanceId, eventType: normalizedEvent.eventType, sourcePayloadSha256: mode === "wrong_receipt" ? "wrong" : hash(normalizedEvent), payload: { prpEvent: normalizedEvent } }; + const normalizedReplacement = { ...replacement, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:2` }; const stop = { eventType: mode === "new_launch" ? "native.process_start_requested" : "native.local_process_stopped", payload: { processPid: mode === "worker_alive" ? process.pid : 99_999_999, processGroupId: mode === "worker_alive" ? process.pid : 99_999_999 } }; let queryIndex = 0; const db = { select() { const rows = [ [{ provider: mode === "remote" ? "daytona" : "local", releasedAt: mode === "unreleased" ? null : new Date() }], - [stop], mode === "missing_receipt" ? [] : [receipt, ...(mode.startsWith("replacement") ? [{ ...receipt, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:2`, sourcePayloadSha256: hash(replacement), payload: { prpEvent: replacement } }] : [])], + [stop], mode === "missing_receipt" ? [] : [receipt, ...(mode.startsWith("replacement") ? [{ ...receipt, + sourceEventId: normalizedReplacement.sourceEventId, sourceSeq: 2, eventType: normalizedReplacement.eventType, + sourcePayloadSha256: hash(normalizedReplacement), payload: { prpEvent: normalizedReplacement } }] : [])], ][queryIndex++] ?? []; const q: any = { from: () => q, where: () => q, orderBy: () => q, limit: () => q, then: Promise.resolve(rows).then.bind(Promise.resolve(rows)) }; return q; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6f7d5deced1..03507c2b2b3 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -95,7 +95,7 @@ import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js"; import { uploadVerifiedProviderPack } from "./provider-pack-upload.js"; import { selectRemotePiCompanion } from "./remote-pi-companion.js"; -import { readNativeLocalProcessStop, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js"; +import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { connectorToolDefinitionsForSkills } from "../connector-runtime.js"; import { setNativeExecutionActive } from "./native-active-runs.js"; @@ -3015,7 +3015,8 @@ export function rebaseRetainedNativeCleanupProviderHome( export async function verifyStoppedNativeSessionForContinuation( db: Db, run: Pick, + "status" | "finishedAt" | "nativeIssueId" | "nativeSessionId" | "runnerInstanceId" | "runnerProfileJson" | + "errorCode" | "createdAt" | "processPid" | "processGroupId" | "processStartedAt">, ): Promise<{ evidence: Record; retire: () => boolean } | null> { try { if (run.runtimeMode !== "native" || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status) || @@ -3024,22 +3025,66 @@ export async function verifyStoppedNativeSessionForContinuation( if (!(["codex", "acpx"] as string[]).includes(execution.provider.kind) || execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId || execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId || - nativeSessionKey(execution) !== run.nativeSessionId || - record(run.runnerProfileJson).nativeToolContractFingerprint !== nativeToolContractFingerprintForTarget("local")) return null; + nativeSessionKey(execution) !== run.nativeSessionId) return null; const scope = nativeSessionScopeKey(execution); const idle = () => !activeNativeSessions.has(run.id) && !executingRunnerdSessionScopes.has(scope) && !initializingSessionToolAuthorities.has(scope) && !warmNativeSessions.has(scope); if (!idle()) return null; const leases = await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id))); - if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt || lease.cleanupStatus === "failed")) return null; - const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id); - if (!stopped) return null; + if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt || + lease.status === "pending_cleanup" || lease.cleanupStatus === "failed")) return null; const root = scopedRunnerdStateRoot(execution); const providerFile = runnerProviderStateFilename(execution); const snapshot = cleanupStateSnapshot(root, providerFile); const identity = record(snapshot.control.identity); - if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId || + let owner = run; + let ownerExecution = execution; + if (identity.runId !== run.id) { + if (run.errorCode !== "native_session_cleanup_quarantined" || typeof identity.runId !== "string") return null; + const [started] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, + "session.started", "session.resumed", "session.reconciled"]))).limit(1); + if (started) return null; + const [retainedRow] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, identity.runId), + eq(heartbeatRuns.agentId, run.agentId), eq(heartbeatRuns.nativeIssueId, run.nativeIssueId), + eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), eq(heartbeatRuns.runnerInstanceId, run.runnerInstanceId), + )).for("update", { noWait: true }).limit(1); + const retained = retainedRow ? await withReleasedNativeRunProcess(db, retainedRow) : null; + if (!retained || retained.runtimeMode !== "native" || + !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(retained.status) || + !retained.finishedAt || retained.finishedAt > run.createdAt || + retained.controllerLeaseExpiresAt && retained.controllerLeaseExpiresAt > new Date() || + !retained.processStartedAt || !run.processStartedAt || + retained.processStartedAt.getTime() !== run.processStartedAt.getTime() || + retained.processPid !== run.processPid || retained.processGroupId !== run.processGroupId) return null; + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.nativeIssueId), + eq(nativeRunFinalizations.runId, retained.id), + )).for("update", { noWait: true }).limit(1); + if (!coordinator || !["committed", "terminal_failure"].includes(coordinator.phase) || coordinator.leaseOwner || + coordinator.nextAttemptAt || coordinator.recoveryState) return null; + ownerExecution = parseNativeExecutionInput(record(retained.runnerProfileJson).nativeExecutionInput); + if (ownerExecution.binding.runId !== retained.id || ownerExecution.binding.companyId !== run.companyId || + ownerExecution.binding.agentId !== run.agentId || ownerExecution.binding.issueId !== run.nativeIssueId || + nativeSessionScopeKey(ownerExecution) !== scope || ownerExecution.workspace.cwd !== execution.workspace.cwd) return null; + const ownerLeases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, retained.id))); + if (ownerLeases.some(lease => lease.provider !== "local" || !lease.releasedAt || + lease.status === "pending_cleanup" || lease.cleanupStatus === "failed")) return null; + owner = retained; + } + let stop = await readNativeLocalProcessStop(db, owner.companyId, owner.id); + if (!stop && owner.processPid && owner.processPid > 1 && owner.processGroupId === owner.processPid && + await recordedLocalExecutionStopped(db, owner)) { + stop = { processPid: owner.processPid, processGroupId: owner.processPid, + ...(owner.processStartedAt ? { processStartedAt: owner.processStartedAt.toISOString() } : {}) }; + } + const stopped = stop; + if (!stopped || activeNativeSessions.has(owner.id)) return null; + if (!durableIdentityMatchesExecution(identity, ownerExecution) || identity.runnerInstanceId !== owner.runnerInstanceId || !["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key => typeof identity[key] === "string" && identity[key] && snapshot.runner[key] === identity[key]) || !Array.isArray(snapshot.control.committedEvents)) return null; @@ -3048,7 +3093,7 @@ export async function verifyStoppedNativeSessionForContinuation( if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(snapshot.control.schema)) || snapshot.runner.schema !== RUNNERD_STATE_SCHEMA || snapshot.provider.schema !== (execution.provider.kind === "codex" ? "paperclip.runner.codex-provider-state.v1" : ACPX_PROVIDER_STATE_SCHEMA) || - !["turn_active", "prepared", "suspended"].includes(String(snapshot.provider.lifecycle)) || + !["turn_active", "prepared", "suspended", "session_open"].includes(String(snapshot.provider.lifecycle)) || snapshot.provider.startupAttempt != null) return null; const pending = JSON.stringify([snapshot.runner.outbox, snapshot.provider.pendingEvents, snapshot.provider.queuedEvents]); if (/session\.(started|resumed|reconciled)/.test(pending)) return null; @@ -3056,36 +3101,42 @@ export async function verifyStoppedNativeSessionForContinuation( const providerEvents = events.filter(event => ["session.started", "session.resumed", "session.reconciled"].includes(String(event.eventType))); if (!providerEvents.length) return null; const receipts = await db.select().from(heartbeatRunEvents).where(and( - eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), - eq(heartbeatRunEvents.sourceInstanceId, run.runnerInstanceId), + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, owner.id), + eq(heartbeatRunEvents.sourceInstanceId, owner.runnerInstanceId!), inArray(heartbeatRunEvents.eventType, ["session.started", "session.resumed", "session.reconciled", "harness.diagnostic"]))); + const providerReceipts = receipts.filter(row => row.eventType !== "harness.diagnostic") + .sort((a, b) => Number(a.sourceSeq) - Number(b.sourceSeq)); + if (providerReceipts.length !== providerEvents.length) return null; const providerPids = new Set(); - for (const event of providerEvents) { + for (const [index, event] of providerEvents.entries()) { const provider = record(event.payload); - if (!validatePrpEvent(event).ok || event.sourceKind !== "runner" || event.sourceInstanceId !== run.runnerInstanceId || - event.runId !== run.id || event.normalizedSessionId !== run.nativeSessionId || + if (!validatePrpEvent(event).ok || event.sourceKind !== "runner" || event.sourceInstanceId !== owner.runnerInstanceId || + event.runId !== owner.id || event.normalizedSessionId !== run.nativeSessionId || typeof provider.providerSessionId !== "string" || !provider.providerSessionId || !cleanupProcessAbsent(provider.processId) || provider.processId === stopped.processPid) return null; if (event.eventType === "session.reconciled" && (!providerPids.has(Number(provider.previousProcessId)) || !cleanupProcessAbsent(provider.previousProcessId))) return null; - const receipt = receipts.find(row => row.sourceEventId === `${run.runnerInstanceId}:${run.id}:${event.sourceSeq}`); + const receipt = providerReceipts[index]; const durable = record(record(receipt?.payload).prpEvent); const durableProvider = record(durable.payload); if (!receipt || !validatePrpEvent(durable).ok || durable.sourceInstanceId !== event.sourceInstanceId || - durable.runId !== run.id || durable.normalizedSessionId !== run.nativeSessionId || + durable.runId !== owner.id || durable.normalizedSessionId !== run.nativeSessionId || // Normalized session-open receipts omit turn/item IDs. Those belong // to the durable runner identity checked above, not the open event. - durable.sourceSeq !== event.sourceSeq || durable.eventType !== event.eventType || + durable.sourceSeq !== receipt.sourceSeq || durable.eventType !== event.eventType || + receipt.eventType !== durable.eventType || receipt.sourceInstanceId !== durable.sourceInstanceId || + receipt.sourceEventId !== `${owner.runnerInstanceId}:${owner.id}:${durable.sourceSeq}` || + durable.sourceEventId !== receipt.sourceEventId || receipt.sourcePayloadSha256 !== nativeSha256(durable) || (durableProvider.processId !== undefined && durableProvider.processId !== provider.processId) || (durableProvider.driverSessionId ?? durableProvider.providerSessionId) !== provider.providerSessionId) return null; providerPids.add(provider.processId); } - if (receipts.filter(row => record(record(row.payload).prpEvent).eventType !== "harness.diagnostic").length !== providerEvents.length) return null; // ACPX's sidecar is not its agent process. Include separately recorded // owners from both the durable journal and the independent DB/checkpoint. // Extra evidence can only add a stop requirement, never waive one. const owners: Record[] = [record(record(run.runnerProfileJson?.sessionCheckpoint).process), + record(record(owner.runnerProfileJson?.sessionCheckpoint).process), snapshot.provider, record(snapshot.provider.identity), record(snapshot.provider.descriptor)]; for (const event of [...events, ...receipts.map(row => record(record(row.payload).prpEvent))]) { const payload = record(event.payload); @@ -3103,16 +3154,18 @@ export async function verifyStoppedNativeSessionForContinuation( if (!cleanupProcessAbsent(pid)) return null; providerPids.add(pid); } - const unchanged = () => idle() && cleanupProcessAbsent(stopped.processPid) && + const unchanged = () => idle() && !activeNativeSessions.has(owner.id) && cleanupProcessAbsent(stopped.processPid) && [...providerPids].every(cleanupProcessAbsent) && cleanupStateSnapshot(root, providerFile).fingerprint === snapshot.fingerprint; return { evidence: { schema: "paperclip.stopped_native_conversation.v1", runId: run.id, + ownerRunId: owner.id, + executionWorkspaceId: execution.binding.executionWorkspaceId, workspaceCwd: execution.workspace.cwd, nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId, processPid: stopped.processPid, providerPids: [...providerPids], stateFingerprint: snapshot.fingerprint }, retire: () => { - try { return unchanged() && completeTerminatedLocalNativeSessionCleanup({ - companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId!, - }); } catch { return false; } + try { return unchanged() && [owner.id, run.id].every(runId => completeTerminatedLocalNativeSessionCleanup({ + companyId: run.companyId, runId, runnerInstanceId: run.runnerInstanceId!, + })); } catch { return false; } }, }; } catch { return null; } diff --git a/server/src/services/native-runtime/quarantined-task-retry.live.test.ts b/server/src/services/native-runtime/quarantined-task-retry.live.test.ts new file mode 100644 index 00000000000..1eafc083a28 --- /dev/null +++ b/server/src/services/native-runtime/quarantined-task-retry.live.test.ts @@ -0,0 +1,141 @@ +import { randomUUID } from "node:crypto"; +import { copyFile, mkdir, readFile, writeFile } from "node:fs/promises"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { and, eq } from "drizzle-orm"; +import { agents, agentWakeupRequests, companies, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { expect, it } from "vitest"; +import { startRunnerApiTestServer } from "../../__tests__/helpers/runner-api-server.js"; +import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; +import { heartbeatService } from "../heartbeat.js"; +import { issueService } from "../issues.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { withReleasedNativeRunProcess } from "../native-local-process-stop.js"; +import { nativeSha256 } from "./canonical.js"; +import { closeIdleWarmNativeSessionsForRestart } from "./native-session-executor.js"; +import { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; + +it.skipIf(process.env.PAPERCLIP_LIVE_QUARANTINED_RETRY !== "true")("starts a fresh real Codex turn through task Retry after inherited quarantine", async () => { + const home = createIsolatedTestHome(); + const keys = ["PAPERCLIP_HOME", "CODEX_HOME", "PAPERCLIP_API_URL", "PAPERCLIP_AGENT_JWT_SECRET", "PAPERCLIP_TELEMETRY_DISABLED", "PAPERCLIP_RUNNER_STATE_DIR"]; + const saved = Object.fromEntries(keys.map(key => [key, process.env[key]])); + let server: Awaited> | undefined; + let heartbeat: ReturnType | undefined; + let companyId: string | undefined; + try { + const codexHome = join(home.home, "codex"), stateBase = join(home.home, "runner-state"); + await mkdir(codexHome, { mode: 0o700 }); + await copyFile(join(process.env.PAPERCLIP_LIVE_CODEX_HOME ?? join(homedir(), ".codex"), "auth.json"), join(codexHome, "auth.json")); + process.env.CODEX_HOME = codexHome; + process.env.PAPERCLIP_HOME = join(home.home, "instance"); + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase; + process.env.PAPERCLIP_AGENT_JWT_SECRET = randomUUID(); + process.env.PAPERCLIP_TELEMETRY_DISABLED = "1"; + server = await startRunnerApiTestServer({ deploymentMode: "authenticated", humanSessions: true }); + process.env.PAPERCLIP_API_URL = server.apiUrl; + const f = await server.fixture({ connectionScenario: "fresh" }), db = server.db; + companyId = f.companyId; + const markers = [randomUUID(), randomUUID()]; + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ status: "todo", executionRunId: null, executionWorkspaceSettings: { mode: "shared_workspace" }, + title: "Preserve work through task Retry", description: "Follow the latest human comment. Use the task workspace. Complete only that requested file step and finish with paperclip_finish, reportedWorkDisposition done. This is a local file task; no external writes are needed.", + }).where(eq(issues.id, f.issueId)); + await db.update(companies).set({ defaultResponsibleUserId: f.userId }).where(eq(companies.id, f.companyId)); + await db.update(agents).set({ adapterConfig: { provider: "codex", model: "gpt-6.1-sol", modelReasoningEffort: "high", + cwd: f.workspace, workspaceStrategy: { type: "project_primary" }, env: { CODEX_HOME: codexHome }, timeoutSec: 180, lifecycleMode: "per_turn" }, + runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 1 } }, + }).where(eq(agents.id, f.agentId)); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true }); + heartbeat = heartbeatService(db); + const service = issueService(db); + const first = await service.addComment(f.issueId, `Write exactly ${markers[0]} and a newline to retry-proof.txt. Read it, then finish.`, { userId: f.userId! }); + const run = await heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: f.userId, + payload: { issueId: f.issueId, commentId: first.id }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, wakeCommentId: first.id, skipIssueComment: true }, + }); + expect(run?.id).toBeTruthy(); + await heartbeat.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + const [original] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id)); + expect({ status: original.status, error: original.error }).toMatchObject({ status: "succeeded", error: null }); + expect(await readFile(join(f.workspace, "retry-proof.txt"), "utf8")).toBe(`${markers[0]}\n`); + const owner = await withReleasedNativeRunProcess(db, original); + expect(owner.processPid).toBeTruthy(); + const execution = parseNativeExecutionInput(owner.runnerProfileJson!.nativeExecutionInput); + const root = join(stateBase, nativeSha256({ schema: "paperclip.native-session-scope.v2", companyId: f.companyId, agentId: f.agentId, + workspace: { kind: "managed", executionWorkspaceId: execution.binding.executionWorkspaceId }, + provider: { driverKind: "codex_app_server", identity: { kind: "codex" } }, normalizedSessionId: owner.nativeSessionId })); + const controlPath = join(root, "control-plane/control-plane-state.json"), providerPath = join(root, "runner/codex-provider-state.json"); + const control = JSON.parse(await readFile(controlPath, "utf8")), provider = JSON.parse(await readFile(providerPath, "utf8")); + control.commands.push({ commandId: "unacknowledged-result", status: "pending", payload: { commandType: "tool.result", result: "Retain this uncertain command without replay." } }); + provider.lifecycle = "session_open"; + await writeFile(controlPath, JSON.stringify(control)); + await writeFile(providerPath, JSON.stringify(provider)); + const retainedBytes = await readFile(controlPath, "utf8"), sourceId = randomUUID(); + const retryComment = await service.addComment(f.issueId, `Read retry-proof.txt and preserve its existing line. Append exactly ${markers[1]} and a newline once. Never repeat the earlier file step. Read the complete file and finish.`, { userId: f.userId! }); + const oldFingerprint = `sha256:${"1".repeat(64)}`; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...owner.runnerProfileJson, nativeToolContractFingerprint: oldFingerprint } }).where(eq(heartbeatRuns.id, owner.id)); + await db.insert(heartbeatRuns).values({ id: sourceId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + nativeIssueId: f.issueId, runtimeMode: "native", nativeSessionId: owner.nativeSessionId, runnerInstanceId: owner.runnerInstanceId, + processPid: owner.processPid, processGroupId: owner.processGroupId, processStartedAt: owner.processStartedAt, + status: "failed", finishedAt: new Date(), errorCode: "native_session_cleanup_quarantined", responsibleUserId: f.userId, + contextSnapshot: { issueId: f.issueId, commentId: retryComment.id, wakeCommentId: retryComment.id, wakeCommentIds: [retryComment.id] }, runnerProfileJson: { ...owner.runnerProfileJson, + nativeExecutionInput: { ...execution, binding: { ...execution.binding, runId: sourceId }, + task: { ...execution.task, prompt: retryComment.body } }, nativeToolContractFingerprint: oldFingerprint }, + }); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId: sourceId, phase: "terminal_failure" }); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", + cause: "native_session_cleanup_quarantined", fingerprint: sourceId, status: "resolved", outcome: "blocked", + nextAction: "Automatic recovery stopped.", evidence: { runId: sourceId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }); + await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + const cookie = await server.humanSession(f.userId!); + const response = await fetch(`${server.apiUrl}/api/issues/${f.issueId}/retry`, { method: "POST", + headers: { "content-type": "application/json", cookie, origin: server.apiUrl }, + body: JSON.stringify({ observedRunId: sourceId, idempotencyKey: randomUUID() }), + }); + const receipt = await response.json() as { requestId: string; accepted: boolean }; + expect(response.status, JSON.stringify(receipt)).toBe(202); + expect(receipt.accepted).toBe(true); + await expect.poll(async () => { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)); + return wake.runId; + }, { timeout: 30_000, interval: 250 }).toBeTruthy(); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)); + await expect.poll(async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)))[0].status, + { timeout: 240_000, interval: 500 }).toBe("succeeded"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)); + expect(successor.nativeSessionId).not.toBe(owner.nativeSessionId); + expect(successor.sessionIdAfter).not.toBe(owner.sessionIdAfter); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: sourceId, forceFreshSession: true }); + expect(await readFile(join(f.workspace, "retry-proof.txt"), "utf8")).toBe(markers.join("\n") + "\n"); + expect(await readFile(controlPath, "utf8")).toBe(retainedBytes); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task).toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null }); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, successor.id)); + expect(coordinator.phase).toBe("committed"); + const proof = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, sourceId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified"))); + expect(proof).toHaveLength(1); + const evidence = { passed: true, platform: `${process.platform}-${process.arch}`, model: "gpt-6.1-sol", effort: "high", + ownerRunId: owner.id, sourceRunId: sourceId, successorRunId: successor.id, freshSession: true, + priorJournalPreserved: true, fileLines: 2, taskStatus: task.status, locksCleared: true, finalization: coordinator.phase }; + if (process.env.PAPERCLIP_LIVE_EVIDENCE_PATH) await writeFile(process.env.PAPERCLIP_LIVE_EVIDENCE_PATH, JSON.stringify(evidence, null, 2) + "\n"); + console.log(JSON.stringify(evidence)); + } catch (error) { + if (server && companyId && process.env.PAPERCLIP_LIVE_EVIDENCE_PATH) { + const runs = await server.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + await writeFile(`${process.env.PAPERCLIP_LIVE_EVIDENCE_PATH}.failure.json`, JSON.stringify({ error: String(error), + runs: runs.map(run => ({ id: run.id, status: run.status, error: run.error, result: run.resultJson, + execution: (run.runnerProfileJson?.nativeExecutionInput as { workspace?: unknown } | undefined)?.workspace, + continuation: run.contextSnapshot?.executionContinuation })), + }, null, 2)); + } + throw error; + } finally { + await heartbeat?.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + await server?.close(); + for (const key of keys) { if (saved[key] === undefined) delete process.env[key]; else process.env[key] = saved[key]; } + home.cleanup(); + } +}, 600_000); diff --git a/server/src/services/quarantined-task-retry.test.ts b/server/src/services/quarantined-task-retry.test.ts new file mode 100644 index 00000000000..56006603a50 --- /dev/null +++ b/server/src/services/quarantined-task-retry.test.ts @@ -0,0 +1,243 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import { hostname, tmpdir } from "node:os"; +import { join } from "node:path"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, agentWakeupRequests, companies, createDb, environmentLeases, executionWorkspaces, heartbeatRunEvents, + heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, workspaceOperations, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { heartbeatService } from "./heartbeat.js"; +import { getExecutionBlocker } from "./execution-blocker.js"; +import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; +import { instanceSettingsService } from "./instance-settings.js"; +import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; +import { nativeSha256 } from "./native-runtime/canonical.js"; +import { verifyStoppedNativeSessionForContinuation } from "./native-runtime/native-session-executor.js"; +import { requestTaskRetry } from "./task-retry.js"; + +describe("task Retry after inherited native cleanup quarantine", () => { + let postgres: Awaited>; + let db: ReturnType; + beforeAll(async () => { + postgres = await startEmbeddedPostgresTestDatabase("quarantined-task-retry-"); + db = createDb(postgres.connectionString); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true }); + }, 90_000); + afterAll(async () => { await db?.$client.end(); await postgres?.cleanup(); }); + + async function seed(base: string, transient = false) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + const ownerId = randomUUID(), sourceId = randomUUID(), nativeSessionId = randomUUID(), runnerInstanceId = randomUUID(); + const workspaceId = randomUUID(); + const execution = (runId: string) => parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId, agentId, issueId, runId, executionWorkspaceId: transient ? runId : workspaceId }, + task: { identifier: "TEST", title: "Continue saved work", description: null, prompt: "Continue saved work", workMode: "standard" }, + workspace: { cwd: base, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: nativeSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: "contract", sha256: "sha", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Continue saved work", criteria: [{ id: "done", requirement: "Save the result" }] } }, + interactionResponses: [], credentialBindings: [], + }); + const processStartedAt = new Date(Date.now() - 120_000); + const processIdentity = { processPid: 99_999_999, processGroupId: 99_999_999, processStartedAt }; + await db.insert(companies).values({ id: companyId, name: "Quarantined Retry", defaultResponsibleUserId: "board", + issuePrefix: `Q${companyId.replaceAll("-", "")}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native worker", adapterType: "paperclip_runner", status: "idle", + runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue saved work", status: "blocked", assigneeAgentId: agentId }); + if (!transient) { + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, + mode: "shared_workspace", strategyType: "project_primary", name: "Saved work", cwd: base }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issueId)); + } + const resultJson = { savedAnswer: "Existing work", recoveredExecutionFailure: { errorCode: "provider_transport_failed" } }; + for (const runId of [ownerId, sourceId]) { + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, + runtimeMode: "native", nativeSessionId, runnerInstanceId, ...processIdentity, + status: runId === ownerId ? "succeeded" : "failed", resultJson: runId === ownerId ? resultJson : null, + errorCode: runId === sourceId ? "native_session_cleanup_quarantined" : null, + createdAt: new Date(Date.now() - (runId === ownerId ? 110_000 : 60_000)), + startedAt: new Date(Date.now() - (runId === ownerId ? 109_000 : 59_000)), + finishedAt: new Date(Date.now() - (runId === ownerId ? 100_000 : 50_000)), + contextSnapshot: { issueId }, runnerProfileJson: { nativeExecutionInput: execution(runId), + nativeToolContractFingerprint: `sha256:${"1".repeat(64)}`, sessionCheckpoint: { sessionId: "retained-provider-thread" } }, + }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, + phase: runId === ownerId ? "committed" : "terminal_failure" }); + await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + provider: "local", status: "failed", leasePolicy: "ephemeral", releasedAt: new Date(Date.now() - 40_000) }); + } + await appendHeartbeatRunEvent(db, { companyId, runId: ownerId, agentId, eventType: PROCESS_IDENTITY_RECORDED, + payload: { ...processIdentity, processStartedAt: processStartedAt.toISOString(), targetKind: "local", hostname: hostname() } }); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", + cause: "native_session_cleanup_quarantined", fingerprint: sourceId, status: "resolved", outcome: "blocked", + nextAction: "Automatic recovery stopped.", evidence: { runId: sourceId, + automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }); + const identity = { runId: ownerId, runnerInstanceId, normalizedSessionId: nativeSessionId, + environmentLeaseId: transient ? ownerId : workspaceId, turnId: `turn-${ownerId}`, itemId: `item-${ownerId}` }; + const event = { schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceKind: "runner", sourceSeq: 5, + sourceEventId: "provider-open", sourceInstanceId: runnerInstanceId, runId: ownerId, + normalizedSessionId: nativeSessionId, turnId: identity.turnId, itemId: identity.itemId, + priority: 0, emittedAt: processStartedAt.toISOString(), eventType: "session.resumed", + payload: { providerSessionId: "retained-provider-thread", processId: 99_999_998 } }; + const normalized = { ...event, sourceSeq: 1, sourceEventId: `${runnerInstanceId}:${ownerId}:1`, turnId: undefined, itemId: undefined, + payload: { providerSessionId: "retained-provider-thread", driverSessionId: "retained-provider-thread" } }; + await appendHeartbeatRunEvent(db, { companyId, runId: ownerId, agentId, eventType: event.eventType, + payload: { prpEvent: normalized }, nativeSource: { sourceInstanceId: runnerInstanceId, + sourceEventId: normalized.sourceEventId, sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: normalized } }); + const root = join(base, nativeSha256({ schema: "paperclip.native-session-scope.v2", companyId, agentId, + workspace: transient ? { kind: "transient", cwd: base, repoUrl: null, repoRef: null, branchName: null } + : { kind: "managed", executionWorkspaceId: workspaceId }, + provider: { driverKind: "codex_app_server", identity: { kind: "codex" } }, normalizedSessionId: nativeSessionId })); + const files = { + "control-plane/control-plane-state.json": { schema: "paperclip.runner.durable.control-plane-state.v1", identity, + committedEvents: [{ envelope: { payload: event } }], commands: [{ status: "pending", payload: { commandType: "tool.result" } }] }, + "runner/runner-state.json": { schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended", outbox: [] }, + "runner/codex-provider-state.json": { schema: "paperclip.runner.codex-provider-state.v1", lifecycle: "session_open", + threadId: "retained-provider-thread", startupAttempt: null, pendingEvents: [], queuedEvents: [] }, + }; + await mkdir(join(root, "control-plane"), { recursive: true }); + await mkdir(join(root, "runner"), { recursive: true }); + for (const [path, value] of Object.entries(files)) await writeFile(join(root, path), JSON.stringify(value)); + await writeFile(join(base, "saved-work.txt"), "Keep the prior result and uncertain action history."); + return { companyId, agentId, issueId, workspaceId, ownerId, sourceId, runnerInstanceId, resultJson, root, files }; + } + + it.each(["preserved", "projectless", "changed_binding", "changed_cwd", "pending_operation", "active_lease", "closed_workspace", "remote_workspace", "foreign_receipt"])( + "admits one fresh successor through canonical Retry without replaying the retained session (%s)", async mode => { + const base = await mkdtemp(join(tmpdir(), "quarantined-retry-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = base; + try { + const f = await seed(base, mode === "projectless"); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const input = { db, companyId: f.companyId, issueId: f.issueId, actorId: "board", actorSource: "local_implicit", observedRunId: f.sourceId }; + const receipt = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() }); + const duplicate = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() }); + expect(duplicate.requestId).toBe(receipt.requestId); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!)]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + const [state] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + expect(successors, JSON.stringify(state.payload)).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.sourceId, forceFreshSession: true, taskRetryRequestId: receipt.requestId }); + if (mode === "changed_binding") await db.update(issues).set({ executionWorkspaceId: null }).where(eq(issues.id, f.issueId)); + if (mode === "changed_cwd") await db.update(executionWorkspaces).set({ cwd: join(base, "another-directory") }).where(eq(executionWorkspaces.id, f.workspaceId)); + if (mode === "closed_workspace") await db.update(executionWorkspaces).set({ status: "closed" }).where(eq(executionWorkspaces.id, f.workspaceId)); + if (mode === "remote_workspace") await db.update(executionWorkspaces).set({ providerType: "plugin" }).where(eq(executionWorkspaces.id, f.workspaceId)); + if (mode === "pending_operation") await db.insert(workspaceOperations).values({ companyId: f.companyId, + executionWorkspaceId: f.workspaceId, heartbeatRunId: f.ownerId, issueId: f.issueId, phase: "workspace_restore", status: "running" }); + if (mode === "active_lease") await db.insert(environmentLeases).values({ companyId: f.companyId, + executionWorkspaceId: f.workspaceId, heartbeatRunId: f.ownerId, issueId: f.issueId, provider: "local", status: "active", leasePolicy: "ephemeral" }); + if (mode === "foreign_receipt") { + const [proof] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceId), + eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified"))); + await db.update(heartbeatRunEvents).set({ payload: { ...proof.payload, successorRunId: randomUUID() } }) + .where(eq(heartbeatRunEvents.id, proof.id)); + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "fresh_isolated_workspace" }); + } else if (mode === "projectless") { + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "fresh_isolated_workspace", + retryIntentId: receipt.requestId, sourceRunId: f.sourceId }); + } else if (mode === "preserved") { + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_stopped_native_workspace", + retryIntentId: receipt.requestId, sourceRunId: f.sourceId, workspaceId: f.workspaceId, cwd: base }); + } else await expect(readFreshWorkspaceReadiness(db, successors[0])).rejects.toThrow(/Verified native continuation workspace/); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + expect(wake).toMatchObject({ status: "coalesced", runId: successors[0].id }); + const [owner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.ownerId)); + expect(owner.resultJson).toEqual(f.resultJson); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(action.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" }); + const proof = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified"))); + expect(proof).toHaveLength(1); + expect(proof[0].payload).toMatchObject({ runId: f.sourceId, ownerRunId: f.ownerId }); + for (const [path, value] of Object.entries(f.files)) expect(await readFile(join(f.root, path), "utf8")).toBe(JSON.stringify(value)); + expect(await readFile(join(base, "saved-work.txt"), "utf8")).toContain("Keep the prior result"); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(base, { recursive: true, force: true }); + } + }); + + it.each(["provider_alive", "worker_alive", "owner_running", "controller_owned", "owner_remote", "owner_unreleased", + "owner_new_launch", "source_new_launch", "foreign_host", "foreign_workspace", "foreign_cwd", "foreign_runner", "missing_provider_receipt", + "pending_provider_launch", "changed_process_birth", "changed_journal", "normalized_receipt_identity", "provider_identity_conflict"])("keeps quarantine for %s", async mode => { + const base = await mkdtemp(join(tmpdir(), "quarantined-retry-fence-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = base; + try { + const f = await seed(base); + if (mode === "provider_alive" || mode === "pending_provider_launch") { + await writeFile(join(f.root, "runner/codex-provider-state.json"), JSON.stringify({ + ...f.files["runner/codex-provider-state.json"], + ...(mode === "provider_alive" ? { processId: process.pid } : { startupAttempt: { phase: "intent" } }), + })); + } + if (mode === "worker_alive") { + await db.update(heartbeatRuns).set({ processPid: process.pid, processGroupId: process.pid }) + .where(eq(heartbeatRuns.companyId, f.companyId)); + const [owner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.ownerId)); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.ownerId, agentId: f.agentId, + eventType: PROCESS_IDENTITY_RECORDED, payload: { targetKind: "local", hostname: hostname(), + processPid: process.pid, processGroupId: process.pid, processStartedAt: owner.processStartedAt!.toISOString() } }); + } + if (mode === "owner_running") await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.ownerId)); + if (mode === "controller_owned") await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller" }).where(eq(nativeRunFinalizations.runId, f.ownerId)); + if (mode === "owner_remote" || mode === "owner_unreleased") await db.update(environmentLeases).set( + mode === "owner_remote" ? { provider: "daytona" } : { releasedAt: null }, + ).where(eq(environmentLeases.heartbeatRunId, f.ownerId)); + if (mode === "owner_new_launch" || mode === "source_new_launch") await appendHeartbeatRunEvent(db, { + companyId: f.companyId, runId: mode === "owner_new_launch" ? f.ownerId : f.sourceId, agentId: f.agentId, eventType: PROCESS_START_REQUESTED, + }); + if (mode === "foreign_host") { + const [identity] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.ownerId), + eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED))); + await db.update(heartbeatRunEvents).set({ payload: { ...identity.payload, hostname: "another-host" } }) + .where(eq(heartbeatRunEvents.id, identity.id)); + } + if (mode === "changed_process_birth") await db.update(heartbeatRuns).set({ processStartedAt: new Date() }).where(eq(heartbeatRuns.id, f.sourceId)); + if (mode === "foreign_runner") await db.update(heartbeatRuns).set({ runnerInstanceId: randomUUID() }).where(eq(heartbeatRuns.id, f.ownerId)); + if (mode === "foreign_workspace" || mode === "foreign_cwd") { + const [owner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.ownerId)); + const input = parseNativeExecutionInput(owner.runnerProfileJson!.nativeExecutionInput); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...owner.runnerProfileJson, + nativeExecutionInput: { ...input, + ...(mode === "foreign_workspace" ? { binding: { ...input.binding, executionWorkspaceId: "another-workspace" } } + : { workspace: { ...input.workspace, cwd: join(base, "another-directory") } }), + }, + } }).where(eq(heartbeatRuns.id, f.ownerId)); + } + if (mode === "missing_provider_receipt") await db.delete(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.runId, f.ownerId), eq(heartbeatRunEvents.eventType, "session.resumed"))); + if (mode === "normalized_receipt_identity") await db.update(heartbeatRunEvents).set({ sourceSeq: 2 }).where(and( + eq(heartbeatRunEvents.runId, f.ownerId), eq(heartbeatRunEvents.eventType, "session.resumed"))); + if (mode === "provider_identity_conflict") { + const [receipt] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.ownerId), + eq(heartbeatRunEvents.eventType, "session.resumed"))); + const event = receipt.payload!.prpEvent as Record; + const changed = { ...event, payload: { providerSessionId: "another-thread", driverSessionId: "another-thread" } }; + await db.update(heartbeatRunEvents).set({ payload: { prpEvent: changed }, sourcePayloadSha256: nativeSha256(changed) }) + .where(eq(heartbeatRunEvents.id, receipt.id)); + } + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceId)); + const proof = await verifyStoppedNativeSessionForContinuation(db, source); + if (mode === "changed_journal") { + expect(proof).not.toBeNull(); + await writeFile(join(f.root, "runner/runner-state.json"), "{}"); + expect(proof!.retire()).toBe(false); + } else expect(proof).toBeNull(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(base, { recursive: true, force: true }); + } + }); +}); diff --git a/tests/e2e/agent-chat-projects.spec.ts b/tests/e2e/agent-chat-projects.spec.ts index 46ea6ca7844..a0effe0666a 100644 --- a/tests/e2e/agent-chat-projects.spec.ts +++ b/tests/e2e/agent-chat-projects.spec.ts @@ -195,6 +195,19 @@ test("split handoff commits relevant plans before execution and never creates ch }) ).status(), ).toBe(422); + await expect.poll(async () => { + const comments = await json(await request.get(`/api/issues/${chat.id}/comments`)); + const completed = comments.filter((comment: any) => comment.authorAgentId === f.agent.id && + comment.body.startsWith("Delegated work completed:")); + return tasks.every((task: any) => completed.some((comment: any) => comment.body.includes(task.identifier))); + }, { timeout: 30_000 }).toBe(true); + await expect.poll(async () => await json(await request.get(`/api/companies/${f.company.id}/live-runs`)), + { timeout: 30_000 }).toEqual([]); + const settledTasks = await json(await request.get(`/api/companies/${f.company.id}/issues`)); + expect(settledTasks.map((task: any) => ({ id: task.id, status: task.status, parentId: task.parentId })) + .sort((a: any, b: any) => a.id.localeCompare(b.id))).toEqual(tasks.map((task: any) => ({ + id: task.id, status: "done", parentId: null, + })).sort((a: any, b: any) => a.id.localeCompare(b.id))); } finally { await f.restore(); }