Add durable PRP transport and recovery (#12100)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The package-local runner can supervise a local process, but it cannot yet survive a broken controller connection. > - A production transport must authenticate both peers without putting the bootstrap secret on the wire. > - Commands and events must remain bounded, ordered, and recoverable across reconnects and crashes. > - Retrying an uncertain side effect is unsafe, so indeterminate outcomes must fail closed instead of running twice. > - This pull request adds those transport and recovery guarantees inside the runner package only. > - The benefit is a durable PRP boundary that can be reviewed before any provider or server integration exists. ## Linked Issues or Issue Description **Subsystem affected** Cross-cutting. This pull request extends private transport infrastructure in `packages/paperclip-runner`. **Problem or motivation** The local runner introduced by #12095 has no authenticated network handshake, durable outbox, reconnect lease, cumulative acknowledgement, or crash-safe command journal. A dropped connection could otherwise lose an event or tempt a controller to repeat a side effect whose outcome is unknown. **Proposed solution** Add an authenticated PRP v1 WebSocket transport, encrypted frames, lease-based reconnects, a bounded durable event outbox, cumulative acknowledgements, and an idempotent command journal. Preserve pending commands before execution and classify the crash window as indeterminate so an uncertain side effect is never repeated automatically. **Alternatives considered** The combined runner branch implements transport together with Codex, semantic tools, and server coordination. That change is too large for one review unit. Keeping transport in memory would make reconnect and crash recovery unverifiable. Re-running a pending command after restart would weaken the at-most-once side-effect boundary. **Roadmap alignment** This work supports the governed tools and self-healing run direction in `ROADMAP.md`. It does not add a production provider, server endpoint, adapter, feature flag, or user-facing behavior. **Additional context** Refs #12095 and #11962. Pull request #12095 was squash-merged first. This branch has been rebased onto the resulting `master` commit, and its current delta is 13 files. ## What Changed - Added a loopback-only WebSocket connection policy with one-time DNS resolution and pinned reconnect addresses. - Added an HMAC mutual-authentication handshake that never sends the bootstrap ticket over the socket. - Added AES-256-GCM secure frames with per-direction keys, monotonic counters, and session-bound authenticated data. - Added one-use bootstrap-ticket handling and lease-based reconnect validation with expiry, revocation, and epoch checks. - Added a private, symlink-resistant state directory with atomic, synchronized state replacement. - Added a bounded durable event outbox, priority-zero reserve, cumulative acknowledgements, and reconnect replay of only the unacknowledged suffix. - Added a bounded command journal with contiguous sequence enforcement, persistent results, and deterministic duplicate responses. Duplicate replay requires a SHA-256 match over the complete canonical command. - Persisted commands before their effects. A crash after persistence but before result storage returns an indeterminate terminal result and does not execute the command again. - Migrated pre-fingerprint command journals by compacting through their persisted controller cursor. Legacy redelivery fails closed instead of reconstructing an incomplete identity or repeating an uncertain effect. - Added strict limits and validation for frames, state, results, outbox entries, command history, and redacted diagnostics. - Added a transport-only `paperclip-runnerd --connect-url` mode. It handles lifecycle commands and rejects provider commands because no provider is present in this pull request. - Added a full disconnect-before-ack fault test that reconnects with the lease, replays identical command and event state, and proves the effect ran once. - Kept provider transports, semantic tools, server integration, and production runtime selection out of this pull request. ## Verification - `pnpm --filter @paperclipai/paperclip-runner check:all` passes. - TypeScript contract tests pass: 8 Node tests and 44 Vitest tests. - Rust tests pass: 33 unit tests, 3 public durable-recovery integration tests, plus the existing 2 local-runner and 3 process-supervisor tests. - The disconnect-before-ack, lease reconnect, duplicate command, malformed state, unknown command, bounds, and crash-window tests pass. - Rust conformance and replay parity checks pass against the shared PRP fixtures. - `cargo clippy --workspace --all-targets -- -A clippy::filter-map-bool-then -D warnings` passes. The narrow allow covers an unchanged replay implementation from the preceding contract pull request. - `pnpm -r typecheck` passes. - `pnpm build` passes. - `pnpm check:token-gates` passes. - `git diff --check` passes. - The delta against `master` is 13 files. The package lockfile is unchanged. - `pnpm test:run` completed locally with 4,686 passing tests, 19 skipped tests, and 30 failures in 8 unchanged server test files. The failures reproduce the established local macOS path-alias, listener, port-range, and workspace-runtime baseline. No changed-file test failed; Linux CI remains the repository handoff authority. - Storybook visual regression is not applicable because this pull request changes no UI or story files. ## Risks Production behavior is unchanged because no server code starts or connects to this transport. The main risks are secret disclosure, forged or replayed frames, state corruption, unbounded disk growth, duplicated side effects, and incorrect recovery. Mutual authentication, encrypted counter-bound frames, private atomic state, explicit bounds, cumulative acknowledgements, a durable command journal, fail-closed indeterminate recovery, and fault-injection tests cover these risks. I checked `ROADMAP.md`. This change is private transport infrastructure for planned control-plane work. It does not duplicate a shipped or public product surface. ## Model Used OpenAI Codex with GPT-5 was used. The exact serving model ID and context size were not exposed. The model used high reasoning, repository tools, GitHub tools, and local code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
6b20cc97cc
commit
b76e36d6cf
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"] }
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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<String, LocalRunnerError> {
|
||||
let index = args
|
||||
|
|
@ -34,8 +39,67 @@ fn usize_value(args: &[String], name: &str, default: usize) -> Result<usize, Loc
|
|||
})
|
||||
}
|
||||
|
||||
struct TransportOnlyExecutor;
|
||||
|
||||
impl CommandExecutor for TransportOnlyExecutor {
|
||||
fn execute(&mut self, command: &Command) -> Result<CommandExecution, DurableRunnerError> {
|
||||
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::<Vec<_>>();
|
||||
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")?,
|
||||
|
|
|
|||
|
|
@ -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<String>) -> 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<u8>);
|
||||
|
||||
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<Self, DurableRunnerError> {
|
||||
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<Option<BootstrapTicket>, 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(())
|
||||
}
|
||||
}
|
||||
|
|
@ -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<CommandExecution, DurableRunnerError>;
|
||||
}
|
||||
|
||||
pub fn run_durable_runner<E: CommandExecutor>(
|
||||
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<LeaseCredential> = 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<E: CommandExecutor>(
|
||||
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<CommandExecution, DurableRunnerError> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
|
|
@ -1,5 +1,6 @@
|
|||
#![forbid(unsafe_code)]
|
||||
|
||||
pub mod durable;
|
||||
pub mod fake_harness;
|
||||
pub mod local_runner;
|
||||
pub mod process_supervisor;
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
}
|
||||
Loading…
Reference in New Issue