diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 52dbcac821e..b64eaaa16f0 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -730,7 +730,16 @@ recovery path with the specific scan cause and an operator action. Generic stranded-work recovery must not grant another budget for these errors. This does not automatically replay historical generic `setup_failed` runs. -### ACP startup handshake bound +### Runner authentication and ACP startup handshake bounds + +The native Runner waits up to the ten-second connection setup budget for the +authenticated welcome. The current execution or reconnect deadline can shorten +that wait. Controller admission and durable lease persistence happen before the +welcome, so a short internal response limit must not discard an otherwise valid +bootstrap ticket during that work. A failed welcome still makes the one-use +ticket unsafe to replay. Authentication proofs, credential expiry, and identity +checks remain required. Steady-state reads return to their 250 ms polling bound +after the welcome. An adapter-backed live path also requires that the ACP startup handshake itself cannot hang forever. The engine bounds the handshake with a fixed startup deadline and a poll of the duplex control-channel disposition. Either condition ends the handshake and reports a closed, typed code, so the issue can reach a settled disposition instead of staying `in_progress` with no observable next action. diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index a6f49522fef..b72eb30ea42 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -175,6 +175,13 @@ provider restart. For that journey, interrupt a disposable deployment after workspace preparation and before its first executor claim, then confirm one provider turn starts on the original run through the normal admission gates. +The [Runner transport regressions](../../packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs) +exercise a real WebSocket welcome delayed beyond two seconds. Authentication +uses the ten-second connection setup budget, clipped by the current lifecycle +deadline. The cases also retain failed-welcome ticket fencing, identity checks, +and reconnect replay without repeated provider effects. These transport cases +do not establish hosted provider acceptance. + Active remote process supervision distinguishes a failed health read from a confirmed exit. The [executor tests](../../server/src/services/native-runtime/native-session-executor.test.ts) exercise timeout recovery with a real listening process, unchanged ownership, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs index 135a5feb962..ae2be51ddfb 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs @@ -26,7 +26,7 @@ use super::{ const SECURE_FRAME_SCHEMA: &str = "paperclip.runner.secure-frame.v1"; const AUTH_TIMEOUT: Duration = Duration::from_secs(2); -const WELCOME_TIMEOUT: Duration = Duration::from_secs(2); +const WELCOME_TIMEOUT: Duration = CONNECT_TOTAL_TIMEOUT; const RUNTIME_READ_TIMEOUT: Duration = Duration::from_millis(250); const CONNECT_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(3); const CONNECT_TOTAL_TIMEOUT: Duration = Duration::from_secs(10); @@ -1106,8 +1106,9 @@ impl AuthenticatedTransport { max_frame_bytes: config.max_frame_bytes, }; // Ticket validation and lease persistence happen before the server - // emits welcome. Keep that work bounded without assuming a loaded - // control-plane event loop can always respond within 250 ms. + // emits welcome. Give that durable admission the connection setup + // budget; a two-second wait can discard a consumed ticket under + // controller load. The execution/reconnect deadline still wins. let welcome_deadline = bounded_operation_deadline(connect_deadline, WELCOME_TIMEOUT)?; let welcome_timeout = welcome_deadline.saturating_duration_since(Instant::now()); if welcome_timeout.is_zero() { @@ -2358,6 +2359,115 @@ mod tests { assert!(validate_control_identity(&envelope, &state, Some(&connection)).is_err()); } + #[test] + fn bootstrap_welcome_waits_for_bounded_controller_admission() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut config = config(port); + config.max_runtime = Duration::from_secs(10); + let state = test_state(&config); + let server_config = config.clone(); + let server_state = state.clone(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + let expires = current_unix_ms().unwrap() + 60_000; + let mut secure = server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + // The authority has the client proof. Durable admission and lease + // persistence may take longer than the old two-second welcome wait. + thread::sleep(Duration::from_secs(3)); + send_secure( + &mut socket, + &mut secure, + &server_config, + &welcome( + &server_state, + "connection_1", + Some("lease-secret"), + expires, + 0, + vec![], + ), + ); + }); + let endpoint = RunnerTransportEndpoint::new(&config.connect_url, &config.run_id).unwrap(); + let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); + let result = AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + Instant::now() + config.max_runtime, + ); + server.join().unwrap(); + let (transport, greeting) = result.unwrap().unwrap(); + assert_eq!(greeting.lease.unwrap().expose().unwrap(), "lease-secret"); + assert_eq!( + transport.socket.read_timeout().unwrap(), + Some(RUNTIME_READ_TIMEOUT) + ); + } + + #[test] + fn bootstrap_welcome_retains_the_shorter_lifecycle_deadline() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let config = config(port); + let state = test_state(&config); + let server_config = config.clone(); + let server_state = state.clone(); + let (release, held) = mpsc::channel(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: current_unix_ms().unwrap() + 60_000, + revocation_epoch: 0, + }, + ); + held.recv_timeout(Duration::from_secs(2)).unwrap(); + }); + let endpoint = RunnerTransportEndpoint::new(&config.connect_url, &config.run_id).unwrap(); + let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); + let started = Instant::now(); + let result = AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + started + Duration::from_secs(1), + ); + release.send(()).unwrap(); + server.join().unwrap(); + let failure = match result { + Err(error) => error, + Ok(_) => panic!("welcome authenticated after the lifecycle deadline"), + }; + assert!(failure.bootstrap_maybe_consumed); + assert!(started.elapsed() < Duration::from_secs(2)); + assert!(failure.error.to_string().contains("deadline elapsed")); + } + #[test] fn authenticates_bootstrap_and_receives_bound_lease() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); diff --git a/packages/paperclip-runner/src/live/runnerd-welcome.integration.test.ts b/packages/paperclip-runner/src/live/runnerd-welcome.integration.test.ts new file mode 100644 index 00000000000..49d44895434 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-welcome.integration.test.ts @@ -0,0 +1,62 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import { createCapabilityRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js"; + +it("starts one provider after delayed bootstrap admission and retains it for a second turn", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-delayed-welcome-")); + const calls = join(root, "provider-calls.log"); + let admissionDelayed = false; + let admissionTimer: ReturnType | undefined; + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "provider.json"), "--call-log", calls, "--record-process-start", "--durable-turn-ids"], + stateDirectory: join(root, "state"), + controlPlaneRegistration: async authority => { + const attach = authority.attachWireConnection.bind(authority); + vi.spyOn(authority, "attachWireConnection").mockImplementation(wire => attach({ + sendJson: value => wire.sendJson(value), + close: code => wire.close(code), + onClose: listener => wire.onClose(listener), + onJson: listener => wire.onJson(value => { + if (!admissionDelayed && (value as { kind?: string }).kind === "auth_response") { + admissionDelayed = true; + admissionTimer = setTimeout(() => listener(value), 3_000); + } else listener(value); + }), + })); + await authority.start(); + return { connectUrl: authority.connectUrl, release: () => authority.stop() }; + }, + }); + const terminalTurns: unknown[] = []; + try { + const opened = await bundle.transport.request("thread/start", { cwd: root, dynamicTools: [] }); + expect(admissionDelayed).toBe(true); + for (const text of ["Complete the first turn.", "Continue the same session."]) { + await bundle.transport.request("turn/start", { input: [{ type: "text", text }] }); + for await (const event of bundle.transport.notifications()) { + if (event.method === "turn/completed") { + expect(event.params).toMatchObject({ threadId: opened.thread.id }); + terminalTurns.push(event.params); + break; + } + } + } + await bundle.transport.close(); + const methods = (await readFile(calls, "utf8")).trim().split("\n"); + expect(methods.filter(method => method === "process-start")).toHaveLength(1); + expect(methods.filter(method => method === "thread/start")).toHaveLength(1); + expect(methods.filter(method => method === "turn/start")).toHaveLength(2); + expect(terminalTurns).toHaveLength(2); + const runner = JSON.parse(await readFile(join(root, "state/runner/runner-state.json"), "utf8")); + expect(runner.lifecycle).toBe("suspended"); + expect(runner.outbox).toEqual([]); + } finally { + if (admissionTimer) clearTimeout(admissionTimer); + await bundle.transport.close(); + await rm(root, { recursive: true, force: true }); + } +}, 30_000);