diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 2fee42e1a17..5ce2a939bf9 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3831,6 +3831,17 @@ provider event, or process-launch evidence may exist. Environment cleanup still requires the same receipts. This historical proof permits explicit Retry or a newer saved user message; it does not replay the cancelled input. +Workspace and AI-capacity waits use the same pre-dispatch classification for +recovery and stop proof. A cancelled legacy run must retain its matching +`workspace_wait` or `ai_connection_wait` receipt with `providerWorkStarted: false`. +The runtime must remain unresolved, and the stage must be `preparing` or +`settled`. Native execution references, workspace recovery obligations, a live +controller lease, retained resources, or dispatch events still reject the proof. +Remote cleanup requires the exact termination receipt. This rule also covers +historical waits without a process ID; neither an old timestamp nor a different +server boot proves that a run stopped. Pause and terminate can then finish, +while a subsequent run still requires ordinary admission and new authorization. + The existing bounded saved-message worker rechecks this proof after restart. Admission atomically settles an unclaimed coordinator and admits one fresh turn, preserving history, unknown action outcomes, and attempt counts. Pauses, approvals, diff --git a/server/src/modules/agent-lifecycle/__tests__/resource-wait-stop.test.ts b/server/src/modules/agent-lifecycle/__tests__/resource-wait-stop.test.ts new file mode 100644 index 00000000000..fc55090b1ac --- /dev/null +++ b/server/src/modules/agent-lifecycle/__tests__/resource-wait-stop.test.ts @@ -0,0 +1,142 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { agents, applyPendingMigrations, closeRegisteredClients, companies, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; +import { createAgentLifecycle, configureAgentLifecycle } from "../../../services/agent-lifecycle.js"; +import { createLifecycleDriver } from "../../../services/agent-lifecycle-driver.js"; +import { agentExecutionsHaveStopped } from "../../../services/agent-execution-stop.js"; +import { canRetryStoppedRun } from "../../../services/cancelled-native-startup.js"; +import { legacyExecutionNeedsReconciliation } from "../../../services/legacy-execution-recovery.js"; +import { legacyControllerBootId } from "../../../services/legacy-controller-lease.js"; +import { PROCESS_START_REQUESTED } from "../../../services/native-local-process-stop.js"; +import { remoteTerminationReceipt } from "../../../services/remote-execution-termination.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_LIFECYCLE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("pre-dispatch resource wait stop proof", () => { + let database: Awaited> | undefined; + let db: Db; + beforeAll(async () => { + if (external) { await applyPendingMigrations(external); db = createDb(external); } + else { database = await startEmbeddedPostgresTestDatabase("paperclip-resource-wait-stop-"); db = createDb(database.connectionString); } + }, 90_000); + afterAll(async () => { if (external) await closeRegisteredClients(external); else await database?.cleanup(); }); + + async function fixture(errorCode = "workspace_busy", executionStage = "preparing") { + const id = randomUUID(); + const [company] = await db.insert(companies).values({ name: "Resource wait", issuePrefix: `W${id.slice(0, 7)}` }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Migrated agent", + adapterType: "paperclip_runner", status: "idle", lifecycleState: "ready" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "cancelled", invocationSource: "on_demand", errorCode, runtimeMode: "legacy", executionStage, + startedAt: new Date("2000-01-01T00:00:00Z"), finishedAt: new Date("2000-01-01T00:00:01Z"), + controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: new Date("2000-01-01T00:01:00Z"), + runnerProfileJson: { adapterDispatch: { adapterType: "claude_local" } }, + resultJson: { + executionRecovery: { kind: errorCode === "workspace_busy" ? "workspace_wait" : "ai_connection_wait", providerWorkStarted: false }, + cancellation: { source: "control_plane", expected: true, initiator: { type: "system" }, + reason: "Waiting for a resource", recordedAt: "2000-01-01T00:00:01Z" }, + }, + }).returning(); + return { agent, run }; + } + + it.each(["workspace_busy", "ai_connection_busy", "ai_connection_pool_exhausted", "provider_limit_wait"].flatMap(errorCode => + ["preparing", "settled"].map(stage => [errorCode, stage])))("finishes a pause after a %s wait at %s", async (errorCode, stage) => { + const { agent, run } = await fixture(errorCode, stage); + const work = configureAgentLifecycle(db, { + requiredPluginIds: async () => [], runPlugin: async () => "complete", + runHost: createLifecycleDriver(db, {} as never).runHost, + }); + try { + expect(legacyExecutionNeedsReconciliation(run)).toBe(false); + expect(await canRetryStoppedRun(db, run)).toBe(true); + await createAgentLifecycle(db).pauseAgent(agent.id); + await work.process(agent.id); + const [paused] = await db.select().from(agents).where(eq(agents.id, agent.id)); + expect(paused).toMatchObject({ status: "paused", lifecycleState: "paused", lifecycleHolds: ["manual"] }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agent.id)))).toHaveLength(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("cancelled"); + } finally { await work.stop(); } + }); + + it.each([ + { resultJson: {} }, + { resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: true } } }, + { resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }, + { errorCode: "cancelled" }, { executionStage: "dispatching" }, { executionStage: null }, + { runtimeMode: "native" }, { runtimeModeResolvedAt: new Date("2000-01-01T00:00:00Z") }, + { processPid: process.pid }, { processStartedAt: new Date("2000-01-01T00:00:00Z") }, + { nativeSessionId: randomUUID() }, { nativeIssueId: randomUUID() }, { sessionIdAfter: "provider-session" }, + { controllerLeaseExpiresAt: new Date("2099-01-01T00:00:00Z") }, + { runnerProfileJson: { nativeExecutionInput: {} } }, { runnerProfileJson: { remoteWorkspace: {} } }, + { runnerProfileJson: { nativeWorkspaceSync: {} } }, + ])("rejects contradictory run evidence: %j", async patch => { + const { agent, run } = await fixture(); + await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, run.id)); + const [changed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(await canRetryStoppedRun(db, changed)).toBe(false); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(false); + }); + + it.each(["restore_unsafe_archive", "restore_lock_timeout"])("preserves an outstanding %s workspace obligation", async workspaceRestoreFailure => { + const { agent, run } = await fixture(); + const [changed] = await db.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, workspaceRestoreFailure, + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1" } } }).where(eq(heartbeatRuns.id, run.id)).returning(); + expect(await canRetryStoppedRun(db, changed)).toBe(false); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(false); + }); + + it.each(["adapter.invoke", PROCESS_START_REQUESTED, "session.started", "provider.event", "tool.execution.started", "source_event"])( + "retains a wait with a recorded %s event", async eventType => { + const { agent, run } = await fixture(); + await db.insert(heartbeatRunEvents).values({ companyId: run.companyId, agentId: agent.id, runId: run.id, + seq: 1, eventType, ...(eventType === "source_event" ? { sourceEventId: "provider-event" } : {}) }); + expect(await canRetryStoppedRun(db, run)).toBe(false); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(false); + }, + ); + + it.each([ + { provider: "local", status: "active" }, + { provider: "local", status: "pending_cleanup", releasedAt: new Date(0) }, + { provider: "local", status: "released", cleanupStatus: "failed", releasedAt: new Date(0) }, + { provider: "local", status: "released", cleanupStatus: "success", releasedAt: new Date(0), metadata: { pendingCleanupInFlight: true } }, + { provider: "ssh", status: "released", cleanupStatus: "success", releasedAt: new Date(0), providerLeaseId: "synthetic-lease" }, + ])("requires resource cleanup proof: %j", async lease => { + const { agent, run } = await fixture(); + await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id, ...lease }); + expect(await canRetryStoppedRun(db, run)).toBe(false); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(false); + }); + + it("rejects a native coordinator even without a current owner", async () => { + const { agent, run } = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: run.companyId, title: "Native ownership" }).returning(); + const [changed] = await db.update(heartbeatRuns).set({ nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, run.id)).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, runId: run.id, issueId: issue.id, phase: "observed" }); + expect(await canRetryStoppedRun(db, changed)).toBe(false); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(false); + }); + + it("accepts a remote resource only after its exact termination receipt is saved", async () => { + const { agent, run } = await fixture(); + const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id, + provider: "ssh", providerLeaseId: "synthetic-remote", status: "released", releasedAt: new Date(0), cleanupStatus: "success" }).returning(); + expect(await canRetryStoppedRun(db, run)).toBe(false); + await db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: + remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, lease.id)); + expect(await canRetryStoppedRun(db, run)).toBe(true); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(true); + }); + + it("does not mistake an unproven HTTP dispatch for a resource wait", async () => { + const { agent, run } = await fixture("cancelled", "dispatching"); + const [changed] = await db.update(heartbeatRuns).set({ resultJson: null, + runnerProfileJson: { adapterDispatch: { adapterType: "http" } } }).where(eq(heartbeatRuns.id, run.id)).returning(); + expect(await canRetryStoppedRun(db, changed)).toBe(false); + expect(await agentExecutionsHaveStopped(db, [agent.id])).toBe(false); + }); +}); diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 1926dafef86..f0b8fbe5dce 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -1,11 +1,10 @@ import { isRetainedWorkspaceWorkerStopProof } from "./native-runtime/native-workspace-platform-proof.js"; import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { createHash } from "node:crypto"; -import { and, eq, gt, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; +import { and, eq, gt, inArray, isNotNull, isNull, lte, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueTreeHolds, issueTreeHoldMembers, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { isUuidLike } from "@paperclipai/shared"; import { claimedAdapterType } from "./conversation-continuation.js"; -import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canContinueCancelledRun, readRunCancellation } from "./run-cancellation.js"; import { legacyControllerBootId } from "./legacy-controller-lease.js"; @@ -13,7 +12,7 @@ import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; import { readFailedPreparationSettlement } from "./failed-preparation-settlement.js"; import { hasVerifiedPreparationLog } from "./verified-preparation-log.js"; -import { executionStartedEventPredicate, noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js"; +import { executionStartedEventPredicate, isPreDispatchResourceWait, noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js"; type Run = Pick new Date()) return false; + if (run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return false; const leases = await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), )); if ((!settled && leases.length === 0 && !historicalBeforeSelection) || leases.some(lease => - lease.provider === "local" + lease.metadata?.pendingCleanupInFlight === true || (lease.provider === "local" ? !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed" - : !hasRemoteTerminationReceipt(lease))) return false; + : !hasRemoteTerminationReceipt(lease)))) return false; // Reject contradictory retained evidence, including a crash after a launch // request but before the PID callback. Provider events never certify a stop. return !await nativeStartupHasExecutionEvidence(db, run); @@ -290,10 +294,7 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co async function nativeStartupHasExecutionEvidence(db: Db, run: Run) { const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), - or(isNotNull(heartbeatRunEvents.sourceEventId), - inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, - "harness.ready", "session.started", "session.resumed", "session.updated", "turn.started", - "provider.event", "provider.rpc_result", "tool.execution.started"])), + executionStartedEventPredicate(), )).limit(1); return Boolean(execution); } diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index b202cafa4ec..0a648c20883 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -20,7 +20,7 @@ import { canonicalExecutionUuid } from "./execution-identity.js"; import { randomUUID } from "node:crypto"; import { and, desc, eq, getTableColumns, inArray, isNull, or, sql, type SQL } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; -import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js"; +import { executionStartedEventPredicate, isPreDispatchResourceWait } from "./pre-dispatch-execution-evidence.js"; import { readRunCancellation } from "./run-cancellation.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { issueRecoveryActionService } from "./issue-recovery-actions.js"; @@ -77,10 +77,7 @@ export function legacyExecutionNeedsReconciliation( && (run.resultJson?.executionCancellation as Record | undefined)?.state === "acknowledged") return false; // Waiting for a subscription or workspace precedes provider execution. It is // a resource wait, not a failed provider attempt or permission to replay work. - if (run.status === "cancelled" && (run.errorCode === "ai_connection_busy" || run.errorCode === "ai_connection_pool_exhausted" || run.errorCode === "provider_limit_wait") && - evidence?.kind === "ai_connection_wait" && evidence.providerWorkStarted === false) return false; - if (run.status === "cancelled" && run.errorCode === "workspace_busy" && - evidence?.kind === "workspace_wait" && evidence.providerWorkStarted === false) return false; + if (isPreDispatchResourceWait(run)) return false; // Setup owns the bounded retry budget for temporary workspace scans. Its // exhaustion needs workspace repair, not reconciliation of provider actions // that the bootstrap evidence proves never started. Keep unknown outcomes held. diff --git a/server/src/services/pre-dispatch-execution-evidence.ts b/server/src/services/pre-dispatch-execution-evidence.ts index b7b9aa93408..a6f712dc950 100644 --- a/server/src/services/pre-dispatch-execution-evidence.ts +++ b/server/src/services/pre-dispatch-execution-evidence.ts @@ -3,6 +3,15 @@ import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinaliza import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; +export function isPreDispatchResourceWait(run: Pick): boolean { + const evidence = run.resultJson?.executionRecovery as Record | undefined; + return run.runtimeMode === "legacy" && run.status === "cancelled" && evidence?.providerWorkStarted === false && + (run.errorCode === "workspace_busy" && evidence.kind === "workspace_wait" || + ["ai_connection_busy", "ai_connection_pool_exhausted", "provider_limit_wait"].includes(run.errorCode ?? "") && + evidence.kind === "ai_connection_wait"); +} + export function executionStartedEventPredicate() { return or(isNotNull(heartbeatRunEvents.sourceEventId), inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, "harness.ready", "session.started",