diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 906e9fb6da2..9a9dfe7507e 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -7,6 +7,18 @@ event needs no operator endpoint. ## Native PRP Run-Log Events +`native.recovery.execution_reclaimed` records an atomic same-run recovery claim +that restores a released task execution binding. The claim locks the task, +coordinator, and original run in that order. It preserves the assignee and checks +current execution/checkout owners, other live work, dependencies, pause holds, +and stop/successor records before restoring the binding. Attachment authorization +keeps its existing requirement for that exact binding. Skipped retry candidates +wait 30 seconds before recheck so they cannot fill every bounded claim batch. +Undated retries sort first. A failed candidate cannot discard earlier committed +claims. Queued waiters do not block the original execution owner from recovery; +they still wait for that owner to settle before normal queue promotion. +This is local run-log data. + Fresh remote Codex model substitution emits `runner.model_fallback` on the system stream with `warn` severity. Its payload contains `requestedModel`, `effectiveModel`, and `codexCliVersion`. It records a preparation choice before diff --git a/server/src/__tests__/native-session-resumption.test.ts b/server/src/__tests__/native-session-resumption.test.ts index 4492e3b7db3..89f1cbb2441 100644 --- a/server/src/__tests__/native-session-resumption.test.ts +++ b/server/src/__tests__/native-session-resumption.test.ts @@ -95,6 +95,9 @@ describe("P6-25 pre-result native session recovery", () => { const bootstrapRetryRunId = "79000000-0000-4000-8000-000000000009"; const observedExpiredRunId = "79000000-0000-4000-8000-000000000010"; const observedLivePidRunId = "79000000-0000-4000-8000-000000000011"; + const fixtureRunIds = [runId, cancelledRunId, exhaustedRunId, missingCheckpointRunId, + initialRunId, bootstrapRetryRunId, observedExpiredRunId, observedLivePidRunId]; + const issueForRun = (id: string) => id === runId ? issueId : id; const persistedProfile = { mode: "native", nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId } }, @@ -118,14 +121,10 @@ describe("P6-25 pre-result native session recovery", () => { adapterType: "codex_local", status: "running", }); - await db.insert(issues).values({ - id: issueId, - companyId, - title: "Resume the same native run", - status: "in_progress", - assigneeAgentId: agentId, - workMode: "standard", - }); + await db.insert(issues).values(fixtureRunIds.map(id => ({ + id: issueForRun(id), companyId, title: "Resume the same native run", + status: "in_progress", assigneeAgentId: agentId, workMode: "standard" as const, + }))); await db.insert(heartbeatRuns).values([ { id: runId, @@ -230,7 +229,10 @@ describe("P6-25 pre-result native session recovery", () => { }, contextSnapshot: { issueId }, })), - ]); + ].map(run => ({ ...run, runtimeMode: "native" as const, + nativeIssueId: issueForRun(run.id), contextSnapshot: { issueId: issueForRun(run.id) }, + }))); + await db.insert(nativeRunFinalizations).values([ { runId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) }, { runId: cancelledRunId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) }, @@ -263,7 +265,10 @@ describe("P6-25 pre-result native session recovery", () => { leaseOwner: "prior-native-owner", leaseExpiresAt: new Date(0), })), - ]); + ].map(value => ({ ...value, issueId: issueForRun(value.runId), + phase: value.phase as typeof nativeRunFinalizations.$inferInsert.phase, + }))); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => temporary?.cleanup()); @@ -276,12 +281,15 @@ describe("P6-25 pre-result native session recovery", () => { ); const eligibleRunId = "79100000-0000-4000-8000-000000000099"; const candidateRunIds = [...heldRunIds, eligibleRunId]; + await db.insert(issues).values(candidateRunIds.map(id => ({ + id, companyId, title: "Independent recovery candidate", status: "in_progress", assigneeAgentId: agentId, + }))); await db.insert(heartbeatRuns).values( candidateRunIds.map((candidateRunId) => ({ id: candidateRunId, companyId, agentId, - nativeIssueId: issueId, + nativeIssueId: candidateRunId, status: "running", runtimeMode: "native", nativePhase: @@ -309,7 +317,7 @@ describe("P6-25 pre-result native session recovery", () => { candidateRunIds.map((candidateRunId) => ({ runId: candidateRunId, companyId, - issueId, + issueId: candidateRunId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0), @@ -341,12 +349,14 @@ describe("P6-25 pre-result native session recovery", () => { ), ).toBe(true); } finally { + await db.delete(heartbeatRunEvents).where(inArray(heartbeatRunEvents.runId, candidateRunIds)); await db .delete(nativeRunFinalizations) .where(inArray(nativeRunFinalizations.runId, candidateRunIds)); await db .delete(heartbeatRuns) .where(inArray(heartbeatRuns.id, candidateRunIds)); + await db.delete(issues).where(inArray(issues.id, candidateRunIds)); } }); @@ -523,7 +533,7 @@ describe("P6-25 pre-result native session recovery", () => { expect.objectContaining({ phase: "terminal_failure", failureCode: "native_session_interrupted" }), ]); await expect(db.select().from(issueRecoveryActions) - .where(eq(issueRecoveryActions.sourceIssueId, issueId))).resolves.toEqual([ + .where(eq(issueRecoveryActions.sourceIssueId, missingCheckpointRunId))).resolves.toEqual([ expect.objectContaining({ cause: "native_session_interrupted", wakePolicy: null }), ]); }); @@ -542,11 +552,13 @@ describe("P6-25 pre-result native session recovery", () => { it("sends exactly one Sentry event when a sweep resolves a run from running to failed", async () => { const freshRunId = "79000000-0000-4000-8000-000000000101"; + await db.insert(issues).values({ id: freshRunId, companyId, title: "Independent failure report", + status: "in_progress", assigneeAgentId: agentId }); await db.insert(heartbeatRuns).values({ id: freshRunId, companyId, agentId, - nativeIssueId: issueId, + nativeIssueId: freshRunId, status: "running", runtimeMode: "native", runtimeModeResolvedAt: new Date(), @@ -556,12 +568,12 @@ describe("P6-25 pre-result native session recovery", () => { binding: { runId: freshRunId }, }, }, - contextSnapshot: { issueId }, + contextSnapshot: { issueId: freshRunId }, }); await db.insert(nativeRunFinalizations).values({ runId: freshRunId, companyId, - issueId, + issueId: freshRunId, phase: "retryable_failure", attempt: 1, }); @@ -578,11 +590,13 @@ describe("P6-25 pre-result native session recovery", () => { it("sends no Sentry event when a sweep finds a run that is already failed", async () => { const freshRunId = "79000000-0000-4000-8000-000000000102"; + await db.insert(issues).values({ id: freshRunId, companyId, title: "Independent failure report", + status: "in_progress", assigneeAgentId: agentId }); await db.insert(heartbeatRuns).values({ id: freshRunId, companyId, agentId, - nativeIssueId: issueId, + nativeIssueId: freshRunId, status: "failed", runtimeMode: "native", runtimeModeResolvedAt: new Date(), @@ -592,12 +606,12 @@ describe("P6-25 pre-result native session recovery", () => { binding: { runId: freshRunId }, }, }, - contextSnapshot: { issueId }, + contextSnapshot: { issueId: freshRunId }, }); await db.insert(nativeRunFinalizations).values({ runId: freshRunId, companyId, - issueId, + issueId: freshRunId, phase: "retryable_failure", attempt: 1, }); diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index a79e626bc21..3572e68daf2 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1,5 +1,6 @@ +import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js"; import type { Db } from "@paperclipai/db"; -import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -171,6 +172,33 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { expect(current.executionRunId).toBe(newOwner.id); }); + it.each(["running", "failed"])("resumes the %s native owner before promoting its queued waiter", async status => { + const { run, issue, agent, company } = await fixture(status); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id, runnerProfileJson: { + nativeExecutionInput: { binding: { runId: run.id } }, + sessionCheckpoint: { providerSessionId: "original-session" }, + } }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: company.id, + issueId: issue.id, phase: "retryable_failure", attempt: 1 }); + const [waiter] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "queued", invocationSource: "assignment", responsibleUserId: "queue-test-user", contextSnapshot: { issueId: issue.id }, + }).returning(); + const deps = callbacks(db); + deps.getAgent.mockResolvedValue(agent); + deps.instanceSettings.getExperimental.mockResolvedValue({ enableNativeRunner: true } as Awaited>); + const queue = createHeartbeatQueue(db, deps); + expect(await queue.claimQueuedRun(waiter)).toBeNull(); + const claims = await claimNativeSessionResumptions({ db, runnerInstanceId: "owner", runIds: [run.id] }); + expect(claims.map(claim => claim.runId)).toEqual([run.id]); + expect(await queue.claimQueuedRun(waiter)).toBeNull(); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run.id)); + await db.update(nativeRunFinalizations).set({ phase: "committed", leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, run.id)); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, issue.id)); + expect(await queue.claimQueuedRun(waiter)).toMatchObject({ id: waiter.id, status: "running" }); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].executionRunId).toBe(waiter.id); + }); + it("cancels at the daily run cap before claiming or dispatching", async () => { const { run, agent } = await fixture("queued"); const cappedAgent = { ...agent, runtimeConfig: { heartbeat: { maxDailyRuns: 0 } } }; diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 936ccd4ca75..29f1ff7e1d0 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -3,10 +3,11 @@ import { hasPendingNativeChildCompletion } from "./native-child-completion-deliv import { dismissAutomaticCompletionReviews, decisionHasRetiredAutomaticReview } from "./automatic-completion-reviews.js"; import { logger } from "../../middleware/logger.js"; import { createHash, randomUUID } from "node:crypto"; -import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, notInArray, or, sql } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { completionContracts, + agents, heartbeatRunEvents, heartbeatRuns, issueWorkProducts, @@ -55,6 +56,12 @@ import { type NativeStatusDecision, } from "./status-arbiter.js"; +import { getNativeReviewAssignment } from "./native-review-participant.js"; +import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; +import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js"; +import { readRunCancellation } from "../run-cancellation.js"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; + export type NativeReconciliationFacts = { equivalentAttentionFamily?: boolean; canonicalReplay?: boolean; @@ -343,7 +350,7 @@ export async function claimNativeSessionResumptions(input: { }): Promise { const now = input.now ?? new Date(); const candidates = await input.db - .select({ runId: heartbeatRuns.id }) + .select({ runId: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: nativeRunFinalizations.issueId }) .from(heartbeatRuns) .innerJoin( nativeRunFinalizations, @@ -371,6 +378,7 @@ export async function claimNativeSessionResumptions(input: { : []), ), ) + .orderBy(sql`${nativeRunFinalizations.nextAttemptAt} asc nulls first`, asc(nativeRunFinalizations.runId)) .limit(input.limit ?? 25); const claims: NativeSessionResumeClaim[] = []; @@ -378,16 +386,21 @@ export async function claimNativeSessionResumptions(input: { const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`; let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null; const claimed = await input.db.transaction(async (tx) => { - const row = await tx.select({ - run: heartbeatRuns, - coordinator: nativeRunFinalizations, - }).from(heartbeatRuns) - .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) - .where(eq(heartbeatRuns.id, candidate.runId)) - .for("update") - .limit(1) - .then((rows) => rows[0] ?? null); - if (!row) return false; + await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); + const [issue] = await tx.select().from(issues).where(and( + eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId), + )).for("update").limit(1); + if (!issue || !["todo", "in_progress", "in_review"].includes(issue.status)) return false; + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, candidate.runId), + eq(nativeRunFinalizations.issueId, issue.id), + )).for("update").limit(1); + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, candidate.runId), + eq(heartbeatRuns.nativeIssueId, issue.id), + )).for("update").limit(1); + if (!run || !coordinator) return false; + const row = { run, coordinator }; if ( row.run.runtimeMode !== "native" || isNativeRunnerOwnershipHeld(row.run) || @@ -404,10 +417,37 @@ export async function claimNativeSessionResumptions(input: { ) return false; + if ((issue.executionRunId && issue.executionRunId !== run.id) || + (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return false; + const [agent] = await tx.select().from(agents).where(and( + eq(agents.companyId, candidate.companyId), eq(agents.id, run.agentId), + )).limit(1); + if (!agent || !(await evaluateAgentInvokabilityFromDb(tx as unknown as Db, agent)).invokable) return false; + if (issue.assigneeAgentId !== run.agentId && !await getNativeReviewAssignment(tx as unknown as Db, { + companyId: run.companyId, issueId: issue.id, agentId: run.agentId, + contextSnapshot: run.contextSnapshot, + })) return false; + if (await isAutomaticRecoverySuppressedByPauseHold(tx as unknown as Db, run.companyId, issue.id)) return false; + if (!(await issueService(tx as unknown as Db).getDependencyReadiness(issue.id, tx)).isDependencyReady) return false; + const [competing] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id), + inArray(heartbeatRuns.status, issue.executionRunId === run.id + ? ["running", "scheduled_retry"] : ["queued", "running", "scheduled_retry"]), + or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + )).limit(1); + if (competing) return false; + const profile = row.run.runnerProfileJson ?? {}; const persistedInput = profile.nativeExecutionInput; const checkpoint = profile.sessionCheckpoint; const failureDetail = record(row.coordinator.failureDetail); + const cancellation = readRunCancellation(run.resultJson); + const nativeCancellation = record(record(run.resultJson).nativeCancellation); + if ((cancellation?.source === "operator" && cancellation.expected) || + (["run", "issue"].includes(String(nativeCancellation.scope)) && + ["pending", "acknowledged"].includes(String(nativeCancellation.dispatchState))) || + typeof failureDetail.successorRunId === "string") return false; const originalFailureCode = typeof failureDetail.originalFailureCode === "string" ? failureDetail.originalFailureCode @@ -534,7 +574,24 @@ export async function claimNativeSessionResumptions(input: { nativePhaseUpdatedAt: now, updatedAt: now, }).where(eq(heartbeatRuns.id, row.run.id)); + await tx.update(issues).set({ + executionRunId: run.id, + executionAgentNameKey: agent.name.trim().toLowerCase() || null, + executionLockedAt: now, + updatedAt: now, + }).where(and(eq(issues.id, issue.id), eq(issues.companyId, run.companyId))); + if (issue.executionRunId !== run.id) await appendHeartbeatRunEvent(tx as unknown as Db, { + companyId: run.companyId, runId: run.id, agentId: run.agentId, + eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info", + message: "Reclaimed the original task execution for native session recovery.", + payload: { issueId: issue.id }, + }); return true; + }).catch((error: unknown) => { + terminalRunToEmit = null; + logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId }, + "failed to claim native session recovery candidate"); + return false; }); // Telemetry is best-effort background work; it must not delay claiming // the remaining candidates in this loop, so fire it and do not await it. @@ -543,6 +600,24 @@ export async function claimNativeSessionResumptions(input: { void reportRunFailure(input.db, terminalRunToEmit); } if (claimed) claims.push({ runId: candidate.runId, leaseOwner }); + else await input.db.transaction(async tx => { + await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); + await tx.update(nativeRunFinalizations).set({ + nextAttemptAt: new Date(now.getTime() + 30_000), + updatedAt: now, + }).where(and( + eq(nativeRunFinalizations.companyId, candidate.companyId), + eq(nativeRunFinalizations.issueId, candidate.issueId), + eq(nativeRunFinalizations.runId, candidate.runId), + eq(nativeRunFinalizations.phase, "retryable_failure"), + isNull(nativeRunFinalizations.resultId), + or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)), + or(isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)), + )); + }).catch((error: unknown) => { + logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId }, + "failed to defer native session recovery candidate"); + }); } return claims; } diff --git a/server/src/services/native-runtime/native-session-resumption-binding.test.ts b/server/src/services/native-runtime/native-session-resumption-binding.test.ts new file mode 100644 index 00000000000..b51ef876549 --- /dev/null +++ b/server/src/services/native-runtime/native-session-resumption-binding.test.ts @@ -0,0 +1,209 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { + agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, + issueRelations, issueTreeHolds, issueTreeHoldMembers, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { claimNativeSessionResumptions, dispatchNativeSessionResumptions } from "./native-finalization-reconciler.js"; +import { stageNativeRunnerWakeAttachments } from "./native-runner-file-handoff.js"; + +describe("native recovery execution binding", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-resume-binding-"); + db = createDb(temporary.connectionString); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => temporary?.cleanup()); + + async function fixture() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Recovery binding", issuePrefix: "N" + companyId.replaceAll("-", "").slice(0, 6).toUpperCase() }); + await db.insert(agents).values({ id: agentId, companyId, name: "Recovery owner", status: "idle", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Resume original work", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ + id: runId, companyId, agentId, issueId, nativeIssueId: issueId, status: "failed", runtimeMode: "native", + finishedAt: new Date(), contextSnapshot: { issueId }, runnerProfileJson: { + nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId, companyId, agentId, issueId, executionWorkspaceId: runId } }, + sessionCheckpoint: { sessionId: "provider-session", providerSessionId: "provider-session", identity: { runId, companyId, agentId, issueId } }, + }, + }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) }); + return { companyId, agentId, issueId, runId }; + } + + it("restores the original execution before the real attachment authorization gate", async () => { + const f = await fixture(); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "owner", runIds: [f.runId] })).toHaveLength(1); + const staged = await stageNativeRunnerWakeAttachments({ db, binding: { + ...f, workspaceRoot: "/remote/workspace", executionTargetKind: "remote", + } }); + expect(staged.attachments).toEqual([]); + await staged.cleanup(); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ + executionRunId: f.runId, executionAgentNameKey: "recovery owner", status: "in_progress", assigneeAgentId: f.agentId, + }); + }); + + it.each(["execution", "checkout", "queued"] as const)("does not replace another %s owner", async (kind) => { + const f = await fixture(); + const otherRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: otherRunId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + status: kind === "queued" ? "queued" : "failed", runtimeMode: "legacy", contextSnapshot: { issueId: f.issueId } }); + if (kind !== "queued") await db.update(issues).set(kind === "execution" + ? { executionRunId: otherRunId } : { checkoutRunId: otherRunId }).where(eq(issues.id, f.issueId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "retryable_failure", leaseOwner: null }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed"); + }); + + it.each(["done", "cancelled", "blocked"])("preserves a task moved to %s", async status => { + const f = await fixture(); + await db.update(issues).set({ status }).where(eq(issues.id, f.issueId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status, executionRunId: null }); + }); + + it("does not recover work after reassignment", async () => { + const f = await fixture(); + const nextAgent = randomUUID(); + await db.insert(agents).values({ id: nextAgent, companyId: f.companyId, name: "New owner", status: "idle" }); + await db.update(issues).set({ assigneeAgentId: nextAgent }).where(eq(issues.id, f.issueId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + }); + + it("preserves an agent pause", async () => { + const f = await fixture(); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("paused"); + }); + + it("waits for a first-class dependency", async () => { + const f = await fixture(); + const blocker = randomUUID(); + await db.insert(issues).values({ id: blocker, companyId: f.companyId, title: "Unfinished prerequisite", status: "todo" }); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker, relatedIssueId: f.issueId, type: "blocks" }); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + }); + + it("preserves a task-tree pause hold", async () => { + const f = await fixture(); + const hold = randomUUID(); + await db.insert(issueTreeHolds).values({ id: hold, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause" }); + await db.insert(issueTreeHoldMembers).values({ holdId: hold, companyId: f.companyId, issueId: f.issueId, + issueTitle: "Paused work", issueStatus: "in_progress" }); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + expect((await db.select().from(issueTreeHolds).where(eq(issueTreeHolds.id, hold)))[0].status).toBe("active"); + }); + + it.each(["run", "issue"])("preserves a recorded native %s stop", async scope => { + const f = await fixture(); + await db.update(heartbeatRuns).set({ resultJson: { nativeCancellation: { scope, dispatchState: "acknowledged" } } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + }); + + it("preserves operator cancellation when terminal status was overwritten", async () => { + const f = await fixture(); + await db.update(heartbeatRuns).set({ resultJson: { cancellation: { + source: "operator", expected: true, initiator: { type: "user" }, reason: "Stop", recordedAt: new Date().toISOString(), + } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + }); + + it("does not revive a superseded source", async () => { + const f = await fixture(); + await db.update(nativeRunFinalizations).set({ failureDetail: { successorRunId: randomUUID() } }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]); + }); + + it.each(["pause", "dependency"])("keeps a %s wait from starving the next bounded claim", async kind => { + const held = await fixture(); + const eligible = await fixture(); + if (kind === "pause") { + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, held.agentId)); + } else { + const blocker = randomUUID(); + await db.insert(issues).values({ id: blocker, companyId: held.companyId, title: "Waiting prerequisite", status: "todo" }); + await db.insert(issueRelations).values({ companyId: held.companyId, issueId: blocker, relatedIssueId: held.issueId, type: "blocks" }); + } + const now = new Date(); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(1) }) + .where(eq(nativeRunFinalizations.runId, eligible.runId)); + const input = { db, runnerInstanceId: "owner", runIds: [held.runId, eligible.runId], limit: 1, now }; + const first = await claimNativeSessionResumptions(input); + const second = await claimNativeSessionResumptions(input); + expect([...first, ...second].map(claim => claim.runId)).toEqual([eligible.runId]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, held.runId)))[0].status).toBe("failed"); + }); + + it.each(["task", "coordinator"])("dispatches earlier claims when a later %s lock times out", async kind => { + const first = await fixture(); + const locked = await fixture(); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(1) }) + .where(eq(nativeRunFinalizations.runId, locked.runId)); + let release!: () => void; + let acquired!: () => void; + const ready = new Promise(resolve => { acquired = resolve; }); + const wait = new Promise(resolve => { release = resolve; }); + const blocker = db.transaction(async tx => { + if (kind === "task") await tx.select().from(issues).where(eq(issues.id, locked.issueId)).for("update"); + else await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, locked.runId)).for("update"); + acquired(); + await wait; + }); + try { + await ready; + const dispatch = vi.fn(); + const claims = await dispatchNativeSessionResumptions({ db, runnerInstanceId: "owner", + runIds: [first.runId, locked.runId], dispatch }); + expect(claims.map(claim => claim.runId)).toEqual([first.runId]); + expect(dispatch).toHaveBeenCalledExactlyOnceWith(claims[0]); + } finally { + release(); + await blocker; + } + }, 15_000); + + it.each(["pause", "dependency"])("reaches undated retries past %s waits at real sweep intervals", async kind => { + const held = await fixture(); + const eligible = await fixture(); + if (kind === "pause") { + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, held.agentId)); + } else { + const blocker = randomUUID(); + await db.insert(issues).values({ id: blocker, companyId: held.companyId, title: "Waiting prerequisite", status: "todo" }); + await db.insert(issueRelations).values({ companyId: held.companyId, issueId: blocker, relatedIssueId: held.issueId, type: "blocks" }); + } + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null }) + .where(eq(nativeRunFinalizations.runId, eligible.runId)); + const now = Date.now(); + const claims = []; + for (const offset of [0, 30_000, 60_000]) claims.push(...await claimNativeSessionResumptions({ + db, runnerInstanceId: "owner", runIds: [held.runId, eligible.runId], limit: 1, now: new Date(now + offset), + })); + expect(claims.map(claim => claim.runId)).toEqual([eligible.runId]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, held.runId)))[0].status).toBe("failed"); + }); + + it("admits only one recovering run for a task", async () => { + const f = await fixture(); + const second = randomUUID(); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(heartbeatRuns).values({ ...source, id: second, runnerProfileJson: { + ...source.runnerProfileJson, nativeExecutionInput: { binding: { runId: second } }, + } }); + await db.insert(nativeRunFinalizations).values({ runId: second, companyId: f.companyId, issueId: f.issueId, phase: "retryable_failure", attempt: 1 }); + const results = await Promise.all([f.runId, second].map(runId => claimNativeSessionResumptions({ db, runnerInstanceId: "owner", runIds: [runId] }))); + expect(results.flat()).toHaveLength(1); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(results.flat()[0]!.runId); + }); +});