diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts index df854e62152..f0352841235 100644 --- a/server/src/__tests__/task-interruption-query.test.ts +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -4,6 +4,7 @@ import { createRunSecretRedactionRegistry } from "../services/run-secret-redacti import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, agentWakeupRequests, companies, createDb, applyPendingMigrations, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; +import { readNativeSessionResumptionAgent } from "../services/native-runtime/native-session-resumption-admission.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { captureInterruptionQuery, seedInterruptionFixture } from "./helpers/task-interruption-fixture.js"; @@ -39,6 +40,49 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => await taskInterruptionPresentation(db, issue)]) { expect(presentation).toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() }); } + // The coordinator also pushes nextAttemptAt while it defers an ineligible + // candidate. Those deferrals, or one far past the failure, never hide Retry. + const otherAgentId = randomUUID(); + await db.insert(agents).values({ id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" }); + const admission = async () => { + const [currentIssue] = await db.select().from(issues).where(eq(issues.id, issue.id)); + const [currentRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + return readNativeSessionResumptionAgent(db, { issue: currentIssue!, run: currentRun!, coordinator: coordinator! }); + }; + expect(await admission()).not.toBeNull(); + const otherRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: otherRunId, companyId, agentId: otherAgentId, status: "cancelled", + finishedAt: new Date(Date.now() - 60_000), createdAt: new Date(Date.now() - 60 * 60_000) }); + const terminatedManagerId = randomUUID(); + await db.insert(agents).values({ id: terminatedManagerId, companyId, name: "Former manager", + adapterType: "paperclip_runner", status: "terminated" }); + for (const [table, change, restore] of [ + ["issue", { assigneeAgentId: otherAgentId }, { assigneeAgentId: agentId }], + ["issue", { status: "blocked" }, { status: "in_progress" }], + ["issue", { executionRunId: otherRunId }, { executionRunId: null }], + ["issue", { checkoutRunId: otherRunId }, { checkoutRunId: null }], + ["run", { contextSnapshot: { nativeQuestionCancellation: { requestId: "q" } } }, { contextSnapshot: {} }], + // Invokability is more than status: an idle agent under a terminated manager is refused. + ["agent", { reportsTo: terminatedManagerId }, { reportsTo: null }], + ] as const) { + if (table === "issue") await db.update(issues).set(change).where(eq(issues.id, issue.id)); + else if (table === "run") await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, runId)); + else await db.update(agents).set(change).where(eq(agents.id, agentId)); + // The scheduler would defer this candidate, so the task offers Retry. + if (table !== "run") expect(await admission()).toBeNull(); + expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id)) + .toMatchObject({ state: "failed", canRetry: true }); + if (table === "issue") await db.update(issues).set(restore).where(eq(issues.id, issue.id)); + else if (table === "run") await db.update(heartbeatRuns).set(restore).where(eq(heartbeatRuns.id, runId)); + else await db.update(agents).set(restore).where(eq(agents.id, agentId)); + } + expect(await admission()).not.toBeNull(); + // A provider cooldown can schedule the resume an hour out; it stays visible. + const cooldown = new Date(Date.now() + 60 * 60_000); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: cooldown }).where(eq(nativeRunFinalizations.runId, runId)); + expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id)) + .toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: cooldown.toISOString() }); // Exhausted recovery is terminal: exactly one Retry returns. await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null }) .where(eq(nativeRunFinalizations.runId, runId)); @@ -46,6 +90,29 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => .toMatchObject({ state: "failed", canRetry: true }); }); + it("decides several pending native resumes in one list read", async () => { + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Resuming list", issuePrefix: `L${companyId.slice(0, 7)}` }); + await db.insert(agents).values([{ id: agentId, companyId, name: "Resuming agent", adapterType: "paperclip_runner" }, + { id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" }]); + const resumeAt = new Date(Date.now() + 30_000); + const list = []; + for (let index = 0; index < 6; index++) { + const [issue] = await db.insert(issues).values({ companyId, title: `Resume ${index}`, status: "in_progress", + assigneeAgentId: index % 2 ? otherAgentId : agentId }).returning(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issue.id, scopeKind: "issue", + status: "failed", runtimeMode: "native", errorCode: "native_session_interrupted", finishedAt: new Date() }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId, phase: "retryable_failure", + attempt: 1, nextAttemptAt: resumeAt }); + list.push(issue); + } + const result = await taskInterruptionPresentations(db, companyId, list); + list.forEach((issue, index) => expect(result.get(issue.id)).toMatchObject(index % 2 + ? { state: "failed", canRetry: true } + : { state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() })); + }); + it("redacts workspace preparation errors on task detail and list reads", async () => { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64)); try { diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index ba43627916a..5d97112d267 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,4 +1,5 @@ import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; +import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; @@ -158,7 +159,7 @@ export async function claimNativeSessionResumptions(input: { if (await retireSupersededNativeRetry(tx as unknown as Db, { run, coordinator, issueExecutionRunId: issue.executionRunId, now, })) return false; - if (!["todo", "in_progress", "in_review"].includes(issue.status)) return false; + if (!(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status)) return false; const row = { run, coordinator }; if ( row.run.runtimeMode !== "native" || diff --git a/server/src/services/native-runtime/native-session-resumption-admission.ts b/server/src/services/native-runtime/native-session-resumption-admission.ts index 6d27fff6f82..e768328c05a 100644 --- a/server/src/services/native-runtime/native-session-resumption-admission.ts +++ b/server/src/services/native-runtime/native-session-resumption-admission.ts @@ -8,6 +8,7 @@ import { getNativeReviewAssignment } from "./native-review-participant.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; import { normalizeAgentNameKey } from "../agent-name-key.js"; +import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js"; export async function restoreNativeSessionExecution(db: Db, input: { issue: Pick; @@ -48,7 +49,7 @@ export async function readNativeSessionResumptionAgent(db: Db, input: { const { issue, run, coordinator } = input; if (issue.companyId !== run.companyId || issue.id !== run.nativeIssueId || coordinator.companyId !== run.companyId || coordinator.issueId !== issue.id || coordinator.runId !== run.id || - !["todo", "in_progress", "in_review"].includes(issue.status) || + !(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status) || (issue.executionRunId && issue.executionRunId !== run.id) || (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return null; const [agent] = await db.select().from(agents).where(and( diff --git a/server/src/services/native-runtime/native-session-resumption-statuses.ts b/server/src/services/native-runtime/native-session-resumption-statuses.ts new file mode 100644 index 00000000000..9c19a2a9dc6 --- /dev/null +++ b/server/src/services/native-runtime/native-session-resumption-statuses.ts @@ -0,0 +1,4 @@ +/** Task statuses from which a native same-run resume may be claimed. The task + * interruption read model shares this set with resumption admission so it + * never advertises a resume the scheduler would refuse. */ +export const NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES = ["todo", "in_progress", "in_review"] as const; diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index c24db5150b4..73c7682f9be 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -216,10 +216,15 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i or ${heartbeatRuns.errorCode} = 'workspace_validation_failed' then ${heartbeatRuns.error} else null end`, errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt, scheduledRetryReason: heartbeatRuns.scheduledRetryReason, + // Candidate only: the coordinator also pushes nextAttemptAt forward while + // it defers an ineligible run. The scheduler's own admission check below + // decides whether this resume is real. nativeResumeAt: sql`(select ${nativeRunFinalizations.nextAttemptAt} from ${nativeRunFinalizations} where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed' - and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure')`.mapWith(value => value == null ? null : new Date(value)), + and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure' + and ${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null + and ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null)`.mapWith(value => value == null ? null : new Date(value)), // The run whose resultJson supplied providerFailure. A scheduled retry // shows its predecessor's reason, so redaction must use that run's registry. providerFailureRunId: sql`case when ${heartbeatRuns.resultJson}->'providerFailure' is null @@ -279,6 +284,38 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i inArray(approvals.status, ["pending", "revision_requested"]))), db.select({ status: companies.status, pauseReason: companies.pauseReason }).from(companies).where(eq(companies.id, companyId)), ]); + // Advertise a pending resume only when the scheduler would admit it, using + // its own check, so the task never hides Retry behind a deferred candidate + // and never offers Retry beside a resume that will run. Rare rows only. + // Candidates are failed native runs inside an active coordinator retry, so + // there are few. Their rows load in three batched reads; admission itself + // runs a few bounded reads per candidate, four at a time. + const resumeCandidates = runs.filter(run => run.nativeResumeAt); + if (resumeCandidates.length) { + const { readNativeSessionResumptionAgent } = await import("./native-runtime/native-session-resumption-admission.js"); + const candidateIds = resumeCandidates.map(run => run.id); + const [fullRuns, coordinators] = await Promise.all([ + db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, candidateIds))), + db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId), + inArray(nativeRunFinalizations.runId, candidateIds))), + ]); + const issueIds = [...new Set(fullRuns.flatMap(run => run.nativeIssueId ? [run.nativeIssueId] : []))]; + const issueRows = issueIds.length ? await db.select().from(issues).where(and(eq(issues.companyId, companyId), + inArray(issues.id, issueIds))) : []; + const runById = new Map(fullRuns.map(run => [run.id, run])); + const coordinatorByRun = new Map(coordinators.map(row => [row.runId, row])); + const issueById = new Map(issueRows.map(row => [row.id, row])); + for (let start = 0; start < resumeCandidates.length; start += 4) { + await Promise.all(resumeCandidates.slice(start, start + 4).map(async candidate => { + const run = runById.get(candidate.id); + const issue = run?.nativeIssueId ? issueById.get(run.nativeIssueId) : undefined; + const coordinator = coordinatorByRun.get(candidate.id); + const admitted = run && issue && coordinator + && await readNativeSessionResumptionAgent(db, { issue, run, coordinator }).catch(() => null); + if (!admitted) candidate.nativeResumeAt = null; + })); + } + } const runIds = runs.map(run => run.id); const actions = runIds.length ? await db.selectDistinctOn([issueRecoveryActions.sourceIssueId]).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.sourceIssueId, ids),