diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 44ff7b33b11..029a0a96de8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -415,6 +415,14 @@ start timestamp, or process identity. A later eligible instruction can adopt the saved messages through normal admission. This rule does not prove a dispatched provider has stopped or change native cancellation acknowledgement. +A system-cancelled workspace retry can also accept a later eligible instruction +when it never acquired execution authority. Admission requires its exact parent +to record a same-company, same-task, same-agent workspace wait before provider +work. Both runs must have no conflicting provider, process, resource or output +evidence. The cancelled retry must have no owner, runtime selection or start +timestamp. Admission preserves the old run and any unknown action outcomes. It +does not replay the cancelled turn or reset recovery limits. + ### Known execution waits at admission A known execution hold is a waiting condition, not a new execution attempt. Every issue wake must read the current effective reconciliation hold under the issue admission lock before creating a run. Resolved recovery bookkeeping can still carry a no-replay hold; only clearing the effective hold makes admission eligible again. The final dispatch gate remains required for changes after admission. diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index f12926a96c3..9b147c04dc3 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -1,6 +1,6 @@ import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { createHash } from "node:crypto"; -import { and, eq, inArray, isNotNull, or } from "drizzle-orm"; +import { and, eq, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { isUuidLike } from "@paperclipai/shared"; import { claimedAdapterType } from "./conversation-continuation.js"; @@ -11,6 +11,7 @@ import { legacyControllerBootId } from "./legacy-controller-lease.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; import { readFailedPreparationSettlement } from "./failed-preparation-settlement.js"; +import { executionStartedEventPredicate, noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js"; type Run = Pick { return isCancelledNativeStartup(db, run, coordinator); } +function noProviderOutputPredicate() { + return and(...[ + heartbeatRuns.runtimeModeResolverVersion, heartbeatRuns.runtimeModeReason, heartbeatRuns.runnerInstanceId, + heartbeatRuns.nativePhase, heartbeatRuns.driverKind, heartbeatRuns.driverVersion, + heartbeatRuns.completionContractId, heartbeatRuns.completionContractSha256, heartbeatRuns.externalRunId, + heartbeatRuns.logStore, heartbeatRuns.logRef, heartbeatRuns.logBytes, heartbeatRuns.logSha256, + heartbeatRuns.stdoutExcerpt, heartbeatRuns.stderrExcerpt, heartbeatRuns.lastOutputAt, + heartbeatRuns.lastOutputStream, heartbeatRuns.lastOutputBytes, heartbeatRuns.exitCode, heartbeatRuns.signal, + ].map(column => isNull(column)), eq(heartbeatRuns.lastOutputSeq, 0), eq(heartbeatRuns.logCompressed, false)); +} + +async function isCancelledUnclaimedWorkspaceWaitRetry(db: Db, run: Run): Promise { + const cancellation = readRunCancellation(run.resultJson); + const issueId = run.issueId ?? run.contextSnapshot?.issueId; + if (run.runtimeMode !== "legacy" || run.startedAt !== null || run.executionStage !== null + || run.runnerProfileJson !== null || typeof issueId !== "string" || !isUuidLike(issueId) + || run.errorCode !== "cancelled" || run.resultJson?.stopReason !== "cancelled" + || cancellation?.source !== "control_plane" || cancellation.expected !== true || cancellation.initiator.type !== "system") return false; + const [source] = await db.select({ agentId: heartbeatRuns.agentId, createdAt: heartbeatRuns.createdAt, + retryOfRunId: heartbeatRuns.retryOfRunId, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.scheduledRetryReason, "workspace_busy"), + sql`${heartbeatRuns.contextSnapshot}->>'retryReason' = 'workspace_busy'`, + sql`${heartbeatRuns.contextSnapshot}->>'retryOfRunId' = ${heartbeatRuns.retryOfRunId}::text`, + sql`(${heartbeatRuns.resultJson} - array['stopReason', 'cancellation', 'startupCancellation', + 'timeoutFired', 'timeoutSource', 'timeoutConfigured', 'effectiveTimeoutSec']) = '{}'::jsonb`, + noRetainedExecutionPredicate(), noProviderOutputPredicate(), isNull(heartbeatRuns.runnerProfileJson), + isNull(heartbeatRuns.usageJson))); + if (!source?.retryOfRunId) return false; + const [parent] = await db.select({ id: heartbeatRuns.id, usageJson: heartbeatRuns.usageJson }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, source.agentId), eq(heartbeatRuns.id, source.retryOfRunId), + sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, + eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "cancelled"), eq(heartbeatRuns.errorCode, "workspace_busy"), + eq(heartbeatRuns.executionStage, "settled"), isNotNull(heartbeatRuns.finishedAt), lte(heartbeatRuns.finishedAt, source.createdAt), + isNull(heartbeatRuns.runtimeModeResolvedAt), isNull(heartbeatRuns.controllerLeaseExpiresAt), + ...[heartbeatRuns.processPid, heartbeatRuns.processGroupId, heartbeatRuns.processStartedAt, + heartbeatRuns.nativeIssueId, heartbeatRuns.nativeSessionId, heartbeatRuns.sessionIdAfter].map(column => isNull(column)), + noProviderOutputPredicate(), + sql`${heartbeatRuns.runnerProfileJson} #>> '{adapterDispatch,adapterType}' = 'paperclip_runner'`, + sql`(${heartbeatRuns.runnerProfileJson} - 'adapterDispatch' - 'chatControlRecoveryAdmission') = '{}'::jsonb`, + sql`${heartbeatRuns.resultJson}->'executionRecovery'->>'kind' = 'workspace_wait'`, + sql`${heartbeatRuns.resultJson}->'executionRecovery'->'providerWorkStarted' = 'false'::jsonb`, + sql`(${heartbeatRuns.resultJson} - 'cancellation' - 'workspaceBusy' - 'executionRecovery') = '{}'::jsonb`, + sql`not exists (select 1 from ${environmentLeases} where ${environmentLeases.companyId} = ${heartbeatRuns.companyId} + and ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id})`, + sql`not exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id})`, + sql`not exists (select 1 from ${heartbeatRunEvents} where ${and(eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), + eq(heartbeatRunEvents.runId, heartbeatRuns.id), executionStartedEventPredicate())})`, + )); + if (!parent || !hasOnlyPreparationAccountingContext(parent.usageJson, issueId)) return false; + const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, run.companyId), inArray(nativeRunResults.runId, [run.id, parent.id]), + )).limit(1); + return !result; +} + /** Caller holds the coordinator and run locks when using this proof to admit * work. Attempt zero is a durable never-claimed receipt: every native executor * commits its first claim before it can start or attach a provider. */ @@ -170,7 +228,8 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co Boolean(run.controllerBootId && run.controllerBootId !== legacyControllerBootId && run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt <= new Date()); const beforeReviewDispatch = isPreDispatchReviewWait(run) && !coordinator; - const beforeSelection = beforeReviewDispatch || historicalBeforeSelection || run.runtimeMode === "legacy" && !run.runtimeModeResolvedAt && + const unclaimedWorkspaceRetry = !coordinator && await isCancelledUnclaimedWorkspaceWaitRetry(db, run); + const beforeSelection = beforeReviewDispatch || unclaimedWorkspaceRetry || historicalBeforeSelection || run.runtimeMode === "legacy" && !run.runtimeModeResolvedAt && !run.nativeSessionId && !coordinator && claimedAdapterType(run) === "paperclip_runner" && cancellation?.beforeNativeSelection === true; const neverClaimed = run.runtimeMode === "native" && coordinator && @@ -179,7 +238,7 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co !coordinator.controllerPid && !coordinator.leaseOwner && !coordinator.leaseExpiresAt && !coordinator.resultId && !coordinator.failureDetail?.successorRunId; if (!beforeSelection && !neverClaimed) return false; - const settled = beforeReviewDispatch || typeof run.resultJson?.startupPreparationSettledAt === "string"; + const settled = beforeReviewDispatch || unclaimedWorkspaceRetry || typeof run.resultJson?.startupPreparationSettledAt === "string"; // The old preparer can still be unwinding even though the run is terminal. if (!settled && run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return false; const leases = await db.select().from(environmentLeases).where(and( diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index ce0880a4175..95e7bc52985 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -1259,6 +1259,118 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes return f; } + async function seedCancelledWorkspaceWaitRetry() { + const f = await seed(), parentRunId = randomUUID(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ id: parentRunId, companyId: f.companyId, agentId: f.agentId, + scopeKind: "issue", issueId: f.issueId, + runtimeMode: "legacy", status: "cancelled", executionStage: "settled", errorCode: "workspace_busy", + createdAt: new Date("2026-09-11T09:57:00Z"), startedAt: new Date("2026-09-11T09:57:00Z"), + finishedAt: new Date("2026-09-11T09:58:00Z"), contextSnapshot: { issueId: f.issueId }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" }, chatControlRecoveryAdmission: {} }, + usageJson: { ledgerScope: { issueId: f.issueId, projectId: null, billingCode: null } }, + resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false } } }); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, nativeSessionId: null, + scopeKind: "issue", issueId: f.issueId, retryOfRunId: parentRunId, scheduledRetryReason: "workspace_busy", + status: "cancelled", errorCode: "cancelled", startedAt: null, processPid: null, + createdAt: new Date("2026-09-11T09:59:00Z"), runnerProfileJson: null, + contextSnapshot: { issueId: f.issueId, source: "issue.comment", retryReason: "workspace_busy", retryOfRunId: parentRunId }, + resultJson: { stopReason: "cancelled", cancellation: { reason: "Cancelled by control plane", source: "control_plane", + expected: true, initiator: { type: "system" }, recordedAt: "2026-09-11T10:00:00Z" }, + startupCancellation: { errorCode: "cancelled", requestedAt: "2026-09-11T10:00:00Z", beforeNativeSelection: false } }, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" }) + .where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + return { ...f, parentRunId }; + } + + it("admits a fresh direction after an unclaimed workspace retry was cancelled", async () => { + const f = await seedCancelledWorkspaceWaitRetry(); + expect(await admit(f)).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId }); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(source).toMatchObject({ status: "cancelled", startedAt: null, processPid: null, runnerProfileJson: null }); + const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(hold.evidence.automaticRecovery).toMatchObject({ replay: "explicit_user_continuation", actionOutcome: "unknown" }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued")))).toHaveLength(1); + }); + + it.each(["message", "retry"])("admits an unclaimed workspace retry through the heartbeat %s path", async kind => { + const f = await seedCancelledWorkspaceWaitRetry(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const successor = await heartbeatService(db).wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", + reason: kind === "retry" ? "retry_failed_run" : "issue_commented", + ...(kind === "retry" ? { failedRunId: f.sourceRunId } : {}), requestedByActorType: "user", requestedByActorId: "board", + payload: { issueId: f.issueId, ...(kind === "message" ? { commentId: f.commentId } : {}) }, + contextSnapshot: { issueId: f.issueId, ...(kind === "message" ? { wakeCommentId: f.commentId } : {}) }, + }); + expect(successor).toMatchObject({ status: "queued", contextSnapshot: { forceFreshSession: true, + previousRunId: f.sourceRunId, explicitUserContinuation: { commentId: kind === "message" ? f.commentId : null } } }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); + + it("resumes saved input after an unclaimed workspace retry exactly once", async () => { + const f = await seedCancelledWorkspaceWaitRetry(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "on_demand", reason: "issue_commented", requestedByActorType: "user", requestedByActorId: "board", + status: "deferred_issue_execution", updatedAt: new Date(0), payload: { issueId: f.issueId, commentId: f.commentId, + executionWait: { reason: "execution_recovery" }, + _paperclipWakeContext: { issueId: f.issueId, wakeCommentId: f.commentId, wakeCommentIds: [f.commentId] } }, + }).returning(); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeExecutionWaitComments(), heartbeat.resumeExecutionWaitComments()]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ forceFreshSession: true, wakeCommentIds: [f.commentId] }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0]) + .toMatchObject({ status: "coalesced", runId: successors[0].id }); + }); + + it.each(["started", "lease", "launch", "foreign_parent", "parent_executed", "parent_reason", "parent_receipt", "profile", "owner", "operator_stop", "output", + "parent_lease", "parent_launch", "parent_native", "retry_binding", "parent_output", "parent_profile", "parent_later", "restore"])( + "keeps cancelled workspace retries held with contradictory %s evidence", async kind => { + const f = await seedCancelledWorkspaceWaitRetry(); + if (kind === "started") await db.update(heartbeatRuns).set({ startedAt: new Date("2026-09-11T09:59:30Z") }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "lease") await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId, + provider: "local", leasePolicy: "ephemeral", status: "released", cleanupStatus: "success", releasedAt: new Date() }); + if (kind === "launch") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, + agentId: f.agentId, runId: f.sourceRunId, seq: 1, eventType: PROCESS_START_REQUESTED, payload: {} }); + if (kind === "foreign_parent") { + const foreign = await seed(); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, retryReason: "workspace_busy", + retryOfRunId: foreign.sourceRunId } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + } + if (kind === "parent_executed" || kind === "parent_receipt") await db.update(heartbeatRuns).set({ resultJson: + kind === "parent_receipt" ? {} : { executionRecovery: { kind: "workspace_wait", providerWorkStarted: true } } }) + .where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "parent_reason") await db.update(heartbeatRuns).set({ errorCode: "cancelled" }).where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "profile") await db.update(heartbeatRuns).set({ runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } } }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "owner") await db.update(heartbeatRuns).set({ controllerBootId: randomUUID() }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "operator_stop") await db.update(heartbeatRuns).set({ resultJson: { stopReason: "cancelled", + cancellation: { reason: "Stopped", source: "operator", expected: true, initiator: { type: "user", userId: "board" }, + recordedAt: "2026-09-11T10:00:00Z" } } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "output") await db.update(heartbeatRuns).set({ lastOutputSeq: 1 }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "parent_lease") await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.parentRunId, + provider: "local", leasePolicy: "ephemeral", status: "released", cleanupStatus: "success", releasedAt: new Date() }); + if (kind === "parent_launch") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, + agentId: f.agentId, runId: f.parentRunId, seq: 1, eventType: PROCESS_START_REQUESTED, payload: {} }); + if (kind === "parent_native") await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId }).where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "retry_binding") await db.update(heartbeatRuns).set({ retryOfRunId: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "parent_output") await db.update(heartbeatRuns).set({ lastOutputSeq: 1 }).where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "parent_profile") await db.update(heartbeatRuns).set({ runnerProfileJson: { + adapterDispatch: { adapterType: "paperclip_runner" }, remoteWorkspace: {} } }).where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "parent_later") await db.update(heartbeatRuns).set({ finishedAt: new Date("2026-09-11T10:01:00Z") }).where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "restore") await db.update(heartbeatRuns).set({ resultJson: { stopReason: "cancelled", + workspaceRestoreFailure: "restore_unsafe_archive" } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await admit(f)).toBeNull(); + if (kind === "lease" || kind === "parent_lease") await db.delete(environmentLeases) + .where(eq(environmentLeases.heartbeatRunId, kind === "lease" ? f.sourceRunId : f.parentRunId)); + }, + ); + it("keeps provider diagnostics out of admission evidence and preserves absent results", async () => { const f = await seedCancelledReviewWait(); const heartbeat = heartbeatService(db);