From b6ec51df04e4ab9a97da9dddcef8a57c7f660ba3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 7 Aug 2026 23:37:33 +0000 Subject: [PATCH] feat(runner): add durable Phase 3 recovery transport Co-Authored-By: Paperclip --- .../devtools/browser/src/App.tsx | 106 +- .../devtools/browser/src/styles.css | 14 + .../devtools/browser/static-replay.spec.ts | 16 +- packages/paperclip-runner/package.json | 6 +- .../runner-core/src/bin/paperclip-runnerd.rs | 61 +- .../runner/crates/runner-core/src/lib.rs | 1 + .../runner/crates/runner-core/src/phase3.rs | 2243 +++++++++++++++++ .../scripts/phase2-browser-server.mjs | 42 +- .../scripts/phase2-browser-server.test.mjs | 25 + .../scripts/record-phase3-evidence.mjs | 49 + .../src/cli/phase3-recovery.ts | 105 + .../paperclip-runner/src/contracts/phase3.ts | 158 ++ packages/paperclip-runner/src/index.ts | 2 + .../src/mock-core/phase3-recovery.test.ts | 379 +++ .../src/mock-core/phase3-recovery.ts | 1255 +++++++++ 15 files changed, 4452 insertions(+), 10 deletions(-) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs create mode 100644 packages/paperclip-runner/scripts/record-phase3-evidence.mjs create mode 100644 packages/paperclip-runner/src/cli/phase3-recovery.ts create mode 100644 packages/paperclip-runner/src/contracts/phase3.ts create mode 100644 packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts create mode 100644 packages/paperclip-runner/src/mock-core/phase3-recovery.ts diff --git a/packages/paperclip-runner/devtools/browser/src/App.tsx b/packages/paperclip-runner/devtools/browser/src/App.tsx index 0ace32e779c..e15236e6645 100644 --- a/packages/paperclip-runner/devtools/browser/src/App.tsx +++ b/packages/paperclip-runner/devtools/browser/src/App.tsx @@ -7,6 +7,7 @@ import type { Phase2RunTrace, Phase2Scenario, } from "../../../src/contracts/phase2"; +import type { Phase3RunTrace } from "../../../src/contracts/phase3"; import { applyPhase2LiveEvent, createPhase2LiveSnapshot, @@ -54,7 +55,7 @@ interface BrowserFixture { source: string; } -type BrowserMode = "live" | "replay"; +type BrowserMode = "recovery" | "live" | "replay"; type LiveStatus = "idle" | "starting" | "running" | "terminal" | "error"; type Phase2StreamRecord = @@ -89,6 +90,94 @@ function humanizeProtocolLabel(value: string): string { return words.length === 0 ? words : `${words[0]?.toUpperCase()}${words.slice(1)}`; } +function RecoveryDiagnostics() { + const [trace, setTrace] = useState(null); + const [status, setStatus] = useState<"idle" | "running" | "complete" | "error">("idle"); + const [error, setError] = useState(null); + + async function runRecovery() { + setStatus("running"); + setError(null); + try { + const response = await fetch("/api/phase3/recovery", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ fault: "lost-ack" }), + }); + const result = (await response.json()) as Phase3RunTrace & { error?: string }; + if (!response.ok) throw new Error(result.error ?? "The recovery trace failed."); + setTrace(result); + setStatus("complete"); + } catch (cause) { + setStatus("error"); + setError(cause instanceof Error ? cause.message : String(cause)); + } + } + + return ( +
+ + + Break recovery on purpose + + Lose one ACK, drop the socket, restart the runner, and finish the same session. + + + + +
+ Recovery status + + {status} + +
+ {error ?
{error}
: null} +

+ Diagnostics show IDs and counters only. Bootstrap tickets and lease tokens are hidden. +

+ + + + + + Recovery state + Inspect connection, lease, outbox, ACK, replay, storage, drain, and revoke state. + + + {trace === null ? ( +

Run the trace to collect recovery diagnostics.

+ ) : ( + <> +
+
Connection
{trace.diagnostics.connection.state}
+
Lease
{trace.diagnostics.connection.leaseId ?? "closed"}
+
Outbox
{trace.diagnostics.outbox.events}
+
Last ACK
{trace.diagnostics.cursors.runnerAckedSourceSeq}
+
Replayed
{trace.diagnostics.recovery.replayDeliveries}
+
Duplicate commands
{trace.diagnostics.commands.duplicateDeliveries}
+
Storage
{trace.diagnostics.outbox.bytes} bytes
+
Peak storage
{trace.diagnostics.outbox.peakBytes} bytes
+
Drain or revoke
{trace.diagnostics.recovery.outcome}
+
Secrets
{trace.diagnostics.security.secretLeakCount} leaks
+
+
+ Same runner and session + {trace.assertions.stableIdentity ? "Preserved" : "Changed"} +
+
+ Secrets redacted + {trace.assertions.secretsRedacted ? "Yes" : "No"} +
+ + )} + + +
+ ); +} + function liveStatusTone(status: LiveStatus, snapshot: SessionSnapshot | null) { if (status === "error") { return "danger" as const; @@ -571,16 +660,23 @@ export function App() {
-

Paperclip Runner Protocol · Phase 2

-

Live local runner

+

Paperclip Runner Protocol · Phase 3

+

Durable recovery diagnostics

- Run the Rust supervisor and fake harness, then compare the live state with replay. + Break the outbound transport, recover the same session, and inspect durable state.

Standalone · no Paperclip core
- {mode === "live" ? : } + {mode === "recovery" ? : mode === "live" ? : }
); } diff --git a/packages/paperclip-runner/devtools/browser/src/styles.css b/packages/paperclip-runner/devtools/browser/src/styles.css index 80f8590be72..74d0e06f602 100644 --- a/packages/paperclip-runner/devtools/browser/src/styles.css +++ b/packages/paperclip-runner/devtools/browser/src/styles.css @@ -342,6 +342,20 @@ select { border-block: 0.0625rem solid var(--border); } +.recovery-grid { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: var(--space-4); + margin: 0; + padding-block: var(--space-5); + border-block: 0.0625rem solid var(--border); +} + +.diagnostic-note { + color: var(--muted-foreground); + font-size: var(--font-size-sm); +} + .snapshot-grid div { min-width: 0; } diff --git a/packages/paperclip-runner/devtools/browser/static-replay.spec.ts b/packages/paperclip-runner/devtools/browser/static-replay.spec.ts index 70ff85d28d9..24bd47ee4c3 100644 --- a/packages/paperclip-runner/devtools/browser/static-replay.spec.ts +++ b/packages/paperclip-runner/devtools/browser/static-replay.spec.ts @@ -3,7 +3,7 @@ import { expect, test } from "@playwright/test"; test("validates and renders the shared Phase 1 fixture", async ({ page }, testInfo) => { await page.goto("/"); await page.getByRole("button", { name: "Static replay" }).click(); - await expect(page.getByRole("heading", { name: "Live local runner" })).toBeVisible(); + await expect(page.getByRole("heading", { name: "Durable recovery diagnostics" })).toBeVisible(); await expect(page.getByTestId("terminal-badge")).toHaveText("Succeeded"); await expect(page.getByTestId("timeline").getByRole("listitem")).toHaveCount(9); await expect(page.getByTestId("timeline")).not.toContainText("workspace_preparing"); @@ -16,6 +16,20 @@ test("validates and renders the shared Phase 1 fixture", async ({ page }, testIn }); }); +test("shows Phase 3 recovery diagnostics without secrets", async ({ page }, testInfo) => { + await page.goto("/"); + await page.getByRole("button", { name: "Recovery", exact: true }).click(); + await page.getByRole("button", { name: "Run Phase 3 recovery" }).click(); + await expect(page.getByText("complete", { exact: true })).toBeVisible(); + await expect(page.getByTestId("phase3-diagnostics")).toContainText("Replayed"); + await expect(page.getByTestId("phase3-diagnostics")).toContainText("Duplicate commands"); + await expect(page.getByText("Preserved", { exact: true })).toBeVisible(); + await expect(page.getByText("Yes", { exact: true })).toBeVisible(); + await expect(page.locator("body")).not.toContainText("leaseToken"); + await expect(page.locator("body")).not.toContainText("bootstrapTicket"); + await page.screenshot({ path: testInfo.outputPath("phase-03-recovery-diagnostics.png"), fullPage: true }); +}); + test("shows duplicate and unsupported-version replay states", async ({ page }) => { await page.goto("/"); await page.getByRole("button", { name: "Static replay" }).click(); diff --git a/packages/paperclip-runner/package.json b/packages/paperclip-runner/package.json index 923de737918..28ecea0e690 100644 --- a/packages/paperclip-runner/package.json +++ b/packages/paperclip-runner/package.json @@ -44,9 +44,13 @@ "replay:phase1": "pnpm run build:typescript && node dist/cli/phase1-replay.js", "trace:phase2": "pnpm run build:typescript && pnpm run build:phase2-binaries && node dist/cli/phase2-mock-core.js", "record:phase2": "pnpm run trace:phase2 -- --quiet --output knowledge/evidence/phase-02-happy-path.json", + "trace:phase3": "pnpm run build:typescript && pnpm run build:phase2-binaries && node dist/cli/phase3-recovery.js", + "diagnose:phase3": "pnpm run trace:phase3", + "record:phase3": "pnpm run build:typescript && pnpm run build:phase2-binaries && node scripts/record-phase3-evidence.mjs", + "record:phase3:lost-ack": "pnpm run trace:phase3 -- --fault lost-ack --output knowledge/evidence/phase-03-lost-ack.json", "browser:dev": "pnpm run build:typescript && pnpm run build:phase2-binaries && vite --config vite.config.ts", "browser:preview": "vite preview --config vite.config.ts", - "verify": "pnpm run build && pnpm run typecheck && pnpm run check:phase1-goldens && pnpm run test && pnpm run check:browser-tokens && pnpm run test:browser && pnpm run check:forbidden-imports && pnpm run docs:validate && pnpm run check:phase0-parity && pnpm run check:phase1-parity && pnpm run trace:phase0 && pnpm run replay:phase1 && pnpm run trace:phase2 -- --quiet", + "verify": "pnpm run build && pnpm run typecheck && pnpm run check:phase1-goldens && pnpm run test && pnpm run check:browser-tokens && pnpm run test:browser && pnpm run check:forbidden-imports && pnpm run docs:validate && pnpm run check:phase0-parity && pnpm run check:phase1-parity && pnpm run trace:phase0 && pnpm run replay:phase1 && pnpm run trace:phase2 -- --quiet && pnpm run trace:phase3", "verify:rootless": "bash scripts/verify-rootless-linux.sh" }, "devDependencies": { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 04d61f2ecf7..f69993d6a9e 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -3,6 +3,9 @@ use std::process::ExitCode; use std::time::Duration; use paperclip_runner_core::phase2::{run_local_runner, Phase2Error, RunnerConfig}; +use paperclip_runner_core::phase3::{ + capture_bootstrap_ticket, run_durable_runner, BootstrapTicket, DurableRunnerConfig, +}; fn value(args: &[String], name: &str) -> Result { let index = args @@ -34,8 +37,54 @@ fn usize_value(args: &[String], name: &str, default: usize) -> Result Result<(), Phase2Error> { +fn duration_value(args: &[String], name: &str, default: u64) -> Result { + Ok(Duration::from_millis( + optional_u64(args, name)?.unwrap_or(default), + )) +} + +fn run_phase3( + args: &[String], + bootstrap_ticket: Option, +) -> Result<(), Phase2Error> { + let config = DurableRunnerConfig { + connect_url: value(args, "--connect-url")?, + state_dir: PathBuf::from(value(args, "--state-dir")?), + runner_instance_id: value(args, "--runner-id")?, + environment_lease_id: value(args, "--environment-lease-id")?, + run_id: value(args, "--run-id")?, + normalized_session_id: value(args, "--session-id")?, + turn_id: value(args, "--turn-id")?, + item_id: value(args, "--item-id")?, + runner_version: value(args, "--runner-version")?, + runner_digest: value(args, "--runner-digest")?, + fake_harness_path: args + .iter() + .any(|argument| argument == "--fake-harness") + .then(|| value(args, "--fake-harness").map(PathBuf::from)) + .transpose()?, + fake_harness_script_path: args + .iter() + .any(|argument| argument == "--fake-harness-script") + .then(|| value(args, "--fake-harness-script").map(PathBuf::from)) + .transpose()?, + max_outbox_bytes: usize_value(args, "--max-outbox-bytes", 64 * 1024)?, + p0_reserve_bytes: usize_value(args, "--p0-reserve-bytes", 32 * 1024)?, + max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?, + reconnect_delay: duration_value(args, "--reconnect-delay-ms", 25)?, + max_runtime: duration_value(args, "--max-runtime-ms", 15_000)?, + }; + let bootstrap_ticket = bootstrap_ticket + .ok_or_else(|| Phase2Error::invalid("runner bootstrap ticket is not available"))?; + run_durable_runner(config, bootstrap_ticket) + .map_err(|error| Phase2Error::invalid(error.to_string())) +} + +fn run(bootstrap_ticket: Option) -> Result<(), Phase2Error> { let args = std::env::args().skip(1).collect::>(); + if args.iter().any(|argument| argument == "--connect-url") { + return run_phase3(&args, bootstrap_ticket); + } run_local_runner(RunnerConfig { run_id: value(&args, "--run-id")?, normalized_session_id: value(&args, "--session-id")?, @@ -53,7 +102,15 @@ fn run() -> Result<(), Phase2Error> { } fn main() -> ExitCode { - match run() { + // Capture and remove the bootstrap capability before argument parsing or child work. + let bootstrap_ticket = match capture_bootstrap_ticket() { + Ok(ticket) => ticket, + Err(error) => { + eprintln!("paperclip-runnerd: {error}"); + return ExitCode::FAILURE; + } + }; + match run(bootstrap_ticket) { Ok(()) => ExitCode::SUCCESS, Err(error) => { eprintln!("paperclip-runnerd: {error}"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs index ff17e2ba528..c60f5ae5a53 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -2,6 +2,7 @@ pub mod phase1; pub mod phase2; +pub mod phase3; use std::error::Error; use std::fmt::{self, Display, Formatter}; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs b/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs new file mode 100644 index 00000000000..74871b3b9c1 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs @@ -0,0 +1,2243 @@ +use std::collections::BTreeMap; +use std::error::Error; +use std::fmt::{self, Display, Formatter}; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; +use std::path::{Path, PathBuf}; +use std::thread; +use std::time::{Duration, Instant}; + +#[cfg(unix)] +use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use crate::phase2::SupervisedProcess; + +const STATE_SCHEMA: &str = "paperclip.runner.phase3.state.v1"; +const PROTOCOL: &str = "paperclip.runner"; +const PROTOCOL_VERSION: u64 = 1; +const STATIC_WEBSOCKET_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const STATIC_WEBSOCKET_ACCEPT: &str = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="; +const MAX_HTTP_HEADER_BYTES: usize = 16 * 1024; +const MAX_DIAGNOSTICS: usize = 32; +const REVOKE_FLUSH_TIMEOUT: Duration = Duration::from_secs(2); +const TEMP_FILE_ATTEMPTS: usize = 16; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Phase3Error(String); + +impl Phase3Error { + pub fn invalid(message: impl Into) -> Self { + Self(message.into()) + } +} + +impl Display for Phase3Error { + fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.0) + } +} + +impl Error for Phase3Error {} + +struct SensitiveString(Vec); + +impl SensitiveString { + fn new(value: String) -> Self { + Self(value.into_bytes()) + } + + fn expose(&self) -> &str { + std::str::from_utf8(&self.0).expect("sensitive value started as valid UTF-8") + } + + fn clear(&mut self) { + self.0.fill(0); + self.0.clear(); + } +} + +impl Drop for SensitiveString { + fn drop(&mut self) { + self.clear(); + } +} + +pub struct BootstrapTicket(SensitiveString); + +impl BootstrapTicket { + fn expose(&self) -> &str { + self.0.expose() + } +} + +pub fn capture_bootstrap_ticket() -> Result, Phase3Error> { + let value = std::env::var_os("PAPERCLIP_RUNNER_BOOTSTRAP_TICKET"); + std::env::remove_var("PAPERCLIP_RUNNER_BOOTSTRAP_TICKET"); + value + .map(|value| { + value + .into_string() + .map(|value| BootstrapTicket(SensitiveString::new(value))) + .map_err(|_| Phase3Error::invalid("runner bootstrap ticket is not valid UTF-8")) + }) + .transpose() +} + +#[derive(Clone, Debug)] +pub struct DurableRunnerConfig { + pub connect_url: String, + pub state_dir: PathBuf, + pub runner_instance_id: String, + pub environment_lease_id: String, + pub run_id: String, + pub normalized_session_id: String, + pub turn_id: String, + pub item_id: String, + pub runner_version: String, + pub runner_digest: String, + pub fake_harness_path: Option, + pub fake_harness_script_path: Option, + pub max_outbox_bytes: usize, + pub p0_reserve_bytes: usize, + pub max_frame_bytes: usize, + pub reconnect_delay: Duration, + pub max_runtime: Duration, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StoredOutboxEvent { + pub source_seq: u64, + pub source_event_id: String, + pub priority: u8, + pub event_type: String, + pub item_id: Option, + pub envelope: Value, + pub byte_size: usize, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProcessedCommand { + pub command_id: String, + pub controller_seq: u64, + pub command_digest: String, + pub status: String, + pub logical_effect_count: u64, + pub result: Value, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DurableRunnerState { + pub schema: String, + pub runner_instance_id: String, + pub environment_lease_id: String, + pub run_id: String, + pub normalized_session_id: String, + pub turn_id: String, + pub item_id: String, + pub lifecycle: String, + pub next_source_seq: u64, + pub acked_source_seq: u64, + pub last_controller_command_seq: u64, + pub reconnect_count: u64, + pub max_outbox_bytes: usize, + pub peak_outbox_bytes: usize, + pub outbox: Vec, + pub processed_commands: BTreeMap, + pub diagnostics: Vec, + pub backpressure: bool, + pub recoverable_failure: Option, + pub unrecoverable_outcome: Option, + pub harness_generation: u64, + pub stop_after_flush: bool, +} + +impl DurableRunnerState { + fn new(config: &DurableRunnerConfig) -> Self { + Self { + schema: STATE_SCHEMA.to_owned(), + runner_instance_id: config.runner_instance_id.clone(), + environment_lease_id: config.environment_lease_id.clone(), + run_id: config.run_id.clone(), + normalized_session_id: config.normalized_session_id.clone(), + turn_id: config.turn_id.clone(), + item_id: config.item_id.clone(), + lifecycle: "connecting".to_owned(), + next_source_seq: 1, + acked_source_seq: 0, + last_controller_command_seq: 0, + reconnect_count: 0, + max_outbox_bytes: config.max_outbox_bytes, + peak_outbox_bytes: 0, + outbox: Vec::new(), + processed_commands: BTreeMap::new(), + diagnostics: Vec::new(), + backpressure: false, + recoverable_failure: None, + unrecoverable_outcome: None, + harness_generation: 1, + stop_after_flush: false, + } + } + + pub fn outbox_bytes(&self) -> usize { + self.outbox.iter().map(|event| event.byte_size).sum() + } + + pub fn highest_source_seq(&self) -> u64 { + self.next_source_seq.saturating_sub(1) + } + + pub fn apply_ack(&mut self, acked_source_seq: u64) -> Result<(), Phase3Error> { + if acked_source_seq < self.acked_source_seq { + return Err(Phase3Error::invalid( + "cumulative ACK cannot move behind the durable cursor", + )); + } + if acked_source_seq > self.highest_source_seq() { + return Err(Phase3Error::invalid( + "cumulative ACK cannot move beyond the produced source cursor", + )); + } + self.acked_source_seq = acked_source_seq; + self.outbox + .retain(|event| event.source_seq > acked_source_seq); + Ok(()) + } + + fn record_diagnostic(&mut self, message: impl Into) { + self.diagnostics.push(redact_text(&message.into())); + if self.diagnostics.len() > MAX_DIAGNOSTICS { + self.diagnostics.remove(0); + } + } +} + +#[derive(Clone, Debug)] +pub struct DurableStateStore { + path: PathBuf, +} + +impl DurableStateStore { + pub fn new(state_dir: &Path) -> Result { + if let Ok(metadata) = fs::symlink_metadata(state_dir) { + if metadata.file_type().is_symlink() || !metadata.file_type().is_dir() { + return Err(Phase3Error::invalid(format!( + "runner state directory {} must be a real directory", + state_dir.display() + ))); + } + } + fs::create_dir_all(state_dir).map_err(|error| { + Phase3Error::invalid(format!( + "failed to create runner state directory {}: {error}", + state_dir.display() + )) + })?; + #[cfg(unix)] + fs::set_permissions(state_dir, fs::Permissions::from_mode(0o700)).map_err(|error| { + Phase3Error::invalid(format!( + "failed to secure runner state directory {}: {error}", + state_dir.display() + )) + })?; + verify_private_directory(state_dir)?; + Ok(Self { + path: state_dir.join("runner-state.json"), + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn load_or_create( + &self, + config: &DurableRunnerConfig, + ) -> Result<(DurableRunnerState, bool), Phase3Error> { + let bytes = match open_private_regular_file(&self.path) { + Ok(mut file) => { + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes).map_err(|error| { + Phase3Error::invalid(format!( + "failed to read durable runner state {}: {error}", + self.path.display() + )) + })?; + bytes + } + Err(error) if error.kind() == io::ErrorKind::NotFound => { + let state = DurableRunnerState::new(config); + self.save(&state)?; + return Ok((state, false)); + } + Err(error) => { + return Err(Phase3Error::invalid(format!( + "failed to open private durable runner state {}: {error}", + self.path.display() + ))) + } + }; + let state: DurableRunnerState = serde_json::from_slice(&bytes).map_err(|error| { + Phase3Error::invalid(format!( + "durable runner state is malformed and cannot be recovered: {error}" + )) + })?; + validate_state_binding(&state, config)?; + Ok((state, true)) + } + + pub fn save(&self, state: &DurableRunnerState) -> Result<(), Phase3Error> { + let bytes = serde_json::to_vec_pretty(state).map_err(|error| { + Phase3Error::invalid(format!("failed to serialize durable runner state: {error}")) + })?; + let (temporary, mut file) = create_private_temporary_file(&self.path)?; + let result = (|| -> Result<(), Phase3Error> { + file.write_all(&bytes) + .and_then(|_| file.sync_all()) + .map_err(|error| { + Phase3Error::invalid(format!("failed to commit durable runner state: {error}")) + })?; + drop(file); + fs::rename(&temporary, &self.path).map_err(|error| { + Phase3Error::invalid(format!("failed to replace durable runner state: {error}")) + })?; + sync_parent_directory(&self.path)?; + Ok(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary); + } + result + } +} + +#[cfg(any(target_os = "linux", target_os = "android"))] +fn effective_user_id() -> io::Result { + Ok(fs::metadata("/proc/self")?.uid()) +} + +fn verify_private_directory(path: &Path) -> Result<(), Phase3Error> { + let metadata = fs::symlink_metadata(path).map_err(|error| { + Phase3Error::invalid(format!( + "failed to inspect runner state directory {}: {error}", + path.display() + )) + })?; + if metadata.file_type().is_symlink() || !metadata.file_type().is_dir() { + return Err(Phase3Error::invalid( + "runner state directory must be a real directory", + )); + } + #[cfg(unix)] + { + if metadata.mode() & 0o777 != 0o700 { + return Err(Phase3Error::invalid( + "runner state directory permissions must be 0700", + )); + } + #[cfg(any(target_os = "linux", target_os = "android"))] + if metadata.uid() + != effective_user_id().map_err(|error| { + Phase3Error::invalid(format!("failed to inspect daemon ownership: {error}")) + })? + { + return Err(Phase3Error::invalid( + "runner state directory must be owned by the daemon user", + )); + } + } + Ok(()) +} + +fn open_private_regular_file(path: &Path) -> io::Result { + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + options.custom_flags(no_follow_flag()); + let file = options.open(path)?; + let metadata = file.metadata()?; + if !metadata.file_type().is_file() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "state path is not a regular file", + )); + } + #[cfg(unix)] + { + if metadata.mode() & 0o777 != 0o600 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "state file permissions must be 0600", + )); + } + #[cfg(any(target_os = "linux", target_os = "android"))] + if metadata.uid() != effective_user_id()? { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "state file must be owned by the daemon user", + )); + } + } + Ok(file) +} + +#[cfg(any(target_os = "linux", target_os = "android"))] +const fn no_follow_flag() -> i32 { + 0x20000 +} + +#[cfg(all(unix, not(any(target_os = "linux", target_os = "android"))))] +const fn no_follow_flag() -> i32 { + 0x100 +} + +fn random_suffix() -> Result { + let mut bytes = [0_u8; 16]; + #[cfg(unix)] + { + File::open("/dev/urandom") + .and_then(|mut source| source.read_exact(&mut bytes)) + .map_err(|error| { + Phase3Error::invalid(format!("failed to obtain state-file randomness: {error}")) + })?; + } + #[cfg(windows)] + { + use std::ffi::c_void; + #[link(name = "bcrypt")] + unsafe extern "system" { + fn BCryptGenRandom( + algorithm: *mut c_void, + buffer: *mut u8, + buffer_length: u32, + flags: u32, + ) -> i32; + } + const BCRYPT_USE_SYSTEM_PREFERRED_RNG: u32 = 0x00000002; + // SAFETY: the system RNG receives a valid writable buffer for its exact length. + let status = unsafe { + BCryptGenRandom( + std::ptr::null_mut(), + bytes.as_mut_ptr(), + bytes.len() as u32, + BCRYPT_USE_SYSTEM_PREFERRED_RNG, + ) + }; + if status < 0 { + return Err(Phase3Error::invalid( + "failed to obtain state-file randomness from BCryptGenRandom", + )); + } + } + #[cfg(not(any(unix, windows)))] + { + return Err(Phase3Error::invalid( + "secure state-file randomness is unsupported on this platform", + )); + } + Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect()) +} + +fn create_private_temporary_file(destination: &Path) -> Result<(PathBuf, File), Phase3Error> { + let parent = destination + .parent() + .ok_or_else(|| Phase3Error::invalid("durable state path has no parent directory"))?; + let filename = destination + .file_name() + .and_then(|value| value.to_str()) + .ok_or_else(|| Phase3Error::invalid("durable state filename is not valid UTF-8"))?; + for _ in 0..TEMP_FILE_ATTEMPTS { + let path = parent.join(format!(".{filename}.{}.tmp", random_suffix()?)); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o600).custom_flags(no_follow_flag()); + match options.open(&path) { + Ok(file) => { + #[cfg(unix)] + if let Err(error) = file.set_permissions(fs::Permissions::from_mode(0o600)) { + let _ = fs::remove_file(&path); + return Err(Phase3Error::invalid(format!( + "failed to secure temporary runner state: {error}" + ))); + } + let metadata = file.metadata().map_err(|error| { + Phase3Error::invalid(format!( + "failed to inspect temporary runner state: {error}" + )) + })?; + if !metadata.file_type().is_file() { + let _ = fs::remove_file(&path); + return Err(Phase3Error::invalid( + "temporary runner state is not a regular file", + )); + } + #[cfg(unix)] + { + if metadata.mode() & 0o777 != 0o600 { + let _ = fs::remove_file(&path); + return Err(Phase3Error::invalid( + "temporary runner state is not private", + )); + } + #[cfg(any(target_os = "linux", target_os = "android"))] + if metadata.uid() + != effective_user_id().map_err(|error| { + Phase3Error::invalid(format!( + "failed to inspect temporary state ownership: {error}" + )) + })? + { + let _ = fs::remove_file(&path); + return Err(Phase3Error::invalid( + "temporary runner state is not daemon-owned", + )); + } + } + return Ok((path, file)); + } + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue, + Err(error) => { + return Err(Phase3Error::invalid(format!( + "failed to create exclusive temporary runner state: {error}" + ))) + } + } + } + Err(Phase3Error::invalid( + "failed to allocate a unique temporary runner state file", + )) +} + +fn sync_parent_directory(path: &Path) -> Result<(), Phase3Error> { + #[cfg(unix)] + { + let parent = path + .parent() + .ok_or_else(|| Phase3Error::invalid("durable state path has no parent directory"))?; + File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(|error| { + Phase3Error::invalid(format!( + "failed to sync durable state parent directory: {error}" + )) + })?; + } + Ok(()) +} + +fn validate_state_binding( + state: &DurableRunnerState, + config: &DurableRunnerConfig, +) -> Result<(), Phase3Error> { + if state.schema != STATE_SCHEMA { + return Err(Phase3Error::invalid( + "unsupported durable runner state schema", + )); + } + for (name, stored, expected) in [ + ( + "runnerInstanceId", + state.runner_instance_id.as_str(), + config.runner_instance_id.as_str(), + ), + ( + "environmentLeaseId", + state.environment_lease_id.as_str(), + config.environment_lease_id.as_str(), + ), + ("runId", state.run_id.as_str(), config.run_id.as_str()), + ( + "normalizedSessionId", + state.normalized_session_id.as_str(), + config.normalized_session_id.as_str(), + ), + ("turnId", state.turn_id.as_str(), config.turn_id.as_str()), + ("itemId", state.item_id.as_str(), config.item_id.as_str()), + ] { + if stored != expected { + return Err(Phase3Error::invalid(format!( + "durable {name} does not match the requested recovery identity" + ))); + } + } + Ok(()) +} + +fn redaction_key(key: &str) -> bool { + let lowered = key.to_ascii_lowercase().replace('-', "_"); + [ + "authorization", + "password", + "secret", + "token", + "api_key", + "apikey", + "credential", + ] + .iter() + .any(|needle| lowered.contains(needle)) +} + +pub fn redact_text(input: &str) -> String { + if input.to_ascii_lowercase().contains("bearer ") { + return "[REDACTED]".to_owned(); + } + input.to_owned() +} + +pub fn sanitize_value(value: &Value) -> Value { + match value { + Value::Object(object) => Value::Object( + object + .iter() + .map(|(key, value)| { + ( + key.clone(), + if redaction_key(key) { + Value::String("[REDACTED]".to_owned()) + } else { + sanitize_value(value) + }, + ) + }) + .collect(), + ), + Value::Array(values) => Value::Array(values.iter().map(sanitize_value).collect()), + Value::String(text) => Value::String(redact_text(text)), + other => other.clone(), + } +} + +fn canonical_json(value: &Value) -> String { + match value { + Value::Object(object) => { + let mut entries = object.iter().collect::>(); + entries.sort_by(|left, right| left.0.cmp(right.0)); + let body = entries + .into_iter() + .map(|(key, value)| { + format!( + "{}:{}", + serde_json::to_string(key).expect("JSON key serialization cannot fail"), + canonical_json(value) + ) + }) + .collect::>() + .join(","); + format!("{{{body}}}") + } + Value::Array(values) => format!( + "[{}]", + values + .iter() + .map(canonical_json) + .collect::>() + .join(",") + ), + scalar => serde_json::to_string(scalar).expect("JSON scalar serialization cannot fail"), + } +} + +fn fnv1a_digest(input: &[u8]) -> String { + let mut digest = 0xcbf29ce484222325_u64; + for byte in input { + digest ^= u64::from(*byte); + digest = digest.wrapping_mul(0x100000001b3); + } + format!("fnv1a64:{digest:016x}") +} + +fn deterministic_time(sequence: u64) -> String { + format!( + "2026-08-07T23:{:02}:{:02}.{:03}Z", + (sequence / 60_000) % 60, + (sequence / 1_000) % 60, + sequence % 1_000 + ) +} + +fn event_envelope( + state: &DurableRunnerState, + source_seq: u64, + event_type: &str, + priority: u8, + payload: Value, + item_id: Option<&str>, +) -> Value { + let source_event_id = format!("event_{}_{source_seq:06}", state.runner_instance_id); + let mut event = json!({ + "schema": "paperclip.prp.event.v1", + "sourceEventId": source_event_id, + "sourceSeq": source_seq, + "sourceInstanceId": state.runner_instance_id, + "sourceKind": "runner", + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "eventType": event_type, + "schemaVersion": 1, + "priority": priority, + "emittedAt": deterministic_time(source_seq), + "payload": sanitize_value(&payload), + }); + if let Some(item_id) = item_id { + event["itemId"] = Value::String(item_id.to_owned()); + } + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "envelopeId": format!("envelope_event_{source_seq:06}"), + "kind": "event", + "runnerInstanceId": state.runner_instance_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "sentAt": deterministic_time(source_seq), + "payload": event, + }) +} + +fn mark_backpressure( + state: &mut DurableRunnerState, + config: &DurableRunnerConfig, +) -> Result<(), Phase3Error> { + if state.backpressure { + return Ok(()); + } + state.backpressure = true; + state.lifecycle = "backpressure".to_owned(); + enqueue_event( + state, + config, + "runner.backpressure", + 0, + json!({ + "reason": "outbox_soft_limit", + "maxOutboxBytes": config.max_outbox_bytes, + "p0ReserveBytes": config.p0_reserve_bytes, + "newTurnsAccepted": false, + }), + None, + ) +} + +fn enqueue_event( + state: &mut DurableRunnerState, + config: &DurableRunnerConfig, + event_type: &str, + priority: u8, + payload: Value, + item_id: Option<&str>, +) -> Result<(), Phase3Error> { + if priority > 2 { + return Err(Phase3Error::invalid("event priority must be P0, P1, or P2")); + } + + if priority == 2 { + if let Some(last) = state.outbox.last_mut() { + if last.priority == 2 + && last.event_type == event_type + && last.item_id.as_deref() == item_id + { + let previous_count = last + .envelope + .pointer("/payload/payload/coalescedCount") + .and_then(Value::as_u64) + .unwrap_or(1); + let compacted = json!({ + "coalescedCount": previous_count + 1, + "latest": sanitize_value(&payload), + }); + last.envelope["payload"]["payload"] = compacted; + last.byte_size = serde_json::to_vec(&last.envelope) + .map_err(|error| Phase3Error::invalid(error.to_string()))? + .len(); + state.peak_outbox_bytes = state.peak_outbox_bytes.max(state.outbox_bytes()); + if state.outbox_bytes() + > config + .max_outbox_bytes + .saturating_sub(config.p0_reserve_bytes) + { + return mark_backpressure(state, config); + } + return Ok(()); + } + } + } + + let source_seq = state.next_source_seq; + let envelope = event_envelope(state, source_seq, event_type, priority, payload, item_id); + let bytes = + serde_json::to_vec(&envelope).map_err(|error| Phase3Error::invalid(error.to_string()))?; + let byte_size = bytes.len(); + let projected = state.outbox_bytes().saturating_add(byte_size); + let non_p0_limit = config + .max_outbox_bytes + .saturating_sub(config.p0_reserve_bytes); + + if priority == 2 && projected > non_p0_limit { + return mark_backpressure(state, config); + } + if priority == 1 && projected > non_p0_limit { + mark_backpressure(state, config)?; + return Err(Phase3Error::invalid( + "P1 event cannot enter the reserved P0 storage region", + )); + } + if projected > config.max_outbox_bytes { + state.lifecycle = "unrecoverable".to_owned(); + state.unrecoverable_outcome = Some("p0_storage_exhausted".to_owned()); + state.record_diagnostic( + "P0 storage reserve is exhausted; the durable session requires operator recovery", + ); + return Err(Phase3Error::invalid( + "P0 storage reserve exhausted; durable recovery is explicit", + )); + } + + state.next_source_seq += 1; + state.outbox.push(StoredOutboxEvent { + source_seq, + source_event_id: format!("event_{}_{source_seq:06}", state.runner_instance_id), + priority, + event_type: event_type.to_owned(), + item_id: item_id.map(str::to_owned), + envelope, + byte_size, + }); + state.peak_outbox_bytes = state.peak_outbox_bytes.max(state.outbox_bytes()); + Ok(()) +} + +fn command_result_envelope(state: &DurableRunnerState, processed: &ProcessedCommand) -> Value { + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "envelopeId": format!("envelope_result_{}", processed.command_id), + "kind": "command_result", + "runnerInstanceId": state.runner_instance_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "sentAt": deterministic_time(processed.controller_seq), + "payload": processed.result, + }) +} + +fn execute_command_effect( + state: &mut DurableRunnerState, + config: &DurableRunnerConfig, + command_type: &str, + payload: &Value, +) -> Result<(String, u64, String), Phase3Error> { + if state.lifecycle == "revoked" { + return Ok(( + "rejected".to_owned(), + 0, + "connection capability is revoked; commands have no logical effect".to_owned(), + )); + } + match command_type { + "run.prepare" => { + state.lifecycle = "ready".to_owned(); + enqueue_event( + state, + config, + "workspace.ready", + 1, + json!({ "workspace": "phase-03-durable-fixture" }), + None, + )?; + Ok(("completed".to_owned(), 1, "run prepared".to_owned())) + } + "session.open" => { + enqueue_event( + state, + config, + "session.started", + 0, + json!({ + "normalizedSessionId": state.normalized_session_id, + "driverSessionId": "driver_phase3_fake", + "resumed": state.reconnect_count > 0, + }), + None, + )?; + Ok(("completed".to_owned(), 1, "session bound".to_owned())) + } + "fault.harness_restart" => { + prove_harness_restart(config)?; + let previous_generation = state.harness_generation; + state.harness_generation += 1; + enqueue_event( + state, + config, + "harness.exited", + 0, + json!({ + "generation": previous_generation, + "reason": "fault_injection_restart", + "recoverable": true, + }), + None, + )?; + enqueue_event( + state, + config, + "harness.ready", + 0, + json!({ + "generation": state.harness_generation, + "driverKind": "fake", + }), + None, + )?; + enqueue_event( + state, + config, + "session.reconciled", + 0, + json!({ + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "outcome": "same_session_resumed", + }), + Some(&state.item_id.clone()), + )?; + Ok(( + "completed".to_owned(), + 1, + "harness restarted and reconciled".to_owned(), + )) + } + "fault.storage_pressure" => { + let item_id = state.item_id.clone(); + for index in 0..250_u64 { + enqueue_event( + state, + config, + "item.delta", + 2, + json!({ + "chunk": format!("bounded-progress-{index:03}"), + "secretToken": "must-not-persist", + }), + Some(&item_id), + )?; + } + mark_backpressure(state, config)?; + Ok(( + "completed".to_owned(), + 1, + "storage pressure applied with P2 coalescing".to_owned(), + )) + } + "turn.start" if state.lifecycle == "draining" || state.backpressure => Ok(( + "rejected".to_owned(), + 0, + "new turns are disabled while draining or backpressured".to_owned(), + )), + "turn.start" => { + let item_id = state.item_id.clone(); + enqueue_event( + state, + config, + "turn.accepted", + 0, + json!({ "turnId": state.turn_id, "sameSession": true }), + None, + )?; + enqueue_event( + state, + config, + "item.started", + 1, + json!({ "kind": "assistant_message" }), + Some(&item_id), + )?; + enqueue_event( + state, + config, + "item.delta", + 2, + json!({ "text": payload.get("text").cloned().unwrap_or_else(|| json!("Phase 3")) }), + Some(&item_id), + )?; + enqueue_event( + state, + config, + "item.completed", + 1, + json!({ "text": "Durable recovery completed." }), + Some(&item_id), + )?; + enqueue_event( + state, + config, + "run.result.proposed", + 0, + json!({ + "schema": "paperclip.prp.result.v1", + "status": "succeeded", + "reportedWorkDisposition": "done", + "summary": "Phase 3 durable transport recovered without duplicate effects.", + }), + None, + )?; + enqueue_event( + state, + config, + "run.terminal", + 0, + json!({ + "schema": "paperclip.prp.terminal.v1", + "turnTerminalState": "completed", + "runTerminalState": "succeeded", + "reportedWorkDisposition": "done", + }), + None, + )?; + state.lifecycle = "terminal".to_owned(); + Ok(("completed".to_owned(), 1, "turn completed".to_owned())) + } + "runner.drain" => { + state.lifecycle = "draining".to_owned(); + enqueue_event( + state, + config, + "runner.draining", + 0, + json!({ "newWorkAccepted": false }), + None, + )?; + Ok(("completed".to_owned(), 1, "runner draining".to_owned())) + } + "runner.shutdown" => { + state.stop_after_flush = true; + enqueue_event( + state, + config, + "runner.stopped", + 0, + json!({ "afterDurableFlush": true }), + None, + )?; + Ok(( + "completed".to_owned(), + 1, + "runner will stop after durable flush".to_owned(), + )) + } + unsupported => Ok(( + "rejected".to_owned(), + 0, + format!("unsupported Phase 3 command: {unsupported}"), + )), + } +} + +fn prove_harness_restart(config: &DurableRunnerConfig) -> Result<(), Phase3Error> { + let fake_harness_path = config.fake_harness_path.as_ref().ok_or_else(|| { + Phase3Error::invalid("fake harness path is required for restart injection") + })?; + let script_path = config.fake_harness_script_path.as_ref().ok_or_else(|| { + Phase3Error::invalid("fake harness script is required for restart injection") + })?; + let args = vec![ + "--script".to_owned(), + script_path.display().to_string(), + "--delay-ms".to_owned(), + "1".to_owned(), + ]; + for generation in 1..=2_u64 { + let mut harness = SupervisedProcess::spawn( + fake_harness_path, + &args, + Duration::from_millis(50), + 64 * 1024, + ) + .map_err(|error| { + Phase3Error::invalid(format!( + "failed to start fake harness generation {generation}: {error}" + )) + })?; + let ready = harness + .receive_stdout_line(Duration::from_secs(2)) + .map_err(|error| Phase3Error::invalid(error.to_string()))? + .ok_or_else(|| { + Phase3Error::invalid(format!( + "fake harness generation {generation} did not become ready" + )) + })?; + let message: Value = serde_json::from_str(&ready).map_err(|error| { + Phase3Error::invalid(format!( + "fake harness generation {generation} emitted malformed ready data: {error}" + )) + })?; + if message.get("type").and_then(Value::as_str) != Some("ready") { + return Err(Phase3Error::invalid(format!( + "fake harness generation {generation} did not emit a ready message" + ))); + } + harness + .terminate_group() + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + } + Ok(()) +} + +fn process_command( + state: &mut DurableRunnerState, + store: &DurableStateStore, + config: &DurableRunnerConfig, + command: &Value, +) -> Result { + let command_id = command + .get("commandId") + .and_then(Value::as_str) + .ok_or_else(|| Phase3Error::invalid("commandId is required"))?; + let controller_seq = command + .get("controllerSeq") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("controllerSeq is required"))?; + let command_type = command + .get("type") + .and_then(Value::as_str) + .ok_or_else(|| Phase3Error::invalid("command type is required"))?; + let canonical = canonical_json(command); + let command_digest = fnv1a_digest(canonical.as_bytes()); + + if let Some(previous) = state.processed_commands.get(command_id) { + if previous.command_digest != command_digest { + return Err(Phase3Error::invalid( + "commandId was reused with a different payload", + )); + } + return Ok(previous.clone()); + } + if controller_seq != state.last_controller_command_seq + 1 { + return Err(Phase3Error::invalid(format!( + "controllerSeq must be {}; received {controller_seq}", + state.last_controller_command_seq + 1 + ))); + } + + let payload = command.get("payload").cloned().unwrap_or(Value::Null); + let effect = execute_command_effect(state, config, command_type, &payload); + let (status, logical_effect_count, detail) = match effect { + Ok(effect) => effect, + Err(error) if state.unrecoverable_outcome.is_some() => { + ("failed".to_owned(), 1, error.to_string()) + } + Err(error) => return Err(error), + }; + let result = sanitize_value(&json!({ + "commandId": command_id, + "controllerSeq": controller_seq, + "status": status, + "logicalEffectCount": logical_effect_count, + "detail": detail, + })); + let processed = ProcessedCommand { + command_id: command_id.to_owned(), + controller_seq, + command_digest, + status, + logical_effect_count, + result, + }; + state.last_controller_command_seq = controller_seq; + state + .processed_commands + .insert(command_id.to_owned(), processed.clone()); + store.save(state)?; + Ok(processed) +} + +struct ParsedWsUrl { + host: String, + authority: String, + port: u16, + path: String, +} + +fn parse_ws_url(input: &str) -> Result { + let remainder = input + .strip_prefix("ws://") + .ok_or_else(|| Phase3Error::invalid("runner core accepts exactly the ws:// scheme"))?; + if remainder.is_empty() + || remainder + .chars() + .any(|character| character.is_ascii_control() || character.is_ascii_whitespace()) + || remainder.contains(['?', '#', '\\']) + { + return Err(Phase3Error::invalid( + "WebSocket URL contains malformed, query, fragment, or path ambiguity", + )); + } + let (authority, path) = remainder + .split_once('/') + .map_or((remainder, "/".to_owned()), |(authority, path)| { + (authority, format!("/{path}")) + }); + if authority.is_empty() || authority.contains(['@', '%']) { + return Err(Phase3Error::invalid( + "WebSocket authority must not contain userinfo or encoding ambiguity", + )); + } + let (host, port) = if authority.starts_with('[') { + let closing = authority + .find(']') + .ok_or_else(|| Phase3Error::invalid("bracketed IPv6 authority is malformed"))?; + let host = &authority[1..closing]; + let port = authority[closing + 1..] + .strip_prefix(':') + .ok_or_else(|| Phase3Error::invalid("bracketed IPv6 authority requires a port"))?; + host.parse::() + .map_err(|_| Phase3Error::invalid("bracketed WebSocket host must be IPv6"))?; + (host, port) + } else { + let (host, port) = authority + .rsplit_once(':') + .ok_or_else(|| Phase3Error::invalid("WebSocket URL must include an explicit port"))?; + if host.is_empty() || host.contains(':') { + return Err(Phase3Error::invalid( + "WebSocket host is empty or an unbracketed IPv6 literal", + )); + } + (host, port) + }; + let port = port + .parse::() + .map_err(|error| Phase3Error::invalid(format!("invalid WebSocket port: {error}")))?; + if port == 0 { + return Err(Phase3Error::invalid("WebSocket port must be non-zero")); + } + Ok(ParsedWsUrl { + host: host.to_owned(), + authority: authority.to_owned(), + port, + path, + }) +} + +struct ResolvedWsTarget { + authority: String, + path: String, + addresses: Vec, +} + +impl ResolvedWsTarget { + fn resolve(input: &str) -> Result { + resolve_ws_target_with(input, |host, port| { + (host, port) + .to_socket_addrs() + .map(|addresses| addresses.collect()) + }) + } +} + +fn resolve_ws_target_with(input: &str, resolver: F) -> Result +where + F: FnOnce(&str, u16) -> io::Result>, +{ + let parsed = parse_ws_url(input)?; + let mut addresses = resolver(&parsed.host, parsed.port).map_err(|error| { + Phase3Error::invalid(format!("failed to resolve WebSocket destination: {error}")) + })?; + addresses.sort_unstable(); + addresses.dedup(); + if addresses.is_empty() { + return Err(Phase3Error::invalid( + "WebSocket destination resolved to no addresses", + )); + } + if addresses.iter().any(|address| !address.ip().is_loopback()) { + return Err(Phase3Error::invalid( + "every WebSocket destination must be loopback (127.0.0.0/8 or ::1)", + )); + } + Ok(ResolvedWsTarget { + authority: parsed.authority, + path: parsed.path, + addresses, + }) +} + +struct WsClient { + stream: TcpStream, + mask_counter: u32, + max_frame_bytes: usize, +} + +fn encode_masked_frame( + opcode: u8, + payload: &[u8], + mask: [u8; 4], + max_frame_bytes: usize, +) -> Result, Phase3Error> { + if payload.len() > max_frame_bytes { + return Err(Phase3Error::invalid( + "outbound WebSocket frame exceeds the limit", + )); + } + let mut frame = vec![0x80 | opcode]; + match payload.len() { + length if length <= 125 => frame.push(0x80 | length as u8), + length if length <= u16::MAX as usize => { + frame.push(0x80 | 126); + frame.extend_from_slice(&(length as u16).to_be_bytes()); + } + length => { + frame.push(0x80 | 127); + frame.extend_from_slice(&(length as u64).to_be_bytes()); + } + } + frame.extend_from_slice(&mask); + frame.extend( + payload + .iter() + .enumerate() + .map(|(index, byte)| byte ^ mask[index % 4]), + ); + Ok(frame) +} + +fn checked_inbound_frame_length(length: u64, max_frame_bytes: usize) -> Result { + let length = usize::try_from(length) + .map_err(|_| Phase3Error::invalid("WebSocket frame length overflow"))?; + if length > max_frame_bytes { + return Err(Phase3Error::invalid( + "inbound WebSocket frame exceeds the limit", + )); + } + Ok(length) +} + +impl WsClient { + fn connect( + target: &ResolvedWsTarget, + authorization: &str, + max_frame_bytes: usize, + ) -> Result { + let mut stream = TcpStream::connect(target.addresses.as_slice()) + .map_err(|error| Phase3Error::invalid(format!("WebSocket connect failed: {error}")))?; + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + stream + .set_write_timeout(Some(Duration::from_secs(2))) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + let mut request = format!( + "GET {} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {}\r\nSec-WebSocket-Version: 13\r\nAuthorization: Bearer {}\r\n\r\n", + target.path, + target.authority, + STATIC_WEBSOCKET_KEY, + authorization + ) + .into_bytes(); + let write_result = stream.write_all(&request).and_then(|_| stream.flush()); + // The HTTP upgrade request contains the bearer capability. Overwrite it before + // propagating success or failure so it does not linger in daemon memory. + request.fill(0); + request.clear(); + write_result.map_err(|error| { + Phase3Error::invalid(format!("WebSocket upgrade write failed: {error}")) + })?; + + let mut response = Vec::new(); + let mut byte = [0_u8; 1]; + while !response.ends_with(b"\r\n\r\n") { + if response.len() >= MAX_HTTP_HEADER_BYTES { + return Err(Phase3Error::invalid( + "WebSocket response headers are too large", + )); + } + stream.read_exact(&mut byte).map_err(|error| { + Phase3Error::invalid(format!("WebSocket upgrade response failed: {error}")) + })?; + response.push(byte[0]); + } + let response = String::from_utf8(response) + .map_err(|error| Phase3Error::invalid(format!("invalid HTTP response: {error}")))?; + if !response.starts_with("HTTP/1.1 101 ") { + let status = response + .lines() + .next() + .unwrap_or("HTTP response unavailable"); + return Err(Phase3Error::invalid(format!( + "WebSocket authentication or upgrade rejected: {status}" + ))); + } + let accept_valid = response.lines().any(|line| { + line.split_once(':').is_some_and(|(name, value)| { + name.eq_ignore_ascii_case("sec-websocket-accept") + && value.trim() == STATIC_WEBSOCKET_ACCEPT + }) + }); + if !accept_valid { + return Err(Phase3Error::invalid( + "WebSocket server returned an invalid acceptance proof", + )); + } + stream + .set_read_timeout(Some(Duration::from_millis(250))) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + Ok(Self { + stream, + mask_counter: 1, + max_frame_bytes, + }) + } + + fn send_json(&mut self, value: &Value) -> Result<(), Phase3Error> { + let bytes = serde_json::to_vec(value).map_err(|error| { + Phase3Error::invalid(format!("frame serialization failed: {error}")) + })?; + self.send_frame(0x1, &bytes) + } + + fn send_frame(&mut self, opcode: u8, payload: &[u8]) -> Result<(), Phase3Error> { + let mask = self.mask_counter.to_be_bytes(); + self.mask_counter = self.mask_counter.wrapping_add(1); + let frame = encode_masked_frame(opcode, payload, mask, self.max_frame_bytes)?; + self.stream + .write_all(&frame) + .and_then(|_| self.stream.flush()) + .map_err(|error| Phase3Error::invalid(format!("WebSocket frame write failed: {error}"))) + } + + fn receive_json(&mut self) -> Result, Phase3Error> { + loop { + let mut header = [0_u8; 2]; + match self.stream.read_exact(&mut header) { + Ok(()) => {} + Err(error) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + return Ok(None); + } + Err(error) => { + return Err(Phase3Error::invalid(format!( + "WebSocket connection closed: {error}" + ))) + } + } + let opcode = header[0] & 0x0f; + let masked = header[1] & 0x80 != 0; + let mut length = u64::from(header[1] & 0x7f); + if length == 126 { + let mut extended = [0_u8; 2]; + self.stream + .read_exact(&mut extended) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + length = u64::from(u16::from_be_bytes(extended)); + } else if length == 127 { + let mut extended = [0_u8; 8]; + self.stream + .read_exact(&mut extended) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + length = u64::from_be_bytes(extended); + } + let length = checked_inbound_frame_length(length, self.max_frame_bytes)?; + let mut mask = [0_u8; 4]; + if masked { + self.stream + .read_exact(&mut mask) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + } + let mut payload = vec![0_u8; length]; + self.stream + .read_exact(&mut payload) + .map_err(|error| Phase3Error::invalid(error.to_string()))?; + if masked { + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % 4]; + } + } + match opcode { + 0x1 => { + let value = serde_json::from_slice(&payload).map_err(|error| { + Phase3Error::invalid(format!("malformed WebSocket JSON: {error}")) + })?; + return Ok(Some(value)); + } + 0x8 => return Err(Phase3Error::invalid("WebSocket peer closed the connection")), + 0x9 => self.send_frame(0xA, &payload)?, + 0xA => {} + _ => return Err(Phase3Error::invalid("unsupported WebSocket frame opcode")), + } + } + } +} + +fn hello_envelope(state: &DurableRunnerState, config: &DurableRunnerConfig) -> Value { + let unacked_range = state + .outbox + .first() + .zip(state.outbox.last()) + .map(|(first, last)| json!([first.source_seq, last.source_seq])); + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "envelopeId": format!("envelope_hello_{}", state.reconnect_count), + "kind": "hello", + "runnerInstanceId": state.runner_instance_id, + "sentAt": deterministic_time(state.reconnect_count), + "payload": { + "protocolMin": 1, + "protocolMax": 1, + "runnerVersion": config.runner_version, + "runnerDigest": config.runner_digest, + "environmentLeaseId": state.environment_lease_id, + "sandboxProvider": "standalone_mock", + "platform": { + "os": std::env::consts::OS, + "arch": std::env::consts::ARCH, + "hostname": "redacted-standalone-runner", + }, + "drivers": [{ + "kind": "fake", + "version": "1.0.0", + "capabilities": { "resume": true, "interrupt": true }, + }], + "resume": { + "lastControllerCommandSeq": state.last_controller_command_seq, + "nextSourceEventSeq": state.next_source_seq, + "ackedSourceSeq": state.acked_source_seq, + "unackedEventRange": unacked_range, + }, + }, + }) +} + +fn send_outbox(client: &mut WsClient, state: &DurableRunnerState) -> Result<(), Phase3Error> { + for event in &state.outbox { + client.send_json(&event.envelope)?; + } + Ok(()) +} + +fn welcome_payload(value: &Value) -> Result<&Value, Phase3Error> { + if value.get("protocol").and_then(Value::as_str) != Some(PROTOCOL) + || value.get("version").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) + || value.get("kind").and_then(Value::as_str) != Some("welcome") + { + return Err(Phase3Error::invalid("expected a PRP v1 welcome envelope")); + } + value + .get("payload") + .ok_or_else(|| Phase3Error::invalid("welcome payload is required")) +} + +pub fn run_durable_runner( + config: DurableRunnerConfig, + bootstrap_ticket: BootstrapTicket, +) -> Result<(), Phase3Error> { + if config.p0_reserve_bytes >= config.max_outbox_bytes { + return Err(Phase3Error::invalid( + "P0 reserve must be smaller than the outbox limit", + )); + } + // Resolve once before a bearer can be sent. Reconnects use only this validated, + // concrete address set, so DNS cannot redirect an authenticated retry. + let target = ResolvedWsTarget::resolve(&config.connect_url)?; + let store = DurableStateStore::new(&config.state_dir)?; + let (mut state, recovered) = store.load_or_create(&config)?; + if recovered { + state.reconnect_count += 1; + state.record_diagnostic("runner process restored the same durable identity"); + enqueue_event( + &mut state, + &config, + "runner.reconciled", + 0, + json!({ + "runnerInstanceId": config.runner_instance_id, + "normalizedSessionId": config.normalized_session_id, + "turnId": config.turn_id, + "itemId": config.item_id, + "outcome": "same_durable_session_resumed", + }), + None, + )?; + store.save(&state)?; + } + let mut bootstrap_ticket = Some(bootstrap_ticket); + let mut connection_lease_token: Option = None; + let started = Instant::now(); + + loop { + if started.elapsed() > config.max_runtime { + state.recoverable_failure = Some("transport_reconnect_deadline_exceeded".to_owned()); + state.lifecycle = "recoverable_failure".to_owned(); + state.record_diagnostic("transport reconnect deadline exceeded"); + store.save(&state)?; + return Err(Phase3Error::invalid( + "transport reconnect deadline exceeded; durable state is preserved", + )); + } + let authorization = connection_lease_token + .as_ref() + .map(SensitiveString::expose) + .or_else(|| bootstrap_ticket.as_ref().map(BootstrapTicket::expose)) + .ok_or_else(|| { + Phase3Error::invalid( + "transport capability is unavailable; a fresh runner bootstrap is required", + ) + })?; + let mut client = match WsClient::connect(&target, authorization, config.max_frame_bytes) { + Ok(client) => client, + Err(error) => { + let text = error.to_string(); + if text.contains("401") || text.contains("403") { + state.recoverable_failure = Some("lease_expired_requires_bootstrap".to_owned()); + state.lifecycle = "recoverable_failure".to_owned(); + state.record_diagnostic( + "connection lease expired; a fresh bootstrap may resume this state", + ); + store.save(&state)?; + return Err(Phase3Error::invalid( + "connection lease expired; durable state requires a fresh bootstrap", + )); + } + state.record_diagnostic(format!("transport reconnect scheduled: {text}")); + store.save(&state)?; + thread::sleep(config.reconnect_delay); + continue; + } + }; + + client.send_json(&hello_envelope(&state, &config))?; + let mut welcome = loop { + match client.receive_json() { + Ok(Some(value)) => break value, + Ok(None) if started.elapsed() <= config.max_runtime => continue, + Ok(None) => { + return Err(Phase3Error::invalid("welcome timed out")); + } + Err(error) => { + state.record_diagnostic(error.to_string()); + store.save(&state)?; + thread::sleep(config.reconnect_delay); + continue; + } + } + }; + let next_lease_token = match welcome.pointer_mut("/payload/connectionLeaseToken") { + Some(Value::String(value)) => { + let token = std::mem::take(value); + *value = "[REDACTED]".to_owned(); + Some(SensitiveString::new(token)) + } + _ => None, + }; + let payload = welcome_payload(&welcome)?; + let selected_version = payload + .get("selectedVersion") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("welcome selectedVersion is required"))?; + if selected_version != PROTOCOL_VERSION { + return Err(Phase3Error::invalid( + "mock core selected an unsupported version", + )); + } + if let Some(token) = next_lease_token { + connection_lease_token = Some(token); + } + // A bootstrap ticket is one-use. Clear it as soon as the authenticated welcome + // has exchanged it for a connection lease. + bootstrap_ticket.take(); + if let Some(acked_source_seq) = payload.get("ackedSourceSeq").and_then(Value::as_u64) { + state.apply_ack(acked_source_seq)?; + } + state.lifecycle = + if state.lifecycle == "connecting" || state.lifecycle == "recoverable_failure" { + "ready".to_owned() + } else { + state.lifecycle.clone() + }; + state.recoverable_failure = None; + store.save(&state)?; + + let initial_delivery = (|| -> Result<(), Phase3Error> { + if let Some(commands) = payload.get("pendingCommands").and_then(Value::as_array) { + for command in commands { + match process_command(&mut state, &store, &config, command) { + Ok(processed) => { + client.send_json(&command_result_envelope(&state, &processed))? + } + Err(error) => { + state.record_diagnostic(error.to_string()); + store.save(&state)?; + } + } + } + } + send_outbox(&mut client, &state) + })(); + if let Err(error) = initial_delivery { + state.record_diagnostic(error.to_string()); + state.reconnect_count += 1; + store.save(&state)?; + thread::sleep(config.reconnect_delay); + continue; + } + + let mut revoke_deadline: Option = None; + let disconnected = loop { + if (state.stop_after_flush || state.lifecycle == "revoked") && state.outbox.is_empty() { + state.lifecycle = if state.lifecycle == "revoked" { + "revoked".to_owned() + } else { + "stopped".to_owned() + }; + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + if revoke_deadline.is_some_and(|deadline| Instant::now() >= deadline) { + state.record_diagnostic( + "revocation flush deadline elapsed; unacked durable events remain preserved", + ); + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + match client.receive_json() { + Ok(None) => continue, + Err(error) => { + state.record_diagnostic(error.to_string()); + if state.lifecycle == "revoked" { + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + state.reconnect_count += 1; + store.save(&state)?; + break true; + } + Ok(Some(message)) => match message.get("kind").and_then(Value::as_str) { + Some("ack") => { + let acked = message + .pointer("/payload/ackedSourceSeq") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("ACK cursor is required"))?; + state.apply_ack(acked)?; + store.save(&state)?; + } + Some("command") => { + let command = message + .get("payload") + .ok_or_else(|| Phase3Error::invalid("command payload is required"))?; + match process_command(&mut state, &store, &config, command) { + Ok(processed) => { + let delivery = client + .send_json(&command_result_envelope(&state, &processed)) + .and_then(|()| send_outbox(&mut client, &state)); + if let Err(error) = delivery { + state.record_diagnostic(error.to_string()); + if state.lifecycle == "revoked" { + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + state.reconnect_count += 1; + store.save(&state)?; + break true; + } + } + Err(error) => { + state.record_diagnostic(error.to_string()); + store.save(&state)?; + } + } + } + Some("revoke") => { + state.lifecycle = "revoked".to_owned(); + state.stop_after_flush = true; + revoke_deadline.get_or_insert_with(|| { + Instant::now() + .checked_add(REVOKE_FLUSH_TIMEOUT) + .unwrap_or_else(Instant::now) + }); + state + .record_diagnostic("connection lease revoked; flushing durable events"); + store.save(&state)?; + if let Err(error) = send_outbox(&mut client, &state) { + state.record_diagnostic(error.to_string()); + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + } + Some("ping") => { + let pong = client.send_json(&json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "pong", + "runnerInstanceId": state.runner_instance_id, + "payload": { + "lifecycle": state.lifecycle, + "outboxBytes": state.outbox_bytes(), + "ackedSourceSeq": state.acked_source_seq, + }, + })); + if let Err(error) = pong { + state.record_diagnostic(error.to_string()); + if state.lifecycle == "revoked" { + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + state.reconnect_count += 1; + store.save(&state)?; + break true; + } + } + _ => { + state.record_diagnostic( + "malformed or unsupported control frame closed the connection", + ); + store.save(&state)?; + if state.lifecycle == "revoked" { + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + break true; + } + }, + } + }; + if disconnected { + thread::sleep(config.reconnect_delay); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + static ENVIRONMENT_TEST_LOCK: Mutex<()> = Mutex::new(()); + + fn config(root: &Path) -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: "ws://127.0.0.1:1/phase3/connect".to_owned(), + state_dir: root.to_path_buf(), + runner_instance_id: "runner_phase3_stable".to_owned(), + environment_lease_id: "lease_phase3_stable".to_owned(), + run_id: "run_phase3_stable".to_owned(), + normalized_session_id: "session_phase3_stable".to_owned(), + turn_id: "turn_phase3_stable".to_owned(), + item_id: "item_phase3_stable".to_owned(), + runner_version: "0.3.0".to_owned(), + runner_digest: "sha256:phase3-approved".to_owned(), + fake_harness_path: None, + fake_harness_script_path: None, + max_outbox_bytes: 16 * 1024, + p0_reserve_bytes: 8 * 1024, + max_frame_bytes: 1024 * 1024, + reconnect_delay: Duration::from_millis(1), + max_runtime: Duration::from_millis(10), + } + } + + fn temporary_root(name: &str) -> PathBuf { + let root = std::env::temp_dir().join(format!( + "paperclip-runner-phase3-{name}-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&root); + fs::create_dir_all(&root).unwrap(); + root + } + + fn command(id: &str, sequence: u64, command_type: &str, payload: Value) -> Value { + json!({ + "schema": "paperclip.prp.command.v1", + "commandId": id, + "controllerSeq": sequence, + "type": command_type, + "issuedAt": deterministic_time(sequence), + "payload": payload, + }) + } + + #[test] + fn durable_state_restores_identity_outbox_and_cumulative_ack() { + let root = temporary_root("restore"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, recovered) = store.load_or_create(&config).unwrap(); + assert!(!recovered); + let session_id = state.normalized_session_id.clone(); + enqueue_event( + &mut state, + &config, + "session.started", + 0, + json!({ "session": session_id }), + None, + ) + .unwrap(); + enqueue_event( + &mut state, + &config, + "run.terminal", + 0, + json!({ "status": "succeeded" }), + None, + ) + .unwrap(); + let first_event_id = state.outbox[0].source_event_id.clone(); + store.save(&state).unwrap(); + + let (mut restored, recovered) = store.load_or_create(&config).unwrap(); + assert!(recovered); + assert_eq!(restored.runner_instance_id, config.runner_instance_id); + assert_eq!(restored.normalized_session_id, config.normalized_session_id); + assert_eq!(restored.outbox[0].source_event_id, first_event_id); + restored.apply_ack(1).unwrap(); + assert_eq!(restored.acked_source_seq, 1); + assert_eq!(restored.outbox.len(), 1); + assert!(restored.apply_ack(0).is_err()); + assert!(restored.apply_ack(3).is_err()); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn duplicate_command_has_one_logical_effect_and_no_new_events() { + let root = temporary_root("command-dedupe"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let value = command("command_prepare", 1, "run.prepare", json!({})); + let first = process_command(&mut state, &store, &config, &value).unwrap(); + let event_count = state.outbox.len(); + let second = process_command(&mut state, &store, &config, &value).unwrap(); + assert_eq!(first.result, second.result); + assert_eq!(first.logical_effect_count, 1); + assert_eq!(state.outbox.len(), event_count); + assert_eq!(state.processed_commands.len(), 1); + assert!(process_command( + &mut state, + &store, + &config, + &command( + "command_prepare", + 1, + "run.prepare", + json!({ "changed": true }) + ) + ) + .is_err()); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn pressure_coalesces_p2_preserves_p0_and_stays_bounded() { + let root = temporary_root("pressure"); + let mut config = config(&root); + config.max_outbox_bytes = 12 * 1024; + config.p0_reserve_bytes = 6 * 1024; + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + process_command( + &mut state, + &store, + &config, + &command("command_pressure", 1, "fault.storage_pressure", json!({})), + ) + .unwrap(); + assert!(state.backpressure); + assert!(state.outbox_bytes() <= config.max_outbox_bytes); + assert_eq!( + state + .outbox + .iter() + .filter(|event| event.event_type == "item.delta") + .count(), + 1 + ); + assert!(state + .outbox + .iter() + .any(|event| event.priority == 0 && event.event_type == "runner.backpressure")); + let persisted = fs::read_to_string(store.path()).unwrap(); + assert!(!persisted.contains("must-not-persist")); + assert!(persisted.contains("[REDACTED]")); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn drain_rejects_new_turn_and_revoke_preserves_unacked_events() { + let root = temporary_root("drain"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + process_command( + &mut state, + &store, + &config, + &command("command_drain", 1, "runner.drain", json!({})), + ) + .unwrap(); + let before = state.outbox.len(); + let rejected = process_command( + &mut state, + &store, + &config, + &command("command_turn", 2, "turn.start", json!({})), + ) + .unwrap(); + assert_eq!(rejected.status, "rejected"); + assert_eq!(rejected.logical_effect_count, 0); + assert_eq!(state.outbox.len(), before); + state.lifecycle = "revoked".to_owned(); + store.save(&state).unwrap(); + let (restored, _) = store.load_or_create(&config).unwrap(); + assert_eq!(restored.lifecycle, "revoked"); + assert_eq!(restored.outbox.len(), before); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn oversized_p0_records_an_explicit_unrecoverable_outcome() { + let root = temporary_root("unrecoverable"); + let mut config = config(&root); + config.max_outbox_bytes = 512; + config.p0_reserve_bytes = 256; + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let result = enqueue_event( + &mut state, + &config, + "run.terminal", + 0, + json!({ "evidence": "x".repeat(4_096) }), + None, + ); + assert!(result.is_err()); + assert_eq!( + state.unrecoverable_outcome.as_deref(), + Some("p0_storage_exhausted") + ); + assert_eq!(state.lifecycle, "unrecoverable"); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn destination_validation_rejects_public_private_wildcard_and_mixed_answers() { + for (label, url, addresses) in [ + ( + "public", + "ws://8.8.8.8:443/phase3/connect", + vec!["8.8.8.8:443".parse::().unwrap()], + ), + ( + "private", + "ws://10.1.2.3:3100/phase3/connect", + vec!["10.1.2.3:3100".parse::().unwrap()], + ), + ( + "wildcard", + "ws://0.0.0.0:3100/phase3/connect", + vec!["0.0.0.0:3100".parse::().unwrap()], + ), + ( + "mixed-answer", + "ws://mixed.test:3100/phase3/connect", + vec![ + "127.0.0.1:3100".parse::().unwrap(), + "192.0.2.9:3100".parse::().unwrap(), + ], + ), + ] { + let result = resolve_ws_target_with(url, |_host, _port| Ok(addresses)); + assert!(result.is_err(), "{label} destination must fail closed"); + } + } + + #[test] + fn destination_validation_rejects_malformed_userinfo_query_and_fragment_before_resolution() { + for url in [ + "wss://127.0.0.1:3100/phase3/connect", + "WS://127.0.0.1:3100/phase3/connect", + "ws://127.0.0.1/phase3/connect", + "ws://127.0.0.1:0/phase3/connect", + "ws://[::1:3100/phase3/connect", + "ws://user@127.0.0.1:3100/phase3/connect", + "ws://127.0.0.1:3100/phase3/connect?ticket=ambiguous", + "ws://127.0.0.1:3100/phase3/connect#fragment", + "ws://127.0.0.1:3100/phase3\\connect", + ] { + let mut resolution_attempted = false; + let result = resolve_ws_target_with(url, |_host, _port| { + resolution_attempted = true; + Ok(vec!["127.0.0.1:3100".parse().unwrap()]) + }); + assert!(result.is_err(), "ambiguous URL must be rejected: {url}"); + assert!( + !resolution_attempted, + "malformed URL reached destination resolution: {url}" + ); + } + } + + #[test] + fn destination_validation_accepts_only_concrete_loopback_answers() { + let target = + resolve_ws_target_with("ws://localhost:3100/phase3/connect", |_host, _port| { + Ok(vec![ + "127.42.0.9:3100".parse().unwrap(), + "[::1]:3100".parse().unwrap(), + ]) + }) + .unwrap(); + assert_eq!(target.addresses.len(), 2); + assert!(target + .addresses + .iter() + .all(|address| address.ip().is_loopback())); + } + + #[test] + fn oversized_outbound_websocket_frame_is_rejected_before_write() { + let error = encode_masked_frame(0x1, b"ninebytes", [0; 4], 8).unwrap_err(); + assert!(error + .to_string() + .contains("outbound WebSocket frame exceeds")); + } + + #[test] + fn oversized_inbound_websocket_frame_is_rejected_before_allocation() { + let error = checked_inbound_frame_length(9, 8).unwrap_err(); + assert!(error + .to_string() + .contains("inbound WebSocket frame exceeds")); + } + + #[test] + fn revoked_runner_rejects_turn_with_zero_effects_and_preserves_existing_outbox() { + let root = temporary_root("revoked-command"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + enqueue_event( + &mut state, + &config, + "run.terminal", + 0, + json!({ "status": "succeeded" }), + None, + ) + .unwrap(); + state.lifecycle = "revoked".to_owned(); + let before = state.outbox.clone(); + let rejected = process_command( + &mut state, + &store, + &config, + &command("command_after_revoke", 1, "turn.start", json!({})), + ) + .unwrap(); + assert_eq!(rejected.status, "rejected"); + assert_eq!(rejected.logical_effect_count, 0); + assert_eq!(state.outbox, before); + let _ = fs::remove_dir_all(root); + } + + #[cfg(unix)] + #[test] + fn atomic_state_write_ignores_preplanted_predictable_symlink_and_uses_private_modes() { + use std::os::unix::fs::symlink; + + let root = temporary_root("state-symlink"); + let victim = root.join("victim.txt"); + fs::write(&victim, "unchanged").unwrap(); + let planted = root.join("runner-state.json.next"); + symlink(&victim, &planted).unwrap(); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + store.save(&DurableRunnerState::new(&config)).unwrap(); + + assert_eq!(fs::read_to_string(&victim).unwrap(), "unchanged"); + assert!(fs::symlink_metadata(&planted) + .unwrap() + .file_type() + .is_symlink()); + assert_eq!(fs::metadata(&root).unwrap().mode() & 0o777, 0o700); + assert_eq!(fs::metadata(store.path()).unwrap().mode() & 0o777, 0o600); + let _ = fs::remove_dir_all(root); + } + + #[cfg(unix)] + #[test] + fn durable_state_load_rejects_a_symlink() { + use std::os::unix::fs::symlink; + + let root = temporary_root("state-load-symlink"); + let config = config(&root); + let victim = root.join("victim.json"); + fs::write( + &victim, + serde_json::to_vec(&DurableRunnerState::new(&config)).unwrap(), + ) + .unwrap(); + symlink(&victim, root.join("runner-state.json")).unwrap(); + let store = DurableStateStore::new(&root).unwrap(); + assert!(store.load_or_create(&config).is_err()); + let _ = fs::remove_dir_all(root); + } + + #[cfg(unix)] + #[test] + fn captured_bootstrap_is_removed_and_absent_from_a_real_child_environment() { + let _guard = ENVIRONMENT_TEST_LOCK.lock().unwrap(); + let secret = format!("bootstrap-environment-regression-{}", std::process::id()); + std::env::set_var("PAPERCLIP_RUNNER_BOOTSTRAP_TICKET", &secret); + let ticket = capture_bootstrap_ticket().unwrap().unwrap(); + assert!(std::env::var_os("PAPERCLIP_RUNNER_BOOTSTRAP_TICKET").is_none()); + + let mut child = SupervisedProcess::spawn( + Path::new("/usr/bin/env"), + &[], + Duration::from_millis(50), + 16 * 1024, + ) + .unwrap(); + let mut environment = Vec::new(); + while let Some(line) = child + .receive_stdout_line(Duration::from_millis(250)) + .unwrap() + { + environment.push(line); + } + let exit = child.wait().unwrap(); + assert!(exit.success); + let dump = environment.join("\n"); + assert!(!dump.contains("PAPERCLIP_RUNNER_BOOTSTRAP_TICKET")); + assert!(!dump.contains(&secret)); + drop(ticket); + } +} diff --git a/packages/paperclip-runner/scripts/phase2-browser-server.mjs b/packages/paperclip-runner/scripts/phase2-browser-server.mjs index c523b1c64f1..2dfe16fcf0b 100644 --- a/packages/paperclip-runner/scripts/phase2-browser-server.mjs +++ b/packages/paperclip-runner/scripts/phase2-browser-server.mjs @@ -1,4 +1,7 @@ import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; export const PHASE2_BROWSER_LIMITS = Object.freeze({ maxActiveRuns: 4, @@ -230,9 +233,18 @@ async function loadPhase2Runner() { return import(runnerModuleUrl); } +async function loadPhase3Runner() { + const runnerModuleUrl = new URL( + "../dist/mock-core/phase3-recovery.js", + import.meta.url, + ).href; + return import(runnerModuleUrl); +} + export function createPhase2BrowserMiddleware(options = {}) { const limits = normalizeLimits(options.limits); const loadRunner = options.loadRunner ?? loadPhase2Runner; + const loadRecoveryRunner = options.loadRecoveryRunner ?? loadPhase3Runner; const runs = new Map(); let startingRuns = 0; @@ -252,11 +264,39 @@ export function createPhase2BrowserMiddleware(options = {}) { return async function middleware(request, response, next) { const url = new URL(request.url ?? "/", "http://phase2.local"); - if (!url.pathname.startsWith("/api/phase2/")) { + if (!url.pathname.startsWith("/api/phase2/") && url.pathname !== "/api/phase3/recovery") { next(); return; } + if (request.method === "POST" && url.pathname === "/api/phase3/recovery") { + let storageDirectory = null; + try { + assertTrustedTransport(request); + rejectDeclaredOversize(request, limits.maxRequestBodyBytes); + const body = await readJson(request, limits.maxRequestBodyBytes); + const recovery = await loadRecoveryRunner(); + const scratchRoot = + process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? + process.env.PAPERCLIP_SCRATCH_DIR ?? + tmpdir(); + storageDirectory = await mkdtemp( + join(scratchRoot, "paperclip-runner-browser-phase3-"), + ); + json(response, 200, await recovery.runPhase3Recovery({ + fault: body.fault ?? "lost-ack", + stateDirectory: storageDirectory, + })); + } catch (error) { + sendError(response, error); + } finally { + if (storageDirectory !== null) { + await rm(storageDirectory, { recursive: true, force: true }); + } + } + return; + } + try { assertTrustedTransport(request); rejectDeclaredOversize(request, limits.maxRequestBodyBytes); diff --git a/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs b/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs index 38da92fa5bd..7f673af4101 100644 --- a/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs +++ b/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs @@ -128,6 +128,31 @@ afterEach(async () => { }); describe("Phase 2 browser transport hardening", () => { + it("applies the same loopback and origin guards to Phase 3 recovery", async () => { + let recoveryCalls = 0; + const loadRecoveryRunner = async () => ({ + async runPhase3Recovery() { + recoveryCalls += 1; + return { schema: "paperclip.runner.phase3.trace.v1" }; + }, + }); + const { port } = await startServer({ loadRecoveryRunner }); + + const denied = await sendRequest(port, { + path: "/api/phase3/recovery", + origin: "https://attacker.example", + body: JSON.stringify({ fault: "lost-ack" }), + }); + const accepted = await sendRequest(port, { + path: "/api/phase3/recovery", + body: JSON.stringify({ fault: "lost-ack" }), + }); + + expect(denied.status).toBe(403); + expect(accepted.status).toBe(200); + expect(recoveryCalls).toBe(1); + }); + it("denies cross-origin starts and applies the guard before route lookup", async () => { const runner = fakeRunner(); const { port } = await startServer({ loadRunner: () => runner.load() }); diff --git a/packages/paperclip-runner/scripts/record-phase3-evidence.mjs b/packages/paperclip-runner/scripts/record-phase3-evidence.mjs new file mode 100644 index 00000000000..b786983a84b --- /dev/null +++ b/packages/paperclip-runner/scripts/record-phase3-evidence.mjs @@ -0,0 +1,49 @@ +import { mkdirSync, writeFileSync } from "node:fs"; +import { resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { PHASE3_FAULTS, runPhase3Recovery } from "../dist/index.js"; + +const packageRoot = fileURLToPath(new URL("..", import.meta.url)); +const evidenceRoot = resolve(packageRoot, "knowledge/evidence"); +const faultRoot = resolve(evidenceRoot, "phase-03-faults"); +mkdirSync(faultRoot, { recursive: true }); + +const summaries = []; +for (const fault of PHASE3_FAULTS) { + const trace = await runPhase3Recovery({ fault }); + writeFileSync( + resolve(faultRoot, `${fault}.json`), + `${JSON.stringify(trace, null, 2)}\n`, + ); + if (fault === "lost-ack") { + writeFileSync( + resolve(evidenceRoot, "phase-03-recovery-trace.json"), + `${JSON.stringify(trace, null, 2)}\n`, + ); + } + summaries.push({ + fault, + outcome: trace.diagnostics.recovery.outcome, + connectionCount: trace.diagnostics.connection.connectionCount, + replayDeliveries: trace.diagnostics.recovery.replayDeliveries, + runnerRestarts: trace.diagnostics.recovery.runnerRestarts, + harnessRestarts: trace.diagnostics.recovery.harnessRestarts, + commands: trace.diagnostics.commands, + outbox: trace.diagnostics.outbox, + security: trace.diagnostics.security, + assertions: trace.assertions, + }); +} + +writeFileSync( + resolve(evidenceRoot, "phase-03-fault-matrix.json"), + `${JSON.stringify({ schema: "paperclip.runner.phase3.evidence.v1", traces: summaries }, null, 2)}\n`, +); + +const passed = summaries.every((summary) => + Object.values(summary.assertions).every(Boolean), +); +process.stdout.write( + `Recorded ${summaries.length} Phase 3 fault traces; every trace passed: ${passed}.\n`, +); diff --git a/packages/paperclip-runner/src/cli/phase3-recovery.ts b/packages/paperclip-runner/src/cli/phase3-recovery.ts new file mode 100644 index 00000000000..069c387e2cb --- /dev/null +++ b/packages/paperclip-runner/src/cli/phase3-recovery.ts @@ -0,0 +1,105 @@ +#!/usr/bin/env node + +import { mkdir, writeFile } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; + +import { + PHASE3_FAULTS, + type Phase3Fault, + type Phase3RunTrace, +} from "../contracts/phase3.js"; +import { runPhase3Recovery } from "../mock-core/phase3-recovery.js"; + +interface CliOptions { + fault: Phase3Fault; + json: boolean; + output: string | null; + stateDirectory: string | null; + keepState: boolean; +} + +function optionValue(args: readonly string[], name: string): string | null { + const index = args.indexOf(name); + if (index < 0) return null; + const value = args[index + 1]; + if (value === undefined || value.startsWith("--")) { + throw new Error(`${name} requires a value`); + } + return value; +} + +function parseOptions(args: readonly string[]): CliOptions { + const fault = optionValue(args, "--fault") ?? "lost-ack"; + if (!PHASE3_FAULTS.includes(fault as Phase3Fault)) { + throw new Error( + `--fault must be one of: ${PHASE3_FAULTS.join(", ")}`, + ); + } + return { + fault: fault as Phase3Fault, + json: args.includes("--json"), + output: optionValue(args, "--output"), + stateDirectory: optionValue(args, "--state-dir"), + keepState: args.includes("--keep-state"), + }; +} + +function assertionLine(name: string, passed: boolean): string { + return `${passed ? "PASS" : "FAIL"} ${name}`; +} + +export function formatPhase3Diagnostics(trace: Phase3RunTrace): string { + const { diagnostics, assertions } = trace; + const lines = [ + `Phase 3 recovery: ${diagnostics.fault}`, + `Outcome: ${diagnostics.recovery.outcome} (${diagnostics.recovery.reason})`, + `Connection: ${diagnostics.connection.state}; ${diagnostics.connection.connectionCount} connection(s), ${diagnostics.connection.reconnectCount} reconnect(s)`, + `Lease: ${diagnostics.connection.leaseId ?? "none"}; expires ${diagnostics.connection.leaseExpiresAt ?? "not available"}`, + `Identity: runner=${diagnostics.identity.runnerInstanceId} session=${diagnostics.identity.normalizedSessionId} turn=${diagnostics.identity.turnId} item=${diagnostics.identity.itemId}`, + `Cursors: runner ACK ${diagnostics.cursors.runnerAckedSourceSeq}; core ACK ${diagnostics.cursors.coreAckedSourceSeq}; next source ${diagnostics.cursors.runnerNextSourceSeq}`, + `Outbox: ${diagnostics.outbox.events} pending / ${diagnostics.outbox.bytes} bytes; peak ${diagnostics.outbox.peakBytes} / ${diagnostics.outbox.maxBytes}; P0 lost ${diagnostics.outbox.p0Lost}`, + `Commands: ${diagnostics.commands.completed} completed, ${diagnostics.commands.rejected} rejected, ${diagnostics.commands.duplicateDeliveries} duplicate delivery(s), ${diagnostics.commands.logicalEffects} logical effect(s)`, + `Recovery: ${diagnostics.recovery.replayDeliveries} replay delivery(s), ${diagnostics.recovery.runnerRestarts} runner restart(s), ${diagnostics.recovery.harnessRestarts} harness restart(s)`, + `Security: ${diagnostics.security.secretLeakCount} persisted secret leak(s)`, + "Assertions:", + assertionLine("stable identity", assertions.stableIdentity), + assertionLine("source cursor continuity", assertions.sourceCursorContinuous), + assertionLine( + "one logical effect per accepted command", + assertions.oneLogicalEffectPerAcceptedCommand, + ), + assertionLine("no duplicate logical events", assertions.noDuplicateLogicalEvents), + assertionLine("P0 events preserved", assertions.p0Preserved), + assertionLine("bounded storage", assertions.boundedStorage), + assertionLine("secrets redacted", assertions.secretsRedacted), + ]; + return `${lines.join("\n")}\n`; +} + +async function main(): Promise { + const options = parseOptions(process.argv.slice(2)); + const trace = await runPhase3Recovery({ + fault: options.fault, + stateDirectory: + options.stateDirectory === null ? undefined : resolve(options.stateDirectory), + keepState: options.keepState, + }); + if (options.output !== null) { + const output = resolve(options.output); + await mkdir(dirname(output), { recursive: true }); + await writeFile(output, `${JSON.stringify(trace, null, 2)}\n`); + } + process.stdout.write( + options.json ? `${JSON.stringify(trace, null, 2)}\n` : formatPhase3Diagnostics(trace), + ); + if (Object.values(trace.assertions).some((passed) => !passed)) { + process.exitCode = 1; + } +} + +main().catch((error) => { + process.stderr.write( + `phase3-recovery: ${error instanceof Error ? error.message : String(error)}\n`, + ); + process.exitCode = 1; +}); diff --git a/packages/paperclip-runner/src/contracts/phase3.ts b/packages/paperclip-runner/src/contracts/phase3.ts new file mode 100644 index 00000000000..0dd50d00d83 --- /dev/null +++ b/packages/paperclip-runner/src/contracts/phase3.ts @@ -0,0 +1,158 @@ +export const PHASE3_FAULTS = [ + "none", + "socket-drop", + "lost-ack", + "duplicate-command", + "runner-restart", + "harness-restart", + "malformed-input", + "lease-expiry", + "storage-pressure", + "drain", + "revoke", +] as const; + +export type Phase3Fault = (typeof PHASE3_FAULTS)[number]; + +export interface Phase3Identity { + runnerInstanceId: string; + environmentLeaseId: string; + runId: string; + normalizedSessionId: string; + turnId: string; + itemId: string; +} + +export interface Phase3StoredEvent { + sourceSeq: number; + sourceEventId: string; + priority: 0 | 1 | 2; + eventType: string; + itemId?: string; + envelope: Record; + byteSize: number; +} + +export interface Phase3ProcessedCommand { + commandId: string; + controllerSeq: number; + commandDigest: string; + status: "completed" | "failed" | "rejected"; + logicalEffectCount: number; + result: Record; +} + +export interface Phase3RunnerState extends Phase3Identity { + schema: "paperclip.runner.phase3.state.v1"; + lifecycle: + | "connecting" + | "ready" + | "terminal" + | "backpressure" + | "draining" + | "revoked" + | "stopped" + | "recoverable_failure" + | "unrecoverable"; + nextSourceSeq: number; + ackedSourceSeq: number; + lastControllerCommandSeq: number; + reconnectCount: number; + maxOutboxBytes: number; + peakOutboxBytes: number; + outbox: Phase3StoredEvent[]; + processedCommands: Record; + diagnostics: string[]; + backpressure: boolean; + recoverableFailure: string | null; + unrecoverableOutcome: string | null; + harnessGeneration: number; + stopAfterFlush: boolean; +} + +export interface Phase3CoreCommand { + schema: "paperclip.prp.command.v1"; + commandId: string; + controllerSeq: number; + type: string; + issuedAt: string; + payload: Record; + status: "pending" | "completed" | "failed" | "rejected"; + result: Record | null; +} + +export interface Phase3CommittedEvent { + sourceSeq: number; + sourceEventId: string; + eventType: string; + priority: 0 | 1 | 2; + envelope: Record; + deliveryCount: number; + logicalEffectCount: number; +} + +export interface Phase3Diagnostics { + schema: "paperclip.runner.phase3.diagnostics.v1"; + fault: Phase3Fault; + connection: { + state: string; + connectionCount: number; + reconnectCount: number; + leaseId: string | null; + leaseExpiresAt: string | null; + }; + identity: Phase3Identity; + cursors: { + runnerAckedSourceSeq: number; + runnerNextSourceSeq: number; + coreAckedSourceSeq: number; + highestCommittedSourceSeq: number; + }; + outbox: { + events: number; + bytes: number; + peakBytes: number; + maxBytes: number; + backpressure: boolean; + p0Committed: number; + p0Lost: number; + }; + commands: { + issued: number; + completed: number; + rejected: number; + logicalEffects: number; + duplicateDeliveries: number; + }; + recovery: { + replayDeliveries: number; + runnerRestarts: number; + harnessRestarts: number; + malformedFrames: number; + freshBootstraps: number; + outcome: "recovered" | "drained" | "revoked" | "unrecoverable"; + reason: string; + }; + security: { + bootstrapTicketPersisted: boolean; + connectionLeaseTokenPersisted: boolean; + secretLeakCount: number; + }; + committedEvents: Phase3CommittedEvent[]; +} + +export interface Phase3RunTrace { + schema: "paperclip.runner.phase3.trace.v1"; + diagnostics: Phase3Diagnostics; + runnerState: Phase3RunnerState; + commands: Phase3CoreCommand[]; + assertions: { + stableIdentity: boolean; + sourceCursorContinuous: boolean; + oneLogicalEffectPerAcceptedCommand: boolean; + noDuplicateLogicalEvents: boolean; + p0Preserved: boolean; + boundedStorage: boolean; + secretsRedacted: boolean; + }; +} diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index ce3ed5a2459..b1c2f1c5d06 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -2,9 +2,11 @@ export * from "./contracts/control-plane-port.js"; export * from "./contracts/harness-driver.js"; export * from "./contracts/native-session-backend.js"; export * from "./contracts/phase2.js"; +export * from "./contracts/phase3.js"; export * from "./contracts/types.js"; export * from "./mock-core/mock-control-plane-adapter.js"; export * from "./mock-core/phase2-local-runner.js"; +export * from "./mock-core/phase3-recovery.js"; export * from "./protocol/phase0-fixture.js"; export * from "./protocol/phase1-contract.js"; export * from "./protocol/phase1-loader.js"; diff --git a/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts b/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts new file mode 100644 index 00000000000..cd2731a8c37 --- /dev/null +++ b/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts @@ -0,0 +1,379 @@ +import { + chmodSync, + lstatSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + symlinkSync, + writeFileSync, +} from "node:fs"; +import { connect, type Socket } from "node:net"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { + Phase3MockCore, + phase3Internals, + runPhase3Recovery, +} from "./phase3-recovery.js"; + +async function upgradeStatus(url: string, ticket: string): Promise { + const parsed = new URL(url); + return new Promise((resolveStatus, rejectStatus) => { + const socket = connect(Number(parsed.port), parsed.hostname); + let response = ""; + socket.once("error", rejectStatus); + socket.once("connect", () => { + socket.write( + [ + `GET ${parsed.pathname} HTTP/1.1`, + `Host: ${parsed.host}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + `Authorization: Bearer ${ticket}`, + "\r\n", + ].join("\r\n"), + ); + }); + socket.on("data", (chunk) => { + response += chunk.toString("utf8"); + if (response.includes("\r\n\r\n")) { + const status = Number(response.match(/^HTTP\/1\.1 (\d{3})/)?.[1]); + socket.destroy(); + resolveStatus(status); + } + }); + }); +} + +async function upgradeSocket(url: string, ticket: string): Promise { + const parsed = new URL(url); + return new Promise((resolveSocket, rejectSocket) => { + const socket = connect(Number(parsed.port), parsed.hostname); + let response = Buffer.alloc(0); + const onError = (error: Error): void => rejectSocket(error); + const onData = (chunk: Buffer): void => { + response = Buffer.concat([response, chunk]); + const boundary = response.indexOf("\r\n\r\n"); + if (boundary === -1) return; + socket.off("error", onError); + socket.off("data", onData); + const status = Number(response.toString("utf8").match(/^HTTP\/1\.1 (\d{3})/)?.[1]); + if (status !== 101) { + socket.destroy(); + rejectSocket(new Error(`WebSocket upgrade returned ${String(status)}`)); + return; + } + resolveSocket(socket); + }; + socket.once("error", onError); + socket.once("connect", () => { + socket.write( + [ + `GET ${parsed.pathname} HTTP/1.1`, + `Host: ${parsed.host}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + `Authorization: Bearer ${ticket}`, + "\r\n", + ].join("\r\n"), + ); + }); + socket.on("data", onData); + }); +} + +function sendMaskedJson(socket: Socket, value: unknown): void { + const payload = Buffer.from(JSON.stringify(value)); + const mask = Buffer.from([0x11, 0x22, 0x33, 0x44]); + const header: number[] = [0x81]; + if (payload.length <= 125) { + header.push(0x80 | payload.length); + } else { + header.push(0x80 | 126, (payload.length >>> 8) & 0xff, payload.length & 0xff); + } + const masked = Buffer.from(payload); + for (let index = 0; index < masked.length; index += 1) { + masked[index] = masked[index]! ^ mask[index % mask.length]!; + } + socket.write(Buffer.concat([Buffer.from(header), mask, masked])); +} + +async function expectSocketClosed(socket: Socket): Promise { + await new Promise((resolveClose, rejectClose) => { + const timer = setTimeout(() => rejectClose(new Error("WebSocket was not rejected.")), 2_000); + socket.once("close", () => { + clearTimeout(timer); + resolveClose(); + }); + }); +} + +function hello(identity: ReturnType): Record { + return { + protocol: "paperclip.runner", + version: 1, + envelopeId: "hello_auth_regression", + kind: "hello", + runnerInstanceId: identity.runnerInstanceId, + payload: { + protocolMin: 1, + protocolMax: 1, + runnerVersion: "0.3.0", + runnerDigest: "sha256:phase3-approved", + environmentLeaseId: identity.environmentLeaseId, + }, + }; +} + +describe.sequential("Phase 3 durable transport and recovery", () => { + it("accepts a short-lived bootstrap ticket once and rejects its reuse", async () => { + const scratch = + process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? + process.env.PAPERCLIP_SCRATCH_DIR ?? + tmpdir(); + const root = mkdtempSync(resolve(scratch, "phase3-ticket-test-")); + const core = new Phase3MockCore({ + stateDirectory: root, + identity: phase3Internals.phase3Identity(), + fault: "none", + }); + try { + await core.start(); + const ticket = core.issueBootstrapTicket(); + + await expect(upgradeStatus(core.connectUrl, ticket)).resolves.toBe(101); + await expect(upgradeStatus(core.connectUrl, ticket)).resolves.toBe(401); + const expiredTicket = core.issueBootstrapTicket(-1); + await expect(upgradeStatus(core.connectUrl, expiredTicket)).resolves.toBe(401); + } finally { + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + it("rejects bootstrap and resumed lease credentials bound to a different runner identity", async () => { + const scratch = + process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? + process.env.PAPERCLIP_SCRATCH_DIR ?? + tmpdir(); + const root = mkdtempSync(resolve(scratch, "phase3-identity-auth-test-")); + const identity = phase3Internals.phase3Identity(); + const core = new Phase3MockCore({ + stateDirectory: root, + identity, + fault: "none", + }); + try { + await core.start(); + + const bootstrap = core.issueBootstrapTicket(); + const bootstrapSocket = await upgradeSocket(core.connectUrl, bootstrap); + sendMaskedJson(bootstrapSocket, { + ...hello(identity), + runnerInstanceId: "runner_wrong_identity", + }); + await expectSocketClosed(bootstrapSocket); + + const leaseToken = "lease_wrong_identity_regression"; + const digest = phase3Internals.tokenDigest(leaseToken); + core.store.state.leases[digest] = { + digest, + leaseId: "lease_wrong_identity", + runnerInstanceId: "runner_other_identity", + expiresAt: new Date(Date.now() + 30_000).toISOString(), + revokedAt: null, + }; + core.store.save(); + const leaseSocket = await upgradeSocket(core.connectUrl, leaseToken); + sendMaskedJson(leaseSocket, hello(identity)); + await expectSocketClosed(leaseSocket); + } finally { + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + it.runIf(process.platform !== "win32")( + "atomically stores mock-core state without following a preplanted predictable symlink", + () => { + const scratch = + process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? + process.env.PAPERCLIP_SCRATCH_DIR ?? + tmpdir(); + const root = mkdtempSync(resolve(scratch, "phase3-core-state-symlink-")); + const stateDirectory = resolve(root, "mock-core"); + const victim = resolve(root, "victim.txt"); + mkdirSync(stateDirectory, { mode: 0o700 }); + chmodSync(stateDirectory, 0o700); + writeFileSync(victim, "unchanged", { mode: 0o600 }); + const planted = resolve(stateDirectory, "mock-core-state.json.next"); + symlinkSync(victim, planted); + try { + const core = new Phase3MockCore({ + stateDirectory, + identity: phase3Internals.phase3Identity(), + fault: "none", + }); + core.issueBootstrapTicket(); + const statePath = resolve(stateDirectory, "mock-core-state.json"); + + expect(readFileSync(victim, "utf8")).toBe("unchanged"); + expect(lstatSync(planted).isSymbolicLink()).toBe(true); + expect(lstatSync(stateDirectory).mode & 0o777).toBe(0o700); + expect(lstatSync(statePath).mode & 0o777).toBe(0o600); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + ); + + it.each(["socket-drop", "lost-ack", "malformed-input"] as const)( + "recovers the %s reconnect path with stable identities and cursors", + async (fault) => { + const trace = await runPhase3Recovery({ fault }); + + expect(trace.diagnostics.recovery.outcome).toBe("recovered"); + expect(trace.diagnostics.connection.connectionCount).toBeGreaterThan(1); + expect(trace.assertions.stableIdentity).toBe(true); + expect(trace.assertions.sourceCursorContinuous).toBe(true); + expect(trace.assertions.noDuplicateLogicalEvents).toBe(true); + expect(trace.assertions.p0Preserved).toBe(true); + expect(trace.assertions.secretsRedacted).toBe(true); + expect(trace.diagnostics.security).toMatchObject({ + bootstrapTicketPersisted: false, + connectionLeaseTokenPersisted: false, + secretLeakCount: 0, + }); + if (fault === "lost-ack") { + expect(trace.diagnostics.recovery.replayDeliveries).toBeGreaterThan(0); + expect( + trace.diagnostics.committedEvents.some((event) => event.deliveryCount > 1), + ).toBe(true); + } + if (fault === "malformed-input") { + expect(trace.diagnostics.recovery.malformedFrames).toBe(1); + expect(trace.runnerState.diagnostics.join(" ")).toContain( + "malformed WebSocket JSON", + ); + } + }, + ); + + it("replays a processed command without repeating its logical effect", async () => { + const trace = await runPhase3Recovery({ fault: "duplicate-command" }); + + expect(trace.diagnostics.commands.duplicateDeliveries).toBeGreaterThan(0); + expect(trace.assertions.oneLogicalEffectPerAcceptedCommand).toBe(true); + expect(trace.runnerState.processedCommands.command_phase3_001?.logicalEffectCount).toBe(1); + expect( + trace.diagnostics.committedEvents.filter( + (event) => event.eventType === "workspace.ready", + ), + ).toHaveLength(1); + }); + + it("restores the same runner session after a real runner process restart", async () => { + const trace = await runPhase3Recovery({ fault: "runner-restart" }); + + expect(trace.diagnostics.recovery.runnerRestarts).toBe(1); + expect(trace.diagnostics.recovery.freshBootstraps).toBe(2); + expect(trace.diagnostics.recovery.replayDeliveries).toBeGreaterThan(0); + expect( + trace.diagnostics.committedEvents.some( + (event) => event.eventType === "runner.reconciled", + ), + ).toBe(true); + expect(trace.assertions.stableIdentity).toBe(true); + expect(trace.assertions.sourceCursorContinuous).toBe(true); + }); + + it("reconciles a restarted harness without replacing session, turn, or item IDs", async () => { + const trace = await runPhase3Recovery({ fault: "harness-restart" }); + const reconciled = trace.diagnostics.committedEvents.find( + (event) => event.eventType === "session.reconciled", + ); + + expect(trace.diagnostics.recovery.harnessRestarts).toBe(1); + expect(reconciled).toBeDefined(); + expect(reconciled?.envelope).toMatchObject({ + runId: "run_phase3_stable", + normalizedSessionId: "session_phase3_stable", + turnId: "turn_phase3_stable", + }); + expect(trace.assertions.stableIdentity).toBe(true); + }); + + it("uses a fresh bootstrap after lease expiry and resumes the durable cursor", async () => { + const trace = await runPhase3Recovery({ fault: "lease-expiry" }); + + expect(trace.diagnostics.recovery.runnerRestarts).toBe(1); + expect(trace.diagnostics.recovery.freshBootstraps).toBe(2); + expect(trace.assertions.sourceCursorContinuous).toBe(true); + expect(trace.diagnostics.recovery.outcome).toBe("recovered"); + }); + + it("bounds storage, coalesces P2, preserves P0, and rejects new turns", async () => { + const trace = await runPhase3Recovery({ fault: "storage-pressure" }); + const deltaEvents = trace.diagnostics.committedEvents.filter( + (event) => event.eventType === "item.delta", + ); + const turn = trace.commands.find((command) => command.type === "turn.start"); + + expect(trace.diagnostics.outbox.backpressure).toBe(true); + expect(trace.diagnostics.outbox.peakBytes).toBeLessThanOrEqual( + trace.diagnostics.outbox.maxBytes, + ); + expect(deltaEvents).toHaveLength(1); + expect(trace.diagnostics.committedEvents.some((event) => event.eventType === "runner.backpressure")) + .toBe(true); + expect(turn?.status).toBe("rejected"); + expect(trace.assertions.boundedStorage).toBe(true); + expect(trace.assertions.p0Preserved).toBe(true); + }); + + it.each(["drain", "revoke"] as const)( + "%s stops new work only after preserving the durable event cursor", + async (fault) => { + const trace = await runPhase3Recovery({ fault }); + + expect(trace.diagnostics.recovery.outcome).toBe(fault === "drain" ? "drained" : "revoked"); + expect(trace.diagnostics.outbox.events).toBe(0); + expect(trace.assertions.sourceCursorContinuous).toBe(true); + expect(trace.assertions.p0Preserved).toBe(true); + if (fault === "drain") { + expect(trace.commands.find((command) => command.type === "turn.start")?.status) + .toBe("rejected"); + } else { + const afterRevoke = trace.commands.find( + (command) => command.commandId === "command_after_revoke", + ); + expect(trace.runnerState.lifecycle).toBe("revoked"); + expect(trace.diagnostics.recovery.replayDeliveries).toBeGreaterThan(0); + expect(afterRevoke?.status).toBe("rejected"); + expect(afterRevoke?.result?.logicalEffectCount).toBe(0); + expect( + trace.diagnostics.committedEvents.filter( + (event) => event.eventType === "turn.accepted", + ), + ).toHaveLength(1); + } + }, + ); + + it("passes only the bootstrap capability and platform basics to the runner", () => { + const environment = phase3Internals.runnerEnvironment("opaque-ticket"); + + expect(environment.PAPERCLIP_RUNNER_BOOTSTRAP_TICKET).toBe("opaque-ticket"); + expect(environment.PAPERCLIP_API_KEY).toBeUndefined(); + expect(environment.OPENAI_API_KEY).toBeUndefined(); + expect(environment.EMAIL_AGENTMAIL_GENERAL_API_KEY).toBeUndefined(); + }); +}); diff --git a/packages/paperclip-runner/src/mock-core/phase3-recovery.ts b/packages/paperclip-runner/src/mock-core/phase3-recovery.ts new file mode 100644 index 00000000000..62d1594dc44 --- /dev/null +++ b/packages/paperclip-runner/src/mock-core/phase3-recovery.ts @@ -0,0 +1,1255 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import { + chmodSync, + closeSync, + constants, + fchmodSync, + fstatSync, + fsyncSync, + lstatSync, + mkdirSync, + mkdtempSync, + openSync, + readFileSync, + renameSync, + rmSync, + unlinkSync, + writeFileSync, + type Stats, +} from "node:fs"; +import { createServer, type IncomingMessage, type Server } from "node:http"; +import { tmpdir } from "node:os"; +import { dirname, resolve } from "node:path"; +import type { Duplex } from "node:stream"; +import { fileURLToPath } from "node:url"; + +import { + PHASE3_FAULTS, + type Phase3CommittedEvent, + type Phase3CoreCommand, + type Phase3Diagnostics, + type Phase3Fault, + type Phase3Identity, + type Phase3RunnerState, + type Phase3RunTrace, +} from "../contracts/phase3.js"; + +const packageRoot = fileURLToPath(new URL("../..", import.meta.url)); +const executableSuffix = process.platform === "win32" ? ".exe" : ""; +const runnerBinary = resolve( + packageRoot, + `runner/target/debug/paperclip-runnerd${executableSuffix}`, +); +const fakeHarnessBinary = resolve( + packageRoot, + `runner/target/debug/fake-harness${executableSuffix}`, +); +const fakeHarnessScript = resolve( + packageRoot, + "protocol/fixtures/phase-02/scripts/happy-path.json", +); +const protocol = "paperclip.runner"; +const protocolVersion = 1; +const websocketGuid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; +const coreStateSchema = "paperclip.runner.phase3.mock-core-state.v1"; +const maxFrameBytes = 1024 * 1024; + +interface BootstrapTicketRecord { + digest: string; + runnerInstanceId: string; + expiresAt: string; + usedAt: string | null; +} + +interface ConnectionLeaseRecord { + digest: string; + leaseId: string; + runnerInstanceId: string; + expiresAt: string; + revokedAt: string | null; +} + +interface StoredCoreState { + schema: typeof coreStateSchema; + identity: Phase3Identity; + tickets: Record; + leases: Record; + commands: Phase3CoreCommand[]; + committedEvents: Phase3CommittedEvent[]; + ackedSourceSeq: number; + connectionCount: number; + commandDeliveryCounts: Record; + replayDeliveries: number; + duplicateCommandResults: number; + freshBootstraps: number; + malformedFrames: number; + lastLeaseId: string | null; + lastLeaseExpiresAt: string | null; +} + +type Authorization = + | { + kind: "bootstrap"; + digest: string; + rawToken: string; + ticket: BootstrapTicketRecord; + } + | { + kind: "lease"; + digest: string; + rawToken: string; + lease: ConnectionLeaseRecord; + }; + +interface MockCoreOptions { + stateDirectory: string; + identity: Phase3Identity; + fault: Phase3Fault; + expectedRunnerVersion?: string; + expectedRunnerDigest?: string; +} + +interface RunnerProcessResult { + code: number | null; + signal: NodeJS.Signals | null; + stdout: string; + stderr: string; +} + +interface RunnerProcessHandle { + child: ChildProcessWithoutNullStreams; + completion: Promise; +} + +function tokenDigest(token: string): string { + return `sha256:${createHash("sha256").update(token).digest("hex")}`; +} + +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map(canonicalJson).join(",")}]`; + } + if (typeof value === "object" && value !== null) { + const object = value as Record; + return `{${Object.keys(object) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalJson(object[key])}`) + .join(",")}}`; + } + return JSON.stringify(value); +} + +function safeDate(offsetMs: number): string { + return new Date(Date.now() + offsetMs).toISOString(); +} + +function initialCoreState(identity: Phase3Identity): StoredCoreState { + return { + schema: coreStateSchema, + identity, + tickets: {}, + leases: {}, + commands: [], + committedEvents: [], + ackedSourceSeq: 0, + connectionCount: 0, + commandDeliveryCounts: {}, + replayDeliveries: 0, + duplicateCommandResults: 0, + freshBootstraps: 0, + malformedFrames: 0, + lastLeaseId: null, + lastLeaseExpiresAt: null, + }; +} + +function isNodeError(error: unknown, code: string): boolean { + return error instanceof Error && "code" in error && error.code === code; +} + +function verifyPrivateDirectory(path: string): void { + const metadata = lstatSync(path); + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error(`Private state directory is not a real directory: ${path}`); + } + if (process.platform !== "win32") { + if ((metadata.mode & 0o777) !== 0o700) { + throw new Error(`Private state directory does not use mode 0700: ${path}`); + } + if (process.geteuid !== undefined && metadata.uid !== process.geteuid()) { + throw new Error(`Private state directory is not owned by the daemon user: ${path}`); + } + } +} + +function verifyPrivateRegularFile(file: Stats, path: string): void { + if (!file.isFile()) { + throw new Error(`Private state path is not a regular file: ${path}`); + } + if (process.platform !== "win32") { + if ((file.mode & 0o777) !== 0o600) { + throw new Error(`Private state file does not use mode 0600: ${path}`); + } + if (process.geteuid !== undefined && file.uid !== process.geteuid()) { + throw new Error(`Private state file is not owned by the daemon user: ${path}`); + } + } +} + +function readPrivateFile(path: string): string | null { + let descriptor: number; + try { + descriptor = openSync(path, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)); + } catch (error) { + if (isNodeError(error, "ENOENT")) return null; + throw error; + } + try { + verifyPrivateRegularFile(fstatSync(descriptor), path); + return readFileSync(descriptor, "utf8"); + } finally { + closeSync(descriptor); + } +} + +function syncParentDirectory(path: string): void { + if (process.platform === "win32") return; + const descriptor = openSync( + dirname(path), + constants.O_RDONLY | (constants.O_DIRECTORY ?? 0) | (constants.O_NOFOLLOW ?? 0), + ); + try { + fsyncSync(descriptor); + } finally { + closeSync(descriptor); + } +} + +function atomicPrivateWrite(path: string, contents: string): void { + const temporary = resolve(dirname(path), `.${path.split(/[\\/]/).at(-1)}.${randomUUID()}.tmp`); + let descriptor: number | null = null; + let created = false; + try { + descriptor = openSync( + temporary, + constants.O_WRONLY | + constants.O_CREAT | + constants.O_EXCL | + (constants.O_NOFOLLOW ?? 0), + 0o600, + ); + created = true; + if (process.platform !== "win32") fchmodSync(descriptor, 0o600); + verifyPrivateRegularFile(fstatSync(descriptor), temporary); + writeFileSync(descriptor, contents, "utf8"); + fsyncSync(descriptor); + closeSync(descriptor); + descriptor = null; + renameSync(temporary, path); + created = false; + syncParentDirectory(path); + } finally { + if (descriptor !== null) closeSync(descriptor); + if (created) { + try { + unlinkSync(temporary); + } catch (error) { + if (!isNodeError(error, "ENOENT")) throw error; + } + } + } +} + +class DurableCoreStore { + readonly path: string; + #state: StoredCoreState; + + constructor(directory: string, identity: Phase3Identity) { + try { + const metadata = lstatSync(directory); + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error(`Private state directory is not a real directory: ${directory}`); + } + } catch (error) { + if (!isNodeError(error, "ENOENT")) throw error; + mkdirSync(directory, { recursive: true, mode: 0o700 }); + } + if (process.platform !== "win32") chmodSync(directory, 0o700); + verifyPrivateDirectory(directory); + this.path = resolve(directory, "mock-core-state.json"); + const stored = readPrivateFile(this.path); + if (stored !== null) { + this.#state = JSON.parse(stored) as StoredCoreState; + if ( + this.#state.schema !== coreStateSchema || + canonicalJson(this.#state.identity) !== canonicalJson(identity) + ) { + throw new Error("Mock core state does not match the requested Phase 3 identity."); + } + } else { + this.#state = initialCoreState(identity); + this.save(); + } + } + + get state(): StoredCoreState { + return this.#state; + } + + save(): void { + atomicPrivateWrite(this.path, `${JSON.stringify(this.#state, null, 2)}\n`); + } +} + +class MockWebSocketConnection { + readonly socket: Duplex; + readonly authorization: Authorization; + lease: ConnectionLeaseRecord | null = null; + #buffer = Buffer.alloc(0); + #closed = false; + #onText: (text: string) => void; + #onClose: () => void; + + constructor( + socket: Duplex, + authorization: Authorization, + onText: (text: string) => void, + onClose: () => void, + ) { + this.socket = socket; + this.authorization = authorization; + this.#onText = onText; + this.#onClose = onClose; + socket.on("data", (chunk: Buffer) => this.#consume(chunk)); + socket.on("close", () => { + if (!this.#closed) { + this.#closed = true; + this.#onClose(); + } + }); + socket.on("error", () => this.close()); + } + + sendJson(value: unknown): void { + this.sendText(JSON.stringify(value)); + } + + sendText(text: string): void { + if (this.#closed) { + return; + } + const payload = Buffer.from(text); + const header: number[] = [0x81]; + if (payload.length <= 125) { + header.push(payload.length); + } else if (payload.length <= 0xffff) { + header.push(126, (payload.length >>> 8) & 0xff, payload.length & 0xff); + } else { + const length = BigInt(payload.length); + header.push(127); + for (let shift = 56n; shift >= 0n; shift -= 8n) { + header.push(Number((length >> shift) & 0xffn)); + } + } + this.socket.write(Buffer.concat([Buffer.from(header), payload])); + } + + close(): void { + if (this.#closed) { + return; + } + this.#closed = true; + this.socket.destroy(); + this.#onClose(); + } + + #consume(chunk: Buffer): void { + this.#buffer = Buffer.concat([this.#buffer, chunk]); + while (this.#buffer.length >= 2) { + const first = this.#buffer[0]!; + const second = this.#buffer[1]!; + const opcode = first & 0x0f; + const masked = (second & 0x80) !== 0; + let length = second & 0x7f; + let cursor = 2; + if (length === 126) { + if (this.#buffer.length < 4) return; + length = this.#buffer.readUInt16BE(2); + cursor = 4; + } else if (length === 127) { + if (this.#buffer.length < 10) return; + const extended = this.#buffer.readBigUInt64BE(2); + if (extended > BigInt(maxFrameBytes)) { + this.close(); + return; + } + length = Number(extended); + cursor = 10; + } + if (length > maxFrameBytes || !masked) { + this.close(); + return; + } + if (this.#buffer.length < cursor + 4 + length) return; + const mask = this.#buffer.subarray(cursor, cursor + 4); + cursor += 4; + const payload = Buffer.from(this.#buffer.subarray(cursor, cursor + length)); + this.#buffer = this.#buffer.subarray(cursor + length); + for (let index = 0; index < payload.length; index += 1) { + payload[index] = payload[index]! ^ mask[index % 4]!; + } + if (opcode === 0x1) { + this.#onText(payload.toString("utf8")); + } else if (opcode === 0x8) { + this.close(); + return; + } else if (opcode === 0x9) { + this.#sendControl(0x0a, payload); + } else if (opcode !== 0x0a) { + this.close(); + return; + } + } + } + + #sendControl(opcode: number, payload: Buffer): void { + if (payload.length > 125 || this.#closed) return; + this.socket.write(Buffer.concat([Buffer.from([0x80 | opcode, payload.length]), payload])); + } +} + +export class Phase3MockCore { + readonly fault: Phase3Fault; + readonly identity: Phase3Identity; + readonly store: DurableCoreStore; + #expectedRunnerVersion: string; + #expectedRunnerDigest: string; + #server: Server | null = null; + #connections = new Set(); + #port: number | null = null; + #faultTriggered = false; + #replayCursorOverrideOnce = false; + #faultTriggerResolve!: () => void; + #faultTrigger = new Promise((resolveFault) => { + this.#faultTriggerResolve = resolveFault; + }); + + constructor(options: MockCoreOptions) { + if (!PHASE3_FAULTS.includes(options.fault)) { + throw new Error(`Unsupported Phase 3 fault: ${options.fault}`); + } + this.fault = options.fault; + this.identity = options.identity; + this.store = new DurableCoreStore(options.stateDirectory, options.identity); + this.#expectedRunnerVersion = options.expectedRunnerVersion ?? "0.3.0"; + this.#expectedRunnerDigest = + options.expectedRunnerDigest ?? "sha256:phase3-approved"; + } + + get connectUrl(): string { + if (this.#port === null) { + throw new Error("Phase 3 mock core is not listening."); + } + return `ws://127.0.0.1:${this.#port}/phase3/connect`; + } + + async start(port = 0): Promise { + if (this.#server !== null) { + throw new Error("Phase 3 mock core is already running."); + } + const server = createServer((_request, response) => { + response.writeHead(404).end(); + }); + this.#server = server; + server.on("upgrade", (request, socket) => this.#upgrade(request, socket)); + await new Promise((resolveListen, rejectListen) => { + server.once("error", rejectListen); + server.listen(port, "127.0.0.1", () => { + server.off("error", rejectListen); + resolveListen(); + }); + }); + const address = server.address(); + if (address === null || typeof address === "string") { + throw new Error("Phase 3 mock core did not bind a TCP port."); + } + this.#port = address.port; + } + + async stop(): Promise { + for (const connection of this.#connections) { + connection.close(); + } + this.#connections.clear(); + const server = this.#server; + this.#server = null; + this.#port = null; + if (server !== null) { + await new Promise((resolveClose) => server.close(() => resolveClose())); + } + } + + issueBootstrapTicket(ttlMs = 5_000): string { + const ticket = `bootstrap_${randomUUID()}`; + const digest = tokenDigest(ticket); + this.store.state.tickets[digest] = { + digest, + runnerInstanceId: this.identity.runnerInstanceId, + expiresAt: safeDate(ttlMs), + usedAt: null, + }; + this.store.state.freshBootstraps += 1; + this.store.save(); + return ticket; + } + + queueCommand( + type: string, + payload: Record = {}, + commandId?: string, + ): Phase3CoreCommand { + const controllerSeq = this.store.state.commands.length + 1; + const command: Phase3CoreCommand = { + schema: "paperclip.prp.command.v1", + commandId: commandId ?? `command_phase3_${controllerSeq.toString().padStart(3, "0")}`, + controllerSeq, + type, + issuedAt: `2026-08-07T23:30:${controllerSeq.toString().padStart(2, "0")}.000Z`, + payload, + status: "pending", + result: null, + }; + this.store.state.commands.push(command); + this.store.save(); + return command; + } + + waitForFaultTrigger(): Promise { + return this.#faultTrigger; + } + + #triggerFault(): void { + if (!this.#faultTriggered) { + this.#faultTriggered = true; + this.#faultTriggerResolve(); + } + } + + #upgrade(request: IncomingMessage, socket: Duplex): void { + if (request.url !== "/phase3/connect") { + socket.write("HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return; + } + const authorization = this.#authorize(request.headers.authorization); + if (authorization === null) { + socket.write("HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return; + } + const websocketKey = request.headers["sec-websocket-key"]; + if (typeof websocketKey !== "string") { + socket.write("HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return; + } + const accept = createHash("sha1") + .update(`${websocketKey}${websocketGuid}`) + .digest("base64"); + socket.write( + [ + "HTTP/1.1 101 Switching Protocols", + "Upgrade: websocket", + "Connection: Upgrade", + `Sec-WebSocket-Accept: ${accept}`, + "\r\n", + ].join("\r\n"), + ); + const connection = new MockWebSocketConnection( + socket, + authorization, + (text) => this.#handleText(connection, text), + () => this.#connections.delete(connection), + ); + this.#connections.add(connection); + } + + #authorize(header: string | undefined): Authorization | null { + const rawToken = header?.match(/^Bearer (.+)$/i)?.[1]; + if (rawToken === undefined) return null; + const digest = tokenDigest(rawToken); + const ticket = this.store.state.tickets[digest]; + if (ticket !== undefined && ticket.usedAt === null && Date.parse(ticket.expiresAt) > Date.now()) { + ticket.usedAt = new Date().toISOString(); + this.store.save(); + return { kind: "bootstrap", digest, rawToken, ticket }; + } + const lease = this.store.state.leases[digest]; + if ( + lease !== undefined && + lease.revokedAt === null && + Date.parse(lease.expiresAt) > Date.now() + ) { + return { kind: "lease", digest, rawToken, lease }; + } + return null; + } + + #handleText(connection: MockWebSocketConnection, text: string): void { + let envelope: Record; + try { + envelope = JSON.parse(text) as Record; + } catch { + this.store.state.malformedFrames += 1; + this.store.save(); + connection.close(); + return; + } + if (envelope.protocol !== protocol || envelope.version !== protocolVersion) { + connection.close(); + return; + } + const kind = envelope.kind; + if (kind === "hello") { + this.#hello(connection, envelope); + return; + } + if (connection.lease === null) { + connection.close(); + return; + } + if (kind === "event") { + this.#event(connection, envelope); + return; + } + if (kind === "command_result") { + this.#commandResult(connection, envelope); + return; + } + if (kind !== "pong") { + connection.close(); + } + } + + #hello(connection: MockWebSocketConnection, envelope: Record): void { + const payload = envelope.payload as Record | undefined; + const authorizedRunnerInstanceId = + connection.authorization.kind === "bootstrap" + ? connection.authorization.ticket.runnerInstanceId + : connection.authorization.lease.runnerInstanceId; + if ( + envelope.runnerInstanceId !== this.identity.runnerInstanceId || + envelope.runnerInstanceId !== authorizedRunnerInstanceId || + payload?.environmentLeaseId !== this.identity.environmentLeaseId || + payload?.runnerVersion !== this.#expectedRunnerVersion || + payload?.runnerDigest !== this.#expectedRunnerDigest || + payload?.protocolMin !== 1 || + payload.protocolMax !== 1 + ) { + connection.close(); + return; + } + + let leaseToken = connection.authorization.rawToken; + let lease: ConnectionLeaseRecord; + if (connection.authorization.kind === "bootstrap") { + leaseToken = `lease_${randomUUID()}`; + const digest = tokenDigest(leaseToken); + lease = { + digest, + leaseId: `connection_lease_${randomUUID()}`, + runnerInstanceId: this.identity.runnerInstanceId, + expiresAt: safeDate(this.fault === "lease-expiry" && !this.#faultTriggered ? 50 : 30_000), + revokedAt: null, + }; + this.store.state.leases[digest] = lease; + } else { + lease = connection.authorization.lease; + } + connection.lease = lease; + this.store.state.connectionCount += 1; + this.store.state.lastLeaseId = lease.leaseId; + this.store.state.lastLeaseExpiresAt = lease.expiresAt; + + const reportedAck = this.#replayCursorOverrideOnce + ? Math.max(0, this.store.state.ackedSourceSeq - 1) + : this.store.state.ackedSourceSeq; + this.#replayCursorOverrideOnce = false; + const suppressPending = + !this.#faultTriggered && + ["socket-drop", "malformed-input", "lease-expiry"].includes(this.fault); + const pending = suppressPending ? [] : this.#nextPendingCommand(); + for (const command of pending) { + this.store.state.commandDeliveryCounts[command.commandId] = + (this.store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; + } + this.store.save(); + connection.sendJson({ + protocol, + version: protocolVersion, + envelopeId: `welcome_${this.store.state.connectionCount}`, + kind: "welcome", + runnerInstanceId: this.identity.runnerInstanceId, + connectionId: `connection_${this.store.state.connectionCount}`, + sentAt: new Date().toISOString(), + payload: { + selectedVersion: 1, + heartbeatIntervalMs: 250, + connectionLeaseId: lease.leaseId, + connectionLeaseToken: leaseToken, + connectionLeaseExpiresAt: lease.expiresAt, + maxFrameBytes, + maxBatchEvents: 100, + ackedSourceSeq: reportedAck, + pendingCommands: pending.map(this.#wireCommand), + }, + }); + + if (suppressPending) { + this.#triggerFault(); + if (this.fault === "malformed-input") { + this.store.state.malformedFrames += 1; + this.store.save(); + connection.sendText('{"kind":'); + } + if (this.fault === "lease-expiry") { + lease.expiresAt = safeDate(-1); + this.store.state.lastLeaseExpiresAt = lease.expiresAt; + this.store.save(); + } + setTimeout(() => connection.close(), 5); + } + } + + #wireCommand(command: Phase3CoreCommand): Omit { + const { status: _status, result: _result, ...wire } = command; + return wire; + } + + #nextPendingCommand(): Phase3CoreCommand[] { + const command = this.store.state.commands.find((candidate) => candidate.status === "pending"); + return command === undefined ? [] : [command]; + } + + #sendNextCommand(connection: MockWebSocketConnection): void { + const [command] = this.#nextPendingCommand(); + if (command === undefined) return; + this.store.state.commandDeliveryCounts[command.commandId] = + (this.store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; + this.store.save(); + connection.sendJson({ + protocol, + version: protocolVersion, + envelopeId: `command_${command.commandId}_${this.store.state.commandDeliveryCounts[command.commandId]}`, + kind: "command", + runnerInstanceId: this.identity.runnerInstanceId, + runId: this.identity.runId, + normalizedSessionId: this.identity.normalizedSessionId, + sentAt: new Date().toISOString(), + payload: this.#wireCommand(command), + }); + } + + #commandResult( + connection: MockWebSocketConnection, + envelope: Record, + ): void { + const result = envelope.payload as Record | undefined; + const commandId = result?.commandId; + if (result === undefined || typeof commandId !== "string") { + connection.close(); + return; + } + const command = this.store.state.commands.find((candidate) => candidate.commandId === commandId); + if (command === undefined) { + connection.close(); + return; + } + if (this.fault === "duplicate-command" && !this.#faultTriggered) { + this.#triggerFault(); + connection.close(); + return; + } + if (command.status !== "pending") { + this.store.state.duplicateCommandResults += 1; + } + const status = result.status; + command.status = + status === "completed" || status === "failed" || status === "rejected" + ? status + : "failed"; + command.result = structuredClone(result); + this.store.save(); + this.#sendNextCommand(connection); + } + + #event(connection: MockWebSocketConnection, envelope: Record): void { + const event = envelope.payload as Record | undefined; + const sourceSeq = event?.sourceSeq; + const sourceEventId = event?.sourceEventId; + const eventType = event?.eventType; + const priority = event?.priority; + if ( + typeof sourceSeq !== "number" || + typeof sourceEventId !== "string" || + typeof eventType !== "string" || + (priority !== 0 && priority !== 1 && priority !== 2) || + event?.sourceInstanceId !== this.identity.runnerInstanceId || + event.runId !== this.identity.runId || + event.normalizedSessionId !== this.identity.normalizedSessionId || + event.turnId !== this.identity.turnId + ) { + connection.close(); + return; + } + const existing = this.store.state.committedEvents.find( + (candidate) => candidate.sourceEventId === sourceEventId, + ); + if (existing !== undefined) { + if (canonicalJson(existing.envelope) !== canonicalJson(envelope)) { + connection.close(); + return; + } + existing.deliveryCount += 1; + this.store.state.replayDeliveries += 1; + } else { + if (sourceSeq !== this.store.state.ackedSourceSeq + 1) { + connection.close(); + return; + } + this.store.state.committedEvents.push({ + sourceSeq, + sourceEventId, + eventType, + priority, + envelope: structuredClone(envelope), + deliveryCount: 1, + logicalEffectCount: 1, + }); + this.store.state.ackedSourceSeq = sourceSeq; + } + this.store.save(); + + if (this.fault === "revoke" && eventType === "run.terminal" && !this.#faultTriggered) { + if (connection.lease !== null) { + connection.lease.revokedAt = new Date().toISOString(); + } + this.queueCommand( + "turn.start", + { turnId: this.identity.turnId, text: "must be rejected after revoke" }, + "command_after_revoke", + ); + this.#triggerFault(); + this.store.save(); + // Revoke before ACK so the runner must flush a genuinely non-empty durable outbox. + connection.sendJson({ + protocol, + version: protocolVersion, + envelopeId: "revoke_phase3", + kind: "revoke", + runnerInstanceId: this.identity.runnerInstanceId, + sentAt: new Date().toISOString(), + payload: { reason: "fault_injection_revoke", drain: true }, + }); + this.#sendNextCommand(connection); + return; + } + + const shouldLoseAck = + !this.#faultTriggered && (this.fault === "lost-ack" || this.fault === "runner-restart"); + if (shouldLoseAck) { + this.#replayCursorOverrideOnce = true; + this.#triggerFault(); + if (this.fault === "lost-ack") { + connection.close(); + } + return; + } + + connection.sendJson({ + protocol, + version: protocolVersion, + envelopeId: `ack_${this.store.state.ackedSourceSeq}`, + kind: "ack", + runnerInstanceId: this.identity.runnerInstanceId, + sentAt: new Date().toISOString(), + payload: { ackedSourceSeq: this.store.state.ackedSourceSeq }, + }); + } +} + +function runnerEnvironment(ticket: string): NodeJS.ProcessEnv { + const environment: NodeJS.ProcessEnv = { + PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: ticket, + }; + for (const key of ["PATH", "SystemRoot", "WINDIR", "PATHEXT"]) { + const value = process.env[key]; + if (value !== undefined) environment[key] = value; + } + return environment; +} + +function spawnRunner(options: { + connectUrl: string; + stateDirectory: string; + identity: Phase3Identity; + ticket: string; + maxOutboxBytes: number; + p0ReserveBytes: number; +}): RunnerProcessHandle { + const args = [ + "--connect-url", + options.connectUrl, + "--state-dir", + options.stateDirectory, + "--runner-id", + options.identity.runnerInstanceId, + "--environment-lease-id", + options.identity.environmentLeaseId, + "--run-id", + options.identity.runId, + "--session-id", + options.identity.normalizedSessionId, + "--turn-id", + options.identity.turnId, + "--item-id", + options.identity.itemId, + "--runner-version", + "0.3.0", + "--runner-digest", + "sha256:phase3-approved", + "--fake-harness", + fakeHarnessBinary, + "--fake-harness-script", + fakeHarnessScript, + "--max-outbox-bytes", + String(options.maxOutboxBytes), + "--p0-reserve-bytes", + String(options.p0ReserveBytes), + "--reconnect-delay-ms", + "20", + "--max-runtime-ms", + "10000", + ]; + const child = spawn(runnerBinary, args, { + cwd: packageRoot, + env: runnerEnvironment(options.ticket), + stdio: "pipe", + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", (chunk: string) => { + stdout = `${stdout}${chunk}`.slice(-16_384); + }); + child.stderr.setEncoding("utf8").on("data", (chunk: string) => { + stderr = `${stderr}${chunk}`.slice(-16_384); + }); + const completion = new Promise((resolveCompletion, rejectCompletion) => { + child.once("error", rejectCompletion); + child.once("exit", (code, signal) => resolveCompletion({ code, signal, stdout, stderr })); + }); + return { child, completion }; +} + +async function waitForProcess( + handle: RunnerProcessHandle, + timeoutMs = 15_000, +): Promise { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + handle.completion, + new Promise((_resolve, reject) => { + timer = setTimeout(() => { + handle.child.kill("SIGKILL"); + reject(new Error("Phase 3 runner timed out.")); + }, timeoutMs); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + +function phase3Identity(): Phase3Identity { + return { + runnerInstanceId: "runner_phase3_stable", + environmentLeaseId: "environment_lease_phase3_stable", + runId: "run_phase3_stable", + normalizedSessionId: "session_phase3_stable", + turnId: "turn_phase3_stable", + itemId: "item_phase3_stable", + }; +} + +function queueScenario(core: Phase3MockCore, fault: Phase3Fault): void { + core.queueCommand("run.prepare", { workspace: "phase-03-durable-fixture" }); + core.queueCommand("session.open", { reuse: "same_session" }); + if (fault === "harness-restart") { + core.queueCommand("fault.harness_restart", {}); + } + if (fault === "storage-pressure") { + core.queueCommand("fault.storage_pressure", {}); + } + if (fault === "drain") { + core.queueCommand("runner.drain", {}); + } + core.queueCommand("turn.start", { + turnId: core.identity.turnId, + text: "Prove Phase 3 durable recovery.", + }); + if (fault !== "revoke") { + core.queueCommand("runner.shutdown", {}); + } +} + +function readRunnerState(stateDirectory: string): Phase3RunnerState { + return JSON.parse( + readFileSync(resolve(stateDirectory, "runner-state.json"), "utf8"), + ) as Phase3RunnerState; +} + +function assertContinuous(events: readonly Phase3CommittedEvent[]): boolean { + return events + .toSorted((left, right) => left.sourceSeq - right.sourceSeq) + .every((event, index) => event.sourceSeq === index + 1); +} + +function secretLeakCount(paths: readonly string[], secrets: readonly string[]): number { + let leaks = 0; + for (const path of paths) { + const source = readFileSync(path, "utf8"); + for (const secret of secrets) { + if (secret.length > 0 && source.includes(secret)) leaks += 1; + } + } + return leaks; +} + +function persistedCapabilityShape(path: string, fieldNames: readonly string[]): boolean { + const source = readFileSync(path, "utf8"); + return fieldNames.some((fieldName) => source.includes(`\"${fieldName}\"`)); +} + +export interface RunPhase3RecoveryOptions { + fault?: Phase3Fault; + stateDirectory?: string; + keepState?: boolean; +} + +export async function runPhase3Recovery( + options: RunPhase3RecoveryOptions = {}, +): Promise { + const fault = options.fault ?? "lost-ack"; + const identity = phase3Identity(); + const scratchRoot = + process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? + process.env.PAPERCLIP_SCRATCH_DIR ?? + tmpdir(); + const root = + options.stateDirectory ?? + mkdtempSync(resolve(scratchRoot, `paperclip-runner-phase3-${fault}-`)); + const runnerStateDirectory = resolve(root, "runner"); + const coreStateDirectory = resolve(root, "mock-core"); + mkdirSync(runnerStateDirectory, { recursive: true, mode: 0o700 }); + const core = new Phase3MockCore({ + stateDirectory: coreStateDirectory, + identity, + fault, + }); + queueScenario(core, fault); + await core.start(); + const tickets: string[] = []; + let runnerRestarts = 0; + const maxOutboxBytes = fault === "storage-pressure" ? 16 * 1024 : 64 * 1024; + const p0ReserveBytes = fault === "storage-pressure" ? 8 * 1024 : 32 * 1024; + + const launch = (): RunnerProcessHandle => { + const ticket = core.issueBootstrapTicket(); + tickets.push(ticket); + return spawnRunner({ + connectUrl: core.connectUrl, + stateDirectory: runnerStateDirectory, + identity, + ticket, + maxOutboxBytes, + p0ReserveBytes, + }); + }; + + let handle = launch(); + let result: RunnerProcessResult; + try { + if (fault === "runner-restart") { + await core.waitForFaultTrigger(); + handle.child.kill("SIGKILL"); + await waitForProcess(handle); + runnerRestarts += 1; + handle = launch(); + } else if (fault === "lease-expiry") { + const expired = await waitForProcess(handle); + if (expired.code === 0) { + throw new Error("Lease-expiry injection did not produce a recoverable restart boundary."); + } + runnerRestarts += 1; + handle = launch(); + } + result = await waitForProcess(handle); + if (result.code !== 0) { + throw new Error( + `Phase 3 runner exited with code ${String(result.code)}: ${result.stderr.trim()}`, + ); + } + } finally { + await core.stop(); + } + + const runnerState = readRunnerState(runnerStateDirectory); + const coreState = core.store.state; + const runnerStatePath = resolve(runnerStateDirectory, "runner-state.json"); + const leaks = secretLeakCount( + [runnerStatePath, core.store.path], + [...tickets, "must-not-persist"], + ); + const bootstrapTicketPersisted = persistedCapabilityShape( + runnerStatePath, + ["bootstrapTicket", "bootstrap_ticket"], + ); + const connectionLeaseTokenPersisted = persistedCapabilityShape( + runnerStatePath, + ["connectionLeaseToken", "connection_lease_token"], + ); + const committedEvents = coreState.committedEvents.toSorted( + (left, right) => left.sourceSeq - right.sourceSeq, + ); + const p0Committed = committedEvents.filter((event) => event.priority === 0).length; + const completedCommands = coreState.commands.filter( + (command) => command.status === "completed", + ); + const rejectedCommands = coreState.commands.filter( + (command) => command.status === "rejected", + ); + const logicalEffects = coreState.commands.reduce( + (sum, command) => + sum + + (typeof command.result?.logicalEffectCount === "number" + ? command.result.logicalEffectCount + : 0), + 0, + ); + const duplicateDeliveries = Object.values(coreState.commandDeliveryCounts).reduce( + (sum, count) => sum + Math.max(0, count - 1), + 0, + ); + const outcome: Phase3Diagnostics["recovery"]["outcome"] = + fault === "drain" + ? "drained" + : fault === "revoke" + ? "revoked" + : runnerState.unrecoverableOutcome === null + ? "recovered" + : "unrecoverable"; + const diagnostics: Phase3Diagnostics = { + schema: "paperclip.runner.phase3.diagnostics.v1", + fault, + connection: { + state: runnerState.lifecycle, + connectionCount: coreState.connectionCount, + reconnectCount: runnerState.reconnectCount, + leaseId: coreState.lastLeaseId, + leaseExpiresAt: coreState.lastLeaseExpiresAt, + }, + identity, + cursors: { + runnerAckedSourceSeq: runnerState.ackedSourceSeq, + runnerNextSourceSeq: runnerState.nextSourceSeq, + coreAckedSourceSeq: coreState.ackedSourceSeq, + highestCommittedSourceSeq: committedEvents.at(-1)?.sourceSeq ?? 0, + }, + outbox: { + events: runnerState.outbox.length, + bytes: runnerState.outbox.reduce((sum, event) => sum + event.byteSize, 0), + peakBytes: runnerState.peakOutboxBytes, + maxBytes: maxOutboxBytes, + backpressure: runnerState.backpressure, + p0Committed, + p0Lost: runnerState.ackedSourceSeq === coreState.ackedSourceSeq ? 0 : p0Committed, + }, + commands: { + issued: coreState.commands.length, + completed: completedCommands.length, + rejected: rejectedCommands.length, + logicalEffects, + duplicateDeliveries, + }, + recovery: { + replayDeliveries: coreState.replayDeliveries, + runnerRestarts, + harnessRestarts: Math.max(0, runnerState.harnessGeneration - 1), + malformedFrames: coreState.malformedFrames, + freshBootstraps: coreState.freshBootstraps, + outcome, + reason: + runnerState.unrecoverableOutcome ?? + runnerState.recoverableFailure ?? + `${fault.replaceAll("-", "_")}_completed`, + }, + security: { + bootstrapTicketPersisted, + connectionLeaseTokenPersisted, + secretLeakCount: leaks, + }, + committedEvents, + }; + const acceptedWithTooManyEffects = coreState.commands.some((command) => { + const effectCount = command.result?.logicalEffectCount; + return command.status === "completed" && effectCount !== 1; + }); + const trace: Phase3RunTrace = { + schema: "paperclip.runner.phase3.trace.v1", + diagnostics, + runnerState, + commands: structuredClone(coreState.commands), + assertions: { + stableIdentity: + canonicalJson(identity) === + canonicalJson({ + runnerInstanceId: runnerState.runnerInstanceId, + environmentLeaseId: runnerState.environmentLeaseId, + runId: runnerState.runId, + normalizedSessionId: runnerState.normalizedSessionId, + turnId: runnerState.turnId, + itemId: runnerState.itemId, + }), + sourceCursorContinuous: + assertContinuous(committedEvents) && + runnerState.ackedSourceSeq === coreState.ackedSourceSeq, + oneLogicalEffectPerAcceptedCommand: !acceptedWithTooManyEffects, + noDuplicateLogicalEvents: committedEvents.every( + (event) => event.logicalEffectCount === 1, + ), + p0Preserved: diagnostics.outbox.p0Lost === 0, + boundedStorage: + diagnostics.outbox.bytes <= maxOutboxBytes && + runnerState.peakOutboxBytes <= maxOutboxBytes, + secretsRedacted: + leaks === 0 && + !bootstrapTicketPersisted && + !connectionLeaseTokenPersisted, + }, + }; + + if (options.keepState !== true && options.stateDirectory === undefined) { + rmSync(root, { recursive: true, force: true }); + } + return trace; +} + +export const phase3Internals = { + runnerBinary, + runnerEnvironment, + tokenDigest, + canonicalJson, + phase3Identity, +};