diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 74422770352..0e93aefda20 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -370,6 +370,13 @@ The active-lock lifecycle is part of the checkout contract: Stale-lock recovery is crash recovery, not a retry loop. Paperclip must not clear or adopt locks held by non-terminal runs. After stale cleanup, a checkout `409` should mean a real live owner, status/assignee mismatch, unresolved blocker, or active gate still prevents checkout. Agents must treat that `409` as an ownership conflict and stop rather than retrying the same checkout. +A queued retry waits for its same-company, same-task and same-agent predecessor +to finish execution and cleanup, including when the task execution lock already +names the retry. This also applies to failures before native runtime selection. +Committed cleanup notifications reconsider that same queued run through ordinary +admission. Failed cleanup, Stop, reassignment and independent recovery gates +retain their existing effect; a terminal run row alone cannot release the wait. + Stopping a queued legacy run before runtime selection clears only its own task locks. It preserves deferred messages without promoting another run. The stop must have a recorded, expected operator cancellation and no runtime selection, diff --git a/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts b/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts index b5153c32d7d..dd82048651f 100644 --- a/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts +++ b/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts @@ -1,8 +1,10 @@ import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, authUsers, companies, companyMemberships, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, authUsers, companies, companyMemberships, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { heartbeatService } from "../services/heartbeat.js"; import { NativeStartupCleanupPendingError, readNativeStartupCleanup } from "../services/native-runtime/native-startup-cleanup.js"; @@ -10,6 +12,9 @@ import { prepareNativeHeartbeatRun } from "../services/native-runtime/prepare-na import { environmentService } from "../services/environments.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { createIsolatedTestHome } from "./helpers/isolated-test-home.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js"; const execute = vi.hoisted(() => vi.fn()); vi.mock("../adapters/index.js", async () => { @@ -76,6 +81,97 @@ suite("native terminal-run cleanup admission", () => { expect(execute).toHaveBeenCalledTimes(3); }, 30_000); + it.each(["continue", "stop", "reassign", "failed cleanup"] as const)( + "settles an automatic legacy retry from the committed cleanup event: %s", async mode => { + const home = createIsolatedTestHome(), previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = home.home; + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), sourceId = randomUUID(); + const successorId = randomUUID(), wakeId = randomUUID(), ownerId = randomUUID(); + const cwd = path.join(home.home, "saved-task"); + await mkdir(cwd, { recursive: true }); + await writeFile(path.join(cwd, "saved.txt"), "Preserved work before preparation failed.\n"); + await db.insert(authUsers).values({ id: ownerId, name: "Task owner", email: `${ownerId}@example.test`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companies).values({ id: companyId, name: "Automatic cleanup retry", issuePrefix: `R${companyId.slice(0, 6)}`, defaultResponsibleUserId: ownerId }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: ownerId, status: "active", membershipRole: "owner" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Continuation worker", status: "idle", adapterType: "process", + adapterConfig: { cwd }, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue preserved work", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: ownerId }); + await db.insert(heartbeatRuns).values({ id: sourceId, companyId, agentId, issueId, status: "failed", runtimeMode: "legacy", + startedAt: new Date(Date.now() - 10_000), finishedAt: new Date(), executionStage: "preparing", errorCode: "adapter_failed", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, contextSnapshot: { issueId } }); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: sourceId, + provider: "daytona", providerLeaseId: randomUUID(), status: "active" }).returning(); + await db.insert(agentWakeupRequests).values({ id: wakeId, companyId, agentId, source: "automation", triggerDetail: "system", + reason: "issue_continuation_needed", requestedByActorType: "system", status: "queued", runId: successorId, + payload: { issueId } }); + await db.insert(heartbeatRuns).values({ id: successorId, companyId, agentId, issueId, status: "queued", invocationSource: "automation", + responsibleUserId: ownerId, retryOfRunId: sourceId, wakeupRequestId: wakeId, + contextSnapshot: { issueId, retryOfRunId: sourceId, source: "issue.continuation_recovery", + wakeReason: "issue_continuation_needed", skipIssueComment: true } }); + await db.update(issues).set({ executionRunId: successorId, executionAgentNameKey: "continuation-worker", executionLockedAt: new Date() }) + .where(eq(issues.id, issueId)); + execute.mockReset(); + execute.mockImplementation(async (context: { runId: string }) => { + expect(context.runId).toBe(successorId); + expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("Preserved work before preparation failed.\n"); + await writeFile(path.join(cwd, "continued.txt"), "Continued once after cleanup.\n"); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + return { exitCode: 0, signal: null, timedOut: false, summary: "Continued preserved work after cleanup." }; + }); + const heartbeat = heartbeatService(db), errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let completedCleanupDispatches = 0; + const cleanupStatus = mode === "failed cleanup" ? "failed" : "success"; + try { + await registerRunDispatchWork(coordinator, { promoteDueScheduledRetries: async () => undefined, + nextScheduledRetryAt: async () => null, schedulingSuppressed: async () => false, + resumeQueuedRunsAfterDrain: async () => { + const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + await heartbeat.resumeQueuedRunsAfterDrain({ companyId }); + if (current?.status === "released" && current.releasedAt && current.cleanupStatus === cleanupStatus) + completedCleanupDispatches++; + }, + }, () => true).ready; + expect((await heartbeat.getRun(successorId))?.status).toBe("queued"); + expect(execute).not.toHaveBeenCalled(); + if (mode === "stop") await heartbeat.cancelRun(successorId, "Cancelled by a board operator", { + resultJson: { cancelledByActorType: "user", cancelledByUserId: ownerId }, + }); + if (mode === "reassign") { + const [replacement] = await db.insert(agents).values({ companyId, name: "New owner", adapterType: "process" }).returning(); + await db.update(issues).set({ assigneeAgentId: replacement.id }).where(eq(issues.id, issueId)); + } + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "released", releasedAt: new Date(), + cleanupStatus }).where(eq(environmentLeases.id, lease.id)), + { companyId, runId: sourceId }); + await vi.waitFor(() => expect(completedCleanupDispatches).toBeGreaterThan(0), { timeout: 10_000 }); + if (mode === "continue") { + await heartbeat.drainActiveRunExecutions(); + expect((await heartbeat.getRun(successorId))?.status).toBe("succeeded"); + expect(execute).toHaveBeenCalledTimes(1); + expect(await readFile(path.join(cwd, "continued.txt"), "utf8")).toBe("Continued once after cleanup.\n"); + const [done] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(done).toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null }); + } else { + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 }); + expect((await heartbeat.getRun(successorId))?.status).toBe(mode === "failed cleanup" ? "queued" : "cancelled"); + expect(execute).not.toHaveBeenCalled(); + } + expect((await heartbeat.getRun(sourceId))?.status).toBe("failed"); + expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("Preserved work before preparation failed.\n"); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); + await heartbeat.drainActiveRunExecutions(); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId)); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + home.cleanup(); + } + }, 30_000, + ); + it("retries a failed startup intent through timer failure and drain before verified cancellation", async () => { const home = createIsolatedTestHome(), originalHome = process.env.PAPERCLIP_HOME; process.env.PAPERCLIP_HOME = home.home; diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index cf6e2f7bffb..9e87edb1c00 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1,6 +1,6 @@ import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js"; import type { Db } from "@paperclipai/db"; -import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; @@ -259,6 +259,94 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { return { company, agent, issue, wake, run }; } + async function pendingPredecessorFixture() { + const f = await fixture("queued"); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + issueId: f.issue.id, status: "failed", runtimeMode: "legacy", finishedAt: new Date(), + executionStage: "preparing", errorCode: "adapter_failed", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + contextSnapshot: { issueId: f.issue.id } }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.company.id, issueId: f.issue.id, + heartbeatRunId: source.id, provider: "daytona", providerLeaseId: crypto.randomUUID(), status: "active" }).returning(); + const [successor] = await db.update(heartbeatRuns).set({ retryOfRunId: source.id, responsibleUserId: "queue-test-user", + contextSnapshot: { issueId: f.issue.id, source: "issue.continuation_recovery", retryOfRunId: source.id, + wakeReason: "issue_continuation_needed" } }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + const deps = callbacks(db); + deps.getAgent.mockResolvedValue(f.agent); + return { ...f, source, lease, successor, deps, queue: createHeartbeatQueue(db, deps) }; + } + + it("keeps an automatic continuation queued while its legacy predecessor still owns cleanup", async () => { + const { successor, lease, queue, deps, ...f } = await pendingPredecessorFixture(); + expect(await queue.claimQueuedRun(successor)).toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successor.id)))[0].status).toBe("queued"); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].executionRunId).toBe(successor.id); + expect(deps.executeRun).not.toHaveBeenCalled(); + + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success" }) + .where(eq(environmentLeases.id, lease.id)); + expect(await queue.claimQueuedRun(successor)).toMatchObject({ id: successor.id, status: "running" }); + expect(await queue.claimQueuedRun(successor)).toBeNull(); + }); + + it("keeps a retry queued after failed cleanup and until its local executor releases", async () => { + const f = await pendingPredecessorFixture(); + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "failed" }) + .where(eq(environmentLeases.id, f.lease.id)); + expect(await f.queue.claimQueuedRun(f.successor)).toBeNull(); + await db.update(environmentLeases).set({ cleanupStatus: "success" }).where(eq(environmentLeases.id, f.lease.id)); + f.deps.liveRunExecutions.has.mockImplementation(id => id === f.source.id); + expect(await f.queue.claimQueuedRun(f.successor)).toBeNull(); + f.deps.liveRunExecutions.has.mockReturnValue(false); + expect(await f.queue.claimQueuedRun(f.successor)).toMatchObject({ id: f.successor.id, status: "running" }); + }); + + it.each(["missing", "foreign company", "other task", "other agent", "native task overrides context"] as const)( + "does not hold a queued run for a %s retry source", async mismatch => { + const f = await pendingPredecessorFixture(); + let sourceId = f.source.id; + if (mismatch === "missing") { + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, sourceId)); + expect(await f.queue.claimQueuedRun(f.successor)).toMatchObject({ id: f.successor.id, status: "running" }); + return; + } else if (mismatch === "foreign company") { + const [company] = await db.insert(companies).values({ name: "Unrelated company", issuePrefix: "OTHER" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Other company worker", adapterType: "process" }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "failed", finishedAt: new Date() }).returning(); + sourceId = source.id; + await db.insert(environmentLeases).values({ companyId: company.id, heartbeatRunId: sourceId, status: "active", provider: "daytona" }); + } else if (mismatch === "other agent") { + const [other] = await db.insert(agents).values({ companyId: f.company.id, name: "Another worker", adapterType: "process" }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: other.id, issueId: f.issue.id, + status: "failed", finishedAt: new Date() }).returning(); + sourceId = source.id; + await db.insert(environmentLeases).values({ companyId: f.company.id, heartbeatRunId: sourceId, status: "active", provider: "daytona" }); + } else { + const [other] = await db.insert(issues).values({ companyId: f.company.id, title: "Another task" }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, issueId: other.id, + ...(mismatch === "native task overrides context" ? { nativeIssueId: other.id } : {}), + contextSnapshot: { issueId: f.issue.id }, status: "failed", finishedAt: new Date() }).returning(); + sourceId = source.id; + await db.insert(environmentLeases).values({ companyId: f.company.id, heartbeatRunId: sourceId, status: "active", provider: "daytona" }); + } + const [successor] = await db.update(heartbeatRuns).set({ retryOfRunId: sourceId }) + .where(eq(heartbeatRuns.id, f.successor.id)).returning(); + expect(await f.queue.claimQueuedRun(successor)).toMatchObject({ id: successor.id, status: "running" }); + }, + ); + + it("retains the current execution owner's hold when retry provenance is unrelated", async () => { + const f = await pendingPredecessorFixture(); + const [unrelated] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "succeeded", finishedAt: new Date() }).returning(); + const [successor] = await db.update(heartbeatRuns).set({ retryOfRunId: unrelated.id }) + .where(eq(heartbeatRuns.id, f.successor.id)).returning(); + await db.update(issues).set({ executionRunId: f.source.id }).where(eq(issues.id, f.issue.id)); + expect(await f.queue.claimQueuedRun(successor)).toBeNull(); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].executionRunId).toBe(f.source.id); + }); + it.each((["process", "paperclip_runner"] as const).flatMap(adapterType => (["coalesced", "claimed-primary", "legacy-envelope", "delete-old", "delete-new", "stopped", "running", "system-primary", "foreign-primary", "same-named-agent", "mixed-author"] as const).map(scenario => ({ adapterType, scenario }))))("protects a queued comment batch across $adapterType/$scenario", async ({ adapterType, scenario }) => { const f = await fixture("queued"); let admissionAgent: typeof agents.$inferSelect = { ...f.agent, adapterType, adapterConfig: adapterType === "paperclip_runner" ? { provider: "codex" } : {} }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index aa0977e5659..dff68ca5b46 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1079,14 +1079,20 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (ownsIssue && issueId && await taskChatRetrySourceCleanupPending(tx, run, issueId)) { return { ownsIssue, blocked: true }; } - const previousRunId = run.scheduledRetryReason === "native_provider_overloaded" + const executionOwnerId = run.scheduledRetryReason === "native_provider_overloaded" ? run.retryOfRunId : ownsIssue ? owner?.executionRunId : null; - if (previousRunId && previousRunId !== run.id) { - const [previous] = await tx.select({ status: heartbeatRuns.status }) + for (const previousRunId of new Set([executionOwnerId, ownsIssue ? run.retryOfRunId : null])) { + if (!previousRunId || previousRunId === run.id) continue; + const [previous] = await tx.select({ status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, + issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, + contextIssueId: sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`, + }) .from(heartbeatRuns).where(and( eq(heartbeatRuns.id, previousRunId), eq(heartbeatRuns.companyId, run.companyId), )); + if (previousRunId !== executionOwnerId && (previous?.agentId !== run.agentId || + (previous.nativeIssueId ?? previous.issueId ?? previous.contextIssueId) !== issueId)) continue; // A terminal result can precede workspace/lease cleanup on this or // another controller. Local absence alone is not a release receipt. if (!isHeartbeatRunTerminalStatus(previous?.status) ||