From ffcde13f5729c685c506393e133f71ccfb7ffd59 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 14:53:15 -0600 Subject: [PATCH] fix(workspaces): continue explicitly from preserved coordinators --- .../services/explicit-native-continuation.ts | 15 ++++++++++- .../legacy-workspace-recovery-migration.ts | 26 +++++++++++++------ .../src/services/workspace-recovery.test.ts | 16 ++++++++++-- 3 files changed, 46 insertions(+), 11 deletions(-) diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index b45214afd1c..5a3493a0dbe 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -6,6 +6,8 @@ import { readTaskRetryIntent } from "./task-retry.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; +import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js"; import { createHash } from "node:crypto"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; @@ -380,8 +382,19 @@ export async function admitExplicitNativeContinuation(input: { // cleanup failed. That immutable result is history, not an active commit. // Keep it intact and require the same controller/process/lease stop proofs // before admitting new user input; never apply or replay the old result. + const legacyReference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); + const migratedSource = coordinator && !coordinator.resultId && !coordinator.leaseOwner + && ["observed", "retryable_failure"].includes(coordinator.phase) && legacyReference + && await hasVerifiedWorkspaceRepair(db, run) + ? await readMigratedLegacyWorkspaceSource(db, { companyId, runId: run.id, reference: legacyReference }) : null; + // Offline migration fenced this exact unfinished turn and preserved its + // coordinator. A verified attachment permits new user input without + // inventing a terminal result or replaying the old session. All stop and + // ownership checks below still apply. + const preservedFailure = migratedSource?.issueId === issueId + && migratedSource.unresolvedFinalization?.phase === coordinator?.phase; if (!cancelledStartup && coordinator && ( - (coordinator.phase !== "terminal_failure" && !committedFailure) || coordinator.leaseOwner || + (coordinator.phase !== "terminal_failure" && !committedFailure && !preservedFailure) || coordinator.leaseOwner || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", run.status === "cancelled" && !coordinator.leaseOwner ? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details." diff --git a/server/src/services/legacy-workspace-recovery-migration.ts b/server/src/services/legacy-workspace-recovery-migration.ts index f9ef179fe41..85c4596e5c7 100644 --- a/server/src/services/legacy-workspace-recovery-migration.ts +++ b/server/src/services/legacy-workspace-recovery-migration.ts @@ -37,25 +37,35 @@ export async function isLegacyWorkspaceTaskRelated(db: Pick, input /** This receipt changes where recovery may save files; it never certifies that * a provider stopped or that its files were recovered. The recovery owner must * still validate allocation provenance and termination before exporting. */ -export async function hasMigratedLegacyWorkspaceSource(db: Pick, input: { +export async function readMigratedLegacyWorkspaceSource(db: Pick, input: { companyId: string; runId: string; reference: NativeWorkspaceSyncReference; -}) { - if (input.reference.state !== "prepared") return false; +}): Promise { + if (input.reference.state !== "prepared") return null; const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.reference.workspaceId), )); - if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return false; + if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return null; const marker = workspace?.metadata?.legacyWorkspaceRecoveryMigration as { schema?: unknown; operationId?: unknown } | undefined; if (marker?.schema !== LEGACY_WORKSPACE_RECOVERY_MIGRATION || typeof marker.operationId !== "string") throw new Error("legacy_workspace_migration_receipt_invalid"); + const [run] = await db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), + )); + if (!isDeepStrictEqual(run?.profile?.legacyWorkspaceRecoveryMigration, { + schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: marker.operationId, workspaceId: input.reference.workspaceId, + })) throw new Error("legacy_workspace_migration_receipt_invalid"); const [operation] = await db.select({ metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and( eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.executionWorkspaceId, input.reference.workspaceId), eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"), eq(workspaceOperations.status, "succeeded"), )); const sources = operation?.metadata?.legacyRecoverySources; - if (!Array.isArray(sources) || !sources.some(source => source && typeof source === "object" - && source.runId === input.runId && isDeepStrictEqual(source.reference, input.reference))) - throw new Error("legacy_workspace_migration_receipt_invalid"); - return true; + const source = Array.isArray(sources) ? sources.find(source => source && typeof source === "object" + && source.runId === input.runId && typeof source.issueId === "string" && isDeepStrictEqual(source.reference, input.reference)) : null; + if (!source) throw new Error("legacy_workspace_migration_receipt_invalid"); + return source as LegacyWorkspaceRecoverySource; +} + +export async function hasMigratedLegacyWorkspaceSource(...args: Parameters) { + return Boolean(await readMigratedLegacyWorkspaceSource(...args)); } diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 7939fa0e5f5..b22ed0178b2 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -249,9 +249,21 @@ describe("automatic retained workspace recovery", () => { const nextRunId = randomUUID(), nextLeaseId = randomUUID(), sourceRun = await run(f.runId); const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "operator", body: "/new" }).returning(); - expect(await admitExplicitNativeContinuation({ db, companyId: f.companyId, issueId: f.issueId, + const continuationInput = { db, companyId: f.companyId, issueId: f.issueId, agentId: sourceRun.agentId, actorType: "user", actorId: "operator", reason: "issue_commented", - commentId: comment.id, successorRunId: nextRunId })).toMatchObject({ previousRunId: f.runId, commentId: comment.id }); + commentId: comment.id, successorRunId: nextRunId }; + if (mode === "expired-owner") { + for (const changed of [{ phase: "retryable_failure", leaseOwner: null }, { phase: "observed", leaseOwner: "new-controller" }]) { + await db.update(nativeRunFinalizations).set(changed).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await admitExplicitNativeContinuation({ ...continuationInput, dryRun: true })).toBeNull(); + } + await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...sourceRun.runnerProfileJson, + legacyWorkspaceRecoveryMigration: { invalid: true } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await admitExplicitNativeContinuation({ ...continuationInput, dryRun: true })).toBeNull(); + await db.update(heartbeatRuns).set({ runnerProfileJson: sourceRun.runnerProfileJson }).where(eq(heartbeatRuns.id, f.runId)); + } + expect(await admitExplicitNativeContinuation(continuationInput)).toMatchObject({ previousRunId: f.runId, commentId: comment.id }); await db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, agentId: sourceRun.agentId, nativeIssueId: f.issueId, runtimeMode: "native", status: "running" }); const storage = { ...getStorageService(), provider: "s3" as const };