diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index a75a065ea27..3df961605db 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -99,25 +99,57 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes comment: "local_cleanup", retry: "cleanup_quarantined" }, { shape: "quarantined after it requested a launch", errorCode: "native_session_cleanup_quarantined", launched: true, comment: "process_identity_missing", retry: "process_identity_missing" }, - { shape: "failed for another reason without a process record", errorCode: "adapter_failed", launched: false, - comment: "process_identity_missing", retry: "process_identity_missing" }, - ])("routes input after a run $shape to the gate that can clear it", async ({ errorCode, launched, comment, retry }) => { + // w-20261110T050818-s20386: a run that failed before it asked for a runner + // (here a failed preparation query) never had a process to stop. + { shape: "failed before it requested a launch", errorCode: "adapter_failed", launched: false, + comment: null, retry: null }, + { shape: "failed after a runner event without a process record", errorCode: "adapter_failed", launched: false, + runnerEvent: true, comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "failed before launch after its native session was prepared", errorCode: "adapter_failed", launched: false, + prepared: true, comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "failed without a process record after native selection", errorCode: "adapter_failed", launched: false, + coordinator: true, comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "timed out before it requested a launch", status: "timed_out", errorCode: "timeout", launched: false, + comment: null, retry: null }, + { shape: "timed out after a runner event without a process record", status: "timed_out", errorCode: "timeout", launched: false, + runnerEvent: true, comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "was interrupted without a process record after native selection", status: "interrupted", errorCode: "process_lost", launched: false, + coordinator: true, comment: "process_identity_missing", retry: "process_identity_missing" }, + ] as Array<{ shape: string; status?: "timed_out" | "interrupted"; errorCode: string; launched: boolean; runnerEvent?: boolean; prepared?: boolean; coordinator?: boolean; + comment: string | null; retry: string | null }>)("routes input after a run $shape to the gate that can clear it", async ({ + status, errorCode, launched, runnerEvent, prepared, coordinator, comment, retry }) => { // s7532/t04: a Retry's run failed at the quarantine fence before it asked // for a runner. Input then waited for a stop record of a process that // never existed ("process_identity_missing", Retry still offered) instead // of the quarantine's own gate, which names the action that clears it. const f = await seed(); - await db.update(heartbeatRuns).set({ processPid: null, errorCode }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(heartbeatRuns).set({ processPid: null, errorCode, ...(status ? { status } : {}) }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + // A run that failed before native selection has no native issue and no + // coordinator. A coordinator row requires the run's native issue (foreign + // key), so the coordinator rows keep it and are held by that evidence. + if (errorCode !== "native_session_cleanup_quarantined" && !coordinator) { + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ nativeIssueId: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + } if (launched) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId, seq: 1, eventType: "native.process_start_requested", stream: "system", level: "info", message: "launch requested", payload: { hostname: "another-host" } }); + if (runnerEvent) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId, + seq: 1, eventType: "session.started", stream: "system", level: "info", message: "runner event", + sourceEventId: "runner-event-1", payload: {} }); + if (prepared) await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, nativeSessionId: randomUUID(), runnerInstanceId: randomUUID() }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); const { result, blocked } = await readAdmission(f, false); - expect(result).toBeNull(); - expect(blocked).toEqual([comment]); + if (comment === null) expect(result).not.toBeNull(); + else expect(result).toBeNull(); + expect(blocked).toEqual(comment === null ? [] : [comment]); const retryBlocked: string[] = []; - expect(await admitExplicitNativeContinuation({ ...f, db, dryRun: true, reason: "retry_failed_run", - failedRunId: f.sourceRunId, onBlocked: reason => retryBlocked.push(reason) })).toBeNull(); - expect(retryBlocked).toEqual([retry]); + const retried = await admitExplicitNativeContinuation({ ...f, db, dryRun: true, reason: "retry_failed_run", + failedRunId: f.sourceRunId, onBlocked: reason => retryBlocked.push(reason) }); + if (retry === null) expect(retried).not.toBeNull(); + else expect(retried).toBeNull(); + expect(retryBlocked).toEqual(retry === null ? [] : [retry]); }); it("keeps admission and the complete locked session profile while omitting prompt data", async () => { diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 78d2c443da1..d831683c307 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -16,7 +16,7 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { canContinueCancelledRun, userStopBoundaryAt } from "./run-cancellation.js"; import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { isCancelledNativeStartup, isFailedNativePreparation, isFailedNativePreparationRetryAncestor } from "./cancelled-native-startup.js"; -import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, nativeRunRefusedBeforeLaunch, recordedLocalExecutionStopped, recordUnidentifiedLocalRunnerStop } from "./native-local-process-stop.js"; +import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, nativeRunNeverLaunched, recordedLocalExecutionStopped, recordUnidentifiedLocalRunnerStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup, nativeSessionCleanupBlock } from "../vendor/paperclip-runner/index.js"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; @@ -514,7 +514,7 @@ export async function admitExplicitNativeContinuation(input: { if (!run.processPid && !run.processGroupId && !await hasNativeLocalProcessStop(db, companyId, run.id) && !await hasHistoricalSuspendedNativeSession(db, run) && - !await nativeRunRefusedBeforeLaunch(db, run) && + !await nativeRunNeverLaunched(db, run) && !(run.runtimeMode === "native" && await recordUnidentifiedLocalRunnerStop(db, run, { record: !input.dryRun }))) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet."); if (run.processPid && !processStopped(run.processPid)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically."); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 2e30bc1f927..3715622b670 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -238,17 +238,30 @@ function record(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; } -/** A native run that an admission fence refused before it requested any - * runner launch (a quarantined prior-session cleanup is checked before the - * provider starts) never had a process: there is nothing to stop. Any launch - * request or identity record keeps the normal stop proofs. */ -export async function nativeRunRefusedBeforeLaunch(db: Db, run: Pick): Promise { - if (run.runtimeMode !== "native" || run.status !== "failed" || run.processPid || run.processGroupId || - run.errorCode !== "native_session_cleanup_quarantined") return false; +/** A settled native run that never had a runner has nothing to stop: + * - an admission fence (a quarantined prior-session cleanup) refused it before + * it requested a launch; or + * - it failed before native selection (for example a refused budget + * reservation or a failed preparation query). Selection records the run's + * native issue, native session, runner instance and coordinator, and a + * runner launch needs them, so without any of them no runner or provider for this run can exist, + * in any build. + * Any launch request, identity record or runner event keeps the normal stop + * proofs. */ +export async function nativeRunNeverLaunched(db: Db, run: Pick): Promise { + if (run.runtimeMode !== "native" || !SETTLED_RUN_STATUSES.includes(run.status) || run.processPid || run.processGroupId) return false; + const quarantinedBeforeLaunch = run.status === "failed" && run.errorCode === "native_session_cleanup_quarantined"; + if (!quarantinedBeforeLaunch) { + // A coordinator row requires the run's native issue (foreign key), so the + // native issue also rules out a coordinator. + if (run.nativeSessionId || run.runnerInstanceId || run.nativeIssueId) return false; + } const [launch] = 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]), + or(isNotNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED])), )).limit(1); return !launch; }