diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 2566a73b00..619485c0fa 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -6,10 +6,12 @@ The package currently exposes only the language-neutral PRP v1 TypeScript contract, provider-neutral structured questions and responses, deterministic fixture validation/replay, structured-result normalization, and the session reducer oracle. It also contains a package-local Rust runner, scripted fake -harness, bounded process supervisor, and cross-language replay oracle. These -Rust binaries are test infrastructure. No server code starts or invokes them. -The package does not add a provider transport, server adapter, semantic-tool -authorization, or production Paperclip behavior. +harness, bounded process supervisor, cross-language replay oracle, and durable +PRP transport. The transport authenticates and encrypts loopback WebSocket +sessions, persists an ACK-driven outbox and command journal, and reconnects with +a short-lived lease. No server code starts or invokes it. The package does not +add a provider, server adapter, semantic-tool authorization, or production +Paperclip behavior. The first provider scope is Codex. The protocol contains provider-neutral event and semantic receipt shapes, but their presence does not authorize a tool or @@ -34,6 +36,11 @@ This command checks Rust formatting, builds and tests the minimal workspace, verifies bounded process cleanup, exercises the fake local runner, and compares the Rust conformance and replay summaries with the shared fixtures. +Durability and failure semantics are documented in +[`runner/DURABLE_TRANSPORT.md`](runner/DURABLE_TRANSPORT.md). The fault suite +drops a connection before its event ACK, reconnects with the bound lease, +replays the same event, and proves the duplicated command effect ran once. + Use `generate:protocol-manifest` after a schema or fixture change, `generate:protocol-types` after a schema change, and `generate:replay-goldens` after an intentional reducer change. Commit generated diff --git a/packages/paperclip-runner/package.json b/packages/paperclip-runner/package.json index 855d57c992..51686f9fc6 100644 --- a/packages/paperclip-runner/package.json +++ b/packages/paperclip-runner/package.json @@ -28,6 +28,7 @@ "test": "pnpm run test:typescript && pnpm run test:rust", "test:typescript": "node --test test/protocol-contract.test.mjs && vitest run", "test:rust": "cargo test --manifest-path runner/Cargo.toml --locked --workspace", + "test:durable": "cargo test --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core durable::", "generate:protocol-manifest": "node scripts/generate-protocol-manifest.mjs", "check:protocol-manifest": "node scripts/generate-protocol-manifest.mjs --check", "generate:protocol-types": "node scripts/generate-protocol-schema-module.mjs", diff --git a/packages/paperclip-runner/runner/Cargo.lock b/packages/paperclip-runner/runner/Cargo.lock index 91e922401f..770faf90ba 100644 --- a/packages/paperclip-runner/runner/Cargo.lock +++ b/packages/paperclip-runner/runner/Cargo.lock @@ -2,24 +2,257 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common", + "inout", +] + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "rand_core 0.6.4", + "typenum", +] + +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", + "subtle", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", +] + +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "itoa" version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + [[package]] name = "memchr" version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "paperclip-runner-core" version = "0.0.0" dependencies = [ + "aes-gcm", + "getrandom 0.3.4", + "hmac", "serde", "serde_json", + "sha2", + "tungstenite", +] + +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures", + "opaque-debug", + "universal-hash", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", ] [[package]] @@ -40,6 +273,50 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.17", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "serde" version = "1.0.229" @@ -67,7 +344,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.4", ] [[package]] @@ -83,6 +360,45 @@ dependencies = [ "zmij", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "3.0.4" @@ -94,12 +410,118 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand", + "sha1", + "thiserror", + "utf-8", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common", + "subtle", +] + +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + +[[package]] +name = "zerocopy" +version = "0.8.56" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.56" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/packages/paperclip-runner/runner/Cargo.toml b/packages/paperclip-runner/runner/Cargo.toml index 47b88d455c..bf407f5011 100644 --- a/packages/paperclip-runner/runner/Cargo.toml +++ b/packages/paperclip-runner/runner/Cargo.toml @@ -9,5 +9,10 @@ license = "MIT" publish = false [workspace.dependencies] +aes-gcm = "0.10" +getrandom = "0.3" +hmac = "0.12" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +sha2 = "0.10" +tungstenite = { version = "0.28", default-features = false, features = ["handshake"] } diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md new file mode 100644 index 0000000000..aae730d796 --- /dev/null +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -0,0 +1,55 @@ +# Durable PRP transport + +This layer gives `paperclip-runnerd` a provider-neutral, package-local PRP v1 +transport. Nothing in the Paperclip server invokes the durable mode yet, and no +provider is installed by this change. + +## Trust boundary + +- The runner accepts only `ws://` destinations whose complete DNS result is + loopback. Resolution happens once and reconnects reuse the pinned addresses. +- A bootstrap ticket is read from `PAPERCLIP_RUNNER_BOOTSTRAP_TICKET`, removed + from the environment immediately, and never sent over the socket. Both peers + prove possession through HMAC-SHA-256. +- A successful bootstrap exchanges the one-use ticket for a connection-bound, + expiring lease held only in memory. Once authentication starts, a failed + bootstrap attempt is not replayed automatically. +- Post-authentication frames use AES-256-GCM with direction-specific keys, + monotonically increasing counters, and session-bound associated data. + Plaintext, replayed, out-of-order, oversized, or incorrectly bound frames + fail closed. +- The durable state directory is private, symlinks are rejected, and updates + use a private temporary file, file sync, atomic rename, and directory sync. + Credentials and lease tokens are never written to this state. + +## Recovery contract + +Events enter the outbox before delivery. A cumulative ACK may advance only to a +source sequence the runner has produced; acknowledged prefixes are removed +atomically from durable state. After disconnect, every remaining event is sent +again with the same identity and source sequence. + +Commands require a contiguous controller sequence. The runner journals a +pending command before invoking its executor and persists its result afterward. +An exact duplicate returns the stored result without repeating the effect. If +the process dies inside the effect window, recovery records an indeterminate +result and refuses to execute that command again. Recent results are bounded; +commands older than the compacted controller cursor fail closed. + +State written before complete-command fingerprints existed is migrated by +compacting its legacy command journal through the last recorded controller +sequence. The runner can recover, but it rejects redelivery of those older +commands instead of guessing an identity or repeating an uncertain effect. + +The outbox has separate hard and reserved limits. P1/P2 events cannot consume +the P0 reserve. When the soft limit is reached, the runner enters backpressure +and rejects the new non-P0 event rather than silently losing it. Exhausting the +P0 reserve is an explicit unrecoverable condition. + +## Current boundary + +Durable mode is selected only when `paperclip-runnerd` receives +`--connect-url`. Its transport-only executor handles runner lifecycle commands +and rejects provider commands with `provider_not_installed`. The existing local +fake-runner mode remains unchanged. Codex execution, semantic tools, server +coordination, and the user-facing adapter belong to later layers. diff --git a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml index fa063349d8..bf26046eb3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml +++ b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml @@ -6,8 +6,13 @@ license.workspace = true publish.workspace = true [dependencies] +aes-gcm.workspace = true +getrandom.workspace = true +hmac.workspace = true serde.workspace = true serde_json.workspace = true +sha2.workspace = true +tungstenite.workspace = true [[bin]] name = "conformance-tracer" 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 268599d852..69c1ea0b50 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 @@ -2,7 +2,12 @@ use std::path::PathBuf; use std::process::ExitCode; use std::time::Duration; +use paperclip_runner_core::durable::{ + capture_bootstrap_ticket, run_durable_runner, Command, CommandExecution, CommandExecutor, + DurableRunnerConfig, DurableRunnerError, +}; use paperclip_runner_core::local_runner::{run_local_runner, LocalRunnerError, RunnerConfig}; +use serde_json::json; fn value(args: &[String], name: &str) -> Result { let index = args @@ -34,8 +39,67 @@ fn usize_value(args: &[String], name: &str, default: usize) -> Result Result { + Ok(CommandExecution::result( + if matches!( + command.command_type.as_str(), + "runner.shutdown" | "runner.suspend" | "runner.drain" + ) { + json!({"status": "completed"}) + } else { + json!({ + "status": "rejected", + "code": "provider_not_installed", + "message": "the durable transport is active, but no provider is installed in this build", + }) + }, + )) + } +} + +fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { + let ticket = capture_bootstrap_ticket() + .map_err(|error| LocalRunnerError::invalid(error.to_string()))? + .ok_or_else(|| { + LocalRunnerError::invalid( + "PAPERCLIP_RUNNER_BOOTSTRAP_TICKET is required for durable mode", + ) + })?; + let duration = |name: &str, default: u64| { + optional_u64(args, name).map(|value| Duration::from_millis(value.unwrap_or(default))) + }; + run_durable_runner( + 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")?, + max_outbox_bytes: usize_value(args, "--max-outbox-bytes", 16 * 1024 * 1024)?, + p0_reserve_bytes: usize_value(args, "--p0-reserve-bytes", 1024 * 1024)?, + max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?, + reconnect_delay: duration("--reconnect-delay-ms", 250)?, + max_runtime: duration("--max-runtime-ms", 60 * 60 * 1000)?, + }, + ticket, + TransportOnlyExecutor, + ) + .map_err(|error| LocalRunnerError::invalid(error.to_string())) +} + fn run() -> Result<(), LocalRunnerError> { let args = std::env::args().skip(1).collect::>(); + if args.iter().any(|argument| argument == "--connect-url") { + return run_durable(&args); + } run_local_runner(RunnerConfig { run_id: value(&args, "--run-id")?, normalized_session_id: value(&args, "--session-id")?, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs new file mode 100644 index 0000000000..cbb1e1ebbd --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs @@ -0,0 +1,159 @@ +mod runner; +mod state; +mod transport; + +use std::error::Error; +use std::fmt::{self, Display, Formatter}; +use std::path::PathBuf; +use std::time::Duration; + +pub use runner::{run_durable_runner, CommandExecution, CommandExecutor}; +pub use state::{ + Command, CommandDisposition, DurableState, DurableStateStore, EventPriority, + StoredCommandResult, StoredOutboxEvent, +}; + +pub const PROTOCOL: &str = "paperclip.runner"; +pub const PROTOCOL_VERSION: u64 = 1; +pub const BOOTSTRAP_TICKET_ENV: &str = "PAPERCLIP_RUNNER_BOOTSTRAP_TICKET"; +const MAX_OUTBOX_BYTES: usize = 512 * 1024 * 1024; +const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DurableRunnerError(String); + +impl DurableRunnerError { + pub fn invalid(message: impl Into) -> Self { + Self(message.into()) + } +} + +impl Display for DurableRunnerError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.0) + } +} + +impl Error for DurableRunnerError {} + +#[derive(Debug)] +struct Secret(Vec); + +impl Secret { + fn new(value: String) -> Self { + Self(value.into_bytes()) + } + + fn expose(&self) -> Result<&str, DurableRunnerError> { + std::str::from_utf8(&self.0) + .map_err(|_| DurableRunnerError::invalid("transport credential is not valid UTF-8")) + } +} + +impl Drop for Secret { + fn drop(&mut self) { + self.0.fill(0); + } +} + +#[derive(Debug)] +pub struct BootstrapTicket(Secret); + +impl BootstrapTicket { + pub fn new(value: String) -> Result { + if value.trim().is_empty() { + return Err(DurableRunnerError::invalid( + "bootstrap ticket must not be empty", + )); + } + Ok(Self(Secret::new(value))) + } + + fn expose(&self) -> Result<&str, DurableRunnerError> { + self.0.expose() + } +} + +pub fn capture_bootstrap_ticket() -> Result, DurableRunnerError> { + let value = match std::env::var(BOOTSTRAP_TICKET_ENV) { + Ok(value) => value, + Err(std::env::VarError::NotPresent) => return Ok(None), + Err(error) => { + return Err(DurableRunnerError::invalid(format!( + "failed to read bootstrap ticket: {error}" + ))) + } + }; + std::env::remove_var(BOOTSTRAP_TICKET_ENV); + BootstrapTicket::new(value).map(Some) +} + +#[derive(Clone, Debug, PartialEq, Eq)] +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 max_outbox_bytes: usize, + pub p0_reserve_bytes: usize, + pub max_frame_bytes: usize, + pub reconnect_delay: Duration, + pub max_runtime: Duration, +} + +impl DurableRunnerConfig { + pub fn validate(&self) -> Result<(), DurableRunnerError> { + for (name, value) in [ + ("connect_url", self.connect_url.as_str()), + ("runner_instance_id", self.runner_instance_id.as_str()), + ("environment_lease_id", self.environment_lease_id.as_str()), + ("run_id", self.run_id.as_str()), + ("normalized_session_id", self.normalized_session_id.as_str()), + ("turn_id", self.turn_id.as_str()), + ("item_id", self.item_id.as_str()), + ("runner_version", self.runner_version.as_str()), + ("runner_digest", self.runner_digest.as_str()), + ] { + if value.is_empty() || value.len() > 512 || value.chars().any(char::is_control) { + return Err(DurableRunnerError::invalid(format!( + "{name} must be a non-empty bounded string without control characters" + ))); + } + } + if self.max_outbox_bytes == 0 + || self.max_outbox_bytes > MAX_OUTBOX_BYTES + || self.p0_reserve_bytes >= self.max_outbox_bytes + { + return Err(DurableRunnerError::invalid( + "P0 reserve must be smaller than an outbox limit no larger than 512 MiB", + )); + } + if !(1024..=MAX_FRAME_BYTES).contains(&self.max_frame_bytes) { + return Err(DurableRunnerError::invalid( + "transport frame limit must be between 1 KiB and 16 MiB", + )); + } + if self.max_runtime.is_zero() { + return Err(DurableRunnerError::invalid( + "durable runner max runtime must be non-zero", + )); + } + if self.reconnect_delay.is_zero() || self.reconnect_delay > Duration::from_secs(60) { + return Err(DurableRunnerError::invalid( + "reconnect delay must be between one millisecond and 60 seconds", + )); + } + if self.max_runtime > Duration::from_secs(7 * 24 * 60 * 60) { + return Err(DurableRunnerError::invalid( + "durable runner max runtime must not exceed seven days", + )); + } + Ok(()) + } +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs new file mode 100644 index 0000000000..dba5f6b57d --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -0,0 +1,416 @@ +use std::thread; +use std::time::Instant; + +use serde_json::{json, Value}; + +use super::state::{ + Command, CommandDisposition, DurableState, DurableStateStore, EventPriority, + StoredCommandResult, +}; +use super::transport::{ + current_unix_ms, validate_control_identity, AuthenticatedTransport, ConnectionMetadata, + LeaseCredential, ResolvedWsTarget, +}; +use super::{BootstrapTicket, DurableRunnerConfig, DurableRunnerError, PROTOCOL, PROTOCOL_VERSION}; + +#[derive(Clone, Debug, PartialEq)] +pub struct CommandExecution { + pub result: Value, + pub events: Vec<(String, EventPriority, Value)>, +} + +impl CommandExecution { + pub fn result(result: Value) -> Self { + Self { + result, + events: Vec::new(), + } + } +} + +pub trait CommandExecutor { + fn execute(&mut self, command: &Command) -> Result; +} + +pub fn run_durable_runner( + config: DurableRunnerConfig, + bootstrap_ticket: BootstrapTicket, + mut executor: E, +) -> Result<(), DurableRunnerError> { + config.validate()?; + let store = DurableStateStore::new(&config.state_dir)?; + let (mut state, recovered) = store.load_or_create(&config)?; + if state.lifecycle == "revoked" || state.lifecycle == "stopped" { + return Ok(()); + } + if recovered { + state.reconnect_count = state.reconnect_count.saturating_add(1); + state.record_diagnostic("runner restored its durable identity after process recovery"); + state.enqueue_event( + &config, + "runner.reconciled", + EventPriority::P0, + json!({"outcome": "same_durable_session_resumed"}), + )?; + store.save(&state)?; + } + // Resolve once. Every reconnect uses the same validated concrete addresses, + // so DNS cannot redirect a retry after the trust decision. + let target = ResolvedWsTarget::resolve(&config.connect_url)?; + let started = Instant::now(); + let mut bootstrap_ticket = Some(bootstrap_ticket); + let mut lease: Option = None; + + loop { + if started.elapsed() >= config.max_runtime { + state.lifecycle = "recoverable_failure".to_owned(); + state.recoverable_failure = Some("transport_reconnect_deadline_exceeded".to_owned()); + state.record_diagnostic( + "transport reconnect deadline elapsed; durable state is preserved", + ); + store.save(&state)?; + return Err(DurableRunnerError::invalid( + "transport reconnect deadline elapsed; durable state is preserved", + )); + } + if lease.as_ref().is_some_and(|credential| { + current_unix_ms().is_ok_and(|now| now >= credential.expires_at_unix_ms) + }) { + state.lifecycle = "recoverable_failure".to_owned(); + state.recoverable_failure = Some("lease_expired_requires_bootstrap".to_owned()); + state.record_diagnostic("connection lease expired; a fresh bootstrap is required"); + store.save(&state)?; + return Err(DurableRunnerError::invalid( + "connection lease expired; a fresh bootstrap is required", + )); + } + + let using_bootstrap = lease.is_none(); + let connection = AuthenticatedTransport::connect( + &target, + &config, + &state, + bootstrap_ticket.as_ref(), + lease.as_ref(), + ); + let (mut transport, welcome) = match connection { + Ok(connection) => connection, + Err(error) => { + state.record_diagnostic(format!("transport reconnect scheduled: {error}")); + store.save(&state)?; + if using_bootstrap && error.bootstrap_maybe_consumed { + return Err(DurableRunnerError::invalid( + "bootstrap connection failed closed; provide a fresh one-use ticket", + )); + } + thread::sleep(config.reconnect_delay); + continue; + } + }; + if let Some(next_lease) = welcome.lease { + lease = Some(next_lease); + // A bootstrap capability is one-use. It is destroyed only after a + // mutually authenticated secure welcome exchanges it for a lease. + bootstrap_ticket.take(); + } + if let Some(acked_source_seq) = welcome.acked_source_seq { + state.apply_ack(acked_source_seq)?; + } + state.lifecycle = "ready".to_owned(); + state.recoverable_failure = None; + store.save(&state)?; + let mut sent_source_seq = state.acked_source_seq; + + let mut stop_after_reply = false; + let mut disconnected = false; + for command in welcome.pending_commands { + let (result, stop) = + process_command(&mut state, &store, &config, &mut executor, &command)?; + stop_after_reply |= stop; + if let Err(error) = transport.send_json(&command_result_envelope(&state, &result)) { + state.record_diagnostic(error.to_string()); + disconnected = true; + break; + } + } + if !disconnected && send_outbox(&mut transport, &state, &mut sent_source_seq).is_err() { + state.record_diagnostic("outbox delivery failed; unacknowledged suffix will replay"); + disconnected = true; + } + if stop_after_reply && !disconnected { + state.lifecycle = "stopped".to_owned(); + store.save(&state)?; + return Ok(()); + } + if disconnected { + state.reconnect_count = state.reconnect_count.saturating_add(1); + store.save(&state)?; + thread::sleep(config.reconnect_delay); + continue; + } + + let connection = welcome.connection; + loop { + if started.elapsed() >= config.max_runtime { + break; + } + if current_unix_ms()? >= connection.expires_at_unix_ms { + state.lifecycle = "recoverable_failure".to_owned(); + state.recoverable_failure = Some("lease_expired_requires_bootstrap".to_owned()); + state.record_diagnostic("active connection lease expired"); + store.save(&state)?; + return Err(DurableRunnerError::invalid( + "active connection lease expired; durable state is preserved", + )); + } + let message = match transport.receive_json() { + Ok(Some(message)) => message, + Ok(None) => continue, + Err(error) => { + state.record_diagnostic(error.to_string()); + state.reconnect_count = state.reconnect_count.saturating_add(1); + store.save(&state)?; + break; + } + }; + if let Err(error) = validate_control_identity(&message, &state, Some(&connection)) { + state.record_diagnostic(format!( + "control identity mismatch closed the connection: {error}" + )); + state.reconnect_count = state.reconnect_count.saturating_add(1); + store.save(&state)?; + break; + } + 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(|| DurableRunnerError::invalid("ACK cursor is required"))?; + state.apply_ack(acked)?; + store.save(&state)?; + } + Some("command") => { + let command: Command = + serde_json::from_value(message.get("payload").cloned().ok_or_else( + || DurableRunnerError::invalid("command payload is required"), + )?) + .map_err(|error| { + DurableRunnerError::invalid(format!("command is malformed: {error}")) + })?; + let (result, stop) = + process_command(&mut state, &store, &config, &mut executor, &command)?; + let delivery = transport + .send_json(&command_result_envelope(&state, &result)) + .and_then(|()| send_outbox(&mut transport, &state, &mut sent_source_seq)); + if let Err(error) = delivery { + state.record_diagnostic(error.to_string()); + state.reconnect_count = state.reconnect_count.saturating_add(1); + store.save(&state)?; + break; + } + if stop { + state.lifecycle = "stopped".to_owned(); + store.save(&state)?; + return Ok(()); + } + } + Some("revoke") => { + let epoch = message + .pointer("/payload/revocationEpoch") + .and_then(Value::as_u64) + .ok_or_else(|| { + DurableRunnerError::invalid("revoke revocation epoch is required") + })?; + if epoch <= connection.revocation_epoch { + return Err(DurableRunnerError::invalid( + "revoke must advance the authenticated revocation epoch", + )); + } + state.lifecycle = "revoked".to_owned(); + state.record_diagnostic("connection capability was revoked"); + store.save(&state)?; + return Ok(()); + } + Some("ping") => { + transport.send_json(&control_envelope( + &state, + &connection, + "pong", + json!({ + "lifecycle": state.lifecycle, + "ackedSourceSeq": state.acked_source_seq, + "outboxBytes": state.outbox_bytes(), + }), + ))?; + } + _ => { + state.record_diagnostic( + "malformed or unsupported control frame closed the connection", + ); + state.reconnect_count = state.reconnect_count.saturating_add(1); + store.save(&state)?; + break; + } + } + } + thread::sleep(config.reconnect_delay); + } +} + +fn process_command( + state: &mut DurableState, + store: &DurableStateStore, + config: &DurableRunnerConfig, + executor: &mut E, + command: &Command, +) -> Result<(StoredCommandResult, bool), DurableRunnerError> { + match state.begin_command(command)? { + CommandDisposition::Replay(result) | CommandDisposition::Reject(result) => { + return Ok((result, false)); + } + CommandDisposition::Execute => {} + } + // Persist the pending marker before any command effect. If the process dies + // in the effect window, recovery returns an indeterminate result and never + // executes the same logical command twice. + store.save(state)?; + let execution = executor.execute(command)?; + for (event_type, priority, payload) in execution.events { + state.enqueue_event(config, event_type, priority, payload)?; + } + let result = state.complete_command(command, execution.result)?; + store.save(state)?; + let stop = matches!( + command.command_type.as_str(), + "runner.shutdown" | "runner.suspend" + ); + Ok((result, stop)) +} + +fn send_outbox( + transport: &mut AuthenticatedTransport, + state: &DurableState, + sent_source_seq: &mut u64, +) -> Result<(), DurableRunnerError> { + for event in &state.outbox { + if event.source_seq <= *sent_source_seq { + continue; + } + transport.send_json(&event.envelope)?; + *sent_source_seq = event.source_seq; + } + Ok(()) +} + +fn command_result_envelope(state: &DurableState, result: &StoredCommandResult) -> Value { + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "command_result", + "runnerInstanceId": state.runner_instance_id, + "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "payload": result, + }) +} + +fn control_envelope( + state: &DurableState, + connection: &ConnectionMetadata, + kind: &str, + payload: Value, +) -> Value { + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": kind, + "runnerInstanceId": state.runner_instance_id, + "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "connectionId": connection.connection_id, + "connectionLeaseId": connection.lease_id, + "payload": payload, + }) +} + +#[cfg(test)] +mod tests { + use std::fs; + use std::path::PathBuf; + use std::time::Duration; + + use super::*; + + struct CountingExecutor { + calls: usize, + } + + impl CommandExecutor for CountingExecutor { + fn execute(&mut self, _command: &Command) -> Result { + self.calls += 1; + Ok(CommandExecution::result(json!({"calls": self.calls}))) + } + } + + fn config(directory: PathBuf) -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: "ws://127.0.0.1:3000/path".to_owned(), + state_dir: directory, + runner_instance_id: "runner_1".to_owned(), + environment_lease_id: "environment_1".to_owned(), + run_id: "run_1".to_owned(), + normalized_session_id: "session_1".to_owned(), + turn_id: "turn_1".to_owned(), + item_id: "item_1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: "sha256:test".to_owned(), + max_outbox_bytes: 64 * 1024, + p0_reserve_bytes: 4096, + max_frame_bytes: 64 * 1024, + reconnect_delay: Duration::from_millis(1), + max_runtime: Duration::from_secs(1), + } + } + + fn command() -> Command { + Command { + schema: "paperclip.prp.command.v1".to_owned(), + command_id: "command_1".to_owned(), + controller_seq: 1, + command_type: "session.open".to_owned(), + issued_at: "2026-08-24T00:00:00.000Z".to_owned(), + deadline_at: None, + precondition: None, + payload: json!({}), + } + } + + #[test] + fn duplicate_delivery_replays_the_durable_result() { + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-command-replay-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&directory); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = CountingExecutor { calls: 0 }; + let first = process_command(&mut state, &store, &config, &mut executor, &command()) + .unwrap() + .0; + let replay = process_command(&mut state, &store, &config, &mut executor, &command()) + .unwrap() + .0; + assert_eq!(executor.calls, 1); + assert_eq!(first, replay); + fs::remove_dir_all(directory).unwrap(); + } +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs new file mode 100644 index 0000000000..9dc4174229 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs @@ -0,0 +1,1176 @@ +use std::collections::BTreeMap; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::path::{Path, PathBuf}; +use std::time::{SystemTime, UNIX_EPOCH}; + +#[cfg(unix)] +use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use super::{DurableRunnerConfig, DurableRunnerError, PROTOCOL, PROTOCOL_VERSION}; + +const STATE_SCHEMA: &str = "paperclip.runner.durable.state.v1"; +const STATE_FILE: &str = "runner-state.json"; +const MAX_RECENT_COMMANDS: usize = 128; +const MAX_DIAGNOSTICS: usize = 32; +const MAX_COMMAND_RESULT_BYTES: usize = 64 * 1024; +const STATE_OVERHEAD_BYTES: usize = 16 * 1024 * 1024; +const TEMP_FILE_ATTEMPTS: usize = 32; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EventPriority { + P0, + P1, + P2, +} + +impl EventPriority { + fn number(self) -> u8 { + match self { + Self::P0 => 0, + Self::P1 => 1, + Self::P2 => 2, + } + } +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Command { + pub schema: String, + pub command_id: String, + pub controller_seq: u64, + #[serde(rename = "type")] + pub command_type: String, + pub issued_at: String, + #[serde(default)] + pub deadline_at: Option, + #[serde(default)] + pub precondition: Option, + #[serde(default)] + pub payload: Value, +} + +impl Command { + pub fn validate(&self) -> Result<(), DurableRunnerError> { + if self.schema != "paperclip.prp.command.v1" { + return Err(DurableRunnerError::invalid( + "command requires the paperclip.prp.command.v1 schema", + )); + } + if self.command_id.is_empty() + || self.command_id.len() > 160 + || self.command_id.chars().any(char::is_control) + { + return Err(DurableRunnerError::invalid( + "commandId is empty, oversized, or contains control characters", + )); + } + if self.issued_at.is_empty() + || self.issued_at.len() > 64 + || self.issued_at.chars().any(char::is_control) + || self.deadline_at.as_ref().is_some_and(|deadline| { + deadline.is_empty() || deadline.len() > 64 || deadline.chars().any(char::is_control) + }) + { + return Err(DurableRunnerError::invalid( + "command timestamps are empty, oversized, or contain control characters", + )); + } + if self.controller_seq == 0 { + return Err(DurableRunnerError::invalid( + "command controllerSeq must be positive", + )); + } + if !self.payload.is_object() { + return Err(DurableRunnerError::invalid( + "command payload must be an object", + )); + } + if self + .precondition + .as_ref() + .is_some_and(|precondition| !precondition.is_object()) + { + return Err(DurableRunnerError::invalid( + "command precondition must be an object", + )); + } + if !matches!( + self.command_type.as_str(), + "run.prepare" + | "run.attach" + | "session.open" + | "turn.start" + | "turn.steer" + | "turn.interrupt" + | "turn.stop" + | "request.resolve" + | "interaction.receipt" + | "semantic_tool.result" + | "session.snapshot" + | "session.close" + | "session.budget.increase" + | "session.destroy" + | "run.cancel" + | "runner.drain" + | "runner.suspend" + | "runner.shutdown" + ) { + return Err(DurableRunnerError::invalid( + "command type is not supported by PRP v1", + )); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StoredOutboxEvent { + pub source_seq: u64, + pub priority: u8, + pub event_type: String, + pub envelope: Value, + pub byte_size: usize, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StoredCommandResult { + pub command_id: String, + pub controller_seq: u64, + pub command_type: String, + pub status: String, + pub result: Value, +} + +#[derive(Clone, Debug, PartialEq)] +pub enum CommandDisposition { + Execute, + Replay(StoredCommandResult), + Reject(StoredCommandResult), +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DurableState { + 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 compacted_through_controller_seq: u64, + pub reconnect_count: u64, + pub max_outbox_bytes: usize, + pub p0_reserve_bytes: usize, + pub peak_outbox_bytes: usize, + pub outbox: Vec, + pub processed_commands: BTreeMap, + #[serde(default)] + pub processed_command_fingerprints: BTreeMap, + pub diagnostics: Vec, + pub backpressure: bool, + pub recoverable_failure: Option, +} + +impl DurableState { + pub(crate) 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, + compacted_through_controller_seq: 0, + reconnect_count: 0, + max_outbox_bytes: config.max_outbox_bytes, + p0_reserve_bytes: config.p0_reserve_bytes, + peak_outbox_bytes: 0, + outbox: Vec::new(), + processed_commands: BTreeMap::new(), + processed_command_fingerprints: BTreeMap::new(), + diagnostics: Vec::new(), + backpressure: false, + recoverable_failure: None, + } + } + + 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 enqueue_event( + &mut self, + config: &DurableRunnerConfig, + event_type: impl Into, + priority: EventPriority, + payload: Value, + ) -> Result { + let event_type = event_type.into(); + if event_type.is_empty() + || event_type.len() > 160 + || event_type.chars().any(char::is_control) + { + return Err(DurableRunnerError::invalid( + "event type is empty, oversized, or contains control characters", + )); + } + if !payload.is_object() { + return Err(DurableRunnerError::invalid( + "durable event payload must be an object", + )); + } + + let source_seq = self.next_source_seq; + let emitted_at = current_timestamp()?; + let envelope = json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "event", + "runnerInstanceId": self.runner_instance_id, + "environmentLeaseId": self.environment_lease_id, + "runId": self.run_id, + "normalizedSessionId": self.normalized_session_id, + "turnId": self.turn_id, + "itemId": self.item_id, + "payload": { + "schema": "paperclip.prp.event.v1", + "sourceEventId": format!("event_{}_{source_seq:016}", self.runner_instance_id), + "sourceSeq": source_seq, + "sourceInstanceId": self.runner_instance_id, + "sourceKind": "runner", + "runId": self.run_id, + "normalizedSessionId": self.normalized_session_id, + "turnId": self.turn_id, + "itemId": self.item_id, + "eventType": event_type, + "schemaVersion": 1, + "priority": priority.number(), + "emittedAt": emitted_at, + "payload": sanitize_value(&payload), + }, + }); + let byte_size = serde_json::to_vec(&envelope) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))? + .len(); + if byte_size > config.max_frame_bytes { + return Err(DurableRunnerError::invalid( + "durable event exceeds the transport frame limit", + )); + } + let projected = self.outbox_bytes().saturating_add(byte_size); + let non_p0_limit = config + .max_outbox_bytes + .saturating_sub(config.p0_reserve_bytes); + + if priority != EventPriority::P0 && projected > non_p0_limit { + self.backpressure = true; + self.lifecycle = "backpressure".to_owned(); + self.record_diagnostic("outbox soft limit reached; non-P0 event rejected"); + return Err(DurableRunnerError::invalid( + "outbox soft limit reached; reserved storage is available only to P0 events", + )); + } + if projected > config.max_outbox_bytes { + self.lifecycle = "unrecoverable".to_owned(); + self.record_diagnostic("P0 outbox reserve exhausted; operator recovery is required"); + return Err(DurableRunnerError::invalid( + "durable outbox limit exhausted", + )); + } + + self.next_source_seq = self + .next_source_seq + .checked_add(1) + .ok_or_else(|| DurableRunnerError::invalid("source sequence exhausted"))?; + self.outbox.push(StoredOutboxEvent { + source_seq, + priority: priority.number(), + event_type, + envelope, + byte_size, + }); + self.peak_outbox_bytes = self.peak_outbox_bytes.max(projected); + Ok(source_seq) + } + + pub fn apply_ack(&mut self, acked_source_seq: u64) -> Result<(), DurableRunnerError> { + if acked_source_seq < self.acked_source_seq { + return Err(DurableRunnerError::invalid( + "cumulative ACK cannot move behind the durable cursor", + )); + } + if acked_source_seq > self.highest_source_seq() { + return Err(DurableRunnerError::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); + if self.backpressure + && self.outbox_bytes() < self.max_outbox_bytes.saturating_sub(self.p0_reserve_bytes) + { + self.backpressure = false; + if self.lifecycle == "backpressure" { + self.lifecycle = "ready".to_owned(); + } + } + Ok(()) + } + + pub fn begin_command( + &mut self, + command: &Command, + ) -> Result { + command.validate()?; + let fingerprint = command_fingerprint(command)?; + if let Some(previous) = self.processed_commands.get(&command.command_id) { + let previous_fingerprint = self + .processed_command_fingerprints + .get(&command.command_id) + .ok_or_else(|| { + DurableRunnerError::invalid( + "durable command journal is missing its identity fingerprint", + ) + })?; + if previous_fingerprint != &fingerprint { + return Err(DurableRunnerError::invalid( + "commandId was reused with different command data", + )); + } + return Ok(CommandDisposition::Replay(previous.clone())); + } + if command.controller_seq <= self.compacted_through_controller_seq { + return Ok(CommandDisposition::Reject(command_result( + command, + "rejected", + json!({ + "code": "command_history_compacted", + "message": "command is older than the bounded replay journal and was not re-executed", + }), + ))); + } + let expected = self + .last_controller_command_seq + .checked_add(1) + .ok_or_else(|| DurableRunnerError::invalid("controller sequence exhausted"))?; + if command.controller_seq != expected { + return Err(DurableRunnerError::invalid(format!( + "controller sequence must be contiguous: expected {expected}, received {}", + command.controller_seq + ))); + } + + self.last_controller_command_seq = command.controller_seq; + self.processed_commands.insert( + command.command_id.clone(), + command_result( + command, + "pending", + json!({ + "code": "execution_indeterminate", + "message": "command was journaled before its effect", + }), + ), + ); + self.processed_command_fingerprints + .insert(command.command_id.clone(), fingerprint); + self.compact_command_history(); + Ok(CommandDisposition::Execute) + } + + pub fn complete_command( + &mut self, + command: &Command, + result: Value, + ) -> Result { + let stored = self + .processed_commands + .get_mut(&command.command_id) + .ok_or_else(|| { + DurableRunnerError::invalid("command was not journaled before completion") + })?; + if stored.controller_seq != command.controller_seq || stored.status != "pending" { + return Err(DurableRunnerError::invalid( + "command completion does not match a pending journal entry", + )); + } + let result = sanitize_value(&result); + let result_bytes = serde_json::to_vec(&result) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))? + .len(); + if result_bytes > MAX_COMMAND_RESULT_BYTES { + return Err(DurableRunnerError::invalid( + "command result exceeds the 64 KiB durable journal limit", + )); + } + stored.status = "completed".to_owned(); + stored.result = result; + Ok(stored.clone()) + } + + pub fn reconcile_pending_commands(&mut self) -> bool { + let mut changed = false; + for command in self.processed_commands.values_mut() { + if command.status == "pending" { + command.status = "indeterminate".to_owned(); + command.result = json!({ + "code": "execution_indeterminate", + "message": "runner recovered after journaling this command; it will not execute twice", + }); + changed = true; + } + } + changed + } + + fn has_legacy_command_journal(&self) -> bool { + !self.processed_commands.is_empty() && self.processed_command_fingerprints.is_empty() + } + + fn compact_legacy_command_journal(&mut self) { + self.processed_commands.clear(); + self.processed_command_fingerprints.clear(); + self.compacted_through_controller_seq = self.last_controller_command_seq; + self.record_diagnostic( + "pre-fingerprint command journal was compacted; prior commands remain non-reexecutable", + ); + } + + pub(crate) 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); + } + } + + fn compact_command_history(&mut self) { + while self.processed_commands.len() > MAX_RECENT_COMMANDS { + let Some(oldest_id) = self + .processed_commands + .values() + .min_by_key(|command| command.controller_seq) + .map(|command| command.command_id.clone()) + else { + break; + }; + if let Some(oldest) = self.processed_commands.remove(&oldest_id) { + self.processed_command_fingerprints.remove(&oldest_id); + self.compacted_through_controller_seq = self + .compacted_through_controller_seq + .max(oldest.controller_seq); + } + } + } +} + +fn command_result(command: &Command, status: &str, result: Value) -> StoredCommandResult { + StoredCommandResult { + command_id: command.command_id.clone(), + controller_seq: command.controller_seq, + command_type: command.command_type.clone(), + status: status.to_owned(), + result, + } +} + +fn command_fingerprint(command: &Command) -> Result { + let value = serde_json::to_value(command).map_err(|error| { + DurableRunnerError::invalid(format!("failed to fingerprint durable command: {error}")) + })?; + let digest = Sha256::digest(canonical_json(&value).as_bytes()); + const HEX: &[u8; 16] = b"0123456789abcdef"; + let mut fingerprint = String::with_capacity(digest.len() * 2); + for byte in digest { + fingerprint.push(HEX[usize::from(byte >> 4)] as char); + fingerprint.push(HEX[usize::from(byte & 0x0f)] as char); + } + Ok(fingerprint) +} + +fn canonical_json(value: &Value) -> String { + match value { + Value::Array(values) => format!( + "[{}]", + values + .iter() + .map(canonical_json) + .collect::>() + .join(",") + ), + Value::Object(values) => { + let mut keys = values.keys().collect::>(); + keys.sort(); + format!( + "{{{}}}", + keys.into_iter() + .map(|key| format!( + "{}:{}", + serde_json::to_string(key).expect("JSON object key should serialize"), + canonical_json(&values[key]) + )) + .collect::>() + .join(",") + ) + } + _ => value.to_string(), + } +} + +#[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.is_dir() { + return Err(DurableRunnerError::invalid(format!( + "runner state directory {} must be a real directory", + state_dir.display() + ))); + } + } + fs::create_dir_all(state_dir).map_err(|error| { + DurableRunnerError::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| { + DurableRunnerError::invalid(format!( + "failed to secure runner state directory {}: {error}", + state_dir.display() + )) + })?; + verify_private_directory(state_dir)?; + Ok(Self { + path: state_dir.join(STATE_FILE), + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn load_or_create( + &self, + config: &DurableRunnerConfig, + ) -> Result<(DurableState, bool), DurableRunnerError> { + let mut bytes = Vec::new(); + match open_private_regular_file(&self.path) { + Ok(mut file) => { + let maximum_state_bytes = + config.max_outbox_bytes.saturating_add(STATE_OVERHEAD_BYTES); + let file_bytes = usize::try_from( + file.metadata() + .map_err(|error| DurableRunnerError::invalid(error.to_string()))? + .len(), + ) + .map_err(|_| DurableRunnerError::invalid("durable state length overflowed"))?; + if file_bytes > maximum_state_bytes { + return Err(DurableRunnerError::invalid( + "durable state exceeds its configured storage bound", + )); + } + file.read_to_end(&mut bytes).map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to read durable state {}: {error}", + self.path.display() + )) + })? + } + Err(error) if error.kind() == io::ErrorKind::NotFound => { + let state = DurableState::new(config); + self.save(&state)?; + return Ok((state, false)); + } + Err(error) => { + return Err(DurableRunnerError::invalid(format!( + "failed to open durable state {}: {error}", + self.path.display() + ))) + } + }; + let mut state: DurableState = serde_json::from_slice(&bytes).map_err(|error| { + DurableRunnerError::invalid(format!( + "durable state is malformed and cannot be recovered: {error}" + )) + })?; + let has_legacy_command_journal = state.has_legacy_command_journal(); + validate_binding(&state, config, has_legacy_command_journal)?; + let mut changed = false; + if has_legacy_command_journal { + state.compact_legacy_command_journal(); + validate_binding(&state, config, false)?; + changed = true; + } + if state.reconcile_pending_commands() { + changed = true; + } + if changed { + self.save(&state)?; + } + Ok((state, true)) + } + + pub fn save(&self, state: &DurableState) -> Result<(), DurableRunnerError> { + let bytes = serde_json::to_vec_pretty(state).map_err(|error| { + DurableRunnerError::invalid(format!("failed to serialize durable state: {error}")) + })?; + if bytes.len() > state.max_outbox_bytes.saturating_add(STATE_OVERHEAD_BYTES) { + return Err(DurableRunnerError::invalid( + "durable state exceeds its configured storage bound", + )); + } + let (temporary, mut file) = create_private_temporary_file(&self.path)?; + let result = (|| -> Result<(), DurableRunnerError> { + file.write_all(&bytes) + .and_then(|_| file.sync_all()) + .map_err(|error| { + DurableRunnerError::invalid(format!("failed to commit durable state: {error}")) + })?; + drop(file); + fs::rename(&temporary, &self.path).map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to atomically replace durable state: {error}" + )) + })?; + #[cfg(unix)] + if let Some(parent) = self.path.parent() { + File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to sync durable state directory: {error}" + )) + })?; + } + Ok(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary); + } + result + } +} + +fn validate_binding( + state: &DurableState, + config: &DurableRunnerConfig, + allow_legacy_command_journal: bool, +) -> Result<(), DurableRunnerError> { + if state.schema != STATE_SCHEMA + || state.runner_instance_id != config.runner_instance_id + || state.environment_lease_id != config.environment_lease_id + || state.run_id != config.run_id + || state.normalized_session_id != config.normalized_session_id + || state.turn_id != config.turn_id + || state.item_id != config.item_id + || state.max_outbox_bytes != config.max_outbox_bytes + || state.p0_reserve_bytes != config.p0_reserve_bytes + { + return Err(DurableRunnerError::invalid( + "durable state binding does not match this runner invocation", + )); + } + let outbox_bytes = state.outbox.iter().try_fold(0_usize, |total, event| { + let serialized = serde_json::to_vec(&event.envelope) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + if event.byte_size != serialized.len() + || event + .envelope + .pointer("/payload/sourceSeq") + .and_then(Value::as_u64) + != Some(event.source_seq) + || event.priority > 2 + { + return Err(DurableRunnerError::invalid( + "durable outbox metadata does not match its envelope", + )); + } + total + .checked_add(event.byte_size) + .ok_or_else(|| DurableRunnerError::invalid("durable outbox size overflowed")) + })?; + let outbox_cursors_are_valid = match (state.outbox.first(), state.outbox.last()) { + (None, None) => state.acked_source_seq == state.highest_source_seq(), + (Some(first), Some(last)) => { + state.acked_source_seq.checked_add(1) == Some(first.source_seq) + && last.source_seq == state.highest_source_seq() + } + _ => false, + }; + let mut command_sequences = state + .processed_commands + .iter() + .map(|(key, command)| { + if key != &command.command_id + || command.controller_seq <= state.compacted_through_controller_seq + || command.controller_seq > state.last_controller_command_seq + || !matches!( + command.status.as_str(), + "pending" | "completed" | "indeterminate" + ) + { + return Err(DurableRunnerError::invalid( + "durable command journal metadata is inconsistent", + )); + } + Ok(command.controller_seq) + }) + .collect::, _>>()?; + let command_fingerprints_are_valid = (state.processed_command_fingerprints.len() + == state.processed_commands.len() + && state + .processed_command_fingerprints + .iter() + .all(|(key, value)| { + state.processed_commands.contains_key(key) + && value.len() == 64 + && value.bytes().all(|byte| byte.is_ascii_hexdigit()) + })) + || (allow_legacy_command_journal + && !state.processed_commands.is_empty() + && state.processed_command_fingerprints.is_empty()); + command_sequences.sort_unstable(); + let command_cursors_are_valid = match (command_sequences.first(), command_sequences.last()) { + (None, None) => state.compacted_through_controller_seq == state.last_controller_command_seq, + (Some(first), Some(last)) => { + state.compacted_through_controller_seq.checked_add(1) == Some(*first) + && *last == state.last_controller_command_seq + && command_sequences + .windows(2) + .all(|pair| pair[0].checked_add(1) == Some(pair[1])) + } + _ => false, + }; + + if state.next_source_seq == 0 + || state.acked_source_seq > state.highest_source_seq() + || !outbox_cursors_are_valid + || state + .outbox + .windows(2) + .any(|pair| pair[0].source_seq.checked_add(1) != Some(pair[1].source_seq)) + || outbox_bytes > state.max_outbox_bytes + || state.peak_outbox_bytes < outbox_bytes + || state.compacted_through_controller_seq > state.last_controller_command_seq + || !command_cursors_are_valid + || !command_fingerprints_are_valid + { + return Err(DurableRunnerError::invalid( + "durable state cursors, bounds, or journals are inconsistent", + )); + } + Ok(()) +} + +fn verify_private_directory(path: &Path) -> Result<(), DurableRunnerError> { + let metadata = fs::symlink_metadata(path) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(DurableRunnerError::invalid( + "durable state directory must not be a symlink", + )); + } + #[cfg(unix)] + if metadata.permissions().mode() & 0o077 != 0 { + return Err(DurableRunnerError::invalid( + "durable state directory must not be accessible by group or other users", + )); + } + 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.is_file() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "durable state path is not a regular file", + )); + } + #[cfg(unix)] + if metadata.permissions().mode() & 0o077 != 0 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "durable state file is accessible by group or other users", + )); + } + Ok(file) +} + +#[cfg(any(target_os = "linux", target_os = "android"))] +const fn no_follow_flag() -> i32 { + 0o400000 +} + +#[cfg(any(target_os = "macos", target_os = "ios", target_os = "freebsd"))] +const fn no_follow_flag() -> i32 { + 0x00000100 +} + +#[cfg(all( + unix, + not(any( + target_os = "linux", + target_os = "android", + target_os = "macos", + target_os = "ios", + target_os = "freebsd" + )) +))] +const fn no_follow_flag() -> i32 { + 0 +} + +fn create_private_temporary_file(path: &Path) -> Result<(PathBuf, File), DurableRunnerError> { + let parent = path + .parent() + .ok_or_else(|| DurableRunnerError::invalid("durable state path has no parent"))?; + let process_id = std::process::id(); + for attempt in 0..TEMP_FILE_ATTEMPTS { + let temporary = parent.join(format!(".{STATE_FILE}.{process_id}.{attempt}.tmp")); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o600).custom_flags(no_follow_flag()); + match options.open(&temporary) { + Ok(file) => return Ok((temporary, file)), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue, + Err(error) => { + return Err(DurableRunnerError::invalid(format!( + "failed to create private durable state temporary file: {error}" + ))) + } + } + } + Err(DurableRunnerError::invalid( + "failed to allocate a private durable state temporary file", + )) +} + +fn sensitive_key(key: &str) -> bool { + let normalized = key.to_ascii_lowercase().replace(['-', '_'], ""); + [ + "authorization", + "cookie", + "password", + "secret", + "token", + "ticket", + "apikey", + "credential", + ] + .iter() + .any(|needle| normalized.contains(needle)) +} + +fn sanitize_value(value: &Value) -> Value { + match value { + Value::Object(object) => Value::Object( + object + .iter() + .map(|(key, value)| { + ( + key.clone(), + if sensitive_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(value) => Value::String(redact_text(value)), + value => value.clone(), + } +} + +fn redact_text(input: &str) -> String { + let (bounded, truncated) = if input.len() > 4096 { + let boundary = input + .char_indices() + .map(|(index, _)| index) + .take_while(|index| *index <= 4096) + .last() + .unwrap_or(0); + (&input[..boundary], true) + } else { + (input, false) + }; + let normalized = bounded.to_ascii_lowercase(); + if [ + "authorization", + "bearer ", + "api_key", + "apikey", + "password=", + "secret=", + "ticket=", + "token=", + ] + .iter() + .any(|marker| normalized.contains(marker)) + { + "[REDACTED diagnostic containing a sensitive marker]".to_owned() + } else if truncated { + format!("{bounded}…[truncated]") + } else { + bounded.to_owned() + } +} + +fn current_timestamp() -> Result { + let duration = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|error| { + DurableRunnerError::invalid(format!("system clock is invalid: {error}")) + })?; + let total_seconds = i64::try_from(duration.as_secs()) + .map_err(|_| DurableRunnerError::invalid("system clock value overflowed"))?; + let days = total_seconds.div_euclid(86_400); + let second_of_day = total_seconds.rem_euclid(86_400); + let shifted = days + 719_468; + let era = shifted.div_euclid(146_097); + let day_of_era = shifted - era * 146_097; + let year_of_era = + (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365; + let mut year = year_of_era + era * 400; + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); + let month_prime = (5 * day_of_year + 2) / 153; + let day = day_of_year - (153 * month_prime + 2) / 5 + 1; + let month = month_prime + if month_prime < 10 { 3 } else { -9 }; + year += i64::from(month <= 2); + if !(0..=9999).contains(&year) { + return Err(DurableRunnerError::invalid( + "system clock is outside the supported RFC 3339 range", + )); + } + let hour = second_of_day / 3600; + let minute = second_of_day % 3600 / 60; + let second = second_of_day % 60; + Ok(format!( + "{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}.{:03}Z", + duration.subsec_millis() + )) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + fn config(state_dir: PathBuf) -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: "ws://127.0.0.1:3000/api/runner/v1/connect/run_1".to_owned(), + state_dir, + runner_instance_id: "runner_1".to_owned(), + environment_lease_id: "environment_1".to_owned(), + run_id: "run_1".to_owned(), + normalized_session_id: "session_1".to_owned(), + turn_id: "turn_1".to_owned(), + item_id: "item_1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: "sha256:test".to_owned(), + max_outbox_bytes: 16_384, + p0_reserve_bytes: 4096, + max_frame_bytes: 65_536, + reconnect_delay: Duration::from_millis(1), + max_runtime: Duration::from_secs(1), + } + } + + fn command(id: &str, sequence: u64) -> Command { + Command { + schema: "paperclip.prp.command.v1".to_owned(), + command_id: id.to_owned(), + controller_seq: sequence, + command_type: "session.open".to_owned(), + issued_at: "2026-08-24T00:00:00.000Z".to_owned(), + deadline_at: None, + precondition: None, + payload: json!({}), + } + } + + fn temporary_directory(label: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "paperclip-runner-durable-{label}-{}", + std::process::id() + )) + } + + #[test] + fn cumulative_ack_is_monotonic_and_bounded() { + let config = config(PathBuf::from("unused")); + let mut state = DurableState::new(&config); + state + .enqueue_event(&config, "runner.connected", EventPriority::P0, json!({})) + .unwrap(); + state + .enqueue_event(&config, "runner.reconnected", EventPriority::P1, json!({})) + .unwrap(); + state.apply_ack(1).unwrap(); + assert_eq!(state.outbox.len(), 1); + assert!(state.apply_ack(0).is_err()); + assert!(state.apply_ack(3).is_err()); + } + + #[test] + fn duplicate_command_replays_without_executing() { + let config = config(PathBuf::from("unused")); + let mut state = DurableState::new(&config); + let command = command("command_1", 1); + assert_eq!( + state.begin_command(&command).unwrap(), + CommandDisposition::Execute + ); + state + .complete_command(&command, json!({"ok": true})) + .unwrap(); + assert!(matches!( + state.begin_command(&command).unwrap(), + CommandDisposition::Replay(result) if result.result == json!({"ok": true}) + )); + } + + #[test] + fn command_gaps_and_identifier_reuse_fail_closed() { + let config = config(PathBuf::from("unused")); + let mut state = DurableState::new(&config); + let mut unknown = command("command_unknown", 1); + unknown.command_type = "future.required.command".to_owned(); + assert!(state.begin_command(&unknown).is_err()); + assert!(state.begin_command(&command("command_2", 2)).is_err()); + let first = command("command_1", 1); + state.begin_command(&first).unwrap(); + state.complete_command(&first, json!({})).unwrap(); + assert!(state.begin_command(&command("command_1", 2)).is_err()); + } + + #[test] + fn recovery_marks_ambiguous_effect_without_reexecution() { + let directory = temporary_directory("ambiguous"); + let _ = fs::remove_dir_all(&directory); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let command = command("command_1", 1); + state.begin_command(&command).unwrap(); + store.save(&state).unwrap(); + + let (mut recovered, existed) = store.load_or_create(&config).unwrap(); + assert!(existed); + assert!(matches!( + recovered.begin_command(&command).unwrap(), + CommandDisposition::Replay(result) if result.status == "indeterminate" + )); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn state_binding_prevents_cross_run_reuse() { + let directory = temporary_directory("binding"); + let _ = fs::remove_dir_all(&directory); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + store.load_or_create(&config).unwrap(); + let mut wrong = config.clone(); + wrong.run_id = "run_2".to_owned(); + assert!(store.load_or_create(&wrong).is_err()); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn event_payloads_are_redacted_before_persistence() { + let config = config(PathBuf::from("unused")); + let mut state = DurableState::new(&config); + state + .enqueue_event( + &config, + "runner.diagnostic", + EventPriority::P1, + json!({"nested": {"api_token": "secret-value"}}), + ) + .unwrap(); + assert_eq!( + state.outbox[0] + .envelope + .pointer("/payload/payload/nested/api_token"), + Some(&Value::String("[REDACTED]".to_owned())) + ); + } + + #[test] + fn outbox_reserves_capacity_for_p0_and_bounds_frames() { + let mut bounds_config = config(PathBuf::from("unused")); + bounds_config.max_outbox_bytes = 1800; + bounds_config.p0_reserve_bytes = 600; + let mut state = DurableState::new(&bounds_config); + while state + .enqueue_event( + &bounds_config, + "item.delta", + EventPriority::P1, + json!({"text": "x".repeat(200)}), + ) + .is_ok() + {} + assert!(state.backpressure); + assert!(state + .enqueue_event( + &bounds_config, + "runner.diagnostic", + EventPriority::P0, + json!({"message": "storage pressure"}), + ) + .is_ok()); + + let mut frame_limited = config(PathBuf::from("unused")); + frame_limited.max_frame_bytes = 1024; + let mut state = DurableState::new(&frame_limited); + assert!(state + .enqueue_event( + &frame_limited, + "item.delta", + EventPriority::P1, + json!({"text": "x".repeat(2048)}), + ) + .is_err()); + assert!(state.outbox.is_empty()); + } +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs new file mode 100644 index 0000000000..8bbab5733b --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs @@ -0,0 +1,1565 @@ +use std::io; +use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use aes_gcm::aead::{Aead, Payload}; +use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; +use hmac::{Hmac, Mac}; +use serde::Deserialize; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use tungstenite::client::{client_with_config, IntoClientRequest}; +use tungstenite::protocol::WebSocketConfig; +use tungstenite::{Message, WebSocket}; + +use super::state::{Command, DurableState}; +use super::{ + BootstrapTicket, DurableRunnerConfig, DurableRunnerError, Secret, PROTOCOL, PROTOCOL_VERSION, +}; + +const SECURE_FRAME_SCHEMA: &str = "paperclip.runner.secure-frame.v1"; +const AUTH_TIMEOUT: Duration = Duration::from_secs(2); + +type HmacSha256 = Hmac; + +#[derive(Clone, Debug)] +struct ParsedWsUrl { + host: String, + authority: String, + port: u16, + path: String, +} + +#[derive(Clone, Debug)] +pub(crate) struct ResolvedWsTarget { + authority: String, + path: String, + addresses: Vec, +} + +impl ResolvedWsTarget { + pub(crate) fn resolve(input: &str) -> Result { + resolve_ws_target_with(input, |host, port| { + (host, port) + .to_socket_addrs() + .map(|addresses| addresses.collect()) + }) + } + + fn request_url(&self) -> String { + format!("ws://{}{}", self.authority, self.path) + } +} + +fn parse_ws_url(input: &str) -> Result { + let remainder = input + .strip_prefix("ws://") + .ok_or_else(|| DurableRunnerError::invalid("runner transport accepts exactly ws://"))?; + if remainder.is_empty() + || remainder + .chars() + .any(|character| character.is_ascii_control() || character.is_ascii_whitespace()) + || remainder.contains(['?', '#', '\\']) + { + return Err(DurableRunnerError::invalid( + "WebSocket URL contains query, fragment, whitespace, 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(DurableRunnerError::invalid( + "WebSocket authority must not contain userinfo or encoding ambiguity", + )); + } + let (host, port) = if authority.starts_with('[') { + let closing = authority + .find(']') + .ok_or_else(|| DurableRunnerError::invalid("bracketed IPv6 authority is malformed"))?; + let host = &authority[1..closing]; + let port = authority[closing + 1..].strip_prefix(':').ok_or_else(|| { + DurableRunnerError::invalid("bracketed IPv6 authority requires a port") + })?; + host.parse::() + .map_err(|_| DurableRunnerError::invalid("bracketed WebSocket host must be IPv6"))?; + (host, port) + } else { + let (host, port) = authority.rsplit_once(':').ok_or_else(|| { + DurableRunnerError::invalid("WebSocket URL requires an explicit port") + })?; + if host.is_empty() || host.contains(':') { + return Err(DurableRunnerError::invalid( + "WebSocket host is empty or contains unbracketed IPv6", + )); + } + (host, port) + }; + let port = port + .parse::() + .map_err(|error| DurableRunnerError::invalid(format!("invalid WebSocket port: {error}")))?; + if port == 0 { + return Err(DurableRunnerError::invalid( + "WebSocket port must be non-zero", + )); + } + Ok(ParsedWsUrl { + host: host.to_owned(), + authority: authority.to_owned(), + port, + path, + }) +} + +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| { + DurableRunnerError::invalid(format!("failed to resolve WebSocket destination: {error}")) + })?; + addresses.sort_unstable(); + addresses.dedup(); + if addresses.is_empty() { + return Err(DurableRunnerError::invalid( + "WebSocket destination resolved to no addresses", + )); + } + if addresses.iter().any(|address| !address.ip().is_loopback()) { + return Err(DurableRunnerError::invalid( + "every WebSocket destination must resolve to loopback", + )); + } + Ok(ResolvedWsTarget { + authority: parsed.authority, + path: parsed.path, + addresses, + }) +} + +#[derive(Debug)] +struct CredentialMaterial { + credential_id: String, + auth_key: [u8; 32], +} + +impl CredentialMaterial { + fn from_token(token: &str) -> Self { + Self { + credential_id: format!( + "sha256:{}", + hex_encode(&digest_domain( + "paperclip-runner-credential-id-v1", + &[token.as_bytes()] + )) + ), + auth_key: digest_domain("paperclip-runner-auth-key-v1", &[token.as_bytes()]), + } + } +} + +impl Drop for CredentialMaterial { + fn drop(&mut self) { + self.auth_key.fill(0); + } +} + +#[derive(Debug)] +pub(crate) struct LeaseCredential { + pub(crate) lease_id: String, + pub(crate) expires_at_unix_ms: u64, + pub(crate) revocation_epoch: u64, + token: Secret, +} + +impl LeaseCredential { + fn expose(&self) -> Result<&str, DurableRunnerError> { + self.token.expose() + } +} + +#[derive(Clone, Debug)] +pub(crate) struct ConnectionMetadata { + pub(crate) connection_id: String, + pub(crate) lease_id: String, + pub(crate) expires_at_unix_ms: u64, + pub(crate) revocation_epoch: u64, +} + +#[derive(Debug)] +pub(crate) struct Welcome { + pub(crate) connection: ConnectionMetadata, + pub(crate) lease: Option, + pub(crate) acked_source_seq: Option, + pub(crate) pending_commands: Vec, +} + +struct SecureChannel { + send_cipher: Aes256Gcm, + receive_cipher: Aes256Gcm, + send_counter: u64, + receive_counter: u64, + session_id: String, +} + +impl SecureChannel { + fn client( + auth_key: &[u8], + challenge: &[u8], + server_proof: &[u8], + client_proof: &[u8], + ) -> Result { + Self::new(auth_key, challenge, server_proof, client_proof, false) + } + + #[cfg(test)] + fn server( + auth_key: &[u8], + challenge: &[u8], + server_proof: &[u8], + client_proof: &[u8], + ) -> Result { + Self::new(auth_key, challenge, server_proof, client_proof, true) + } + + fn new( + auth_key: &[u8], + challenge: &[u8], + server_proof: &[u8], + client_proof: &[u8], + server_direction: bool, + ) -> Result { + let session_binding = digest_domain( + "paperclip-runner-session-binding-v1", + &[challenge, server_proof, client_proof], + ); + let client_to_server = hmac_domain( + auth_key, + "paperclip-runner-client-to-core-key-v1", + &[&session_binding], + ); + let server_to_client = hmac_domain( + auth_key, + "paperclip-runner-core-to-client-key-v1", + &[&session_binding], + ); + let (send_key, receive_key) = if server_direction { + (server_to_client, client_to_server) + } else { + (client_to_server, server_to_client) + }; + Ok(Self { + send_cipher: Aes256Gcm::new_from_slice(&send_key) + .map_err(|_| DurableRunnerError::invalid("failed to initialize encryption"))?, + receive_cipher: Aes256Gcm::new_from_slice(&receive_key) + .map_err(|_| DurableRunnerError::invalid("failed to initialize decryption"))?, + send_counter: 0, + receive_counter: 0, + session_id: format!("sha256:{}", hex_encode(&session_binding)), + }) + } + + fn nonce(direction: &[u8; 4], counter: u64) -> [u8; 12] { + let mut nonce = [0_u8; 12]; + nonce[..4].copy_from_slice(direction); + nonce[4..].copy_from_slice(&counter.to_be_bytes()); + nonce + } + + fn aad(&self, direction: &str, counter: u64) -> Vec { + format!( + "{SECURE_FRAME_SCHEMA}\0{}\0{direction}\0{counter}", + self.session_id + ) + .into_bytes() + } + + fn encrypt( + &mut self, + plaintext: &[u8], + server_direction: bool, + ) -> Result { + let counter = self.send_counter; + let direction = if server_direction { b"P3S1" } else { b"P3C1" }; + let label = if server_direction { + "core_to_client" + } else { + "client_to_core" + }; + let nonce = Self::nonce(direction, counter); + let aad = self.aad(label, counter); + let ciphertext = self + .send_cipher + .encrypt( + Nonce::from_slice(&nonce), + Payload { + msg: plaintext, + aad: &aad, + }, + ) + .map_err(|_| DurableRunnerError::invalid("secure transport encryption failed"))?; + self.send_counter = self + .send_counter + .checked_add(1) + .ok_or_else(|| DurableRunnerError::invalid("secure send counter exhausted"))?; + Ok(json!({ + "schema": SECURE_FRAME_SCHEMA, + "counter": counter, + "ciphertext": hex_encode(&ciphertext), + })) + } + + fn decrypt( + &mut self, + frame: &Value, + server_direction: bool, + ) -> Result { + if frame.get("schema").and_then(Value::as_str) != Some(SECURE_FRAME_SCHEMA) { + return Err(DurableRunnerError::invalid( + "unauthenticated plaintext control frame was rejected", + )); + } + let counter = frame + .get("counter") + .and_then(Value::as_u64) + .ok_or_else(|| DurableRunnerError::invalid("secure frame counter is required"))?; + if counter != self.receive_counter { + return Err(DurableRunnerError::invalid( + "secure frame counter was replayed or arrived out of order", + )); + } + let ciphertext = frame + .get("ciphertext") + .and_then(Value::as_str) + .ok_or_else(|| DurableRunnerError::invalid("secure frame ciphertext is required"))?; + let ciphertext = hex_decode(ciphertext)?; + let direction = if server_direction { b"P3C1" } else { b"P3S1" }; + let label = if server_direction { + "client_to_core" + } else { + "core_to_client" + }; + let nonce = Self::nonce(direction, counter); + let aad = self.aad(label, counter); + let plaintext = self + .receive_cipher + .decrypt( + Nonce::from_slice(&nonce), + Payload { + msg: &ciphertext, + aad: &aad, + }, + ) + .map_err(|_| DurableRunnerError::invalid("secure frame authentication failed"))?; + self.receive_counter = self + .receive_counter + .checked_add(1) + .ok_or_else(|| DurableRunnerError::invalid("secure receive counter exhausted"))?; + serde_json::from_slice(&plaintext).map_err(|error| { + DurableRunnerError::invalid(format!("secure JSON is malformed: {error}")) + }) + } +} + +pub(crate) struct AuthenticatedTransport { + socket: WebSocket, + secure_channel: SecureChannel, + max_frame_bytes: usize, +} + +#[derive(Debug)] +pub(crate) struct ConnectFailure { + error: DurableRunnerError, + pub(crate) bootstrap_maybe_consumed: bool, +} + +impl std::fmt::Display for ConnectFailure { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.error.fmt(formatter) + } +} + +impl ConnectFailure { + fn retryable(error: DurableRunnerError) -> Self { + Self { + error, + bootstrap_maybe_consumed: false, + } + } + + fn after_auth_started(error: DurableRunnerError, credential_kind: &str) -> Self { + Self { + error, + bootstrap_maybe_consumed: credential_kind == "bootstrap", + } + } +} + +impl AuthenticatedTransport { + pub(crate) fn connect( + target: &ResolvedWsTarget, + config: &DurableRunnerConfig, + state: &DurableState, + bootstrap: Option<&BootstrapTicket>, + lease: Option<&LeaseCredential>, + ) -> Result<(Self, Welcome), ConnectFailure> { + let (credential_token, credential_kind, expected_lease) = match (lease, bootstrap) { + (Some(lease), _) => ( + lease.expose().map_err(ConnectFailure::retryable)?, + "lease", + Some(lease), + ), + (None, Some(bootstrap)) => ( + bootstrap.expose().map_err(ConnectFailure::retryable)?, + "bootstrap", + None, + ), + (None, None) => { + return Err(ConnectFailure::retryable(DurableRunnerError::invalid( + "a bootstrap or unexpired connection lease is required", + ))) + } + }; + let credential = CredentialMaterial::from_token(credential_token); + let stream = TcpStream::connect(target.addresses.as_slice()) + .map_err(|error| { + DurableRunnerError::invalid(format!("WebSocket connect failed: {error}")) + }) + .map_err(ConnectFailure::retryable)?; + stream + .set_read_timeout(Some(AUTH_TIMEOUT)) + .and_then(|()| stream.set_write_timeout(Some(AUTH_TIMEOUT))) + .map_err(|error| DurableRunnerError::invalid(error.to_string())) + .map_err(ConnectFailure::retryable)?; + let request = target + .request_url() + .into_client_request() + .map_err(|error| { + DurableRunnerError::invalid(format!("invalid WebSocket request: {error}")) + }) + .map_err(ConnectFailure::retryable)?; + let websocket_config = WebSocketConfig::default() + .max_message_size(Some(config.max_frame_bytes)) + .max_frame_size(Some(config.max_frame_bytes)); + let (mut socket, _) = client_with_config(request, stream, Some(websocket_config)) + .map_err(|error| { + DurableRunnerError::invalid(format!("WebSocket upgrade failed: {error}")) + }) + .map_err(ConnectFailure::retryable)?; + + let authenticate = + || -> Result<(Self, Welcome), DurableRunnerError> { + let client_nonce = random_nonce()?; + send_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_hello", + "payload": { + "credentialId": credential.credential_id, + "credentialKind": credential_kind, + "clientNonce": client_nonce, + "protocolMin": PROTOCOL_VERSION, + "protocolMax": PROTOCOL_VERSION, + "runnerInstanceId": state.runner_instance_id, + "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "runnerVersion": config.runner_version, + "runnerDigest": config.runner_digest, + "resume": { + "lastControllerCommandSeq": state.last_controller_command_seq, + "nextSourceEventSeq": state.next_source_seq, + "ackedSourceSeq": state.acked_source_seq, + }, + }, + }), + config.max_frame_bytes, + )?; + + let challenge_value = receive_plain(&mut socket, config.max_frame_bytes)?; + validate_envelope_kind(&challenge_value, "auth_challenge")?; + let challenge: AuthChallenge = + serde_json::from_value(challenge_value.get("payload").cloned().ok_or_else( + || DurableRunnerError::invalid("challenge payload is required"), + )?) + .map_err(|error| { + DurableRunnerError::invalid(format!("invalid auth challenge: {error}")) + })?; + validate_challenge( + &challenge, + state, + config, + &credential, + credential_kind, + &client_nonce, + expected_lease, + )?; + let signing_bytes = challenge_signing_bytes(&challenge); + verify_hmac_hex( + &credential.auth_key, + "paperclip-runner-server-proof-v1", + &[&signing_bytes], + &challenge.server_proof, + )?; + let client_proof = hex_encode(&hmac_domain( + &credential.auth_key, + "paperclip-runner-client-proof-v1", + &[&signing_bytes, challenge.server_proof.as_bytes()], + )); + send_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_response", + "payload": { + "credentialId": credential.credential_id, + "clientNonce": client_nonce, + "serverNonce": challenge.server_nonce, + "clientProof": client_proof, + }, + }), + config.max_frame_bytes, + )?; + let secure_channel = SecureChannel::client( + &credential.auth_key, + &signing_bytes, + challenge.server_proof.as_bytes(), + client_proof.as_bytes(), + )?; + let mut transport = Self { + socket, + secure_channel, + max_frame_bytes: config.max_frame_bytes, + }; + transport + .socket + .get_mut() + .set_read_timeout(Some(Duration::from_millis(250))) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + let mut welcome_value = transport.receive_json()?.ok_or_else(|| { + DurableRunnerError::invalid("authenticated welcome timed out") + })?; + let welcome = + validate_welcome(&mut welcome_value, state, credential_kind, expected_lease)?; + Ok((transport, welcome)) + }; + authenticate().map_err(|error| ConnectFailure::after_auth_started(error, credential_kind)) + } + + pub(crate) fn send_json(&mut self, value: &Value) -> Result<(), DurableRunnerError> { + let bytes = serde_json::to_vec(value) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + if bytes.len() > self.max_frame_bytes { + return Err(DurableRunnerError::invalid( + "outbound secure frame exceeds the configured limit", + )); + } + let frame = self.secure_channel.encrypt(&bytes, false)?; + send_plain(&mut self.socket, &frame, self.max_frame_bytes) + } + + pub(crate) fn receive_json(&mut self) -> Result, DurableRunnerError> { + let frame = match receive_plain_optional(&mut self.socket, self.max_frame_bytes)? { + Some(frame) => frame, + None => return Ok(None), + }; + self.secure_channel.decrypt(&frame, false).map(Some) + } +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct AuthChallenge { + credential_id: String, + credential_kind: String, + client_nonce: String, + server_nonce: String, + runner_instance_id: String, + environment_lease_id: String, + run_id: String, + normalized_session_id: String, + turn_id: String, + item_id: String, + runner_version: String, + runner_digest: String, + selected_version: u64, + credential_expires_at_unix_ms: u64, + credential_lease_id: Option, + revocation_epoch: u64, + server_proof: String, +} + +fn validate_challenge( + challenge: &AuthChallenge, + state: &DurableState, + config: &DurableRunnerConfig, + credential: &CredentialMaterial, + credential_kind: &str, + client_nonce: &str, + expected_lease: Option<&LeaseCredential>, +) -> Result<(), DurableRunnerError> { + for (field, actual, expected) in [ + ( + "credentialId", + challenge.credential_id.as_str(), + credential.credential_id.as_str(), + ), + ( + "credentialKind", + challenge.credential_kind.as_str(), + credential_kind, + ), + ("clientNonce", challenge.client_nonce.as_str(), client_nonce), + ( + "runnerInstanceId", + challenge.runner_instance_id.as_str(), + state.runner_instance_id.as_str(), + ), + ( + "environmentLeaseId", + challenge.environment_lease_id.as_str(), + state.environment_lease_id.as_str(), + ), + ("runId", challenge.run_id.as_str(), state.run_id.as_str()), + ( + "normalizedSessionId", + challenge.normalized_session_id.as_str(), + state.normalized_session_id.as_str(), + ), + ("turnId", challenge.turn_id.as_str(), state.turn_id.as_str()), + ("itemId", challenge.item_id.as_str(), state.item_id.as_str()), + ( + "runnerVersion", + challenge.runner_version.as_str(), + config.runner_version.as_str(), + ), + ( + "runnerDigest", + challenge.runner_digest.as_str(), + config.runner_digest.as_str(), + ), + ] { + if actual != expected { + return Err(DurableRunnerError::invalid(format!( + "authentication challenge {field} does not match this session" + ))); + } + } + if challenge.server_nonce.is_empty() + || challenge.selected_version != PROTOCOL_VERSION + || challenge.credential_expires_at_unix_ms <= current_unix_ms()? + { + return Err(DurableRunnerError::invalid( + "authentication challenge is expired or selected an unsupported protocol", + )); + } + match expected_lease { + Some(lease) + if challenge.credential_lease_id.as_deref() == Some(lease.lease_id.as_str()) + && challenge.credential_expires_at_unix_ms == lease.expires_at_unix_ms + && challenge.revocation_epoch == lease.revocation_epoch => {} + None if challenge.credential_lease_id.is_none() => {} + _ => { + return Err(DurableRunnerError::invalid( + "authentication challenge changed the credential lease binding", + )) + } + } + Ok(()) +} + +fn challenge_signing_bytes(challenge: &AuthChallenge) -> Vec { + let lease_id = challenge.credential_lease_id.as_deref().unwrap_or(""); + [ + challenge.credential_id.as_str(), + challenge.credential_kind.as_str(), + challenge.client_nonce.as_str(), + challenge.server_nonce.as_str(), + challenge.runner_instance_id.as_str(), + challenge.environment_lease_id.as_str(), + challenge.run_id.as_str(), + challenge.normalized_session_id.as_str(), + challenge.turn_id.as_str(), + challenge.item_id.as_str(), + challenge.runner_version.as_str(), + challenge.runner_digest.as_str(), + lease_id, + ] + .iter() + .fold(Vec::new(), |mut output, part| { + output.extend_from_slice(&(part.len() as u64).to_be_bytes()); + output.extend_from_slice(part.as_bytes()); + output + }) + .into_iter() + .chain(challenge.selected_version.to_be_bytes()) + .chain(challenge.credential_expires_at_unix_ms.to_be_bytes()) + .chain(challenge.revocation_epoch.to_be_bytes()) + .collect() +} + +fn validate_welcome( + value: &mut Value, + state: &DurableState, + credential_kind: &str, + expected_lease: Option<&LeaseCredential>, +) -> Result { + validate_control_identity(value, state, None)?; + validate_envelope_kind(value, "welcome")?; + let connection_id = required_string(value, "connectionId")?.to_owned(); + let connection_lease_id = required_string(value, "connectionLeaseId")?.to_owned(); + let payload = value + .get_mut("payload") + .ok_or_else(|| DurableRunnerError::invalid("welcome payload is required"))?; + if payload.get("selectedVersion").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) + || payload.get("connectionLeaseId").and_then(Value::as_str) + != Some(connection_lease_id.as_str()) + { + return Err(DurableRunnerError::invalid( + "welcome protocol or lease identity is inconsistent", + )); + } + let expires_at_unix_ms = payload + .get("connectionLeaseExpiresAtUnixMs") + .and_then(Value::as_u64) + .ok_or_else(|| DurableRunnerError::invalid("welcome lease expiry is required"))?; + if expires_at_unix_ms <= current_unix_ms()? { + return Err(DurableRunnerError::invalid( + "welcome carried an expired connection lease", + )); + } + let revocation_epoch = payload + .get("connectionLeaseRevocationEpoch") + .and_then(Value::as_u64) + .ok_or_else(|| DurableRunnerError::invalid("welcome revocation epoch is required"))?; + if let Some(expected) = expected_lease { + if connection_lease_id != expected.lease_id + || expires_at_unix_ms != expected.expires_at_unix_ms + || revocation_epoch != expected.revocation_epoch + { + return Err(DurableRunnerError::invalid( + "welcome changed the authenticated connection lease binding", + )); + } + } + let lease = match payload.get_mut("connectionLeaseToken") { + Some(Value::String(token)) if credential_kind == "bootstrap" && !token.is_empty() => { + let token = std::mem::take(token); + Some(LeaseCredential { + lease_id: connection_lease_id.clone(), + expires_at_unix_ms, + revocation_epoch, + token: Secret::new(token), + }) + } + None | Some(Value::Null) if credential_kind == "lease" => None, + _ => { + return Err(DurableRunnerError::invalid( + "bootstrap welcome must exchange the ticket for a connection lease token", + )) + } + }; + let pending_commands = payload + .get("pendingCommands") + .and_then(Value::as_array) + .map(|commands| { + commands + .iter() + .cloned() + .map(|command| { + serde_json::from_value(command).map_err(|error| { + DurableRunnerError::invalid(format!("invalid pending command: {error}")) + }) + }) + .collect::, _>>() + }) + .transpose()? + .unwrap_or_default(); + Ok(Welcome { + connection: ConnectionMetadata { + connection_id, + lease_id: connection_lease_id, + expires_at_unix_ms, + revocation_epoch, + }, + lease, + acked_source_seq: payload.get("ackedSourceSeq").and_then(Value::as_u64), + pending_commands, + }) +} + +pub(crate) fn validate_control_identity( + value: &Value, + state: &DurableState, + connection: Option<&ConnectionMetadata>, +) -> Result<(), DurableRunnerError> { + if value.get("protocol").and_then(Value::as_str) != Some(PROTOCOL) + || value.get("version").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) + { + return Err(DurableRunnerError::invalid( + "control envelope protocol identity is invalid", + )); + } + for (field, expected) in [ + ("runnerInstanceId", state.runner_instance_id.as_str()), + ("environmentLeaseId", state.environment_lease_id.as_str()), + ("runId", state.run_id.as_str()), + ("normalizedSessionId", state.normalized_session_id.as_str()), + ("turnId", state.turn_id.as_str()), + ("itemId", state.item_id.as_str()), + ] { + if required_string(value, field)? != expected { + return Err(DurableRunnerError::invalid(format!( + "control envelope {field} does not match the authenticated session" + ))); + } + } + if let Some(connection) = connection { + if required_string(value, "connectionId")? != connection.connection_id + || required_string(value, "connectionLeaseId")? != connection.lease_id + || current_unix_ms()? >= connection.expires_at_unix_ms + { + return Err(DurableRunnerError::invalid( + "control envelope connection lease is mismatched or expired", + )); + } + } + Ok(()) +} + +fn validate_envelope_kind(value: &Value, kind: &str) -> Result<(), DurableRunnerError> { + 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(kind) + { + return Err(DurableRunnerError::invalid(format!( + "expected a PRP v1 {kind} envelope" + ))); + } + Ok(()) +} + +fn required_string<'a>(value: &'a Value, field: &str) -> Result<&'a str, DurableRunnerError> { + value + .get(field) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .ok_or_else(|| DurableRunnerError::invalid(format!("{field} is required"))) +} + +fn send_plain( + socket: &mut WebSocket, + value: &Value, + max_frame_bytes: usize, +) -> Result<(), DurableRunnerError> { + let bytes = serde_json::to_vec(value) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + if bytes.len() > max_frame_bytes { + return Err(DurableRunnerError::invalid( + "outbound WebSocket message exceeds the configured limit", + )); + } + let text = + String::from_utf8(bytes).map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + socket + .send(Message::Text(text.into())) + .map_err(map_websocket_error) +} + +fn receive_plain( + socket: &mut WebSocket, + max_frame_bytes: usize, +) -> Result { + receive_plain_optional(socket, max_frame_bytes)? + .ok_or_else(|| DurableRunnerError::invalid("WebSocket message timed out")) +} + +fn receive_plain_optional( + socket: &mut WebSocket, + max_frame_bytes: usize, +) -> Result, DurableRunnerError> { + loop { + match socket.read() { + Ok(Message::Text(text)) => { + if text.len() > max_frame_bytes { + return Err(DurableRunnerError::invalid( + "inbound WebSocket message exceeds the configured limit", + )); + } + return serde_json::from_slice(text.as_bytes()) + .map(Some) + .map_err(|error| { + DurableRunnerError::invalid(format!("malformed WebSocket JSON: {error}")) + }); + } + Ok(Message::Ping(payload)) => socket + .send(Message::Pong(payload)) + .map_err(map_websocket_error)?, + Ok(Message::Pong(_)) => {} + Ok(Message::Close(_)) => { + return Err(DurableRunnerError::invalid( + "WebSocket peer closed the connection", + )) + } + Ok(_) => { + return Err(DurableRunnerError::invalid( + "binary and continuation WebSocket messages are not accepted", + )) + } + Err(tungstenite::Error::Io(error)) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + return Ok(None) + } + Err(error) => return Err(map_websocket_error(error)), + } + } +} + +fn map_websocket_error(error: tungstenite::Error) -> DurableRunnerError { + DurableRunnerError::invalid(format!("WebSocket transport failed: {error}")) +} + +fn random_nonce() -> Result { + let mut bytes = [0_u8; 32]; + getrandom::fill(&mut bytes).map_err(|error| { + DurableRunnerError::invalid(format!("secure randomness failed: {error}")) + })?; + Ok(hex_encode(&bytes)) +} + +pub(crate) fn current_unix_ms() -> Result { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|error| DurableRunnerError::invalid(format!("system clock is invalid: {error}")))? + .as_millis(); + u64::try_from(millis).map_err(|_| DurableRunnerError::invalid("system clock overflowed")) +} + +fn digest_domain(domain: &str, parts: &[&[u8]]) -> [u8; 32] { + let mut digest = Sha256::new(); + digest.update((domain.len() as u64).to_be_bytes()); + digest.update(domain.as_bytes()); + for part in parts { + digest.update((part.len() as u64).to_be_bytes()); + digest.update(part); + } + digest.finalize().into() +} + +fn hmac_domain(key: &[u8], domain: &str, parts: &[&[u8]]) -> [u8; 32] { + let mut mac = + ::new_from_slice(key).expect("HMAC accepts keys of every length"); + mac.update(&(domain.len() as u64).to_be_bytes()); + mac.update(domain.as_bytes()); + for part in parts { + mac.update(&(part.len() as u64).to_be_bytes()); + mac.update(part); + } + mac.finalize().into_bytes().into() +} + +fn verify_hmac_hex( + key: &[u8], + domain: &str, + parts: &[&[u8]], + expected: &str, +) -> Result<(), DurableRunnerError> { + let expected = hex_decode(expected)?; + let mut mac = + ::new_from_slice(key).expect("HMAC accepts keys of every length"); + mac.update(&(domain.len() as u64).to_be_bytes()); + mac.update(domain.as_bytes()); + for part in parts { + mac.update(&(part.len() as u64).to_be_bytes()); + mac.update(part); + } + mac.verify_slice(&expected) + .map_err(|_| DurableRunnerError::invalid("transport authentication proof is invalid")) +} + +fn hex_encode(input: &[u8]) -> String { + input.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn hex_decode(input: &str) -> Result, DurableRunnerError> { + if !input.len().is_multiple_of(2) { + return Err(DurableRunnerError::invalid("hex value has an odd length")); + } + input + .as_bytes() + .chunks_exact(2) + .map(|pair| { + let high = hex_nibble(pair[0])?; + let low = hex_nibble(pair[1])?; + Ok(high << 4 | low) + }) + .collect() +} + +fn hex_nibble(byte: u8) -> Result { + match byte { + b'0'..=b'9' => Ok(byte - b'0'), + b'a'..=b'f' => Ok(byte - b'a' + 10), + b'A'..=b'F' => Ok(byte - b'A' + 10), + _ => Err(DurableRunnerError::invalid( + "hex value contains invalid characters", + )), + } +} + +#[cfg(test)] +mod tests { + use std::net::{IpAddr, Ipv4Addr, TcpListener}; + use std::path::PathBuf; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::thread; + + use tungstenite::accept; + + use super::*; + + fn config(port: u16) -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: format!("ws://127.0.0.1:{port}/api/runner/v1/connect/run_1"), + state_dir: PathBuf::from("unused"), + runner_instance_id: "runner_1".to_owned(), + environment_lease_id: "environment_1".to_owned(), + run_id: "run_1".to_owned(), + normalized_session_id: "session_1".to_owned(), + turn_id: "turn_1".to_owned(), + item_id: "item_1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: "sha256:test".to_owned(), + max_outbox_bytes: 64 * 1024, + p0_reserve_bytes: 4096, + max_frame_bytes: 64 * 1024, + reconnect_delay: Duration::from_millis(1), + max_runtime: Duration::from_secs(1), + } + } + + fn test_state(config: &DurableRunnerConfig) -> DurableState { + DurableState::new(config) + } + + struct ServerCredential<'a> { + token: &'a str, + kind: &'a str, + lease_id: Option<&'a str>, + expires_at_unix_ms: u64, + revocation_epoch: u64, + } + + fn server_authenticate( + socket: &mut WebSocket, + config: &DurableRunnerConfig, + state: &DurableState, + server_credential: ServerCredential<'_>, + ) -> SecureChannel { + let hello = receive_plain(socket, config.max_frame_bytes).unwrap(); + let payload = hello.get("payload").unwrap(); + let credential = CredentialMaterial::from_token(server_credential.token); + assert_eq!(payload["credentialId"], credential.credential_id); + assert_eq!(payload["credentialKind"], server_credential.kind); + let mut challenge = AuthChallenge { + credential_id: credential.credential_id.clone(), + credential_kind: server_credential.kind.to_owned(), + client_nonce: payload["clientNonce"].as_str().unwrap().to_owned(), + server_nonce: random_nonce().unwrap(), + runner_instance_id: state.runner_instance_id.clone(), + environment_lease_id: state.environment_lease_id.clone(), + run_id: state.run_id.clone(), + normalized_session_id: state.normalized_session_id.clone(), + turn_id: state.turn_id.clone(), + item_id: state.item_id.clone(), + runner_version: config.runner_version.clone(), + runner_digest: config.runner_digest.clone(), + selected_version: PROTOCOL_VERSION, + credential_expires_at_unix_ms: server_credential.expires_at_unix_ms, + credential_lease_id: server_credential.lease_id.map(str::to_owned), + revocation_epoch: server_credential.revocation_epoch, + server_proof: String::new(), + }; + let signing = challenge_signing_bytes(&challenge); + challenge.server_proof = hex_encode(&hmac_domain( + &credential.auth_key, + "paperclip-runner-server-proof-v1", + &[&signing], + )); + send_plain( + socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_challenge", + "payload": { + "credentialId": &challenge.credential_id, + "credentialKind": &challenge.credential_kind, + "clientNonce": &challenge.client_nonce, + "serverNonce": &challenge.server_nonce, + "runnerInstanceId": &challenge.runner_instance_id, + "environmentLeaseId": &challenge.environment_lease_id, + "runId": &challenge.run_id, + "normalizedSessionId": &challenge.normalized_session_id, + "turnId": &challenge.turn_id, + "itemId": &challenge.item_id, + "runnerVersion": &challenge.runner_version, + "runnerDigest": &challenge.runner_digest, + "selectedVersion": challenge.selected_version, + "credentialExpiresAtUnixMs": challenge.credential_expires_at_unix_ms, + "credentialLeaseId": &challenge.credential_lease_id, + "revocationEpoch": challenge.revocation_epoch, + "serverProof": &challenge.server_proof, + }, + }), + config.max_frame_bytes, + ) + .unwrap(); + let response = receive_plain(socket, config.max_frame_bytes).unwrap(); + let client_proof = response["payload"]["clientProof"] + .as_str() + .unwrap() + .to_owned(); + verify_hmac_hex( + &credential.auth_key, + "paperclip-runner-client-proof-v1", + &[&signing, challenge.server_proof.as_bytes()], + &client_proof, + ) + .unwrap(); + SecureChannel::server( + &credential.auth_key, + &signing, + challenge.server_proof.as_bytes(), + client_proof.as_bytes(), + ) + .unwrap() + } + + fn send_secure( + socket: &mut WebSocket, + secure: &mut SecureChannel, + config: &DurableRunnerConfig, + value: &Value, + ) { + let encrypted = secure + .encrypt(&serde_json::to_vec(value).unwrap(), true) + .unwrap(); + send_plain(socket, &encrypted, config.max_frame_bytes).unwrap(); + } + + fn receive_secure( + socket: &mut WebSocket, + secure: &mut SecureChannel, + config: &DurableRunnerConfig, + ) -> Value { + let frame = receive_plain(socket, config.max_frame_bytes).unwrap(); + secure.decrypt(&frame, true).unwrap() + } + + fn welcome( + state: &DurableState, + connection_id: &str, + lease_token: Option<&str>, + expires_at_unix_ms: u64, + acked_source_seq: u64, + pending_commands: Vec, + ) -> Value { + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "welcome", + "runnerInstanceId": state.runner_instance_id, + "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "connectionId": connection_id, + "connectionLeaseId": "lease_1", + "payload": { + "selectedVersion": PROTOCOL_VERSION, + "connectionLeaseId": "lease_1", + "connectionLeaseToken": lease_token, + "connectionLeaseExpiresAtUnixMs": expires_at_unix_ms, + "connectionLeaseRevocationEpoch": 1, + "ackedSourceSeq": acked_source_seq, + "pendingCommands": pending_commands, + }, + }) + } + + fn control(state: &DurableState, connection_id: &str, kind: &str, payload: Value) -> Value { + json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": kind, + "runnerInstanceId": state.runner_instance_id, + "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "connectionId": connection_id, + "connectionLeaseId": "lease_1", + "payload": payload, + }) + } + + #[test] + fn url_resolution_rejects_non_loopback_and_ambiguous_inputs() { + let public = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), 80); + assert!( + resolve_ws_target_with("ws://example.test:80/path", |_, _| Ok(vec![public])).is_err() + ); + for input in [ + "wss://127.0.0.1:80/path", + "ws://user@127.0.0.1:80/path", + "ws://127.0.0.1/path", + "ws://127.0.0.1:80/path?ticket=secret", + ] { + assert!( + resolve_ws_target_with(input, |_, _| Ok(vec![])).is_err(), + "{input}" + ); + } + } + + #[test] + fn secure_frames_reject_replay_and_tampering() { + let key = [7_u8; 32]; + let mut client = SecureChannel::client(&key, b"challenge", b"server", b"client").unwrap(); + let mut server = SecureChannel::server(&key, b"challenge", b"server", b"client").unwrap(); + let frame = client.encrypt(br#"{"ok":true}"#, false).unwrap(); + assert_eq!(server.decrypt(&frame, true).unwrap(), json!({"ok": true})); + assert!(server.decrypt(&frame, true).is_err()); + + let mut client = SecureChannel::client(&key, b"challenge", b"server", b"client").unwrap(); + let mut server = SecureChannel::server(&key, b"challenge", b"server", b"client").unwrap(); + let mut frame = client.encrypt(br#"{"ok":true}"#, false).unwrap(); + frame["ciphertext"] = Value::String("00".repeat(32)); + assert!(server.decrypt(&frame, true).is_err()); + assert!(hex_decode("éé").is_err()); + assert!(server.decrypt(&json!({"kind": "command"}), true).is_err()); + } + + #[test] + fn control_identity_mismatch_fails_closed() { + let config = config(3000); + let state = test_state(&config); + let mut envelope = control(&state, "connection_1", "ack", json!({"ackedSourceSeq": 0})); + let connection = ConnectionMetadata { + connection_id: "connection_1".to_owned(), + lease_id: "lease_1".to_owned(), + expires_at_unix_ms: current_unix_ms().unwrap() + 60_000, + revocation_epoch: 1, + }; + validate_control_identity(&envelope, &state, Some(&connection)).unwrap(); + envelope["runId"] = Value::String("run_from_another_binding".to_owned()); + assert!(validate_control_identity(&envelope, &state, Some(&connection)).is_err()); + } + + #[test] + fn authenticates_bootstrap_and_receives_bound_lease() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let config = config(port); + let state = test_state(&config); + let server_config = config.clone(); + let server_state = state.clone(); + let handle = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + let hello = receive_plain(&mut socket, server_config.max_frame_bytes).unwrap(); + let payload = hello.get("payload").unwrap(); + let credential = CredentialMaterial::from_token("bootstrap-secret"); + assert_eq!(payload["credentialId"], credential.credential_id); + let expires = current_unix_ms().unwrap() + 60_000; + let mut challenge = AuthChallenge { + credential_id: credential.credential_id.clone(), + credential_kind: "bootstrap".to_owned(), + client_nonce: payload["clientNonce"].as_str().unwrap().to_owned(), + server_nonce: "server-nonce".to_owned(), + runner_instance_id: server_state.runner_instance_id.clone(), + environment_lease_id: server_state.environment_lease_id.clone(), + run_id: server_state.run_id.clone(), + normalized_session_id: server_state.normalized_session_id.clone(), + turn_id: server_state.turn_id.clone(), + item_id: server_state.item_id.clone(), + runner_version: server_config.runner_version.clone(), + runner_digest: server_config.runner_digest.clone(), + selected_version: PROTOCOL_VERSION, + credential_expires_at_unix_ms: expires, + credential_lease_id: None, + revocation_epoch: 0, + server_proof: String::new(), + }; + let signing = challenge_signing_bytes(&challenge); + challenge.server_proof = hex_encode(&hmac_domain( + &credential.auth_key, + "paperclip-runner-server-proof-v1", + &[&signing], + )); + send_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_challenge", + "payload": { + "credentialId": challenge.credential_id, + "credentialKind": challenge.credential_kind, + "clientNonce": challenge.client_nonce, + "serverNonce": challenge.server_nonce, + "runnerInstanceId": challenge.runner_instance_id, + "environmentLeaseId": challenge.environment_lease_id, + "runId": challenge.run_id, + "normalizedSessionId": challenge.normalized_session_id, + "turnId": challenge.turn_id, + "itemId": challenge.item_id, + "runnerVersion": challenge.runner_version, + "runnerDigest": challenge.runner_digest, + "selectedVersion": challenge.selected_version, + "credentialExpiresAtUnixMs": challenge.credential_expires_at_unix_ms, + "credentialLeaseId": challenge.credential_lease_id, + "revocationEpoch": challenge.revocation_epoch, + "serverProof": challenge.server_proof, + }, + }), + server_config.max_frame_bytes, + ) + .unwrap(); + let response = receive_plain(&mut socket, server_config.max_frame_bytes).unwrap(); + let client_proof = response["payload"]["clientProof"].as_str().unwrap(); + verify_hmac_hex( + &credential.auth_key, + "paperclip-runner-client-proof-v1", + &[&signing, challenge.server_proof.as_bytes()], + client_proof, + ) + .unwrap(); + let mut secure = SecureChannel::server( + &credential.auth_key, + &signing, + challenge.server_proof.as_bytes(), + client_proof.as_bytes(), + ) + .unwrap(); + let welcome = json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "welcome", + "runnerInstanceId": server_state.runner_instance_id, + "environmentLeaseId": server_state.environment_lease_id, + "runId": server_state.run_id, + "normalizedSessionId": server_state.normalized_session_id, + "turnId": server_state.turn_id, + "itemId": server_state.item_id, + "connectionId": "connection_1", + "connectionLeaseId": "lease_1", + "payload": { + "selectedVersion": PROTOCOL_VERSION, + "connectionLeaseId": "lease_1", + "connectionLeaseToken": "lease-secret", + "connectionLeaseExpiresAtUnixMs": expires, + "connectionLeaseRevocationEpoch": 1, + "ackedSourceSeq": 0, + "pendingCommands": [], + }, + }); + let encrypted = secure + .encrypt(&serde_json::to_vec(&welcome).unwrap(), true) + .unwrap(); + send_plain(&mut socket, &encrypted, server_config.max_frame_bytes).unwrap(); + }); + + let target = ResolvedWsTarget::resolve(&config.connect_url).unwrap(); + let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); + let (_, welcome) = + AuthenticatedTransport::connect(&target, &config, &state, Some(&ticket), None).unwrap(); + assert_eq!(welcome.connection.lease_id, "lease_1"); + assert_eq!(welcome.lease.unwrap().expose().unwrap(), "lease-secret"); + handle.join().unwrap(); + } + + #[test] + fn reconnect_replays_unacked_events_and_not_command_effects() { + struct EventExecutor { + session_open_calls: Arc, + } + + impl super::super::CommandExecutor for EventExecutor { + fn execute( + &mut self, + command: &Command, + ) -> Result { + if command.command_type == "session.open" { + let calls = self.session_open_calls.fetch_add(1, Ordering::SeqCst) + 1; + return Ok(super::super::CommandExecution { + result: json!({"status": "completed", "calls": calls}), + events: vec![( + "provider.notice.recorded".to_owned(), + super::super::EventPriority::P1, + json!({"message": "durable event"}), + )], + }); + } + Ok(super::super::CommandExecution::result( + json!({"status": "completed"}), + )) + } + } + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut config = config(port); + config.max_runtime = Duration::from_secs(5); + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-reconnect-fault-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&directory); + config.state_dir = directory.clone(); + let state = test_state(&config); + let expires = current_unix_ms().unwrap() + 60_000; + let open_command = json!({ + "schema": "paperclip.prp.command.v1", + "commandId": "command_open", + "controllerSeq": 1, + "type": "session.open", + "issuedAt": "2026-08-24T00:00:00.000Z", + "payload": {}, + }); + let shutdown_command = json!({ + "schema": "paperclip.prp.command.v1", + "commandId": "command_shutdown", + "controllerSeq": 2, + "type": "runner.shutdown", + "issuedAt": "2026-08-24T00:00:01.000Z", + "payload": {}, + }); + let server_config = config.clone(); + let server_state = state.clone(); + let server_open = open_command.clone(); + let server = thread::spawn(move || { + let (first_stream, _) = listener.accept().unwrap(); + let mut first = accept(first_stream).unwrap(); + let mut first_secure = server_authenticate( + &mut first, + &server_config, + &server_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + send_secure( + &mut first, + &mut first_secure, + &server_config, + &welcome( + &server_state, + "connection_1", + Some("lease-secret"), + expires, + 0, + vec![server_open.clone()], + ), + ); + let first_result = receive_secure(&mut first, &mut first_secure, &server_config); + let first_event = receive_secure(&mut first, &mut first_secure, &server_config); + assert_eq!(first_result["kind"], "command_result"); + assert_eq!(first_event["kind"], "event"); + drop(first); + + let (second_stream, _) = listener.accept().unwrap(); + let mut second = accept(second_stream).unwrap(); + let mut second_secure = server_authenticate( + &mut second, + &server_config, + &server_state, + ServerCredential { + token: "lease-secret", + kind: "lease", + lease_id: Some("lease_1"), + expires_at_unix_ms: expires, + revocation_epoch: 1, + }, + ); + send_secure( + &mut second, + &mut second_secure, + &server_config, + &welcome( + &server_state, + "connection_2", + None, + expires, + 0, + vec![server_open], + ), + ); + let replayed_result = receive_secure(&mut second, &mut second_secure, &server_config); + let replayed_event = receive_secure(&mut second, &mut second_secure, &server_config); + assert_eq!(replayed_result, first_result); + assert_eq!(replayed_event, first_event); + send_secure( + &mut second, + &mut second_secure, + &server_config, + &control( + &server_state, + "connection_2", + "ack", + json!({"ackedSourceSeq": 1}), + ), + ); + send_secure( + &mut second, + &mut second_secure, + &server_config, + &control(&server_state, "connection_2", "command", shutdown_command), + ); + let shutdown_result = receive_secure(&mut second, &mut second_secure, &server_config); + assert_eq!(shutdown_result["kind"], "command_result"); + }); + + let session_open_calls = Arc::new(AtomicUsize::new(0)); + super::super::run_durable_runner( + config, + BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(), + EventExecutor { + session_open_calls: session_open_calls.clone(), + }, + ) + .unwrap(); + server.join().unwrap(); + assert_eq!(session_open_calls.load(Ordering::SeqCst), 1); + let store = super::super::DurableStateStore::new(&directory).unwrap(); + let state_bytes = std::fs::read(store.path()).unwrap(); + let final_state: DurableState = serde_json::from_slice(&state_bytes).unwrap(); + assert_eq!(final_state.acked_source_seq, 1); + assert!(final_state.outbox.is_empty()); + assert_eq!(final_state.reconnect_count, 1); + std::fs::remove_dir_all(directory).unwrap(); + } +} 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 5d163fdfbf..926aa0c540 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -1,5 +1,6 @@ #![forbid(unsafe_code)] +pub mod durable; pub mod fake_harness; pub mod local_runner; pub mod process_supervisor; diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs new file mode 100644 index 0000000000..6ac1f3f914 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs @@ -0,0 +1,185 @@ +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use paperclip_runner_core::durable::{ + Command, CommandDisposition, DurableRunnerConfig, DurableStateStore, +}; +use serde_json::json; + +static NEXT_TEMPORARY_DIRECTORY: AtomicU64 = AtomicU64::new(0); + +fn config(state_dir: PathBuf) -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: "ws://127.0.0.1:3000/api/runner/v1/connect/run_1".to_owned(), + state_dir, + runner_instance_id: "runner_1".to_owned(), + environment_lease_id: "environment_1".to_owned(), + run_id: "run_1".to_owned(), + normalized_session_id: "session_1".to_owned(), + turn_id: "turn_1".to_owned(), + item_id: "item_1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: "sha256:test".to_owned(), + max_outbox_bytes: 16_384, + p0_reserve_bytes: 4096, + max_frame_bytes: 65_536, + reconnect_delay: Duration::from_millis(1), + max_runtime: Duration::from_secs(1), + } +} + +fn command() -> Command { + Command { + schema: "paperclip.prp.command.v1".to_owned(), + command_id: "command_1".to_owned(), + controller_seq: 1, + command_type: "session.open".to_owned(), + issued_at: "2026-08-24T00:00:00.000Z".to_owned(), + deadline_at: None, + precondition: None, + payload: json!({}), + } +} + +fn temporary_directory() -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system clock must follow the Unix epoch") + .as_nanos(); + let sequence = NEXT_TEMPORARY_DIRECTORY.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!( + "paperclip-runner-public-recovery-{}-{nonce}-{sequence}", + std::process::id() + )) +} + +#[test] +fn public_store_never_reexecutes_a_journaled_command_after_recovery() { + let directory = temporary_directory(); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).expect("create private state store"); + let (mut state, existed) = store.load_or_create(&config).expect("create durable state"); + assert!(!existed); + + let command = command(); + assert_eq!( + state.begin_command(&command).expect("journal command"), + CommandDisposition::Execute + ); + store + .save(&state) + .expect("persist command before its external effect"); + + let (mut recovered, existed) = store + .load_or_create(&config) + .expect("recover durable state"); + assert!(existed); + assert!(matches!( + recovered + .begin_command(&command) + .expect("look up recovered command"), + CommandDisposition::Replay(result) + if result.status == "indeterminate" + && result.result["code"] == "execution_indeterminate" + )); + + fs::remove_dir_all(directory).expect("remove integration-test state"); +} + +#[test] +fn duplicate_replay_requires_the_complete_command_identity() { + let directory = temporary_directory(); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).expect("create private state store"); + let (mut state, _) = store.load_or_create(&config).expect("create durable state"); + let original = command(); + + assert_eq!( + state.begin_command(&original).expect("journal command"), + CommandDisposition::Execute + ); + state + .complete_command(&original, json!({"ok": true})) + .expect("complete command"); + + let mut changed_payload = original.clone(); + changed_payload.payload = json!({"changed": true}); + let mut changed_precondition = original.clone(); + changed_precondition.precondition = Some(json!({"ifMatch": "different"})); + let mut changed_issued_at = original.clone(); + changed_issued_at.issued_at = "2026-08-24T00:00:01.000Z".to_owned(); + let mut changed_deadline = original.clone(); + changed_deadline.deadline_at = Some("2026-08-24T00:01:00.000Z".to_owned()); + let mut changed_type = original.clone(); + changed_type.command_type = "session.close".to_owned(); + let mut changed_sequence = original.clone(); + changed_sequence.controller_seq = 2; + + for conflicting in [ + changed_payload, + changed_precondition, + changed_issued_at, + changed_deadline, + changed_type, + changed_sequence, + ] { + let error = state + .begin_command(&conflicting) + .expect_err("changed duplicate must fail closed"); + assert!(error + .to_string() + .contains("commandId was reused with different command data")); + } + assert!(matches!( + state.begin_command(&original).expect("replay exact command"), + CommandDisposition::Replay(result) if result.result == json!({"ok": true}) + )); + + fs::remove_dir_all(directory).expect("remove integration-test state"); +} + +#[test] +fn pre_fingerprint_journal_recovers_without_reexecuting_old_commands() { + let directory = temporary_directory(); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).expect("create private state store"); + let (mut state, _) = store.load_or_create(&config).expect("create durable state"); + let command = command(); + assert_eq!( + state.begin_command(&command).expect("journal command"), + CommandDisposition::Execute + ); + store.save(&state).expect("persist pending command"); + + let mut legacy: serde_json::Value = + serde_json::from_slice(&fs::read(store.path()).expect("read current durable state")) + .expect("parse current durable state"); + legacy + .as_object_mut() + .expect("durable state must be an object") + .remove("processedCommandFingerprints"); + fs::write( + store.path(), + serde_json::to_vec_pretty(&legacy).expect("serialize legacy state"), + ) + .expect("write simulated pre-fingerprint state"); + + let (mut recovered, existed) = store + .load_or_create(&config) + .expect("migrate pre-fingerprint state"); + assert!(existed); + assert!(recovered.processed_commands.is_empty()); + assert!(recovered.processed_command_fingerprints.is_empty()); + assert_eq!(recovered.compacted_through_controller_seq, 1); + assert!(matches!( + recovered + .begin_command(&command) + .expect("reject migrated command without reexecution"), + CommandDisposition::Reject(result) + if result.result["code"] == "command_history_compacted" + )); + + fs::remove_dir_all(directory).expect("remove integration-test state"); +}