diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a67dda11b8d..138f2f91559 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -951,6 +951,8 @@ The task thread exposes the guarded Retry action for failed or timed-out convers A continuation cancelled by the queued-run gate while waiting for review did not start a provider. Its matching dispatch-gate receipt and unclaimed execution fields keep this deliberate wait out of unknown-action reconciliation. Pending review and decision gates still apply. If an older server already created a recovery hold for this exact state, Retry or a new user message can request one fresh turn after the server verifies there is no provider, coordinator, or unfinished cleanup. Saved user input is reconsidered through the same admission path; the cancelled turn is not replayed. +Other legacy queued-run gate cancellations also stay out of unknown-action reconciliation when the terminal run has no retained execution authority, session, workspace recovery, resources, or execution events. The same proof removes the effect of an old false recovery hold only for its exact company, task, and run; the audit row remains. Started or conflicting work retains its hold. Review waits retain their guarded Retry path. New work still passes the current approval, dependency, budget, pause, and ownership gates. + A successful connected run can still have newer admitted input waiting behind it. Task Retry resumes that input's existing durable chat receipt through the inbound owner, instead of replaying the successful run as a failed attempt. The original diff --git a/server/src/__tests__/heartbeat-undispatched-execution-wait.test.ts b/server/src/__tests__/heartbeat-undispatched-execution-wait.test.ts index 40ad08c67f4..4dc7bb99676 100644 --- a/server/src/__tests__/heartbeat-undispatched-execution-wait.test.ts +++ b/server/src/__tests__/heartbeat-undispatched-execution-wait.test.ts @@ -151,6 +151,16 @@ const support = await getEmbeddedPostgresTestSupport(); }); }); + it.each(["nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "nativeWorkspaceFinalizationOwner", "sessionCheckpoint"])( + "retains an admission-wait hold with a retained %s profile", async key => { + const f = await seed(true); + const [run] = await db.update(heartbeatRuns).set({ runnerProfileJson: { [key]: { retained: true } } }) + .where(eq(heartbeatRuns.id, f.runId)).returning(); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(true); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ recoveryActionId: f.actionId }); + }, + ); + it.each(["started", "runtime_selected", "process", "session", "controller", "stage", "receipt", "receipt_issue", "receipt_action", "lease", "coordinator", "launch", "provider", "workspace"])( "retains a hold with conflicting or incomplete %s evidence", async kind => { const f = await seed(true); diff --git a/server/src/__tests__/legacy-continuation-authority.test.ts b/server/src/__tests__/legacy-continuation-authority.test.ts index 9d8bbd35dc9..2581f087528 100644 --- a/server/src/__tests__/legacy-continuation-authority.test.ts +++ b/server/src/__tests__/legacy-continuation-authority.test.ts @@ -1,10 +1,13 @@ import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db"; import { heartbeatService } from "../services/heartbeat.js"; import { recoveryService } from "../services/recovery/service.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; +import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js"; +import { getExecutionBlocker } from "../services/execution-blocker.js"; describe("legacy continuation persisted authority", () => { let temporary: Awaited>; @@ -33,6 +36,103 @@ describe("legacy continuation persisted authority", () => { } return { companyId, agentId, issueId, runId, createRecovery, runs, actions, finish }; } + it("does not freeze a never-started dispatch cancellation or retain its historical hold", async () => { + const f = await fixture(); + await db.update(heartbeatRuns).set({ status: "queued", resultJson: null }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId, + runId: f.runId, expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_assignee_changed" }); + const run = (await f.runs())[0]; + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(false); + await terminalizeLegacyExecution({ db, run, status: "cancelled", fromStatuses: ["cancelled"], reconcileIfNeeded: true }); + expect(await f.actions()).toHaveLength(0); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board", + fingerprint: `legacy-execution:${run.id}`, status: "resolved", outcome: "blocked", nextAction: "Reconcile", + evidence: { runId: run.id, automaticRecovery: { replay: "blocked" } } }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + for (const errorCode of ["issue_not_found", "issue_not_in_progress", "issue_waiting_for_response", + "issue_dependencies_blocked", "issue_blocked", + "issue_execution_lock_changed", "issue_terminal_status", "issue_review_participant_changed"]) { + const changed = { ...run, errorCode }; + await db.update(heartbeatRuns).set(changed).where(eq(heartbeatRuns.id, run.id)); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, changed)).toBe(false); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + } + const reviewWait = { ...run, errorCode: "issue_continuation_waiting_on_review", resultJson: { + stopReason: "issue_continuation_waiting_on_review", timeoutSource: "stale_queued_run_gate", + } }; + await db.update(issues).set({ assigneeAgentId: f.agentId }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set(reviewWait).where(eq(heartbeatRuns.id, run.id)); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, reviewWait)).toBe(false); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ canRetry: true }); + for (const patch of [{ finishedAt: null }, { startedAt: new Date() }, { runtimeModeResolvedAt: new Date() }, + { processPid: 123456789 }, { processGroupId: 123456789 }, { processStartedAt: new Date() }, + { sessionIdAfter: "provider-session" }, { controllerBootId: randomUUID() }, + { controllerLeaseExpiresAt: new Date() }, { executionStage: "adapter_running" }, { errorCode: "cancelled" }, + { resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } }, + { resultJson: { workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1" } } }, + ...["nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "nativeWorkspaceFinalizationOwner", "sessionCheckpoint"] + .map(key => ({ runnerProfileJson: { [key]: { retained: true } } }))]) { + const changed = { ...run, ...patch }; + await db.update(heartbeatRuns).set(changed).where(eq(heartbeatRuns.id, run.id)); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, changed)).toBe(true); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + } + await db.update(heartbeatRuns).set(run).where(eq(heartbeatRuns.id, run.id)); + await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: run.id, agentId: f.agentId, + seq: 2, eventType: "adapter.invoke" }); + await db.update(heartbeatRuns).set({ nextEventSeq: 3 }).where(eq(heartbeatRuns.id, run.id)); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(true); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + }); + it("keeps historical holds for missing, foreign, or conflicting sources and workspace recovery", async () => { + const f = await fixture(), foreign = await fixture(); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, foreign.issueId)); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), + errorCode: "issue_dependencies_blocked", resultJson: null }).where(eq(heartbeatRuns.id, f.runId)); + const evidence = { runId: f.runId, automaticRecovery: { replay: "blocked" } }; + const [hold] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board", + fingerprint: `legacy-execution:${f.runId}`, status: "resolved", outcome: "blocked", nextAction: "Reconcile", evidence }).returning(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + for (const changed of [{ ...evidence, runId: "malformed" }, { ...evidence, runId: randomUUID() }, + { ...evidence, runId: foreign.runId }, { automaticRecovery: evidence.automaticRecovery }, + { ...evidence, workspaceRestoreFailure: "restore_unsafe_archive" }, + { ...evidence, workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1" } }]) { + await db.update(issueRecoveryActions).set({ evidence: changed }).where(eq(issueRecoveryActions.id, hold.id)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + } + await db.update(issueRecoveryActions).set({ evidence: { sourceRunId: f.runId, + automaticRecovery: evidence.automaticRecovery } }).where(eq(issueRecoveryActions.id, hold.id)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Another task" }).returning(); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: other.id } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + await db.update(heartbeatRuns).set({ issueId: f.issueId, + contextSnapshot: { issueId: f.issueId } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + expect(await f.actions()).toHaveLength(1); + }); + it.each(["released environment", "provider event"])("retains a hold with a recorded %s", async retained => { + const f = await fixture(); + const [run] = await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), + errorCode: "issue_dependencies_blocked", resultJson: null }).where(eq(heartbeatRuns.id, f.runId)).returning(); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board", + fingerprint: `legacy-execution:${f.runId}`, status: "active", nextAction: "Reconcile", evidence: { runId: f.runId } }); + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(false); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + if (retained === "released environment") await db.insert(environmentLeases).values({ companyId: f.companyId, + heartbeatRunId: f.runId, issueId: f.issueId, provider: "local", status: "released", releasedAt: new Date(), cleanupStatus: "success" }); + else { + await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.runId, agentId: f.agentId, + seq: 1, eventType: "provider.diagnostic", sourceEventId: "provider-event-1" }); + await db.update(heartbeatRuns).set({ nextEventSeq: 2 }).where(eq(heartbeatRuns.id, f.runId)); + } + expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(true); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + }); it.each([1, 2, 3, 4, 5])("deduplicates concurrent replay with stale prose-derived liveness (%s)", async () => { const f = await fixture(); const recovery = f.createRecovery(); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 4e57d9b0c88..146e551f358 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -1,4 +1,5 @@ -import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; +import { isPreDispatchReviewWait, neverStartedDispatchCancellationPredicate } from "./pre-dispatch-review-wait.js"; +import { canonicalExecutionUuid } from "./execution-identity.js"; import { preDispatchExecutionWaitRecoveryActionPredicate } from "./pre-dispatch-execution-wait.js"; import { and, desc, eq, gt, inArray, not, or, sql } from "drizzle-orm"; import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js"; @@ -97,6 +98,17 @@ export function executionBlockerPredicate() { return and( not(conversationRecoveryActionPredicate()!), not(preDispatchExecutionWaitRecoveryActionPredicate()), + sql`not (${issueRecoveryActions.cause} = 'legacy_execution_requires_reconciliation' and exists ( + select 1 from ${heartbeatRuns} + where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId} + and ${heartbeatRuns.id} = ${canonicalExecutionUuid(sql`coalesce(${issueRecoveryActions.evidence}->>'runId', ${issueRecoveryActions.evidence}->>'sourceRunId')`)} + and ${issueRecoveryActions.sourceIssueId} = ${canonicalExecutionUuid(sql`${heartbeatRuns.contextSnapshot}->>'issueId'`)} + and (${heartbeatRuns.issueId} is null or ${heartbeatRuns.issueId} = ${issueRecoveryActions.sourceIssueId}) + and ${neverStartedDispatchCancellationPredicate()} + and ${heartbeatRuns.errorCode} <> 'issue_continuation_waiting_on_review' + and coalesce(${issueRecoveryActions.evidence}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive' + and coalesce(${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema', '') <> ${LEGACY_WORKSPACE_RECOVERY_SCHEMA} + ))`, or(unreconciledNativeExportPredicate(), and( inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]), or(inArray(issueRecoveryActions.status, ["active", "escalated"]), diff --git a/server/src/services/pre-dispatch-execution-evidence.ts b/server/src/services/pre-dispatch-execution-evidence.ts new file mode 100644 index 00000000000..b7b9aa93408 --- /dev/null +++ b/server/src/services/pre-dispatch-execution-evidence.ts @@ -0,0 +1,32 @@ +import { and, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations } from "@paperclipai/db"; +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 executionStartedEventPredicate() { + return 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"])); +} + +/** Both dispatch cancellation policies require this same retained-evidence + * proof. Each caller still owns its terminal status and admission receipt. */ +export function noRetainedExecutionPredicate() { + return and( + isNull(heartbeatRuns.startedAt), isNull(heartbeatRuns.runtimeModeResolvedAt), + isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), isNull(heartbeatRuns.processStartedAt), + isNull(heartbeatRuns.nativeIssueId), isNull(heartbeatRuns.nativeSessionId), isNull(heartbeatRuns.sessionIdAfter), + isNull(heartbeatRuns.controllerBootId), isNull(heartbeatRuns.controllerLeaseExpiresAt), isNull(heartbeatRuns.executionStage), + sql`coalesce(${heartbeatRuns.resultJson}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive'`, + sql`coalesce(${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema', '') <> ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}`, + sql`not (coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) ?| array[ + 'nativeExecutionInput', 'nativeWorkspaceSync', 'remoteWorkspace', 'nativeWorkspaceFinalizationOwner', 'sessionCheckpoint'])`, + sql`not exists (select 1 from ${nativeRunFinalizations} + where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id})`, + sql`not exists (select 1 from ${environmentLeases} + where ${environmentLeases.companyId} = ${heartbeatRuns.companyId} and ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id})`, + sql`not exists (select 1 from ${heartbeatRunEvents} + where ${and(eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), + eq(heartbeatRunEvents.runId, heartbeatRuns.id), executionStartedEventPredicate())})`, + ); +} diff --git a/server/src/services/pre-dispatch-execution-wait.ts b/server/src/services/pre-dispatch-execution-wait.ts index 228d1a12d74..b91bbc878be 100644 --- a/server/src/services/pre-dispatch-execution-wait.ts +++ b/server/src/services/pre-dispatch-execution-wait.ts @@ -1,7 +1,7 @@ import { and, eq, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { heartbeatRuns, issueRecoveryActions, type Db } from "@paperclipai/db"; import { canonicalExecutionUuid } from "./execution-identity.js"; -import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; +import { noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js"; import { LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; /** Admission denied by an existing hold is not a provider attempt. Require the @@ -18,27 +18,7 @@ function preDispatchExecutionWaitPredicate() { AND (${heartbeatRuns.issueId} IS NULL OR ${heartbeatRuns.issueId} = ${canonicalExecutionUuid(sql`${heartbeatRuns.contextSnapshot}->>'issueId'`)}) AND ${canonicalExecutionUuid(sql`${heartbeatRuns.resultJson}->'executionWait'->>'recoveryActionId'`)} IS NOT NULL - AND ${heartbeatRuns.startedAt} IS NULL AND ${heartbeatRuns.runtimeModeResolvedAt} IS NULL - AND ${heartbeatRuns.processPid} IS NULL AND ${heartbeatRuns.processGroupId} IS NULL - AND ${heartbeatRuns.processStartedAt} IS NULL AND ${heartbeatRuns.nativeIssueId} IS NULL - AND ${heartbeatRuns.nativeSessionId} IS NULL AND ${heartbeatRuns.sessionIdAfter} IS NULL - AND ${heartbeatRuns.controllerBootId} IS NULL AND ${heartbeatRuns.controllerLeaseExpiresAt} IS NULL - AND ${heartbeatRuns.executionStage} IS NULL - AND coalesce(${heartbeatRuns.resultJson}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive' - AND coalesce(${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema', '') <> ${LEGACY_WORKSPACE_RECOVERY_SCHEMA} - AND NOT EXISTS (SELECT 1 FROM ${environmentLeases} - WHERE ${environmentLeases.companyId} = ${heartbeatRuns.companyId} - AND ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id}) - AND NOT EXISTS (SELECT 1 FROM ${nativeRunFinalizations} - WHERE ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} - AND ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}) - AND NOT EXISTS (SELECT 1 FROM ${heartbeatRunEvents} - WHERE ${heartbeatRunEvents.companyId} = ${heartbeatRuns.companyId} - AND ${heartbeatRunEvents.runId} = ${heartbeatRuns.id} - AND (${heartbeatRunEvents.sourceEventId} IS NOT NULL OR ${heartbeatRunEvents.eventType} IN - ('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'))), false)`; + AND ${noRetainedExecutionPredicate()}, false)`; } export async function isPreDispatchExecutionWaitVerified(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise { diff --git a/server/src/services/pre-dispatch-review-wait.ts b/server/src/services/pre-dispatch-review-wait.ts index 7448d5f8e55..68454275e1e 100644 --- a/server/src/services/pre-dispatch-review-wait.ts +++ b/server/src/services/pre-dispatch-review-wait.ts @@ -1,13 +1,43 @@ -import { and, eq, inArray, isNotNull, or } from "drizzle-orm"; +import { and, eq, inArray, isNotNull } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; -import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; +import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; +import { executionStartedEventPredicate, noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js"; +import type { QueuedRunStalenessErrorCode } from "../modules/run-dispatch/index.js"; type Run = typeof heartbeatRuns.$inferSelect; type DispatchFields = "startedAt" | "runtimeModeResolvedAt" | "processPid" | "processGroupId" | "processStartedAt" | "nativeIssueId" | "nativeSessionId" | "sessionIdAfter" | "controllerBootId" | "controllerLeaseExpiresAt" | "executionStage" | "contextSnapshot"; +const staleDispatchReconciliationExemptions = { + issue_not_found: true, + issue_not_in_progress: true, + issue_assignee_changed: true, + issue_waiting_for_response: true, + issue_continuation_waiting_on_review: true, + issue_dependencies_blocked: true, + issue_blocked: true, + issue_execution_lock_changed: true, + issue_terminal_status: true, + issue_review_participant_changed: true, + // This code requires its own executionWait receipt in pre-dispatch-execution-wait.ts. + execution_reconciliation_required: false, +} satisfies Record; +const staleDispatchErrorCodes = Object.entries(staleDispatchReconciliationExemptions) + .filter(([, exempt]) => exempt).map(([errorCode]) => errorCode); + +/** A dispatch cancellation is not an interrupted provider attempt. Historical + * rows need no new receipt, but every retained execution reference must agree. */ +export function neverStartedDispatchCancellationPredicate() { + return and( + eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "cancelled"), + inArray(heartbeatRuns.errorCode, staleDispatchErrorCodes), + isNotNull(heartbeatRuns.finishedAt), + noRetainedExecutionPredicate(), + ); +} + /** The queued-run gate owns this receipt before execution authority is claimed. * A deliberate review or connection response wait has no provider actions to reconcile. The error code * alone is not proof; missing or conflicting dispatch evidence retains the hold. */ @@ -39,6 +69,17 @@ export function isPreDispatchReviewWait( * retained execution or cleanup evidence. Historical false holds must be * checked against the same evidence that gates explicit user admission. */ export async function isPreDispatchReviewWaitVerified(db: Db, run: Run): Promise { + if (run.runtimeMode === "legacy" && run.status === "cancelled" && run.finishedAt !== null && + staleDispatchErrorCodes.some(errorCode => errorCode === run.errorCode) && + [run.startedAt, run.runtimeModeResolvedAt, run.processPid, run.processGroupId, run.processStartedAt, + run.nativeIssueId, run.nativeSessionId, run.sessionIdAfter, run.controllerBootId, + run.controllerLeaseExpiresAt, run.executionStage].every(value => value === null) && + run.resultJson?.workspaceRestoreFailure !== "restore_unsafe_archive" && !hasRequiredWorkspaceRecovery(run.resultJson)) { + const [unstarted] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), neverStartedDispatchCancellationPredicate(), + )).limit(1); + if (unstarted) return true; + } if (!isPreDispatchReviewWait(run)) return false; const [coordinator] = await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), @@ -52,10 +93,7 @@ export async function isPreDispatchReviewWaitVerified(db: Db, run: Run): Promise : !hasRemoteTerminationReceipt(lease))) return false; 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 !execution; } diff --git a/tests/e2e/legacy-failure-continuation.spec.ts b/tests/e2e/legacy-failure-continuation.spec.ts index 46ce639e5b5..2fc7d49508d 100644 --- a/tests/e2e/legacy-failure-continuation.spec.ts +++ b/tests/e2e/legacy-failure-continuation.spec.ts @@ -40,6 +40,10 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag } })); if (action === "automatic_message_on_reactivation") { await json(await request.patch(`/api/companies/${company.id}`, { data: { status: "paused" } })); + // Pause cleanup cancels pending wakeups. Finish that operation before + // seeding the historical saved input that reactivation must recover. + await expect.poll(async () => (await json(await request.get(`/api/agents/${agent.id}`))).lifecycleState, + { timeout: 45_000 }).toBe("paused"); } const sourceRunId = randomUUID(); // Seed the historical incident, then exercise ordinary recovery entry points.