mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
fix(native): send a re-wake whose crashed predecessor's warm Runner cannot be adopted to a fresh sandbox instead of failing it (#21001)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native run on a warm remote sandbox leaves its Runner `ready` in the sandbox. The next run of the task adopts that Runner live, after it proves that the previous owner finished > - When the server is killed after a run parked on a question, the warm owner lived only in the killed server's memory. The Runner in the sandbox was never suspended > - A re-wake on the replacement server resumed that sandbox and tried live adoption. A proof failed, and the new run failed with `runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate` and needed a Retry, although it did no provider work > - #20863 already sends such a run to a fresh sandbox when the predecessor's Runner cannot be retired, but the live-adoption path of a resumed sandbox (`remote-warm-session-recovery.ts`) was not connected to it, and the failed run's inherited Runner identity blocked the "workspace unchanged" proof that the fallback needs > - This pull request connects the two: a refusal of live adoption after the predecessor is durably terminal with an accepted result goes to the #20863 fresh-sandbox path, and an inherited Runner identity no longer counts as a process of the follow-up > - The benefit is that a re-wake seconds after a crash restart continues without a Retry ## Linked Issues or Issue Description Refs #20863, #20844, #20928, #20979, #20953, #20972 **What happened?** Restart-chaos wave `w-20261112T012841-s7000` (master `4b24352712`, volume-free mode, Daytona warm, Codex, 10 tasks, K1 and K15 with a fresh home and overlap), task t08 (W2Q): - Run `58f57bf1` asked its question and succeeded at 01:29:56. Its warm owner stayed in the server's memory. - K1 killed the server at 01:30:06. The replacement was ready at 01:30:15. - A board comment woke the task at 01:30:17. Run `8f812df6` started at 01:30:19 and resumed the same sandbox (`run.startup.reuse` `verified_reuse`). - The first live-adoption inspection passed. The second check (`revalidate`, in backend setup, before the Runner attach) refused at 01:30:21 (stack: `remote-warm-session-recovery.ts:217` from `revalidate`). The run failed `native_session_interrupted` with `INSTRUCTION_COLLECTION_UNAVAILABLE`. - The failed run kept its full save because the unchanged-files proof refused with `process_recorded`: the run carried its predecessor's Runner process identity. Retry (`6968dc0f`) worked. **Expected behavior** The predecessor finished with an accepted result before the kill, and the new run did no provider work. The task must continue without a Retry. **Steps to reproduce** 1. Finish a warm remote run that parks on a question. 2. Kill the server (SIGKILL) and start a replacement. 3. Wake the task within seconds. The new run resumes the warm sandbox. 4. Any live-adoption proof fails, and the run fails terminally with `prior_owner_terminal_state_indeterminate`. **Which proof failed** The old refusal had no reason. The wave evidence proves that it was `revalidate`, after a passed inspection, and not the SIGTERM of the next K15 fault (that server logged "shutdown signal received" only at 01:31:09). `revalidate` checks the predecessor's coordinator row, the current run's coordinator row and lease, and the task. The evidence cannot tell which one changed in the first seconds after the restart. This pull request names the failed proof (`warm_recovery_refused=<reason>`), so the next occurrence identifies it. The fix covers the class: every refusal that concerns the predecessor or the sandbox. ## What Changed - `server/src/services/native-runtime/remote-warm-session-recovery.ts`: - Each refusal names its proof from a closed list: `identity_incomplete`, `target_not_resumed_sandbox`, `owner_unsettled`, `execution_mismatch`, `checkpoint_unsettled`, `journal_unsettled`, `controller_not_stopped`, `prior_lease_mismatch`, `owner_changed`, `current_lease_changed`, `current_run_not_admitted`, `current_owner_changed`. - After the predecessor's run is `succeeded` and its coordinator is `committed` with a result and no lease, a refusal about the predecessor, its journal or the sandbox throws `NativePredecessorRetirementUnprovenError` (predecessor run, sandbox). The existing #20863 path then marks the sandbox `replaceBeforeReuse` and re-queues the run in the uncharged `interrupted_preparation` lane. The fresh sandbox is used only when the workspace save is clean and the harness backup is verified (unchanged #20863 safety rules); otherwise the sandbox is resumed as before. - A refusal because the new run is no longer admitted (stopped, reassigned, paused, company inactive) or no longer owned by this controller stays an ordinary failure, so a fallback cannot become a path around Stop or reassignment. - `server/src/services/remote-workspace-unchanged-proof.ts`: a run whose recorded Runner process identity is exactly a settled predecessor's identity of the same session (`carriesInheritedRunnerIdentity`, from #20979), with no `native.process_identity_recorded` event of its own, is no longer refused with `process_recorded`. A provider session of its own (`sessionIdAfter`) and every other check still refuse. - `doc/execution-semantics.md`: document both rules. ## Verification - Red on `origin/master` (`8818402e7a`): - `cd server && npx vitest run src/services/native-runtime/native-controller-recovery.integration.test.ts -t "crash restart"`: `AssertionError: expected Error: runner_state_identity_mismatch: pr… to be an instance of NativePredecessorRetirementUnprovenError`. - `npx vitest run src/services/remote-workspace-unchanged-release.test.ts -t "follow-up that carries"`: `expected false to be true` for the inherited-identity case; the own-identity and own-record cases passed (still refused). - Green with the fix: `native-controller-recovery.integration`, `remote-workspace-unchanged-release`, `native-recovered-owner-followup` and `heartbeat-native-predispatch-retry` pass (149 tests). Also run: `native-session-executor.test`, `explicit-native-continuation.test`, `task-workspace-binding.test`, `native-runner-restart-recovery.integration.test` (see the PR comment for the result). `tsc --noEmit` for the server passes. - The new integration case is the staging shape: a predecessor that parked on a question (unanswered `semantic_tool.input` before its terminal), a replacement controller, a resumed sandbox. It checks the journal refusal, a predecessor row taken by another worker between inspection and revalidation, a changed current lease, and that a cancelled current run is not sent to the fallback. - Chaos: the before/after rerun of seed 7000 is in a PR comment. ## Risks - The fallback starts the next attempt on a fresh sandbox instead of reusing the warm Runner. That costs one cold start, only after live adoption is refused. - The unchanged-files proof accepts an inherited Runner identity. The identity must match a settled predecessor of the same session exactly (PID, group and start time), and the run must have no own identity record, no execution event, no runner checkpoint, no workspace operation and no result. These are the same facts #20979 uses for restart recovery. - Error text of this refusal gains `; warm_recovery_refused=<reason>`. Callers match the `runner_state_identity_mismatch: ` prefix only. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
0cfab73520
commit
533f135780
5 files changed
+171
-21
No files matched your search
@@ -3526,6 +3526,25 @@ reason uses only fixed lists of known names: command types, command statuses,
|
||||
owner phases, recovery states, lifecycle states, and fence reasons. Any other
|
||||
stored value appears as `other`. The reason never includes payloads or paths.
|
||||
|
||||
Live adoption of a warm Runner on a resumed sandbox names its failed proof as
|
||||
`warm_recovery_refused=<reason>`, for example `journal_unsettled`,
|
||||
`owner_changed` or `current_lease_changed`. After the predecessor's run row
|
||||
proves `succeeded` and its coordinator proves `committed` with an accepted
|
||||
result and no lease, such a refusal does not fail the task. The new run did no
|
||||
provider work, so it fails with `native_predecessor_retirement_unproven` and
|
||||
its next attempt starts on a fresh sandbox restored from the saved workspace,
|
||||
under the same safety rules as any unretired predecessor. This covers a server
|
||||
that was killed after a run parked on a question: its warm Runner stays `ready`
|
||||
in the sandbox, and only the killed server could have adopted it. A refusal
|
||||
because the new run itself is no longer admitted (stopped, reassigned, paused
|
||||
or no longer owned by this controller) stays an ordinary failure.
|
||||
|
||||
A follow-up on a warm session starts with its predecessor's Runner process
|
||||
identity. That inherited identity alone does not count as a process of the
|
||||
follow-up when it proves that it left the workspace unchanged. The identity
|
||||
must equal a settled predecessor's identity of the same session, and the
|
||||
follow-up must have no `native.process_identity_recorded` event of its own.
|
||||
|
||||
Resume source selection skips a failed follow-up that never started provider
|
||||
work. Such a run holds only the provider checkpoint that it inherited. Without
|
||||
the skip, a retry selects it, and controller recovery fails with
|
||||
|
||||
@@ -18,7 +18,8 @@ import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
|
||||
import { assertNativeRetainedSandboxBinding, createNativeControllerRecovery, deferNativeControllerRecovery, deferNativeRunForBusyAgentDirectory, NATIVE_AGENT_DIRECTORY_BUSY_WAIT, importNativeControllerRecovery, NativeControllerRecoveryWaitingError, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js";
|
||||
import { claimNativeRestartRecoveries } from "./native-restart-recovery.js";
|
||||
import { claimNativeRestartRecoveries, currentNativeControllerIdentity } from "./native-restart-recovery.js";
|
||||
import { NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js";
|
||||
import { AgentDirectoryPreparationDeferral, deferRunForBusyAgentDirectory, prepareWithAgentDirectoryDeferral } from "../heartbeat/agent-directory-busy.js";
|
||||
import { AgentDirectoryOperationBusyError } from "../agent-directory-working-copies.js";
|
||||
import { readNativeControllerRecoveryInventory } from "./native-controller-recovery-inventory.js";
|
||||
@@ -521,6 +522,83 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d
|
||||
expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "duplicate", runIds: [runId], now: deadline })).toEqual([]);
|
||||
});
|
||||
|
||||
it("sends a re-wake after a crash restart to a fresh sandbox when the parked predecessor's warm Runner cannot be adopted", async () => {
|
||||
const fixture = await seed(), runId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID();
|
||||
const previous = parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null },
|
||||
binding: { companyId, agentId, issueId: fixture.scope.issueId, runId: fixture.scope.runId, executionWorkspaceId: randomUUID() },
|
||||
task: { identifier: "NCD-1", title: "Answer the parked question", description: null,
|
||||
prompt: "Continue after the answer.", workMode: "standard" },
|
||||
workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: fixture.scope.normalizedSessionId, driverKind: "codex_app_server",
|
||||
protocolVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } },
|
||||
completionContract: { id: randomUUID(), sha256: "fixture", schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: { revision: "1", objective: "Continue after the answer.", criteria: [{ id: "objective", requirement: "Save the result." }] } },
|
||||
credentialBindings: [], interactionResponses: [],
|
||||
});
|
||||
const current = { ...previous, binding: { ...previous.binding, runId } };
|
||||
const checkpoint: PersistedNativeSession = { ...fixture.snapshot, driverKind: "codex_app_server",
|
||||
providerSessionId: "parked-provider-session", activeTurnId: null, pendingRuntimeRequests: [] };
|
||||
const metadata = createNativeControllerRecovery(checkpoint, compatibility, { runId: fixture.scope.runId,
|
||||
runnerInstanceId: fixture.native.runnerInstanceId, normalizedSessionId: fixture.scope.normalizedSessionId });
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.profile, nativeExecutionInput: previous,
|
||||
sessionCheckpoint: checkpoint, nativeControllerRecovery: metadata } }).where(eq(heartbeatRuns.id, fixture.scope.runId));
|
||||
await db.update(nativeRunFinalizations).set({ controllerPid: 2_147_483_000, controllerProcessStartedAt: new Date(Date.now() - 120_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.scope.runId));
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: fixture.scope.issueId,
|
||||
nativeSessionId: fixture.scope.normalizedSessionId, status: "running", runtimeMode: "native",
|
||||
runnerProfileJson: { nativeExecutionInput: current } });
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId: fixture.scope.issueId, runId, phase: "observed",
|
||||
leaseOwner: "replacement-controller", leaseExpiresAt: new Date(Date.now() + 60_000),
|
||||
controllerBootId: controller.bootId, controllerPid: controller.pid });
|
||||
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, fixture.scope.issueId));
|
||||
await db.insert(environments).values({ id: environmentId, name: `Parked ${environmentId}`, driver: "sandbox" });
|
||||
await db.insert(environmentLeases).values([
|
||||
{ companyId, environmentId, issueId: fixture.scope.issueId, heartbeatRunId: fixture.scope.runId, provider: "daytona",
|
||||
providerLeaseId: "warm-sandbox", status: "released", metadata: { remoteCwd: "/workspace" } },
|
||||
{ id: leaseId, companyId, environmentId, issueId: fixture.scope.issueId, heartbeatRunId: runId, provider: "daytona",
|
||||
providerLeaseId: "warm-sandbox", status: "active", metadata: { remoteCwd: "/workspace" } },
|
||||
]);
|
||||
const identity = { runId: fixture.scope.runId, normalizedSessionId: fixture.scope.normalizedSessionId,
|
||||
runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId,
|
||||
turnId: fixture.native.turnId, itemId: fixture.native.itemId };
|
||||
const parked = { identity, commands: [{ type: "turn.start", status: "completed" }], committedEvents: [
|
||||
{ eventType: "semantic_tool.input", sourceEventId: "question", envelope: identity },
|
||||
{ eventType: "run.terminal", envelope: identity }] };
|
||||
const inspect = (readControlState: () => Record<string, unknown> = () => parked) => inspectRemoteWarmSessionRecovery({
|
||||
db, execution: current, identity, readControlState,
|
||||
target: { kind: "remote", transport: "sandbox", environmentId, leaseId, providerKey: "daytona", remoteCwd: "/workspace",
|
||||
sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "warm-sandbox" } } as never,
|
||||
matchesExecution: saved => saved.binding.runId === fixture.scope.runId,
|
||||
loadCheckpoint: () => checkpoint, canReuseProcess: () => true,
|
||||
});
|
||||
const fallback = (error: unknown, reason: string) => {
|
||||
expect(error).toBeInstanceOf(NativePredecessorRetirementUnprovenError);
|
||||
expect(error).toMatchObject({ predecessorRunId: fixture.scope.runId, providerLeaseId: "warm-sandbox" });
|
||||
expect((error as Error).message).toContain(`warm_recovery_refused=${reason}`);
|
||||
};
|
||||
|
||||
fallback(await inspect(() => ({ ...parked, commands: [...parked.commands, { type: "runner.suspend", status: "pending" }] }))
|
||||
.catch(error => error), "journal_unsettled");
|
||||
|
||||
const recovered = (await inspect())!;
|
||||
await expect(recovered.revalidate("replacement-controller")).resolves.toBeUndefined();
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: "native-finalizer:replacement", leaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.scope.runId));
|
||||
fallback(await recovered.revalidate("replacement-controller").catch(error => error), "owner_changed");
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.scope.runId));
|
||||
await db.update(environmentLeases).set({ status: "pending_cleanup" }).where(eq(environmentLeases.id, leaseId));
|
||||
fallback(await recovered.revalidate("replacement-controller").catch(error => error), "current_lease_changed");
|
||||
await db.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, leaseId));
|
||||
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, runId));
|
||||
const stopped = await recovered.revalidate("replacement-controller").catch(error => error);
|
||||
expect(stopped).not.toBeInstanceOf(NativePredecessorRetirementUnprovenError);
|
||||
expect(stopped.message).toBe("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; warm_recovery_refused=current_run_not_admitted");
|
||||
});
|
||||
|
||||
// The Nov 10 staging shape: a preparation-only run's epoch was abandoned, so
|
||||
// the warm follow-up restored the last save into a new allocation of the same
|
||||
// sandbox. The committed source's Runner state stays in its own directory.
|
||||
|
||||
@@ -4,6 +4,7 @@ import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/executio
|
||||
import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js";
|
||||
import { currentNativeControllerIdentity, evaluateNativeControllerTakeover, recordLegacyHostFallbackProof } from "./native-restart-recovery.js";
|
||||
import { NativeControllerRecoveryWaitingError, readNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js";
|
||||
|
||||
export type RemoteWarmSessionRecovery = {
|
||||
identity: Record<string, string>;
|
||||
@@ -114,6 +115,10 @@ export function describeFailedAttachment(failureMessage: unknown): string | null
|
||||
return `This run could not attach to its provider session: ${cause}. A later run may start a new provider thread. See ${SETTLEMENT_DOC}`;
|
||||
}
|
||||
|
||||
type WarmRecoveryRefusal = "identity_incomplete" | "target_not_resumed_sandbox" | "owner_unsettled" | "execution_mismatch"
|
||||
| "checkpoint_unsettled" | "journal_unsettled" | "controller_not_stopped" | "prior_lease_mismatch"
|
||||
| "current_run_not_admitted" | "current_owner_changed" | "owner_changed" | "current_lease_changed";
|
||||
|
||||
export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
db: Db;
|
||||
execution: NativeExecutionInput;
|
||||
@@ -124,13 +129,23 @@ export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
loadCheckpoint: (previous: NativeExecutionInput) => PersistedNativeSession | null;
|
||||
canReuseProcess: (previous: NativeExecutionInput) => boolean;
|
||||
}): Promise<RemoteWarmSessionRecovery | null> {
|
||||
const deny = (): never => { throw new Error("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"); };
|
||||
const { binding } = input.execution;
|
||||
const { target } = input;
|
||||
let settledPredecessor: { runId: string; providerLeaseId: string } | null = null;
|
||||
const deny = (reason: WarmRecoveryRefusal): never => {
|
||||
const refusal = new Error(`runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; warm_recovery_refused=${reason}`);
|
||||
if (settledPredecessor) {
|
||||
throw new NativePredecessorRetirementUnprovenError(settledPredecessor.runId, settledPredecessor.providerLeaseId, refusal);
|
||||
}
|
||||
throw refusal;
|
||||
};
|
||||
const refuseCurrentRun = (reason: WarmRecoveryRefusal): never => {
|
||||
throw new Error(`runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; warm_recovery_refused=${reason}`);
|
||||
};
|
||||
const identity: Record<string, string> = {};
|
||||
for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) {
|
||||
const value = input.identity[key];
|
||||
if (typeof value !== "string" || !value) return deny();
|
||||
if (typeof value !== "string" || !value) return deny("identity_incomplete");
|
||||
identity[key] = value;
|
||||
}
|
||||
// A Runner lives in the sandbox it started in. A created or replacement
|
||||
@@ -142,28 +157,29 @@ export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
}
|
||||
if (target.transport !== "sandbox" || !target.leaseId || !target.environmentId ||
|
||||
!target.providerKey || !target.sandboxLeaseAcquisition || target.sandboxLeaseAcquisition.outcome !== "resumed" ||
|
||||
identity.runId === binding.runId) return deny();
|
||||
identity.runId === binding.runId) return deny("target_not_resumed_sandbox");
|
||||
const providerLeaseId = target.sandboxLeaseAcquisition.providerLeaseId;
|
||||
const [prior] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations })
|
||||
.from(heartbeatRuns).innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.where(and(eq(heartbeatRuns.id, identity.runId), eq(heartbeatRuns.companyId, binding.companyId),
|
||||
eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId)));
|
||||
if (!prior || prior.run.status !== "succeeded" || !prior.owner.resultId || !prior.owner.controllerPid ||
|
||||
!prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny();
|
||||
!prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny("owner_unsettled");
|
||||
const previous = parseNativeExecutionInput(prior.run.runnerProfileJson?.nativeExecutionInput);
|
||||
if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny();
|
||||
if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny("execution_mismatch");
|
||||
if (prior.owner.leaseOwner && prior.owner.leaseExpiresAt && prior.owner.leaseExpiresAt > new Date()) {
|
||||
throw new NativeControllerRecoveryWaitingError(prior.owner.leaseExpiresAt);
|
||||
}
|
||||
if (prior.owner.phase !== "committed" || prior.owner.leaseOwner) return deny();
|
||||
if (prior.owner.phase !== "committed" || prior.owner.leaseOwner) return deny("owner_unsettled");
|
||||
settledPredecessor = { runId: identity.runId, providerLeaseId };
|
||||
const checkpoint = input.loadCheckpoint(previous);
|
||||
if (!checkpoint || checkpoint.identity.runId !== identity.runId ||
|
||||
checkpoint.identity.companyId !== binding.companyId || checkpoint.identity.issueId !== binding.issueId ||
|
||||
checkpoint.identity.agentId !== binding.agentId || checkpoint.identity.sessionId !== identity.normalizedSessionId ||
|
||||
checkpoint.driverKind !== previous.session.driverKind || !checkpoint.providerSessionId ||
|
||||
checkpoint.activeTurnId !== null || checkpoint.pendingRuntimeRequests?.length !== 0) return deny();
|
||||
checkpoint.activeTurnId !== null || checkpoint.pendingRuntimeRequests?.length !== 0) return deny("checkpoint_unsettled");
|
||||
const control = input.readControlState();
|
||||
if (!isSettledRemoteWarmControlState(control, identity)) return deny();
|
||||
if (!isSettledRemoteWarmControlState(control, identity)) return deny("journal_unsettled");
|
||||
const assertPriorStopped = async () => {
|
||||
const persistedRecovery = readNativeControllerRecovery(prior.run.runnerProfileJson?.nativeControllerRecovery,
|
||||
prior.run.runnerProfileJson?.sessionCheckpoint);
|
||||
@@ -171,7 +187,7 @@ export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
// durable controllers are fenced by the database lease and current turn.
|
||||
const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), requireStoppedController: true,
|
||||
remoteLeaseFenced: persistedRecovery !== null });
|
||||
if (!stopped.allowed) return deny();
|
||||
if (!stopped.allowed) return deny("controller_not_stopped");
|
||||
await recordLegacyHostFallbackProof(input.db, prior.run, stopped, "remote_warm_session_recovery");
|
||||
};
|
||||
await assertPriorStopped();
|
||||
@@ -181,7 +197,7 @@ export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
eq(environmentLeases.provider, target.providerKey), eq(environmentLeases.providerLeaseId, providerLeaseId),
|
||||
));
|
||||
if (priorLeases.length !== 1 || priorLeases[0].metadata?.remoteCwd !== target.remoteCwd ||
|
||||
!["released", "expired", "active"].includes(priorLeases[0].status)) return deny();
|
||||
!["released", "expired", "active"].includes(priorLeases[0].status)) return deny("prior_lease_mismatch");
|
||||
return {
|
||||
identity, checkpoint, previousExecution: previous, reuseProcess: input.canReuseProcess(previous),
|
||||
remote: { providerLeaseId, remoteCwd: target.remoteCwd },
|
||||
@@ -206,15 +222,20 @@ export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
if (!current || current.run.status !== "running" || current.run.resultJson?.nativeCancellation ||
|
||||
current.company.status !== "active" || current.agent.lifecycleState !== "ready" ||
|
||||
["paused", "terminated", "pending_approval"].includes(current.agent.status) ||
|
||||
!priorOwner || priorOwner.leaseOwner || priorOwner.phase !== "committed" ||
|
||||
priorOwner.resultId !== prior.owner.resultId || priorOwner.controllerBootId !== prior.owner.controllerBootId ||
|
||||
current.owner.leaseOwner !== leaseOwner || !current.owner.leaseExpiresAt ||
|
||||
current.owner.leaseExpiresAt <= new Date() || current.owner.resultId ||
|
||||
current.owner.controllerBootId !== controller.bootId || current.owner.controllerPid !== controller.pid ||
|
||||
current.issue.executionRunId !== binding.runId || current.issue.assigneeAgentId !== binding.agentId ||
|
||||
current.issue.status !== "in_progress" || !lease || lease.status !== "active" ||
|
||||
current.issue.status !== "in_progress") return refuseCurrentRun("current_run_not_admitted");
|
||||
if (current.owner.leaseOwner !== leaseOwner || !current.owner.leaseExpiresAt ||
|
||||
current.owner.leaseExpiresAt <= new Date() || current.owner.resultId ||
|
||||
current.owner.controllerBootId !== controller.bootId || current.owner.controllerPid !== controller.pid) {
|
||||
return refuseCurrentRun("current_owner_changed");
|
||||
}
|
||||
if (!priorOwner || priorOwner.leaseOwner || priorOwner.phase !== "committed" ||
|
||||
priorOwner.resultId !== prior.owner.resultId || priorOwner.controllerBootId !== prior.owner.controllerBootId) {
|
||||
return deny("owner_changed");
|
||||
}
|
||||
if (!lease || lease.status !== "active" ||
|
||||
lease.environmentId !== target.environmentId || lease.provider !== target.providerKey ||
|
||||
lease.providerLeaseId !== providerLeaseId || lease.metadata?.remoteCwd !== target.remoteCwd) return deny();
|
||||
lease.providerLeaseId !== providerLeaseId || lease.metadata?.remoteCwd !== target.remoteCwd) return deny("current_lease_changed");
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -6,6 +6,8 @@ import {
|
||||
import { remoteWorkspaceRunReferenceSchema, type RemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
import { runOwnerFenceHeldSql } from "./remote-workspace-writer-hold.js";
|
||||
import { carriesInheritedRunnerIdentity } from "./native-runtime/native-inherited-runner-retry.js";
|
||||
import { PROCESS_IDENTITY_RECORDED } from "./native-local-process-stop.js";
|
||||
|
||||
/**
|
||||
* A writer's prepared files equal its saved version only while nothing ran in
|
||||
@@ -44,6 +46,15 @@ const FILE_NEUTRAL_OPERATION_PHASES = ["remote_workspace_restore", "workspace_fi
|
||||
/** Terminal states of a run that ended without a result. */
|
||||
const ENDED_WITHOUT_RESULT = ["failed", "cancelled", "interrupted", "timed_out"];
|
||||
|
||||
async function carriesOnlyInheritedRunnerIdentity(tx: Pick<Db, "select">,
|
||||
run: Parameters<typeof carriesInheritedRunnerIdentity>[1]): Promise<boolean> {
|
||||
if (!await carriesInheritedRunnerIdentity(tx, run)) return false;
|
||||
const [own] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED))).limit(1);
|
||||
return !own;
|
||||
}
|
||||
|
||||
export type UnchangedProofRefusal =
|
||||
| "baseline_missing" | "authority_mismatch" | "writer_not_ended" | "writer_owner_live" | "process_recorded"
|
||||
| "reference_mismatch" | "provider_session_started" | "execution_recorded" | "result_recorded"
|
||||
@@ -68,15 +79,18 @@ export async function proveRemoteWorkspaceUnchanged(tx: Pick<Db, "select">, inpu
|
||||
if (!authority.dirty || !authority.location || authority.epoch !== baseline.epoch || authority.snapshotId !== baseline.snapshotId)
|
||||
return { proven: false, reason: "authority_mismatch" };
|
||||
|
||||
const [run] = await tx.select({ status: heartbeatRuns.status, processPid: heartbeatRuns.processPid,
|
||||
const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
nativeSessionId: heartbeatRuns.nativeSessionId, createdAt: heartbeatRuns.createdAt,
|
||||
status: heartbeatRuns.status, processPid: heartbeatRuns.processPid,
|
||||
processGroupId: heartbeatRuns.processGroupId, processStartedAt: heartbeatRuns.processStartedAt,
|
||||
sessionIdAfter: heartbeatRuns.sessionIdAfter, runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
||||
ownerFenceHeld: runOwnerFenceHeldSql() }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, runId)));
|
||||
if (!run || !ENDED_WITHOUT_RESULT.includes(run.status)) return { proven: false, reason: "writer_not_ended" };
|
||||
if (!input.callerOwnsWriter && run.ownerFenceHeld) return { proven: false, reason: "writer_owner_live" };
|
||||
if (run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null || run.sessionIdAfter !== null)
|
||||
return { proven: false, reason: "process_recorded" };
|
||||
if (run.sessionIdAfter !== null) return { proven: false, reason: "process_recorded" };
|
||||
if ((run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null)
|
||||
&& !await carriesOnlyInheritedRunnerIdentity(tx, run)) return { proven: false, reason: "process_recorded" };
|
||||
const profile = object(run.runnerProfileJson);
|
||||
const reference = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace);
|
||||
if (profile.nativeWorkspaceSync != null || !reference.success || reference.data.state !== "prepared"
|
||||
|
||||
@@ -194,6 +194,24 @@ describe.skipIf(!support.supported)("remote workspace writer that ended without
|
||||
? { dirty: false, writerRunId: null, snapshotId: f.savedA.snapshotId } : { dirty: true, writerRunId: f.runB });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "the settled predecessor's warm Runner identity", own: false, event: false, released: true },
|
||||
{ label: "a Runner identity of its own", own: true, event: false, released: false },
|
||||
{ label: "the predecessor's identity and its own identity record", own: false, event: true, released: false },
|
||||
])("hands on the workspace of a follow-up that carries $label only when it never had its own Runner", async ({ own, event, released }) => {
|
||||
const f = await preparedFollowUp();
|
||||
const nativeSessionId = randomUUID(), processStartedAt = new Date(Date.now() - 60_000);
|
||||
await db.update(heartbeatRuns).set({ nativeSessionId, processPid: 41_001, processGroupId: 41_001, processStartedAt })
|
||||
.where(eq(heartbeatRuns.id, f.runA));
|
||||
await db.update(heartbeatRuns).set({ nativeSessionId, processPid: own ? 41_002 : 41_001,
|
||||
processGroupId: own ? 41_002 : 41_001, processStartedAt }).where(eq(heartbeatRuns.id, f.runB));
|
||||
if (event) await f.event("native.process_identity_recorded");
|
||||
await f.end({ status: "failed", errorCode: "native_session_interrupted", controllerLeaseExpiresAt: new Date(Date.now() + 60_000) });
|
||||
expect(await releaseUnchangedFallbackWorkspace(db, { id: f.runB, companyId: f.companyId })).toBe(released);
|
||||
expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject(released
|
||||
? { dirty: false, writerRunId: null, snapshotId: f.savedA.snapshotId } : { dirty: true, writerRunId: f.runB });
|
||||
});
|
||||
|
||||
it.each([1, 0])("queues no fallback attempt beside another writer (epoch change %i)", async epochChange => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.end({ status: "failed", errorCode: "adapter_failed" });
|
||||
|
||||
Reference in new issue
Block a user