mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 05:16:50 +01:00
fix(native): recover remote sessions while preserving warm runs (#16451)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native agents keep remote processes and conversation history across warm turns. > - A completed turn can leave handoff authority only in controller memory. > - Losing that controller can block a follow-up even while its sandbox survives. > - This pull request verifies durable ownership and reconnects the exact prior process before handing it to the next run. > - Healthy follow-ups stay warm. Recovery preserves saved work without replaying a completed turn. ## Linked Issues or Issue Description **What happened?** A remote warm turn can finish without suspended backup authority. A replacement controller rejects that prior owner even when its exact process and completed checkpoint remain. A single unavailable identity probe can also stop an otherwise valid reattachment. **Expected behavior** Keep healthy follow-ups warm. Recover a completed session after proving that its controller stopped and that the successor holds current authority. Reuse compatible processes; otherwise save and resume the conversation. Retry unavailable identity reads within one shared deadline. **Steps to reproduce** Use `doc/testing-native-remote-continuity.md`. Its Linux cases execute real turns through PostgreSQL, TLS ingress, and the Rust runner. They kill a separate controller process, recover old checkpoint metadata, and change process configuration. The provider is deterministic. Related: #16432 and #16441 fix live warm-owner handoffs. #16341 adds recovery diagnostics. #14695 covers provider process retention. ## What Changed - Preserve warm process reuse between healthy remote turns. - Verify a committed prior result, stopped controller, matching checkpoint and sandbox, and current task/coordinator ownership before recovering an orphaned warm session. - Reconnect through the prior run route, then use the existing authenticated run attachment to change authority. - Persist a process-configuration digest. Close and save recovered sessions when old or changed configuration cannot support reuse. - Retry unavailable state, marker, and process reads within one 30-second deadline. Keep signals and definitive identity failures single-attempt. - Record closed recovery diagnostics, including unreadable or malformed saved state. - Retire the recovery-only ownership check after verified attachment, so later warm turns use their current authority. - Add real-process Linux regressions and refusal controls. - Expose the existing source-only file-recovery form for retained native holds, with the same board authority, evidence, and exact stop-receipt requirements as legacy holds. Recording recovered files does not change task status or replay a run. ## Verification - `pnpm -r typecheck` passed; the server check was repeated for the third-turn fix. - `pnpm build` passed after the final UI change. The UI typecheck, token gates, and map checks also pass. - Four Linux continuity cases passed: warm reuse, controller loss, legacy checkpoint metadata, and changed configuration. Each creates one conversation and starts exactly three turns, including a healthy follow-up after recovery. The fixture uses expired prior leases; the third turn keeps the runner and provider processes. - Greptile caught the retained recovery gate; the third-turn regression now covers it. An initial test omitted the heartbeat’s saved input and left a test process after failure; the fixture was corrected and its disposable container reset before the passing run. - The current focused executor/restart group passes all 749 tests, including the original failure classifications. - Native file-recovery and UI tests pass (52 tests), including blocked tasks and both native and legacy holds. - After merging current master and retaining its equivalent CI expression check, the CI policy tests pass (15 tests) and all four three-turn Linux continuity cases pass again on `dec418ae55ff21b4fb067132613658cce445f181`. - `pnpm test:run` completed its general server group with 20,572 passes, 103 skips, and three failures: the skill-audit child process and two HTTP test timeouts. The same three suites then passed in isolation (89 tests) without code changes. That local command stopped before later groups; the complete final-head CI, including general, serialized, Runner, build, and browser gates, is green. - Greptile reviewed the combined head `dec418ae55ff21b4fb067132613658cce445f181` at 5/5 with no actionable findings; both earlier findings are resolved. All final-head CI gates and the security scan passed. Staging deployment and retained-incident disposition remain deployment acceptance work. ## Risks - Recovery requires exact durable evidence. Live conflicting controllers, replacement sandboxes, Stop/pause, and uncertain effects must remain blocked. - After controller loss, tool-contract or process-configuration changes can restart the provider. Normal healthy follow-ups retain their processes. - The new idle-recovery path has Linux fake-provider evidence. Live Codex/OpenCode and retained quarantined-state recovery still need staging qualification. - No database schema, telemetry, or runner protocol changes. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository inspection, code execution, and browser tools. The session does not expose a more specific model version or context-window size. ## 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 (targeted checks and isolated reruns; full-run limitation recorded above) - [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 - [x] All Paperclip CI gates are green on `dec418ae55ff21b4fb067132613658cce445f181` - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
59a70dc9e2
commit
db11d899cf
19 files changed
+971
-72
No files matched your search
@@ -1521,6 +1521,18 @@ size. Remote checkpoint archive and expanded-size limits remain 64 MiB.
|
||||
|
||||
### Warm sandbox continuity
|
||||
|
||||
Healthy native warm turns retain their runner and provider processes. After a
|
||||
controller crash, continuation can recover a completed remote session from its
|
||||
saved checkpoint and exact sandbox process identity. The prior controller must
|
||||
be stopped, its result committed, and the successor must hold current task and
|
||||
coordinator authority. Recovery reconnects through the prior run's route before
|
||||
the authenticated run attachment changes authority. Unavailable identity probes
|
||||
have a shared 30-second retry budget; an identity conflict never permits replay.
|
||||
Saved process configuration must match before reuse. Old checkpoints, changed
|
||||
configuration, and controller-owned credential or instruction resources use
|
||||
verified suspension and saved-history resume. This recovery can restart a
|
||||
provider; it does not suspend every healthy warm turn.
|
||||
|
||||
For the native runner, warm mode requests a reusable sandbox lease **before**
|
||||
lease acquisition. The environment's explicit runner lifecycle overrides the
|
||||
agent default; an inherited lifecycle uses the current agent setting on each
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
# Native runner continuity through controller and process loss
|
||||
|
||||
## Goal and delivery boundary
|
||||
|
||||
Ship a verified fix to the selected staging instance and recover the two reported
|
||||
tasks from their retained work. Normal follow-up messages, server restarts, and
|
||||
recoverable runner loss must not strand a task behind
|
||||
`native_continuation_requires_reconciliation` or
|
||||
`runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate`.
|
||||
Removing an error label, clearing a hold without evidence, or blindly starting
|
||||
another provider does not meet this goal.
|
||||
|
||||
The user authorized implementation, review, merge, staging deployment, and task
|
||||
recovery. The supported release covers the affected native Codex and native
|
||||
OpenCode remote execution paths, plus shared lifecycle regression controls.
|
||||
It preserves existing company, ownership, Stop, pause, approval, budget, and
|
||||
workspace boundaries. Genuine conflicting ownership or an unresolved external
|
||||
effect must remain safe and actionable. This is not a promise that infrastructure
|
||||
can never fail; it is a checkable promise that the specified failures do not lose
|
||||
work or require users to diagnose runner internals.
|
||||
|
||||
Warm process reuse is a firm requirement. The user superseded the earlier
|
||||
save-and-suspend choice: healthy follow-up turns must keep the existing provider
|
||||
process. Durable ownership and recovery must coexist with that behavior. The
|
||||
suspend-every-turn candidate is withdrawn from delivery; PR #16451 is draft and
|
||||
has not been merged or deployed.
|
||||
|
||||
## Acceptance matrix
|
||||
|
||||
| Journey or fault | Required result | Evidence |
|
||||
| --- | --- | --- |
|
||||
| Completed turn followed immediately by another user message | Next turn starts once with the saved history, workspace, and message | Real assembled-service regression and staging journey |
|
||||
| Healthy remote warm follow-up | Same runner and provider process, with a verified handoff to the new run | Process-start identity and call-log assertions across real turns |
|
||||
| Controller restart while a session is idle, including immediately after completion | Durable evidence suffices to resume or safely replace the runner; no process-local ownership dependency strands the task | Hard and graceful restart fault tests |
|
||||
| Restart during startup or an active turn | Adopt the exact live execution or fence it before one bounded successor; preserve accepted results and pending messages | Real process/transport boundary fault tests |
|
||||
| Remote process exits or identity cannot initially be read | Distinguish a transient probe failure, a dead execution, and a conflicting owner; reconcile automatically when evidence permits | Linux remote-process tests and persisted recovery timeline |
|
||||
| Crash during checkpoint, recovery, or workspace copy-back | Resume the persistence obligation without replaying completed provider work; retain the sole recoverable source until saved | Restart/concurrency regression with independently checked file contents |
|
||||
| Existing saved or quarantined state from the affected release | Recover eligible old state without requiring a newly created task or discarding its history | Compatibility fixture and recovery of both reported tasks |
|
||||
| Duplicate recovery workers, lost acknowledgements, and newer user messages | At most one successor and one delivery of each message; no replay of a completed external action | Durable ledger and side-effect counter assertions |
|
||||
| Wrong company/session, a genuinely live competing owner, Stop/pause, revoked access, or unknown external effect | No unsafe adoption, signalling, destruction, or speculative action replay; concrete remaining blocker | Negative controls |
|
||||
| Deployment | Running staging application and any changed runner artifact match the tested revisions; both tasks have useful results or their genuine task-level disposition | Live revision, run, file, and task inspection |
|
||||
|
||||
Use the existing recovery budgets (three provider attempts including the original,
|
||||
30-second failure retry interval) and control-transition contract (60-second
|
||||
deadline, 15-second reconciliation, resolution or explicit blocker within 90 seconds
|
||||
with healthy dependencies). Provider work duration is not bounded by this timer.
|
||||
Restarts must not reset budgets. Any necessary departure requires explicit evidence
|
||||
and an updated contract, not a silent relaxation of acceptance.
|
||||
|
||||
## Investigation and implementation slices
|
||||
|
||||
1. Establish both baseline failure paths from persisted evidence. The remote
|
||||
identity failure still needs its exact failing operation established; do not
|
||||
infer process death from the generic error alone.
|
||||
2. Close the completed-turn ownership/checkpoint gap in its existing lifecycle
|
||||
owner, including compatibility with retained old state. Test prevention and
|
||||
recovery, not only an isolated predicate.
|
||||
3. Close the demonstrated remote recovery gap using existing fencing, finalization,
|
||||
workspace retention, and continuation services. Keep diagnostics precise enough
|
||||
to distinguish probe unavailability from conflicting identity.
|
||||
4. Qualify the changed boundary locally first, then on a representative remote
|
||||
Linux target and staging. Prepare reviewable changes and required repository
|
||||
checks; deploy the verified revision and recover the original tasks.
|
||||
|
||||
The first discriminating experiment is a completed native remote turn whose
|
||||
controller is lost before a follow-up. Recreate the old persisted state through
|
||||
the real lifecycle, then attempt the follow-up with an empty process-local owner
|
||||
map. Preserve real persistence, transport, and process boundaries where mocks
|
||||
would hide the failure. Existing recovery E2E and native restart suites are the
|
||||
starting harnesses; past passing evidence does not qualify this change.
|
||||
|
||||
## Current state
|
||||
|
||||
- Branch: `codex/native-runner-recovery`, based on `afe5b3418d`.
|
||||
- PR: https://github.com/paperclipai/paperclip/pull/16451 (draft, unmerged).
|
||||
- Normal warm turns retain their processes. The previous suspend-every-turn
|
||||
implementation is withdrawn. Earlier suspension tests are historical only.
|
||||
- The base includes #16432 (idle expiry serialized with admission) and #16441
|
||||
(exact warm-owner identity independent of configuration changes).
|
||||
- The revised executor can recover a completed remote session after its
|
||||
controller dies. It checks the committed prior result, frozen scope, saved
|
||||
checkpoint, exact remote identity, and current task/coordinator lease. It
|
||||
reconnects through the prior route before authenticated run attachment.
|
||||
- A saved process-configuration digest controls reuse after controller loss.
|
||||
Legacy metadata or changed configuration uses verified close and checkpoint
|
||||
followed by saved-history resume. Healthy warm follow-ups stay open.
|
||||
- Four Linux container/TLS/PostgreSQL cases pass: ordinary warm reuse, actual
|
||||
controller SIGKILL, older checkpoint metadata, and changed environment.
|
||||
They assert one thread creation and exactly three turn starts, including a
|
||||
healthy follow-up after recovery. The hard-restart
|
||||
cases also reject a still-live prior controller and a replacement sandbox.
|
||||
Normal follow-ups use one provider process; recovery may refresh the provider.
|
||||
These are real runner/process tests with a deterministic fake Codex, not live
|
||||
Daytona or live model evidence. The prior lease fixture uses the expired status
|
||||
observed on staging. The third turn retains the runner and provider processes.
|
||||
- Full workspace typecheck and production build pass on the warm candidate.
|
||||
The server and UI typechecks and production build also pass after the final
|
||||
changes. All 749 focused executor/restart tests pass; the full required test
|
||||
suite is running.
|
||||
- Greptile found a stale recovery ownership check retained across later warm
|
||||
turns. Recovery admission now retires after verified attachment; the three-turn
|
||||
regression passes. The earlier saved-state diagnostic finding is also fixed.
|
||||
The third-turn fix received 5/5. Fresh review of the final changes is pending.
|
||||
- The existing source-only file-recovery form now supports retained native holds
|
||||
as well as legacy holds. It preserves board authority, exact stop-receipt
|
||||
validation, task status, and the prohibition on replaying the original run.
|
||||
All 52 UI/native recovery tests, token gates, and map checks pass.
|
||||
- Fresh read-only staging records confirm both original tasks remain blocked.
|
||||
The Codex failure preceded its coordinator claim. The OpenCode failure followed
|
||||
controller takeover, with an active turn in the saved checkpoint. Its original
|
||||
log lacks the failed probe reason; process death is not established.
|
||||
- Both retained sandboxes were restored for file inspection. The completed
|
||||
website commit is intact; the other workspace retains a benchmark script and
|
||||
lockfile changes. Backup and task disposition remain in progress. No agent
|
||||
turn, task status, or deployment has been changed.
|
||||
- The selected tenant has an existing fleet rollout deferred by active work.
|
||||
Do not override it. Remaining: full tests and review, staging qualification,
|
||||
retained-state recovery, and useful disposition of both tasks.
|
||||
|
||||
Deferred: unrelated skill-import setup failures, database query optimization,
|
||||
general UI redesign, and broad provider qualification unrelated to the changed
|
||||
shared lifecycle.
|
||||
@@ -165,6 +165,14 @@ proofs, encryption keys, environment variables, provider credentials, command
|
||||
arguments, or an unsanitized stderr stream. Detailed failed-attempt diagnostics
|
||||
remain in the bounded `native_run_finalizations.recovery_history` ledger.
|
||||
|
||||
`native.remote_runner_recovery_failed` records the closed `stage` and `reason`
|
||||
when remote identity verification fails, including admission failures before the
|
||||
execution coordinator opens. Unavailable probes are retried within a shared
|
||||
30-second deadline; identity conflicts are not retried. The event distinguishes
|
||||
an unreadable identity from a confirmed missing process and contains no command
|
||||
output or credentials. Saved-state read failures use `state_read_failed`; malformed
|
||||
saved JSON uses `state_invalid`. It does not itself authorize replacement or cleanup.
|
||||
|
||||
For a prepared remote workspace interrupted before its first executor claim,
|
||||
that ledger records `remote_execution_never_claimed` with a
|
||||
`bootstrap_incomplete` disposition. The provider attempt remains zero during
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
# Native remote continuity regression
|
||||
|
||||
The opt-in case in
|
||||
`server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts`
|
||||
runs the real executor, PostgreSQL, TLS ingress, Rust runner, remote file transfer,
|
||||
and backup validation against a deterministic fake Codex process in Linux. It
|
||||
checks the same runner process identity and one provider process across three warm
|
||||
turns, provider history reuse, and verified suspension at idle close. Separate
|
||||
cases kill the actual controller process and recover the next turn. They reject
|
||||
a still-live controller and a replacement sandbox before reconnecting. Legacy
|
||||
checkpoints and changed process configuration require verified suspension and
|
||||
replacement, while preserving the conversation and starting each turn once.
|
||||
Recovery can restart a provider to refresh its tool contract; process reuse is
|
||||
required for ordinary healthy follow-ups.
|
||||
Each recovery case includes another healthy turn after the recovered turn and
|
||||
asserts that it does not start another runner or provider process.
|
||||
It does not qualify a live model or a sandbox provider API.
|
||||
|
||||
Use a disposable Linux container with Node.js, `sh`, `tar`, `base64`, `ps`, and
|
||||
`setsid`. Build `paperclip-runnerd` and `fake-codex-app-server` from the checkout's
|
||||
`packages/paperclip-runner/runner/Cargo.toml` for that container's architecture.
|
||||
Keep the Linux runner binary on the host as well: the product hashes and stages
|
||||
that artifact. Put the fake provider at `/opt/continuity/fake-codex-app-server`
|
||||
inside the container. Do not put credentials in this container.
|
||||
|
||||
Start the container with `--init` and publish container port 43127 on localhost.
|
||||
Read its assigned port with `docker port CONTAINER 43127`; an automatic host port
|
||||
can change after a container restart. The test opens a local TLS tunnel to this
|
||||
port. Create a test-only certificate and key:
|
||||
|
||||
```sh
|
||||
openssl req -x509 -newkey rsa:2048 -nodes \
|
||||
-keyout /tmp/continuity-test.key -out /tmp/continuity-test.crt \
|
||||
-days 1 -subj '/CN=localhost' \
|
||||
-addext 'subjectAltName=IP:127.0.0.1,DNS:localhost'
|
||||
```
|
||||
|
||||
Create an executable wrapper on the host at `/tmp/continuity-fake-codex`:
|
||||
|
||||
```sh
|
||||
#!/bin/sh
|
||||
if [ "$1" = '--version' ]; then
|
||||
echo 'codex-cli 0.999.0'
|
||||
exit 0
|
||||
fi
|
||||
exec /opt/continuity/fake-codex-app-server \
|
||||
--state-file-in-codex-home --durable-turn-ids --record-process-start \
|
||||
--call-log "$PWD/provider-calls.log" "$@"
|
||||
```
|
||||
|
||||
Run the test with the container name, current host port, and Linux artifact path:
|
||||
|
||||
```sh
|
||||
NODE_EXTRA_CA_CERTS=/tmp/continuity-test.crt \
|
||||
PAPERCLIP_TEST_RUNNER_TLS_KEY=/tmp/continuity-test.key \
|
||||
PAPERCLIP_TEST_RUNNER_CONTAINER=CONTAINER \
|
||||
PAPERCLIP_TEST_RUNNER_PORT=HOST_PORT \
|
||||
PAPERCLIP_TEST_LINUX_RUNNER_BINARY=/path/to/linux/paperclip-runnerd \
|
||||
PAPERCLIP_TEST_FAKE_CODEX_WRAPPER=/tmp/continuity-fake-codex \
|
||||
pnpm exec vitest run \
|
||||
server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts \
|
||||
-t 'preserves remote warm continuity' --maxWorkers=1
|
||||
```
|
||||
|
||||
The caller owns the disposable container. Remove it after the run, including
|
||||
after failure; failed cases can retain remote processes and files for inspection.
|
||||
The test cleans up its database and host session directories.
|
||||
@@ -108,7 +108,7 @@ retry and some operator controls require board. Completion or cancellation of a
|
||||
task does not clear its exact unresolved native workspace owner hold. Physical
|
||||
owner release still requires the [operator stop-verification procedure](../../doc/native-workspace-finalization-recovery.md).
|
||||
|
||||
For a resolved legacy action that retains a file hold, the same resolution API
|
||||
For a resolved legacy or native action that retains a file hold, the same resolution API
|
||||
supports source-only file repair. Board runtime management and recovery-resolution
|
||||
authority are required. The request identifies the exact source action/run and
|
||||
records recovered-file evidence; the server verifies the original stop receipt.
|
||||
|
||||
@@ -160,7 +160,7 @@ The recovery-card suite covers workspace divergence, reissue, reconcile, and
|
||||
confirmation. [Workspace export](../../ui/src/components/WorkspaceExportRecovery.test.tsx)
|
||||
covers quiet restoration, legacy/operator assistance and resolved states.
|
||||
|
||||
For a retained legacy file hold, open the source task after its recovery action
|
||||
For a retained legacy or native file hold, open the source task after its recovery action
|
||||
has resolved but its files still need recovery. An authorized board operator can
|
||||
record the verified recovered files and backup link without changing the current
|
||||
task status or starting another run. The server must independently validate the
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { createConnection, type Socket } from "node:net";
|
||||
import { createServer } from "node:tls";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
|
||||
export function dockerCommandRunner(container: string): CommandManagedRuntimeRunner {
|
||||
return {
|
||||
async execute(input) {
|
||||
const args = ["exec", "-i", "-u", "root"];
|
||||
if (input.cwd) args.push("-w", input.cwd);
|
||||
for (const [name, value] of Object.entries(input.env ?? {})) args.push("-e", `${name}=${value}`);
|
||||
args.push(container, input.command, ...(input.args ?? []));
|
||||
const child = spawn("docker", args, { stdio: ["pipe", "pipe", "pipe"] });
|
||||
const startedAt = new Date().toISOString();
|
||||
let stdout = "", stderr = "", timedOut = false;
|
||||
child.stdout.setEncoding("utf8").on("data", chunk => { stdout += chunk; });
|
||||
child.stderr.setEncoding("utf8").on("data", chunk => { stderr += chunk; });
|
||||
const timer = setTimeout(() => { timedOut = true; child.kill("SIGKILL"); }, input.timeoutMs ?? 30_000);
|
||||
child.stdin.on("error", () => undefined);
|
||||
child.stdin.end(input.stdin);
|
||||
try {
|
||||
return await new Promise<Awaited<ReturnType<CommandManagedRuntimeRunner["execute"]>>>((resolve, reject) => {
|
||||
child.on("error", reject);
|
||||
child.on("close", (exitCode, signal) => resolve({ exitCode, signal, timedOut, stdout, stderr, pid: child.pid ?? null, startedAt }));
|
||||
});
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export async function createRunnerTestTlsTunnel(input: { key: string; cert: string; port: number }) {
|
||||
const sockets = new Set<Socket>();
|
||||
const server = createServer({ key: await readFile(input.key), cert: await readFile(input.cert) }, socket => {
|
||||
const upstream = createConnection({ host: "127.0.0.1", port: input.port });
|
||||
for (const connection of [socket, upstream]) {
|
||||
sockets.add(connection);
|
||||
connection.on("close", () => sockets.delete(connection));
|
||||
connection.on("error", () => { socket.destroy(); upstream.destroy(); });
|
||||
}
|
||||
socket.pipe(upstream).pipe(socket);
|
||||
});
|
||||
await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") throw new Error("Test TLS listener unavailable");
|
||||
return { port: address.port, close: async () => {
|
||||
for (const socket of sockets) socket.destroy();
|
||||
await new Promise<void>(resolve => server.close(() => resolve()));
|
||||
} };
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
import { createDb } from "@paperclipai/db";
|
||||
import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js";
|
||||
import { dockerCommandRunner } from "./docker-command-runner.js";
|
||||
|
||||
process.once("message", async (input: {
|
||||
connectionString: string;
|
||||
container: string;
|
||||
ingressUrl: string;
|
||||
options: Parameters<typeof executePaperclipNativeSession>[0];
|
||||
}) => {
|
||||
const target = input.options.runnerExecutionTarget;
|
||||
if (target?.kind !== "remote") throw new Error("Expected remote worker target");
|
||||
try {
|
||||
const result = await executePaperclipNativeSession({
|
||||
...input.options,
|
||||
db: createDb(input.connectionString),
|
||||
runnerExecutionTarget: {
|
||||
...target,
|
||||
runner: dockerCommandRunner(input.container),
|
||||
getRunnerIngressEndpoint: async ({ path }) => {
|
||||
const endpoint = {
|
||||
kind: "authenticated_websocket" as const,
|
||||
websocketUrl: `${input.ingressUrl}${path}`, secretHeaders: [], generation: input.container,
|
||||
refresh: async () => endpoint, close: async () => undefined,
|
||||
};
|
||||
return endpoint;
|
||||
},
|
||||
},
|
||||
onLog: async (stream, text) => { process.send?.({ log: { stream, text } }); },
|
||||
});
|
||||
process.send?.({ result });
|
||||
} catch (error) {
|
||||
process.send?.({ error: error instanceof Error ? error.stack : String(error) });
|
||||
}
|
||||
});
|
||||
setInterval(() => undefined, 60_000);
|
||||
@@ -208,6 +208,13 @@ describe("native controller takeover fencing", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it.each([true, false])("requires a stopped idle controller even after its run lease was released: alive=%s", async alive => {
|
||||
await expect(evaluateNativeControllerTakeover({
|
||||
owner: owner({ leaseOwner: null, leaseExpiresAt: null }), now, requireStoppedController: true,
|
||||
isProcessAlive: () => alive, readProcessStartedAt: async () => recordedStart,
|
||||
})).resolves.toEqual({ allowed: !alive, reason: alive ? "controller_still_alive" : "controller_process_dead" });
|
||||
});
|
||||
|
||||
it("takes over a recycled controller PID", async () => {
|
||||
await expect(
|
||||
evaluateNativeControllerTakeover({
|
||||
|
||||
@@ -177,11 +177,12 @@ export async function evaluateNativeControllerTakeover(input: {
|
||||
} | null;
|
||||
isProcessAlive?: (pid: number) => boolean;
|
||||
readProcessStartedAt?: (pid: number) => Promise<Date | null>;
|
||||
requireStoppedController?: boolean;
|
||||
}): Promise<{ allowed: boolean; reason: string }> {
|
||||
const { owner, now } = input;
|
||||
const isAlive = input.isProcessAlive ?? processIsAlive;
|
||||
const readStartedAt = input.readProcessStartedAt ?? observedProcessStart;
|
||||
if (!owner.leaseOwner || !owner.leaseExpiresAt) {
|
||||
if ((!owner.leaseOwner || !owner.leaseExpiresAt) && !input.requireStoppedController) {
|
||||
return { allowed: true, reason: "unowned" };
|
||||
}
|
||||
|
||||
@@ -190,7 +191,7 @@ export async function evaluateNativeControllerTakeover(input: {
|
||||
return {
|
||||
allowed: false,
|
||||
reason:
|
||||
owner.leaseExpiresAt <= now
|
||||
owner.leaseExpiresAt !== null && owner.leaseExpiresAt <= now
|
||||
? "expired_lease_without_controller_identity"
|
||||
: "live_legacy_lease_without_controller_identity",
|
||||
};
|
||||
@@ -199,7 +200,7 @@ export async function evaluateNativeControllerTakeover(input: {
|
||||
return {
|
||||
allowed: true,
|
||||
reason:
|
||||
owner.leaseExpiresAt <= now
|
||||
owner.leaseExpiresAt !== null && owner.leaseExpiresAt <= now
|
||||
? "expired_lease_controller_process_dead"
|
||||
: "controller_process_dead",
|
||||
};
|
||||
|
||||
+218
-7
@@ -1,9 +1,10 @@
|
||||
import { enableRunFailureReporting } from "../../__tests__/helpers/run-failure-reporting.js";
|
||||
import { hasNativeLocalProcessStop } from "../native-local-process-stop.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createRequire } from "node:module";
|
||||
import { spawn } from "node:child_process";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdtemp, readFile, rename, rm } from "node:fs/promises";
|
||||
import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { createServer, type Server } from "node:http";
|
||||
import { tmpdir } from "node:os";
|
||||
import { resolve } from "node:path";
|
||||
@@ -12,6 +13,9 @@ import { and, eq } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
completionContracts,
|
||||
environmentLeases,
|
||||
environments,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
heartbeatRunEvents,
|
||||
@@ -38,10 +42,15 @@ import {
|
||||
setupRunnerPrpWebSocketServer,
|
||||
} from "../../realtime/runner-prp-ws.js";
|
||||
import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { buildNativeExecutionInput } from "./native-execution-input.js";
|
||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||
import { executePaperclipNativeSession, NativeControllerDetachedForRestartError } from "./native-session-executor.js";
|
||||
import { materializeAsset } from "./runtime-context.js";
|
||||
import { createRunnerTestTlsTunnel, dockerCommandRunner } from "../../__tests__/helpers/docker-command-runner.js";
|
||||
import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js";
|
||||
import { finalizeNativeRun } from "./native-run-finalizer.js";
|
||||
import { closeIdleWarmNativeSessionsForRestart, executePaperclipNativeSession, NativeControllerDetachedForRestartError } from "./native-session-executor.js";
|
||||
|
||||
const mockCaptureRunFailure = vi.hoisted(() => vi.fn());
|
||||
vi.mock("../../sentry.js", async () => {
|
||||
@@ -140,6 +149,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
let server: Server | null = null;
|
||||
let apiUrl: string;
|
||||
let originalPaperclipHome: string | undefined;
|
||||
let isolatedHome: ReturnType<typeof createIsolatedTestHome>;
|
||||
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
@@ -150,7 +160,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
"native-runner-restart-recovery-",
|
||||
);
|
||||
runtimeRoot = await mkdtemp(resolve(tmpdir(), "native-restart-runtime-"));
|
||||
paperclipHome = await mkdtemp(resolve(tmpdir(), "native-restart-home-"));
|
||||
isolatedHome = createIsolatedTestHome();
|
||||
paperclipHome = isolatedHome.home;
|
||||
originalPaperclipHome = process.env.PAPERCLIP_HOME;
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
const controllerStartedAt = await readProcessStartedAt(process.pid);
|
||||
@@ -196,10 +207,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
runnerPrpWebSocketInternals.resetForTests();
|
||||
await closeServer(server);
|
||||
await temporary.cleanup();
|
||||
await Promise.all([
|
||||
rm(runtimeRoot, { recursive: true, force: true }),
|
||||
rm(paperclipHome, { recursive: true, force: true }),
|
||||
]);
|
||||
await rm(runtimeRoot, { recursive: true, force: true });
|
||||
isolatedHome.cleanup();
|
||||
if (originalPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = originalPaperclipHome;
|
||||
});
|
||||
@@ -293,6 +302,208 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
};
|
||||
}
|
||||
|
||||
it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed"])("preserves remote warm continuity across completed turns: %s", async (scenario) => {
|
||||
const hardRestart = scenario !== "warm";
|
||||
const container = process.env.PAPERCLIP_TEST_RUNNER_CONTAINER!;
|
||||
const tunnel = await createRunnerTestTlsTunnel({ port: Number(process.env.PAPERCLIP_TEST_RUNNER_PORT),
|
||||
key: process.env.PAPERCLIP_TEST_RUNNER_TLS_KEY!, cert: process.env.NODE_EXTRA_CA_CERTS! });
|
||||
const runner = dockerCommandRunner(container);
|
||||
const fixture = await seedRun(`REMOTE-WARM-CONTINUITY-${scenario}`);
|
||||
const boundContract = async (runId: string) => {
|
||||
const [row] = await fixture.db.select({ contract: completionContracts }).from(heartbeatRuns)
|
||||
.innerJoin(completionContracts, eq(completionContracts.id, heartbeatRuns.completionContractId))
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
return { id: row.contract.id, sha256: row.contract.canonicalSha256,
|
||||
schemaVersion: row.contract.schemaVersion, contract: row.contract.contractJson as never };
|
||||
};
|
||||
const workspace = resolve(runtimeRoot, fixture.issueId);
|
||||
await mkdir(workspace, { recursive: true });
|
||||
const context = nativeRuntimeContextFixture();
|
||||
context.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", content: Buffer.from("Complete the assigned conversation turn.\n"), mode: 0o444 }]);
|
||||
context.aggregateDigest = canonicalNativeRuntimeContextDigest(context);
|
||||
const execution = buildNativeExecutionInput({
|
||||
companyId, agentId, runId: fixture.runId,
|
||||
issue: { id: fixture.issueId, identifier: "NRR-REMOTE-WARM-CONTINUITY", title: "Continue the same conversation", description: null, workMode: "standard" },
|
||||
taskPrompt: "Complete the first turn.",
|
||||
workspace: { id: randomUUID(), cwd: workspace, repoUrl: null, repoRef: null, branchName: null },
|
||||
normalizedSessionId: fixture.native.normalizedSessionId,
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 },
|
||||
runtimeContext: context,
|
||||
completionContract: await boundContract(fixture.runId),
|
||||
});
|
||||
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: {
|
||||
...fixture.preparedProfile, nativeExecutionInput: execution,
|
||||
} }).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
const remoteCwd = `/tmp/paperclip-continuity-${fixture.issueId}`;
|
||||
expect((await runner.execute({ command: "mkdir", args: ["-p", remoteCwd] })).exitCode).toBe(0);
|
||||
const target = {
|
||||
kind: "remote" as const, transport: "sandbox" as const,
|
||||
providerKey: "test-container", providerLeaseId: container,
|
||||
environmentId: randomUUID(), leaseId: randomUUID(), remoteCwd, runner,
|
||||
effectiveCapabilities: { runnerWebSocketIngress: true, reusableLeases: true,
|
||||
nativeSyncIn: false, nativeSyncOut: false, persistentProcessSessions: false,
|
||||
independentControlCommands: true, incrementalSessionOutput: false,
|
||||
concurrentSyncOperations: false, duplexCommandStream: false },
|
||||
sandboxLeaseAcquisition: { outcome: "created" as const, providerLeaseId: container },
|
||||
getRunnerIngressEndpoint: async ({ path }: { path: string }) => {
|
||||
const endpoint = { kind: "authenticated_websocket" as const,
|
||||
websocketUrl: `wss://127.0.0.1:${tunnel.port}${path}`, secretHeaders: [], generation: container,
|
||||
refresh: async () => endpoint, close: async () => undefined };
|
||||
return endpoint;
|
||||
},
|
||||
};
|
||||
const logs: string[] = [];
|
||||
await fixture.db.insert(environments).values({ id: target.environmentId, name: `Continuity ${fixture.issueId}`, driver: "sandbox" });
|
||||
await fixture.db.insert(environmentLeases).values({ id: target.leaseId, companyId, environmentId: target.environmentId,
|
||||
heartbeatRunId: fixture.runId, issueId: fixture.issueId, provider: target.providerKey,
|
||||
providerLeaseId: container, leasePolicy: "reuse_by_environment", metadata: { remoteCwd } });
|
||||
const options = {
|
||||
db: fixture.db, runnerInstanceId: fixture.native.runnerInstanceId,
|
||||
useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerRemoteBinaryPath: process.env.PAPERCLIP_TEST_LINUX_RUNNER_BINARY!,
|
||||
runnerRemoteCodexPath: process.env.PAPERCLIP_TEST_FAKE_CODEX_WRAPPER!,
|
||||
runnerEnvironment: { PATH: "/usr/local/bin:/usr/bin:/bin", PAPERCLIP_CONFIGURED_ENV_KEYS: '["CONTINUITY_TEST_CONFIG"]', CONTINUITY_TEST_CONFIG: "first" },
|
||||
onLog: async (_stream: "stdout" | "stderr", text: string) => { logs.push(text); },
|
||||
};
|
||||
let controller: ReturnType<typeof spawn> | undefined;
|
||||
try {
|
||||
const firstOptions = { ...options, execution, runnerExecutionTarget: target };
|
||||
const first = hardRestart ? await new Promise<{ exitCode: number | null }>((resolveResult, reject) => {
|
||||
controller = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"), resolve("server/src/__tests__/helpers/remote-warm-controller.test-worker.ts")],
|
||||
{ stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env });
|
||||
const timer = setTimeout(() => { controller?.kill("SIGKILL"); reject(new Error("Warm controller startup exceeded 60 seconds")); }, 60_000);
|
||||
controller.stdout?.on("data", data => logs.push(String(data)));
|
||||
controller.stderr?.on("data", data => logs.push(String(data)));
|
||||
controller.on("error", error => { clearTimeout(timer); reject(error); });
|
||||
controller.on("exit", (code, signal) => { clearTimeout(timer); reject(new Error(`Warm controller exited: ${code}/${signal}: ${logs.join("").slice(-3_000)}`)); });
|
||||
controller.on("message", (message: { log?: { text: string }; result?: { exitCode: number | null }; error?: string }) => {
|
||||
if (message.log) logs.push(message.log.text);
|
||||
if (message.error) { clearTimeout(timer); reject(new Error(message.error)); }
|
||||
if (message.result) { clearTimeout(timer); resolveResult(message.result); }
|
||||
});
|
||||
controller.send?.({ connectionString: temporary.connectionString, container, ingressUrl: `wss://127.0.0.1:${tunnel.port}`,
|
||||
options: { ...firstOptions, db: undefined, onLog: undefined, runnerExecutionTarget: { ...target, runner: undefined } } });
|
||||
}) : await executePaperclipNativeSession(firstOptions);
|
||||
expect(first.exitCode).toBe(0);
|
||||
await finalizeNativeRun({ db: fixture.db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
||||
const snapshot = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-state.json`] });
|
||||
expect(snapshot.exitCode).toBe(0);
|
||||
const saved = JSON.parse(snapshot.stdout);
|
||||
expect(saved.lifecycle).toBe("ready");
|
||||
const processBefore = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] });
|
||||
expect(processBefore.exitCode).toBe(0);
|
||||
const historyBefore = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] });
|
||||
const providerBefore = JSON.parse(historyBefore.stdout);
|
||||
const nextRunId = randomUUID();
|
||||
const nextLeaseId = randomUUID();
|
||||
const [nextRun] = await fixture.db.insert(heartbeatRuns).values({
|
||||
id: nextRunId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: fixture.issueId,
|
||||
invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId: fixture.issueId },
|
||||
nativeSessionId: fixture.native.normalizedSessionId,
|
||||
}).returning();
|
||||
await fixture.db.update(issues).set({ executionRunId: nextRunId, status: "in_progress" }).where(eq(issues.id, fixture.issueId));
|
||||
await fixture.db.update(environmentLeases).set({ status: "expired", releasedAt: new Date() }).where(eq(environmentLeases.id, target.leaseId));
|
||||
await fixture.db.insert(environmentLeases).values({ id: nextLeaseId, companyId, environmentId: target.environmentId,
|
||||
heartbeatRunId: nextRunId, issueId: fixture.issueId, provider: target.providerKey,
|
||||
providerLeaseId: container, leasePolicy: "reuse_by_environment", metadata: { remoteCwd } });
|
||||
const next = await prepareNativeHeartbeatRun({ db: fixture.db, run: nextRun,
|
||||
issue: { id: fixture.issueId, title: "Continue the same conversation", description: null, reviewPolicy: null },
|
||||
environmentLeaseId: nextLeaseId });
|
||||
await fixture.db.insert(nativeRunFinalizations).values({ runId: nextRunId, companyId, issueId: fixture.issueId, phase: "observed" });
|
||||
const nextOptions = { ...options, runnerInstanceId: next.runnerInstanceId,
|
||||
runnerEnvironment: scenario === "changed" ? { ...options.runnerEnvironment, CONTINUITY_TEST_CONFIG: "second" } : options.runnerEnvironment,
|
||||
execution: { ...execution, completionContract: await boundContract(nextRunId), binding: { ...execution.binding, runId: nextRunId }, task: { ...execution.task, prompt: "Complete the second conversation turn." } },
|
||||
runnerExecutionTarget: { ...target, leaseId: nextLeaseId, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: container } },
|
||||
};
|
||||
if (controller) {
|
||||
await expect(executePaperclipNativeSession(nextOptions)).rejects.toThrow("prior_owner_terminal_state_indeterminate");
|
||||
const exited = new Promise<void>(resolveExit => controller!.once("exit", () => resolveExit()));
|
||||
controller.kill("SIGKILL");
|
||||
await exited;
|
||||
controller = undefined;
|
||||
await expect(executePaperclipNativeSession({ ...nextOptions,
|
||||
runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget,
|
||||
sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "unrelated-sandbox" } },
|
||||
})).rejects.toThrow("prior_owner_terminal_state_indeterminate");
|
||||
}
|
||||
if (scenario === "legacy") {
|
||||
const checkpoints = resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner", "sessions");
|
||||
let changed = 0;
|
||||
for (const name of await readdir(checkpoints)) {
|
||||
if (!name.endsWith(".json")) continue;
|
||||
const path = resolve(checkpoints, name);
|
||||
const envelope = JSON.parse(await readFile(path, "utf8"));
|
||||
if (envelope.snapshot?.identity?.runId !== fixture.runId) continue;
|
||||
delete envelope.processConfigDigest;
|
||||
await writeFile(path, JSON.stringify(envelope), { mode: 0o600 });
|
||||
changed++;
|
||||
}
|
||||
expect(changed).toBe(1);
|
||||
}
|
||||
const [nextPrepared] = await fixture.db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(eq(heartbeatRuns.id, nextRunId));
|
||||
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...nextPrepared.profile,
|
||||
nativeExecutionInput: nextOptions.execution } }).where(eq(heartbeatRuns.id, nextRunId));
|
||||
const second = await executePaperclipNativeSession(nextOptions);
|
||||
expect(second.exitCode).toBe(0);
|
||||
const historyAfter = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] });
|
||||
const providerAfter = JSON.parse(historyAfter.stdout);
|
||||
expect(providerAfter.threadId).toBe(providerBefore.threadId);
|
||||
expect(providerAfter.nextTurn).toBe(2);
|
||||
const calls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] });
|
||||
expect(calls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(2);
|
||||
expect(calls.stdout.split("\n").filter(line => line === "thread/start")).toHaveLength(1);
|
||||
const providerStarts = calls.stdout.split("\n").filter(line => line === "process-start").length;
|
||||
expect(providerStarts).toBeGreaterThanOrEqual(1);
|
||||
expect(providerStarts).toBeLessThanOrEqual(["legacy", "changed"].includes(scenario) ? 3 : hardRestart ? 2 : 1);
|
||||
const processAfter = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] });
|
||||
expect(processAfter.exitCode).toBe(0);
|
||||
if (["legacy", "changed"].includes(scenario)) expect(processAfter.stdout).not.toBe(processBefore.stdout);
|
||||
else expect(processAfter.stdout).toBe(processBefore.stdout);
|
||||
await finalizeNativeRun({ db: fixture.db, runId: nextRunId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
||||
const thirdRunId = randomUUID();
|
||||
const thirdLeaseId = randomUUID();
|
||||
const [thirdRun] = await fixture.db.insert(heartbeatRuns).values({
|
||||
id: thirdRunId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: fixture.issueId,
|
||||
invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId: fixture.issueId },
|
||||
nativeSessionId: fixture.native.normalizedSessionId,
|
||||
}).returning();
|
||||
await fixture.db.update(issues).set({ executionRunId: thirdRunId, status: "in_progress" }).where(eq(issues.id, fixture.issueId));
|
||||
await fixture.db.update(environmentLeases).set({ status: "expired", releasedAt: new Date() }).where(eq(environmentLeases.id, nextLeaseId));
|
||||
await fixture.db.insert(environmentLeases).values({ id: thirdLeaseId, companyId, environmentId: target.environmentId,
|
||||
heartbeatRunId: thirdRunId, issueId: fixture.issueId, provider: target.providerKey,
|
||||
providerLeaseId: container, leasePolicy: "reuse_by_environment", metadata: { remoteCwd } });
|
||||
const thirdNative = await prepareNativeHeartbeatRun({ db: fixture.db, run: thirdRun,
|
||||
issue: { id: fixture.issueId, title: "Continue the same conversation", description: null, reviewPolicy: null },
|
||||
environmentLeaseId: thirdLeaseId });
|
||||
await fixture.db.insert(nativeRunFinalizations).values({ runId: thirdRunId, companyId, issueId: fixture.issueId, phase: "observed" });
|
||||
const third = await executePaperclipNativeSession({ ...nextOptions, runnerInstanceId: thirdNative.runnerInstanceId,
|
||||
execution: { ...nextOptions.execution, completionContract: await boundContract(thirdRunId),
|
||||
binding: { ...execution.binding, runId: thirdRunId }, task: { ...execution.task, prompt: "Complete the third conversation turn." } },
|
||||
runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, leaseId: thirdLeaseId },
|
||||
});
|
||||
expect(third.exitCode).toBe(0);
|
||||
const finalHistory = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] });
|
||||
expect(JSON.parse(finalHistory.stdout)).toMatchObject({ threadId: providerBefore.threadId, nextTurn: 3 });
|
||||
const finalCalls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] });
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(3);
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "thread/start")).toHaveLength(1);
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "process-start")).toHaveLength(providerStarts);
|
||||
const finalProcess = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] });
|
||||
expect(finalProcess.stdout).toBe(processAfter.stdout);
|
||||
expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: 1, busy: 0, failed: 0 });
|
||||
const suspended = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-state.json`] });
|
||||
expect(suspended.exitCode).toBe(0);
|
||||
expect(JSON.parse(suspended.stdout).lifecycle).toBe("suspended");
|
||||
} catch (error) {
|
||||
await writeFile(resolve(runtimeRoot, "remote-continuity-failure.log"), logs.join(""));
|
||||
throw new Error(`Remote continuity regression failed: ${logs.filter(line => line.startsWith("[paperclip-runner]")).join("").slice(-4_000)}`, { cause: error });
|
||||
} finally {
|
||||
controller?.kill("SIGKILL");
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
await tunnel.close();
|
||||
}
|
||||
}, 180_000);
|
||||
|
||||
async function persistRestartEvidence(input: {
|
||||
fixture: Awaited<ReturnType<typeof seedRun>>;
|
||||
runnerPid: number;
|
||||
|
||||
@@ -587,6 +587,33 @@ describe("remote controller restart adoption", () => {
|
||||
).rejects.toThrow("native_remote_recovery_lease_mismatch");
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
});
|
||||
it.each([
|
||||
["runner_state", "timeout"],
|
||||
["runner_state", "transport"],
|
||||
["identity_marker", "timeout"],
|
||||
["identity_marker", "transport"],
|
||||
["process_verification", "timeout"],
|
||||
["process_verification", "transport"],
|
||||
] as const)("reattaches after a transient %s %s without starting another runner", async (stage, failure) => {
|
||||
const { target, execute } = fixture();
|
||||
const original = execute.getMockImplementation()!;
|
||||
const operation = stage === "identity_marker"
|
||||
? "paperclip-runner-recovery-identity" : "paperclip-runner-recovery-check";
|
||||
let probes = 0;
|
||||
execute.mockImplementation(async request => {
|
||||
if ((stage === "runner_state" ? request.args[1].includes("base64") : request.args[2] === operation) && probes++ === 0) {
|
||||
if (failure === "transport") throw new Error("temporary transport outage");
|
||||
return { stdout: "", stderr: "", exitCode: -1, timedOut: true };
|
||||
}
|
||||
return original(request);
|
||||
});
|
||||
const adopted = await verifyRemoteRunnerReattachment({
|
||||
claim, target: target as never, identity, runId: "run", normalizedSessionId: "session",
|
||||
});
|
||||
expect(adopted.pid).toBe(123);
|
||||
expect(probes).toBe(2);
|
||||
expect(execute.mock.calls.every(([request]) => !request.args.join(" ").match(/nohup|mkdir|rm -|mv -|kill -"/))).toBe(true);
|
||||
});
|
||||
it("rejects a marker that cannot prove its process generation", async () => {
|
||||
const { target, execute } = fixture();
|
||||
const original = execute.getMockImplementation()!;
|
||||
@@ -599,6 +626,16 @@ describe("remote controller restart adoption", () => {
|
||||
remoteRunnerRecovery: { stage: "identity_marker", reason: "fingerprint_missing" },
|
||||
});
|
||||
});
|
||||
it.each(["failed read", "invalid JSON"])("records a closed recovery diagnostic for saved-state %s", async failure => {
|
||||
const { target, execute } = fixture();
|
||||
execute.mockResolvedValueOnce({ stdout: Buffer.from("invalid JSON").toString("base64"), stderr: "",
|
||||
exitCode: failure === "failed read" ? 1 : 0, timedOut: false });
|
||||
const error = await verifyRemoteRunnerReattachment({ claim, target: target as never, identity,
|
||||
runId: "run", normalizedSessionId: "session" }).catch(error => error);
|
||||
expect(readRemoteRunnerRecoveryError(error)).toEqual({ stage: "runner_state",
|
||||
reason: failure === "failed read" ? "state_read_failed" : "state_invalid" });
|
||||
expect(execute).toHaveBeenCalledOnce();
|
||||
});
|
||||
it.each([
|
||||
[31, "marker_missing"], [32, "marker_unsafe"], [33, "marker_read_failed"],
|
||||
] as const)("records the closed reason for marker read exit %s", async (exitCode, reason) => {
|
||||
@@ -8815,13 +8852,16 @@ describe("native warm session supervision", () => {
|
||||
describe("native session bounded recovery", () => {
|
||||
it("persists closed remote recovery evidence without replacing saved run results", async () => {
|
||||
const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
|
||||
const onEvent = vi.fn(async () => undefined);
|
||||
const remoteRunnerRecovery = { stage: "identity_marker", reason: "fingerprint_missing" } as const;
|
||||
const failure = new RemoteRunnerRecoveryError(remoteRunnerRecovery);
|
||||
state.execute.mockReset().mockRejectedValueOnce(failure);
|
||||
state.upsertRecoveryAction.mockReset().mockResolvedValue({});
|
||||
const service = vi.spyOn(issueServiceModule, "issueService").mockReturnValue({ update: vi.fn(async () => null) } as unknown as ReturnType<typeof issueServiceModule.issueService>);
|
||||
try {
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(execution, {}, {}, updates), execution, runnerInstanceId: "runner" })).rejects.toBe(failure);
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(execution, {}, {}, updates), execution, runnerInstanceId: "runner", onEvent })).rejects.toBe(failure);
|
||||
expect(onEvent).toHaveBeenCalledWith({ eventType: "native.remote_runner_recovery_failed", level: "error",
|
||||
message: "Could not verify the existing remote runner.", payload: { remoteRunnerRecovery } });
|
||||
expect(updates.find(update => update.table === nativeRunFinalizations && update.values.failureDetail)?.values.failureDetail)
|
||||
.toMatchObject({ remoteRunnerRecovery });
|
||||
const resultJson = updates.find(update => update.table === heartbeatRuns && update.values.error)?.values.resultJson as SQL;
|
||||
@@ -9714,6 +9754,42 @@ describe("native process ownership", () => {
|
||||
});
|
||||
|
||||
describe("runnerd provider runtime wiring", () => {
|
||||
it("keeps remote warm turns open while retaining verified close semantics", async () => {
|
||||
const current = scopedNativeExecution("remote-warm-settlement");
|
||||
current.session = {
|
||||
...current.session,
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 },
|
||||
};
|
||||
state.execute.mockResolvedValueOnce({
|
||||
result: { summary: "completed" },
|
||||
terminal: { runTerminalState: "succeeded" },
|
||||
turnId: "turn",
|
||||
normalizedSessionId: current.session.normalizedSessionId,
|
||||
providerSessionId: "retained-provider-thread",
|
||||
driverKind: "codex_app_server",
|
||||
driverVersion: "1",
|
||||
nativeEventCount: 1,
|
||||
highestContiguousSourceSeq: 1,
|
||||
});
|
||||
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(current),
|
||||
execution: current,
|
||||
runnerInstanceId: "runner-remote-warm-settlement",
|
||||
useRunnerd: true,
|
||||
runnerExecutionTarget: {
|
||||
kind: "remote", transport: "sandbox", remoteCwd: "/workspace",
|
||||
environmentId: "environment", leaseId: "lease",
|
||||
runner: { execute: vi.fn() },
|
||||
} as never,
|
||||
});
|
||||
|
||||
expect(state.execute).toHaveBeenLastCalledWith(expect.objectContaining({
|
||||
keepSessionOpen: true,
|
||||
requireSessionCloseBeforeReturn: true,
|
||||
}));
|
||||
});
|
||||
|
||||
function scopedNativeExecution(suffix: string): NativeExecutionInputV1 {
|
||||
return {
|
||||
...execution,
|
||||
|
||||
@@ -161,6 +161,8 @@ import {
|
||||
REMOTE_RECOVERY_MARKER_READ_SCRIPT, REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT,
|
||||
RemoteRunnerRecoveryError, attachRemoteRunnerRecoveryDiagnostic, readRemoteRunnerRecoveryError, remoteRecoveryProbeReason,
|
||||
} from "./remote-runner-recovery-diagnostics.js";
|
||||
import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js";
|
||||
import { inspectRemoteWarmSessionRecovery, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js";
|
||||
import { persistActivity, publishActivity } from "../activity-log.js";
|
||||
import { commitNativeStatusDecision } from "./status-decision-committer.js";
|
||||
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
|
||||
@@ -6023,6 +6025,7 @@ function persistWarmNativeCheckpoint(
|
||||
execution: NativeExecutionInput,
|
||||
configDigest: string,
|
||||
snapshot: PersistedNativeSession,
|
||||
processConfigDigest?: string,
|
||||
): void {
|
||||
const path = nativeSessionCheckpointPath(execution);
|
||||
const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`;
|
||||
@@ -6031,6 +6034,7 @@ function persistWarmNativeCheckpoint(
|
||||
JSON.stringify({
|
||||
schema: "paperclip.native-session-supervisor.v1",
|
||||
configDigest,
|
||||
processConfigDigest,
|
||||
updatedAt: new Date().toISOString(),
|
||||
snapshot,
|
||||
}),
|
||||
@@ -6040,6 +6044,16 @@ function persistWarmNativeCheckpoint(
|
||||
chmodSync(path, 0o600);
|
||||
}
|
||||
|
||||
function warmNativeProcessConfigDigest(input: Parameters<typeof executePaperclipNativeSession>[0]): string | undefined {
|
||||
if (input.managedGitHub || input.runnerEnvironment?.PAPERCLIP_GITHUB_BROKER_TOKEN || input.instructionWorkingCopy) return undefined;
|
||||
return nativeSha256({
|
||||
environment: configuredEnvironment(input.runnerEnvironment),
|
||||
managedAiCredentialIdentity: input.managedAiCredentialIdentity ?? null,
|
||||
githubAuthenticationMode: input.runnerEnvironment?.PAPERCLIP_GITHUB_AUTH_MODE ?? null,
|
||||
networkAccess: input.runnerEnvironment?.PAPERCLIP_RUNNER_NETWORK_ACCESS === "enabled",
|
||||
});
|
||||
}
|
||||
|
||||
function loadWarmNativeCheckpoint(
|
||||
execution: NativeExecutionInput,
|
||||
configDigest: string,
|
||||
@@ -7625,6 +7639,15 @@ export async function executePaperclipNativeSession(input: {
|
||||
return await executePaperclipNativeSessionWithinScope(preparedInput);
|
||||
} catch (error) {
|
||||
executionFailure = error;
|
||||
const remoteRunnerRecovery = readRemoteRunnerRecoveryError(error);
|
||||
if (remoteRunnerRecovery) {
|
||||
await input.onEvent?.({
|
||||
eventType: "native.remote_runner_recovery_failed",
|
||||
level: "error",
|
||||
message: "Could not verify the existing remote runner.",
|
||||
payload: { remoteRunnerRecovery },
|
||||
}).catch(() => undefined);
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) {
|
||||
@@ -7744,16 +7767,39 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs });
|
||||
}
|
||||
let retainedTransition: VerifiedWarmTransitionBinding | undefined;
|
||||
let remoteWarmRecovery: RemoteWarmSessionRecovery | undefined;
|
||||
if (input.useRunnerd && !input.restartRecovery && !input.backend &&
|
||||
input.runnerExecutionTarget?.kind === "remote" &&
|
||||
input.execution.session.lifecyclePolicy.mode === "warm" &&
|
||||
!idleWarmNativeSessionOwnerRunId(input)) {
|
||||
const root = scopedRunnerdStateRoot(input.execution);
|
||||
const identity = readRunnerdDurableIdentity(root);
|
||||
if (durableIdentityMatchesSession(identity, input.execution) && identity.runId !== input.execution.binding.runId &&
|
||||
runnerdAuthorityLifecycleWithVerifiedBackup({ root, identity, execution: input.execution, allowVerifiedBackup: true }) !== "suspended") {
|
||||
remoteWarmRecovery = await inspectRemoteWarmSessionRecovery({
|
||||
db: input.db, execution: input.execution, target: input.runnerExecutionTarget, identity,
|
||||
readControlState: () => record(JSON.parse(readBoundedNativeFile(
|
||||
resolve(root, "control-plane", "control-plane-state.json"), NATIVE_CONTROL_PLANE_STATE_MAX_BYTES,
|
||||
"runner_durable_identity_too_large").toString("utf8"))),
|
||||
matchesExecution: previous => nativeSessionScopeKey(previous) === nativeSessionScopeKey(input.execution) &&
|
||||
nativeSessionConfigDigest(previous, "remote") === nativeSessionConfigDigest(input.execution, "remote"),
|
||||
loadCheckpoint: previous => loadWarmNativeCheckpoint(previous, nativeSessionConfigDigest(previous, "remote"), "remote"),
|
||||
canReuseProcess: previous => {
|
||||
const expected = warmNativeProcessConfigDigest(input);
|
||||
if (!expected) return false;
|
||||
const envelope = record(JSON.parse(readBoundedNativeFile(nativeSessionCheckpointPath(previous),
|
||||
NATIVE_WARM_CHECKPOINT_MAX_BYTES, "native_session_supervisor_checkpoint_too_large").toString("utf8")));
|
||||
return envelope.processConfigDigest === expected;
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
if (input.useRunnerd) {
|
||||
retainedTransition = await migrateRunnerdStateRootForExecution({
|
||||
db: input.db,
|
||||
execution: input.execution,
|
||||
allowVerifiedBackup: input.runnerExecutionTarget?.kind === "remote",
|
||||
// A retained warm runner is deliberately still ready rather than
|
||||
// suspended. Only the exact idle in-process owner may rotate that
|
||||
// prior-run authority; after a hard restart the map is empty and the
|
||||
// durable-state verifier continues to require a suspended runner.
|
||||
retainedWarmOwnerRunId: idleWarmNativeSessionOwnerRunId(input),
|
||||
retainedWarmOwnerRunId: idleWarmNativeSessionOwnerRunId(input) ?? remoteWarmRecovery?.identity.runId,
|
||||
allowLocalRecovery: input.runnerExecutionTarget?.kind !== "remote",
|
||||
onLog: input.onLog,
|
||||
restartRecovery: input.restartRecovery,
|
||||
@@ -8643,7 +8689,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
input.db,
|
||||
input.execution.binding,
|
||||
));
|
||||
const runnerdBackend =
|
||||
let remoteWarmRecoveryPending = remoteWarmRecovery !== undefined;
|
||||
let runnerdBackend =
|
||||
input.useRunnerd && input.backend === undefined
|
||||
? await timeController("backend_setup", () => createRunnerdBackend({
|
||||
...input,
|
||||
@@ -8653,10 +8700,34 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
execution: input.execution,
|
||||
runnerInstanceId: effectiveRunnerInstanceId,
|
||||
durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId,
|
||||
remoteWarmRecovery: remoteWarmRecovery ? {
|
||||
...remoteWarmRecovery,
|
||||
revalidate: async () => {
|
||||
if (remoteWarmRecoveryPending) await remoteWarmRecovery.revalidate(leaseOwner);
|
||||
},
|
||||
} : undefined,
|
||||
trace,
|
||||
toolTrace,
|
||||
}))
|
||||
: null;
|
||||
if (remoteWarmRecovery) {
|
||||
await remoteWarmRecovery.revalidate(leaseOwner);
|
||||
const recovered = await runnerdBackend?.recoverSession?.(remoteWarmRecovery.checkpoint, {
|
||||
signal: new AbortController().signal,
|
||||
});
|
||||
if (!recovered?.recovered || !recovered.session) throw new Error("runner_state_identity_mismatch: remote_warm_session_unproven");
|
||||
if (remoteWarmRecovery.reuseProcess) {
|
||||
existingWarmSession = recovered.session;
|
||||
persistedWarmSession = null;
|
||||
} else {
|
||||
await recovered.session.close({ reason: "recovered warm session requires current process configuration" });
|
||||
persistWarmNativeCheckpoint(remoteWarmRecovery.previousExecution,
|
||||
nativeSessionConfigDigest(remoteWarmRecovery.previousExecution, "remote"), await recovered.session.snapshot());
|
||||
persistedWarmSession = loadWarmNativeCheckpoint(input.execution, warmConfigDigest!, "remote");
|
||||
runnerdBackend = await createRunnerdBackend({ ...input, runnerInstanceId: effectiveRunnerInstanceId,
|
||||
durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, toolTrace });
|
||||
}
|
||||
}
|
||||
const remoteCleanupLease = input.runnerExecutionTarget?.kind === "remote" &&
|
||||
input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.leaseId
|
||||
? await input.db.select({ provider: environmentLeases.provider, providerLeaseId: environmentLeases.providerLeaseId })
|
||||
@@ -8759,8 +8830,6 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
keepSessionOpen: warmSessionId !== null,
|
||||
sessionGoalControl: input.sessionGoalControl,
|
||||
resumeSessionGoalHeartbeat: input.resumeSessionGoalHeartbeat,
|
||||
// Every durable runner must finish its bounded suspension before
|
||||
// the next run verifies and rotates the saved authority.
|
||||
requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined,
|
||||
onSessionClosed: input.instructionWorkingCopy?.collectStopped,
|
||||
onCheckpoint: async (snapshot) => {
|
||||
@@ -8774,6 +8843,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
input.execution,
|
||||
warmConfigDigest,
|
||||
snapshot,
|
||||
warmNativeProcessConfigDigest(input),
|
||||
);
|
||||
}
|
||||
if (
|
||||
@@ -8824,6 +8894,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
);
|
||||
},
|
||||
onSession: async (session) => {
|
||||
if (session && session === existingWarmSession) remoteWarmRecoveryPending = false;
|
||||
if (session && input.managedAiCredentialHome) {
|
||||
managedCredentialSession = runnerdBackend?.bindManagedSession(session) ?? session;
|
||||
}
|
||||
@@ -10609,25 +10680,35 @@ export async function syncRemoteRunnerDirectoryOut(input: {
|
||||
export async function readRemoteRunnerState(input: {
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
stateDirectory: string;
|
||||
recoveryDeadline?: number;
|
||||
}): Promise<Record<string, unknown>> {
|
||||
const statePath = posix.join(input.stateDirectory, "runner-state.json");
|
||||
const escapedPath = statePath.replaceAll("'", "'\\''");
|
||||
const result = await input.runner.execute({
|
||||
const request = {
|
||||
command: "sh",
|
||||
args: ["-c", `test -f '${escapedPath}' && base64 < '${escapedPath}'`],
|
||||
bypassSession: true,
|
||||
timeoutMs: 10_000,
|
||||
});
|
||||
};
|
||||
const result = input.recoveryDeadline === undefined
|
||||
? await input.runner.execute(request)
|
||||
: await readRemoteRunnerRecoveryProbe({
|
||||
runner: input.runner, request, stage: "runner_state",
|
||||
deadline: input.recoveryDeadline, readOnly: true,
|
||||
});
|
||||
if (result.exitCode !== 0 || result.timedOut) {
|
||||
throw new Error(
|
||||
const error = new Error(
|
||||
`runner_remote_state_unavailable: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`,
|
||||
);
|
||||
throw input.recoveryDeadline === undefined ? error : attachRemoteRunnerRecoveryDiagnostic(error,
|
||||
{ stage: "runner_state", reason: result.timedOut ? "probe_timeout" : "state_read_failed" });
|
||||
}
|
||||
try {
|
||||
return record(JSON.parse(Buffer.from(result.stdout.replace(/\s+/g, ""), "base64").toString("utf8")));
|
||||
} catch (error) {
|
||||
throw input.recoveryDeadline === undefined ? error : attachRemoteRunnerRecoveryDiagnostic(error,
|
||||
{ stage: "runner_state", reason: "state_invalid" });
|
||||
}
|
||||
return record(
|
||||
JSON.parse(
|
||||
Buffer.from(result.stdout.replace(/\s+/g, ""), "base64").toString("utf8"),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
async function readRemoteRunnerProviderState(input: {
|
||||
@@ -10700,14 +10781,15 @@ export function parseRemoteRunnerProcessIdentity(
|
||||
|
||||
/** Verify remote ownership before exposing the transport's authenticated adoption path. */
|
||||
export async function verifyRemoteRunnerReattachment(input: {
|
||||
claim: Extract<
|
||||
claim: Pick<Extract<
|
||||
NativeRestartRecoveryClaim,
|
||||
{ kind: "reattach_remote_runner" }
|
||||
>;
|
||||
>, "runId" | "remote">;
|
||||
target: AdapterExecutionTarget | null | undefined;
|
||||
identity: Record<string, unknown>;
|
||||
runId: string;
|
||||
normalizedSessionId: string;
|
||||
requireIdle?: boolean;
|
||||
}) {
|
||||
const { claim, target, identity } = input;
|
||||
if (
|
||||
@@ -10725,6 +10807,7 @@ export async function verifyRemoteRunnerReattachment(input: {
|
||||
throw new Error("native_remote_recovery_lease_mismatch");
|
||||
}
|
||||
const runner = target.runner;
|
||||
const probeDeadline = Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS;
|
||||
const stateDirectory = posix.join(
|
||||
target.remoteCwd,
|
||||
".paperclip-runtime",
|
||||
@@ -10733,7 +10816,11 @@ export async function verifyRemoteRunnerReattachment(input: {
|
||||
createHash("sha256").update(input.normalizedSessionId).digest("hex"),
|
||||
"runner",
|
||||
);
|
||||
const runnerState = await readRemoteRunnerState({ runner, stateDirectory });
|
||||
const runnerState = await readRemoteRunnerState({ runner, stateDirectory, recoveryDeadline: probeDeadline });
|
||||
if (input.requireIdle && (runnerState.lifecycle !== "ready" ||
|
||||
!Array.isArray(runnerState.outbox) || runnerState.outbox.length !== 0 || runnerState.pendingTerminalDelivery !== null)) {
|
||||
throw new Error("runner_state_identity_mismatch: remote_warm_session_not_idle");
|
||||
}
|
||||
for (const field of [
|
||||
"runId",
|
||||
"normalizedSessionId",
|
||||
@@ -10751,17 +10838,19 @@ export async function verifyRemoteRunnerReattachment(input: {
|
||||
}
|
||||
}
|
||||
const identityPath = posix.join(stateDirectory, "runner-process.identity");
|
||||
const marker = await runner.execute({
|
||||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
REMOTE_RECOVERY_MARKER_READ_SCRIPT,
|
||||
"paperclip-runner-recovery-identity",
|
||||
identityPath,
|
||||
],
|
||||
bypassSession: true,
|
||||
timeoutMs: 10_000,
|
||||
}).catch(error => { throw attachRemoteRunnerRecoveryDiagnostic(error, { stage: "identity_marker", reason: "probe_transport_failed" }); });
|
||||
const marker = await readRemoteRunnerRecoveryProbe({
|
||||
runner, stage: "identity_marker", deadline: probeDeadline, readOnly: true,
|
||||
request: {
|
||||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
REMOTE_RECOVERY_MARKER_READ_SCRIPT,
|
||||
"paperclip-runner-recovery-identity",
|
||||
identityPath,
|
||||
],
|
||||
bypassSession: true,
|
||||
},
|
||||
});
|
||||
const nonce = marker.stdout.split("\n")[0] ?? "";
|
||||
const runnerInstanceId = identity.runnerInstanceId as string;
|
||||
const processIdentity =
|
||||
@@ -10778,28 +10867,30 @@ export async function verifyRemoteRunnerReattachment(input: {
|
||||
if (!processIdentity) throw new RemoteRunnerRecoveryError({ stage: "identity_marker", reason: "marker_invalid" });
|
||||
if (!processIdentity.processStartFingerprint) throw new RemoteRunnerRecoveryError({ stage: "identity_marker", reason: "fingerprint_missing" });
|
||||
const expectedStartFingerprint = processIdentity.processStartFingerprint;
|
||||
const check = async (signal?: NodeJS.Signals) => {
|
||||
const result = await runner.execute({
|
||||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
`${REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT}; if ${REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT}; then :; else exit 46; fi; test "$process_fingerprint" = "$5" || exit 47` +
|
||||
(signal ? '; kill -"$6" "$pid"' : ""),
|
||||
"paperclip-runner-recovery-check",
|
||||
identityPath,
|
||||
nonce,
|
||||
runnerInstanceId,
|
||||
String(processIdentity.pid),
|
||||
expectedStartFingerprint,
|
||||
...(signal ? [signal.slice(3)] : []),
|
||||
],
|
||||
bypassSession: true,
|
||||
timeoutMs: 10_000,
|
||||
}).catch(error => { throw attachRemoteRunnerRecoveryDiagnostic(error, { stage: "process_verification", reason: "probe_transport_failed" }); });
|
||||
const check = async (signal?: NodeJS.Signals, deadline = Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS) => {
|
||||
const result = await readRemoteRunnerRecoveryProbe({
|
||||
runner, stage: "process_verification", deadline, readOnly: !signal,
|
||||
request: {
|
||||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
`${REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT}; if ${REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT}; then :; else exit 46; fi; test "$process_fingerprint" = "$5" || exit 47` +
|
||||
(signal ? '; kill -"$6" "$pid"' : ""),
|
||||
"paperclip-runner-recovery-check",
|
||||
identityPath,
|
||||
nonce,
|
||||
runnerInstanceId,
|
||||
String(processIdentity.pid),
|
||||
expectedStartFingerprint,
|
||||
...(signal ? [signal.slice(3)] : []),
|
||||
],
|
||||
bypassSession: true,
|
||||
},
|
||||
});
|
||||
return { alive: result.exitCode === 0 && !result.timedOut,
|
||||
reason: remoteRecoveryProbeReason("process_verification", result.exitCode, result.timedOut) };
|
||||
};
|
||||
const verified = await check();
|
||||
const verified = await check(undefined, probeDeadline);
|
||||
if (!verified.alive) throw new RemoteRunnerRecoveryError({ stage: "process_verification", reason: verified.reason });
|
||||
// These checks authorize an attempt to reconnect, not the task itself. The
|
||||
// transport still requires authentication against the exact durable PRP key.
|
||||
@@ -11151,6 +11242,7 @@ export async function createRunnerdBackend(input: {
|
||||
chatAttachmentReadScope?: NativeChatAttachmentReadScope;
|
||||
restartRecovery?: NativeRestartRecoveryClaim;
|
||||
durableEnvironmentLeaseId?: string;
|
||||
remoteWarmRecovery?: Omit<RemoteWarmSessionRecovery, "revalidate"> & { revalidate: () => Promise<void> };
|
||||
onSpawn?: (meta: {
|
||||
pid: number;
|
||||
processGroupId: number | null;
|
||||
@@ -11744,7 +11836,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
)
|
||||
return;
|
||||
selectedRemoteMode = requiredMode;
|
||||
if (input.restartRecovery?.kind === "reattach_remote_runner") {
|
||||
if (input.restartRecovery?.kind === "reattach_remote_runner" || input.remoteWarmRecovery) {
|
||||
await verifyRemoteRunner(requiredMode);
|
||||
if (requiresRemoteProviderPack && stagedRemoteProviderPackRoot) {
|
||||
await verifyRemoteProviderPack(stagedRemoteProviderPackRoot);
|
||||
@@ -12784,7 +12876,17 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
? input.restartRecovery.process
|
||||
: null;
|
||||
const adoptedProcess =
|
||||
input.restartRecovery?.kind === "reattach_remote_runner"
|
||||
input.remoteWarmRecovery
|
||||
? await (async () => {
|
||||
await input.remoteWarmRecovery!.revalidate();
|
||||
return verifyRemoteRunnerReattachment({
|
||||
claim: { runId: input.remoteWarmRecovery!.identity.runId, remote: input.remoteWarmRecovery!.remote },
|
||||
target, identity: input.remoteWarmRecovery!.identity,
|
||||
runId: input.remoteWarmRecovery!.identity.runId,
|
||||
normalizedSessionId: nativeSessionKey(input.execution), requireIdle: true,
|
||||
});
|
||||
})()
|
||||
: input.restartRecovery?.kind === "reattach_remote_runner"
|
||||
? await verifyRemoteRunnerReattachment({
|
||||
claim: input.restartRecovery,
|
||||
target,
|
||||
@@ -12988,7 +13090,14 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
input.execution.interactionResponses.length > 0
|
||||
? "allow_replacement_after_governed_wait"
|
||||
: undefined),
|
||||
prpIdentity: {
|
||||
prpIdentity: input.remoteWarmRecovery ? {
|
||||
runnerInstanceId: input.remoteWarmRecovery.identity.runnerInstanceId,
|
||||
environmentLeaseId: input.remoteWarmRecovery.identity.environmentLeaseId,
|
||||
runId: input.remoteWarmRecovery.identity.runId,
|
||||
normalizedSessionId: input.remoteWarmRecovery.identity.normalizedSessionId,
|
||||
turnId: input.remoteWarmRecovery.identity.turnId,
|
||||
itemId: input.remoteWarmRecovery.identity.itemId,
|
||||
} : {
|
||||
runnerInstanceId: effectiveRunnerInstanceId,
|
||||
environmentLeaseId: effectiveEnvironmentLeaseId,
|
||||
runId: input.execution.binding.runId,
|
||||
@@ -13050,7 +13159,9 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
adoptExistingRunner: adoptedProcess,
|
||||
},
|
||||
target,
|
||||
binding: input.execution.binding,
|
||||
binding: input.remoteWarmRecovery
|
||||
? { ...input.execution.binding, runId: input.remoteWarmRecovery.identity.runId }
|
||||
: input.execution.binding,
|
||||
runnerIngressAuthorized: input.runnerIngressAuthorized,
|
||||
runnerPublicUrl: input.runnerPublicUrl,
|
||||
runnerCaBundlePath: input.runnerCaBundlePath,
|
||||
@@ -13066,6 +13177,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
return targetPath;
|
||||
},
|
||||
beforeConnect: async () => {
|
||||
await input.remoteWarmRecovery?.revalidate();
|
||||
if (retainedTransition && recoveryPending) {
|
||||
const current = await verifyWarmTransitionRestart(input);
|
||||
if (current.stateFingerprint !== retainedTransition.stateFingerprint) {
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
export const REMOTE_RUNNER_RECOVERY_ERROR_CODE = "runner_remote_process_identity_unavailable";
|
||||
|
||||
const stages = ["identity_marker", "process_verification"] as const;
|
||||
const stages = ["runner_state", "identity_marker", "process_verification"] as const;
|
||||
const reasons = [
|
||||
"marker_missing", "marker_unsafe", "marker_read_failed", "marker_invalid",
|
||||
"fingerprint_missing", "process_not_running", "process_identity_changed",
|
||||
"fingerprint_unavailable", "process_generation_changed", "probe_timeout",
|
||||
"probe_transport_failed", "probe_failed",
|
||||
"state_read_failed", "state_invalid",
|
||||
] as const;
|
||||
|
||||
export interface RemoteRunnerRecoveryDiagnostic {
|
||||
@@ -78,7 +79,7 @@ export function remoteRecoveryProbeReason(stage: RemoteRunnerRecoveryDiagnostic[
|
||||
if (exitCode === 31) return "marker_missing";
|
||||
if (exitCode === 32) return "marker_unsafe";
|
||||
if (exitCode === 33) return "marker_read_failed";
|
||||
} else {
|
||||
} else if (stage === "process_verification") {
|
||||
if (exitCode === 41) return "marker_missing";
|
||||
if (exitCode === 42) return "marker_unsafe";
|
||||
if (exitCode === 43) return "marker_read_failed";
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import { readRemoteRunnerRecoveryError } from "./remote-runner-recovery-diagnostics.js";
|
||||
import { readRemoteRunnerRecoveryProbe } from "./remote-runner-recovery-probe.js";
|
||||
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
const timeout = { exitCode: null, signal: null, timedOut: true, stdout: "", stderr: "", pid: null, startedAt: "2026-10-15T00:00:00Z" };
|
||||
|
||||
it("shares the remaining deadline across timed-out reads", async () => {
|
||||
vi.useFakeTimers();
|
||||
const execute = vi.fn(async (request) => {
|
||||
await new Promise(resolve => setTimeout(resolve, request.timeoutMs));
|
||||
return timeout;
|
||||
});
|
||||
const result = readRemoteRunnerRecoveryProbe({
|
||||
runner: { execute } as CommandManagedRuntimeRunner,
|
||||
request: { command: "sh" }, stage: "identity_marker",
|
||||
readOnly: true, deadline: Date.now() + 15_000,
|
||||
}).catch(readRemoteRunnerRecoveryError);
|
||||
await vi.runAllTimersAsync();
|
||||
expect(await result).toEqual({ stage: "identity_marker", reason: "probe_timeout" });
|
||||
expect(execute.mock.calls.map(([request]) => request.timeoutMs)).toEqual([10_000, 4_750]);
|
||||
});
|
||||
|
||||
it("never retries a signal when its acknowledgement is lost", async () => {
|
||||
const execute = vi.fn().mockRejectedValue(new Error("connection lost"));
|
||||
const result = await readRemoteRunnerRecoveryProbe({
|
||||
runner: { execute }, request: { command: "sh" }, stage: "process_verification",
|
||||
readOnly: false, deadline: Date.now() + 30_000,
|
||||
}).catch(readRemoteRunnerRecoveryError);
|
||||
expect(result).toEqual({ stage: "process_verification", reason: "probe_transport_failed" });
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("returns a definitive ownership conflict without retrying it", async () => {
|
||||
const conflict = { ...timeout, timedOut: false, exitCode: 47 };
|
||||
const execute = vi.fn().mockResolvedValue(conflict);
|
||||
expect(await readRemoteRunnerRecoveryProbe({
|
||||
runner: { execute }, request: { command: "sh" }, stage: "process_verification",
|
||||
readOnly: true, deadline: Date.now() + 30_000,
|
||||
})).toBe(conflict);
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
@@ -0,0 +1,39 @@
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import {
|
||||
attachRemoteRunnerRecoveryDiagnostic,
|
||||
RemoteRunnerRecoveryError,
|
||||
type RemoteRunnerRecoveryDiagnostic,
|
||||
} from "./remote-runner-recovery-diagnostics.js";
|
||||
|
||||
export const REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS = 30_000;
|
||||
|
||||
export async function readRemoteRunnerRecoveryProbe(input: {
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
request: Parameters<CommandManagedRuntimeRunner["execute"]>[0];
|
||||
stage: RemoteRunnerRecoveryDiagnostic["stage"];
|
||||
deadline: number;
|
||||
readOnly: boolean;
|
||||
}) {
|
||||
let lastError: unknown = new RemoteRunnerRecoveryError({
|
||||
stage: input.stage, reason: "probe_timeout",
|
||||
});
|
||||
for (let attempt = 0; attempt < (input.readOnly ? 3 : 1); attempt++) {
|
||||
const remaining = input.deadline - Date.now();
|
||||
if (remaining <= 0) break;
|
||||
try {
|
||||
const result = await input.runner.execute({
|
||||
...input.request,
|
||||
timeoutMs: Math.min(10_000, remaining),
|
||||
});
|
||||
if (!result.timedOut) return result;
|
||||
lastError = new RemoteRunnerRecoveryError({ stage: input.stage, reason: "probe_timeout" });
|
||||
} catch (error) {
|
||||
lastError = attachRemoteRunnerRecoveryDiagnostic(error, {
|
||||
stage: input.stage, reason: "probe_transport_failed",
|
||||
});
|
||||
}
|
||||
if (!input.readOnly || attempt === 2 || input.deadline - Date.now() <= 250) break;
|
||||
await new Promise<void>(resolve => setTimeout(resolve, 250));
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { currentNativeControllerIdentity, evaluateNativeControllerTakeover } from "./native-restart-recovery.js";
|
||||
|
||||
export type RemoteWarmSessionRecovery = {
|
||||
identity: Record<string, string>;
|
||||
checkpoint: PersistedNativeSession;
|
||||
previousExecution: NativeExecutionInput;
|
||||
reuseProcess: boolean;
|
||||
remote: { providerLeaseId: string; remoteCwd: string };
|
||||
revalidate: (leaseOwner: string) => Promise<void>;
|
||||
};
|
||||
|
||||
export async function inspectRemoteWarmSessionRecovery(input: {
|
||||
db: Db;
|
||||
execution: NativeExecutionInput;
|
||||
target: Extract<AdapterExecutionTarget, { kind: "remote" }>;
|
||||
identity: Record<string, unknown>;
|
||||
readControlState: () => Record<string, unknown>;
|
||||
matchesExecution: (previous: NativeExecutionInput) => boolean;
|
||||
loadCheckpoint: (previous: NativeExecutionInput) => PersistedNativeSession | null;
|
||||
canReuseProcess: (previous: NativeExecutionInput) => boolean;
|
||||
}): Promise<RemoteWarmSessionRecovery> {
|
||||
const deny = (): never => { throw new Error("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"); };
|
||||
const { binding } = input.execution;
|
||||
const { target } = input;
|
||||
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();
|
||||
identity[key] = value;
|
||||
}
|
||||
if (target.transport !== "sandbox" || !target.leaseId || !target.environmentId ||
|
||||
!target.providerKey || !target.sandboxLeaseAcquisition || target.sandboxLeaseAcquisition.outcome !== "resumed" ||
|
||||
identity.runId === binding.runId) return deny();
|
||||
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.phase !== "committed" ||
|
||||
!prior.owner.resultId || prior.owner.leaseOwner || !prior.owner.controllerPid ||
|
||||
!prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny();
|
||||
const previous = parseNativeExecutionInput(prior.run.runnerProfileJson?.nativeExecutionInput);
|
||||
if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny();
|
||||
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();
|
||||
const control = input.readControlState();
|
||||
const commands = control.commands;
|
||||
const events = control.committedEvents;
|
||||
const terminal = Array.isArray(events) ? events.at(-1) : undefined;
|
||||
if (!Array.isArray(commands) || commands.some(command => command.status !== "completed") ||
|
||||
terminal?.eventType !== "run.terminal" || terminal?.envelope?.runId !== identity.runId ||
|
||||
terminal?.envelope?.normalizedSessionId !== identity.normalizedSessionId ||
|
||||
Object.entries(identity).some(([key, value]) => (control.identity as Record<string, unknown>)?.[key] !== value)) return deny();
|
||||
const assertPriorStopped = async () => {
|
||||
const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), requireStoppedController: true });
|
||||
if (!stopped.allowed) return deny();
|
||||
};
|
||||
await assertPriorStopped();
|
||||
const priorLeases = await input.db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, binding.companyId), eq(environmentLeases.heartbeatRunId, identity.runId),
|
||||
eq(environmentLeases.issueId, binding.issueId), eq(environmentLeases.environmentId, target.environmentId),
|
||||
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();
|
||||
return {
|
||||
identity, checkpoint, previousExecution: previous, reuseProcess: input.canReuseProcess(previous),
|
||||
remote: { providerLeaseId, remoteCwd: target.remoteCwd },
|
||||
revalidate: async leaseOwner => {
|
||||
await assertPriorStopped();
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
const [current] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations, issue: issues, agent: agents, company: companies })
|
||||
.from(heartbeatRuns).innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.innerJoin(issues, eq(issues.id, heartbeatRuns.nativeIssueId))
|
||||
.innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId)))
|
||||
.innerJoin(companies, eq(companies.id, heartbeatRuns.companyId))
|
||||
.where(and(eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId),
|
||||
eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId)));
|
||||
const [lease] = await input.db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.id, target.leaseId!), eq(environmentLeases.companyId, binding.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, binding.runId), eq(environmentLeases.issueId, binding.issueId),
|
||||
));
|
||||
const [priorOwner] = await input.db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.runId, identity.runId), eq(nativeRunFinalizations.companyId, binding.companyId),
|
||||
eq(nativeRunFinalizations.issueId, binding.issueId),
|
||||
));
|
||||
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" ||
|
||||
lease.environmentId !== target.environmentId || lease.provider !== target.providerKey ||
|
||||
lease.providerLeaseId !== providerLeaseId || lease.metadata?.remoteCwd !== target.remoteCwd) return deny();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -7,10 +7,11 @@ import type { ExecutionBlocker, Issue } from "@paperclipai/shared";
|
||||
import { WorkspaceFileRecovery } from "./WorkspaceFileRecovery";
|
||||
const resolve = vi.hoisted(() => vi.fn());
|
||||
vi.mock("../api/issues", () => ({ issuesApi: { resolveRecoveryAction: resolve } }));
|
||||
const blocker = { recoveryActionId: "held-source", runId: "original-run", cause: "legacy_execution_requires_reconciliation",
|
||||
const baseBlocker = { recoveryActionId: "held-source", runId: "original-run",
|
||||
workspaceRepairRequired: true, canRecordWorkspaceRepair: true, canRetry: false, canContinue: false, nextAction: "Recover the original files." } as ExecutionBlocker;
|
||||
|
||||
describe("workspace file recovery", () => {
|
||||
describe.each(["legacy_execution_requires_reconciliation", "native_continuation_requires_reconciliation"] as const)("workspace file recovery: %s", cause => {
|
||||
const blocker = { ...baseBlocker, cause };
|
||||
let root: Root, container: HTMLDivElement, client: QueryClient;
|
||||
const onRecorded = vi.fn();
|
||||
beforeEach(() => {
|
||||
@@ -37,7 +38,7 @@ describe("workspace file recovery", () => {
|
||||
await act(async () => Array.from(container.querySelectorAll("button")).find(b => b.textContent === "Record file recovery")!.click());
|
||||
await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); });
|
||||
}
|
||||
it.each(["in_progress", "in_review", "done", "cancelled"] as Issue["status"][])("records verified source files without changing the %s task", async issueStatus => {
|
||||
it.each(["blocked", "in_progress", "in_review", "done", "cancelled"] as Issue["status"][])("records verified source files without changing the %s task", async issueStatus => {
|
||||
await mount({ issueStatus });
|
||||
const button = Array.from(container.querySelectorAll("button")).find(b => b.textContent === "Record file recovery")!;
|
||||
expect(button.disabled).toBe(true); await enterEvidence(); expect(button.disabled).toBe(true);
|
||||
|
||||
@@ -30,7 +30,8 @@ export function WorkspaceFileRecovery({ issueId, issueStatus, blocker, canManage
|
||||
}),
|
||||
onSuccess: onRecorded,
|
||||
});
|
||||
if (!blocker.workspaceRepairRequired || !blocker.canRecordWorkspaceRepair || blocker.cause !== "legacy_execution_requires_reconciliation"
|
||||
if (!blocker.workspaceRepairRequired || !blocker.canRecordWorkspaceRepair ||
|
||||
!["legacy_execution_requires_reconciliation", "native_continuation_requires_reconciliation"].includes(blocker.cause)
|
||||
|| !blocker.recoveryActionId || !blocker.runId) return null;
|
||||
return <section aria-label="Record workspace file recovery" className="flex w-full flex-col gap-2">
|
||||
{record.isSuccess ? <p role="status">File recovery recorded. The task status is unchanged. Continue the task when it is ready.</p>
|
||||
|
||||
Reference in new issue
Block a user