build: the cloud frontend stands alone

The Durable Object, the sandbox daemon, and their deploy assets
leave the rhema workspace for a repository of their own. The DO
was already a deployment artifact (publish = false, worker feature
wasm-only), and rhema-sandbox-server's only consumer was the image
beside it — both are product, not framework. The harness and the
seam stay behind: rhema-cloud-agent and the sandbox family join
the rhema repository's publish list, and this tree consumes them
from crates.io at 0.2.6, the same arrangement the claw keeps its
framework crates under.

The sandboxd image build carries the fix its extraction needed:
the old Dockerfile still staged the claw's workspace (claw and
tools manifests, integrations globs, the vendored crossterm) and
missed rhema-provider-registry's manifest entirely. The manifest
stage is now this workspace's two members, fetched from the
registry.

Not yet buildable: the rhema 0.2.6 batch (rhema-sandbox,
-local, -http, rhema-cloud-agent) must publish first — then one
cargo check generates Cargo.lock and CI goes green.

crates/rhema-cloudflare-code/Cargo.toml

@@ -0,0 +1,67 @@
+[package]
+name = "rhema-cloudflare-code"
+description = "The coding agent as a Cloudflare Durable Object: WebSocket chat, DO-storage checkpoints, bash in a Cloudflare Sandbox"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+authors.workspace = true
+# A deployment artifact, not a library: the worker feature only
+# compiles on wasm32, so crates.io has nothing it can serve. The
+# deployable ships from deploy/cloudflare instead.
+publish = false
+
+[features]
+default = []
+# The workers-rs glue (the `worker` module). The feature only takes
+# effect on wasm32 — workerd is the one host that provides the
+# bindings — so native `--all-features` builds stay clean.
+worker = []
+
+[dependencies]
+# The harness the DO assembles: assembly, gate, sandbox seam — shared
+# with the other cloud frontends, tested on the host without wasm. The
+# crate lives in the rhema repository and ships from crates.io.
+rhema-cloud-agent = { version = "0.2.6" }
+# The toolkit pieces the worker touches directly: the fail-safe
+# turn-level approval and the skill shape
+rhema-agent-tools = { version = "0.2.6", default-features = false }
+# The engine's session type the checkpoints wrap
+rhema-core = { version = "0.2.6" }
+# The transport seam the fetch transport implements, and the Sleeper
+# the Delay sleeper implements
+rhema-provider = { version = "0.2.6" }
+# The Anthropic dialect over a caller-supplied transport; the native
+# reqwest feature stays off — workerd has no sockets
+rhema-provider-anthropic = { version = "0.2.6", default-features = false }
+# The revision-guarded record the DO checkpoints sessions through
+rhema-rsd = { version = "0.2.6", default-features = false }
+# The sandbox seam the DO's bash runs through: `SandboxDo` speaks the
+# shared protocol types
+rhema-sandbox = { version = "0.2.6", default-features = false }
+# The protocol's HTTPS client: the `HttpTransport` seam this worker
+# implements over host fetch, plus the `HttpSandbox` the `http` backend
+# drives — reqwest behind its `native` feature, off here, workerd has
+# no sockets
+rhema-sandbox-http = { version = "0.2.6", default-features = false }
+rhema-transcript = { version = "0.2.6" }
+# The typed errors the backend choice and the turn's durable pieces report
+thiserror = { workspace = true }
+async-trait = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+# Pinned to 0.8 (verified against this tree): the DurableObject trait,
+# SqlStorage, WebSocket, and Delay APIs this crate uses are 0.8-shaped;
+# the crate is 0.x and its API moves, so the pin is deliberate.
+worker = { version = "0.8.7" }
+# The shim's file payloads ride base64
+base64 = "0.22"
+# The !Send fetch-stream bridge (spawn_local) and the turn task
+wasm-bindgen-futures = "0.4"
+futures = { version = "0.3", default-features = false, features = ["std", "async-await"] }
+
+[lints]
+workspace = true

crates/rhema-cloudflare-code/src/backend.rs

@@ -0,0 +1,134 @@
+//! The deployment's sandbox backend selection: which backend a turn's
+//! `bash` runs in, decoded from the deployment environment — a
+//! deployment decision, never `SessionConfig` (a per-session
+//! client-chosen URL would be an SSRF on a server that executes
+//! commands). The platform side — reading workerd's env and building
+//! the client the choice names — lives in the `worker` module; this
+//! module is plain data and pure functions, so the choice matrix stays
+//! testable off-wasm, where workerd does not exist.
+#![forbid(unsafe_code)]
+
+/// The env var naming the sandbox backend.
+pub const BACKEND_VAR: &str = "SANDBOX_BACKEND";
+
+/// The env var carrying the `http` backend's daemon base URL.
+pub const URL_VAR: &str = "SANDBOX_URL";
+
+/// The sandbox backend a deployment selected.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SandboxBackend {
+    /// The platform's own sandbox: the session's SandboxShim Durable
+    /// Object. The default when [`BACKEND_VAR`] is unset.
+    Cloudflare,
+    /// The sandbox protocol over plain HTTPS to a daemon at a base URL
+    /// (`https://box.example:7600`, no trailing slash); the shim's DO
+    /// binding is never touched.
+    Http(String),
+}
+
+/// The deployment env the selection reads, as plain values: the DO
+/// reads workerd's env once per connection and hands the values here,
+/// so this stays free of the `worker` crate.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct SandboxEnv {
+    /// [`BACKEND_VAR`]'s value; absent means the default.
+    pub backend: Option<String>,
+    /// [`URL_VAR`]'s value; required under `http`.
+    pub url: Option<String>,
+}
+
+/// Why sandbox-backend selection failed: a deployment config mistake —
+/// a misspelled kind, a URL left unset — never a runtime fault.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum BackendError {
+    /// `http` was selected but [`URL_VAR`] is absent or blank, so no
+    /// daemon address exists to run `bash` against.
+    #[error(
+        "{URL_VAR} is not set: {BACKEND_VAR}=http runs bash on the sandbox \
+         daemon at that URL (e.g. https://box.example:7600)"
+    )]
+    MissingUrl,
+    /// [`BACKEND_VAR`] named a kind this crate does not implement.
+    #[error("unknown {BACKEND_VAR} `{0}` (want `cloudflare` or `http`)")]
+    UnknownKind(String),
+}
+
+/// Decodes the deployment's backend choice. Unknown kinds and a
+/// missing URL under `http` are config errors, not silent fallbacks —
+/// a deployment that misspells `http` must not quietly get the DO
+/// shim.
+pub fn select_backend(env: &SandboxEnv) -> Result<SandboxBackend, BackendError> {
+    match env.backend.as_deref().unwrap_or("cloudflare") {
+        "cloudflare" => Ok(SandboxBackend::Cloudflare),
+        "http" => match env.url.as_deref().filter(|url| !url.trim().is_empty()) {
+            Some(url) => Ok(SandboxBackend::Http(url.to_owned())),
+            None => Err(BackendError::MissingUrl),
+        },
+        other => Err(BackendError::UnknownKind(other.to_owned())),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn an_absent_backend_is_the_do_shim() {
+        assert_eq!(
+            select_backend(&SandboxEnv::default()).ok(),
+            Some(SandboxBackend::Cloudflare)
+        );
+    }
+
+    #[test]
+    fn cloudflare_names_the_do_shim_explicitly() {
+        let env = SandboxEnv {
+            backend: Some("cloudflare".into()),
+            url: Some("https://ignored.example".into()),
+        };
+        assert_eq!(select_backend(&env).ok(), Some(SandboxBackend::Cloudflare));
+    }
+
+    #[test]
+    fn http_yields_the_daemon_url() {
+        let env = SandboxEnv {
+            backend: Some("http".into()),
+            url: Some("https://box.example:7600".into()),
+        };
+        assert_eq!(
+            select_backend(&env).ok(),
+            Some(SandboxBackend::Http("https://box.example:7600".into()))
+        );
+    }
+
+    #[test]
+    fn http_without_a_url_is_a_config_error() {
+        let missing = SandboxEnv {
+            backend: Some("http".into()),
+            url: None,
+        };
+        let empty = SandboxEnv {
+            backend: Some("http".into()),
+            url: Some(String::new()),
+        };
+        for env in [missing, empty] {
+            let err = select_backend(&env).expect_err("the missing URL surfaces");
+            let message = err.to_string();
+            assert!(message.contains(URL_VAR), "names the var: {message}");
+        }
+    }
+
+    #[test]
+    fn an_unknown_backend_is_a_config_error() {
+        let env = SandboxEnv {
+            backend: Some("vercel".into()),
+            url: None,
+        };
+        let err = select_backend(&env).expect_err("the unknown kind surfaces");
+        let message = err.to_string();
+        assert!(
+            message.contains(BACKEND_VAR) && message.contains("vercel"),
+            "names the var and the value: {message}"
+        );
+    }
+}

crates/rhema-cloudflare-code/src/lib.rs

@@ -0,0 +1,62 @@
+//! rhema-cloudflare-code — the coding agent as a Cloudflare Durable
+//! Object.
+//!
+//! One [`worker::CodingAgent`](worker::CodingAgent) instance per
+//! session id. The agent itself is the ordinary
+//! [`CodingAgent`](rhema_coding_agent::CodingAgent) driver over the
+//! ordinary step engine — this crate is a frontend, next to the
+//! terminal and the ACP adapter. The frontend-independent harness —
+//! the filesystem-less assembly, the client approval gate, the sandbox
+//! seam, the per-session config — lives in
+//! [`rhema-cloud-agent`](../rhema-cloud-agent); this crate adds only
+//! what Cloudflare demands:
+//!
+//! - the model rides [`worker::FetchTransport`], a
+//!   [`Transport`](rhema_provider::Transport) over workerd's host
+//!   fetch, bridging the `!Send` JS byte stream into plain chunks;
+//! - `bash` runs in a Cloudflare Sandbox container bound to the same
+//!   session name, reached through the shared
+//!   [`Sandbox`](rhema_cloud_agent::Sandbox) seam
+//!   ([`worker::SandboxDo`] implements it) — or, when the deployment
+//!   sets `SANDBOX_BACKEND=http`, on any machine speaking the sandbox
+//!   protocol, through the shared HTTP sandbox client over host fetch
+//!   (the selection lives in [`backend`]);
+//! - the session checkpoints into the DO's SQLite storage — at the
+//!   commit hook, before every side effect, and after each turn —
+//!   revision-guarded, through a
+//!   [`rhema_rsd::Store`](rhema_rsd::Store) implementation, so any
+//!   eviction resumes through the engine's existing revive machinery;
+//! - the per-session [`SessionConfig`](rhema_cloud_agent::SessionConfig)
+//!   persists in the DO's meta table, set over `POST /config`.
+//!
+//! # Module hygiene
+//!
+//! The platform lives at the edges, in the `worker` module and nowhere
+//! else: wasm-bindgen, `worker`, and host-fetch dependencies are absent
+//! from the harness — which moved to `rhema-cloud-agent` precisely so
+//! the compiler polices that boundary. What stays here compiles only
+//! on wasm32 (feature `worker`).
+//!
+//! The crate as a whole carries `#![deny(missing_docs)]` but **not**
+//! `#![forbid(unsafe_code)]`: the `worker` module's `#[durable_object]`
+//! macro emits `unsafe` internally, and a crate-level `forbid` cannot
+//! be overridden by an inner `allow`. That exception is sanctioned and
+//! documented here (the precedent: `vendor/crossterm` is excluded from
+//! the workspace for toolchain reasons).
+//!
+//! # Non-goals (v1)
+//!
+//! The terminal UI, the claw daemon, and the local-only provider stack
+//! do not port to wasm. One provider adapter (Anthropic). Memory tools
+//! and the subagents extension wait for storage-backed discovery; the
+//! alarm-driven step queue is a v2 refinement of turn driving.
+
+pub mod backend;
+
+pub use rhema_cloud_agent::{
+    AgentAssembly, ApprovalChannel, ApprovalPolicy, ClientGate, ExecOutcome, ExecRequest,
+    PolicyApproval, Sandbox, SandboxError, SessionConfig, ShellSandbox, SkillSeed, WorkspaceSeed,
+};
+
+#[cfg(all(feature = "worker", target_arch = "wasm32"))]
+pub mod worker;

crates/rhema-cloudflare-code/src/worker/approval.rs

@@ -0,0 +1,137 @@
+//! The WebSocket approval channel: the async round-trip workerd needs.
+//!
+//! The gate ([`rhema_cloud_agent::ClientGate`]) consults this channel at an
+//! await point; the ask goes out on the socket, the answer comes back
+//! through the socket's pump, which resolves the matching one-shot by
+//! call id. No thread is ever blocked — there is only one thread, and
+//! the pump runs on it while this ask waits. A client that never
+//! answers is denied at the timeout, via a host-timer delay, never a
+//! tokio sleep.
+#![forbid(unsafe_code)]
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use async_trait::async_trait;
+use futures::FutureExt;
+use rhema_core::ToolCall;
+
+use rhema_cloud_agent::{ApprovalChannel, ServerMessage};
+
+/// What one live connection shares between the pump and the turn: the
+/// socket answers go out on, the pending approvals, and the turn's
+/// cancel flag.
+pub struct Live {
+    /// The server side of the accepted socket. Cloneable; sends are
+    /// synchronous host calls.
+    pub socket: worker::WebSocket,
+    /// Approvals awaiting their client, by call id.
+    pub pending: Mutex<HashMap<String, futures::channel::oneshot::Sender<bool>>>,
+    /// The current turn's cancel flag; a stop message or a disconnect
+    /// flips it. An `Arc` because the turn's context holds the same flag.
+    pub cancelled: Arc<std::sync::atomic::AtomicBool>,
+    /// Whether a turn is running; a second chat message is refused.
+    pub busy: std::sync::atomic::AtomicBool,
+}
+
+impl Live {
+    /// Sends one protocol message to the client.
+    pub fn send(&self, message: &ServerMessage) {
+        let Ok(text) = serde_json::to_string(message) else {
+            return;
+        };
+        let _ = self.socket.send_with_str(text);
+    }
+}
+
+/// The channel: asks the connected client, denies on timeout.
+pub struct WsChannel {
+    live: Arc<Live>,
+    timeout: Duration,
+}
+
+impl WsChannel {
+    /// A channel over `live`, denying asks `timeout` old.
+    pub fn new(live: Arc<Live>, timeout: Duration) -> Self {
+        Self { live, timeout }
+    }
+
+    /// Resolves a pending approval by call id — the pump's path for an
+    /// `approve` message.
+    pub fn answer(&self, call_id: &str, approved: bool) {
+        if let Some(sender) = self
+            .live
+            .pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .remove(call_id)
+        {
+            let _ = sender.send(approved);
+        }
+    }
+
+    /// Drops every pending approval — the disconnect path.
+    pub fn abandon_all(&self) {
+        self.live
+            .pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clear();
+    }
+}
+
+/// The command text of a call, for the prompt the client shows.
+fn command_of(call: &ToolCall) -> String {
+    call.args
+        .get("command")
+        .and_then(serde_json::Value::as_str)
+        .map(str::to_owned)
+        .unwrap_or_default()
+}
+
+#[async_trait]
+impl ApprovalChannel for WsChannel {
+    async fn ask(&self, call: &ToolCall) -> Option<bool> {
+        let (sender, receiver) = futures::channel::oneshot::channel();
+        self.live
+            .pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(call.id.clone(), sender);
+        self.live.send(&ServerMessage::ApprovalRequest {
+            id: call.id.clone(),
+            tool: call.name.clone(),
+            command: command_of(call),
+        });
+        // The answer arrives through the pump; the timeout arrives
+        // through the host's timer. Either way this task suspends and
+        // the socket stays live — no thread is held. The timer's
+        // `!Send` future runs on the local task queue, handing back a
+        // plain one-shot: the JS handle stays inside the bridge.
+        let (timed_out, timed_out_rx) = futures::channel::oneshot::channel::<()>();
+        let timeout = self.timeout;
+        wasm_bindgen_futures::spawn_local(async move {
+            worker::Delay::from(timeout).await;
+            let _ = timed_out.send(());
+        });
+        let timeout = futures::future::select(receiver.fuse(), timed_out_rx.fuse()).await;
+        self.live
+            .pending
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .remove(&call.id);
+        match timeout {
+            futures::future::Either::Left((Ok(answer), _)) => Some(answer),
+            // The client disconnected mid-ask: denied.
+            futures::future::Either::Left((Err(_), _)) => None,
+            // Timed out: denied, and the client is told.
+            futures::future::Either::Right(_) => {
+                self.live.send(&ServerMessage::ApprovalExpired {
+                    id: call.id.clone(),
+                });
+                None
+            }
+        }
+    }
+}

crates/rhema-cloudflare-code/src/worker/mod.rs

@@ -0,0 +1,546 @@
+//! The workers-rs glue — the one module where the platform lives.
+//!
+//! [`CodingAgent`] is one Durable Object per session id. Its
+//! WebSocket surface speaks a small JSON protocol (chat, approval
+//! answers, stop; streaming events out), its HTTP surface serves the
+//! checkpoint introspection, and every turn is the ordinary coding
+//! agent over the ordinary engine: the Anthropic adapter on
+//! [`FetchTransport`], `bash` in the session's sandbox — the shim DO
+//! ([`SandboxDo`]) or the `http` daemon the deployment selects —
+//! checkpoints through [`DoStore`]. Approval asks and their answers
+//! cross between the pump and the turn task only through channels —
+//! workerd runs everything on one thread.
+//!
+//! Turn driving (v1): a turn runs inside the client's live connection.
+//! Checkpoints land at the commit hook — before every side effect —
+//! and after each turn, so an eviction resumes through the engine's
+//! revive machinery; the alarm-driven queue that would let turns
+//! outlive requests is a v2 refinement (SPEC.md).
+//!
+//! Layout: the [`do_class`] submodule holds the `#[durable_object]`
+//! macro invocation — the one place the sanctioned `unsafe` exception
+//! applies, and the one place `missing_docs` is off, because the macro
+//! generates the wasm-bindgen constructor no doc comment can reach.
+//! Everything hand-written lives outside it, under the crate's lints.
+#![forbid(unsafe_code)]
+
+mod approval;
+mod sandbox;
+mod sandbox_http;
+mod sleeper;
+mod storage;
+mod transport;
+
+pub use approval::{Live, WsChannel};
+pub use sandbox::SandboxDo;
+pub use sleeper::DelaySleeper;
+pub use storage::DoStore;
+pub use transport::FetchTransport;
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use futures::StreamExt;
+use worker::{Env, Error, Method, Request, Response, Result, State, WebSocketPair};
+
+use rhema_agent_tools::DenyAll;
+use rhema_core::{Event, EventSink, ModelProvider, Session};
+use rhema_provider::Retrying;
+use rhema_provider_anthropic::AnthropicProvider;
+use rhema_rsd::{ResourceKey, Store, StoreError};
+
+use crate::backend::{BACKEND_VAR, SandboxBackend, SandboxEnv, URL_VAR, select_backend};
+use rhema_cloud_agent::{
+    AgentAssembly, ClientMessage, Sandbox, SandboxError, ServerMessage, SessionConfig,
+    ShellSandbox, WorkspaceSeed,
+};
+use sandbox::SandboxSource;
+use storage::session_key;
+
+/// How long an approval ask waits for its client before the gate
+/// denies. The client sees `approval_expired` when it happens.
+const APPROVAL_TIMEOUT: Duration = Duration::from_secs(120);
+
+/// The streaming sink: every engine event serialized to the socket.
+struct WsSink {
+    live: Arc<Live>,
+}
+
+impl EventSink for WsSink {
+    fn emit(&mut self, event: Event<'_>) {
+        if let Some(message) = ServerMessage::from_event(event) {
+            self.live.send(&message);
+        }
+    }
+}
+
+/// The routes of the DO. Free-standing so the hand-written logic keeps
+/// the crate's documentation lint; only the class module below is
+/// exempt from it.
+async fn route(state: &State, env: &Env, req: Request) -> Result<Response> {
+    let path = req.path();
+    match path.as_str() {
+        "/healthz" => Response::from_json(&serde_json::json!({ "ok": true })),
+        "/session" => snapshot(state).await,
+        "/config" if req.method() == Method::Post => set_config(req, state).await,
+        "/ws" => {
+            let upgrade = req
+                .headers()
+                .get("Upgrade")?
+                .is_some_and(|value| value.eq_ignore_ascii_case("websocket"));
+            if !upgrade {
+                return Response::error("the /ws route upgrades to a WebSocket", 426);
+            }
+            chat_socket(state, env).await
+        }
+        _ => Response::error("not found", 404),
+    }
+}
+
+/// The class the `#[durable_object]` macro turns into a JS
+/// constructor. `missing_docs` is expected to fire here and only here:
+/// the macro generates the wasm-bindgen constructor and fetch wrapper,
+/// which no doc comment can reach, and this module holds nothing else.
+/// The sanctioned `unsafe` exception (the macro emits some) is scoped
+/// to this module too.
+#[expect(
+    missing_docs,
+    reason = "the durable_object macro generates the \
+    wasm-bindgen constructor and fetch wrapper, which no doc comment can reach"
+)]
+mod do_class {
+    use super::route;
+    use worker::wasm_bindgen;
+    use worker::{DurableObject, Env, Request, Response, Result, State, durable_object};
+
+    #[durable_object]
+    pub struct CodingAgent {
+        pub(crate) state: State,
+        pub(crate) env: Env,
+    }
+
+    impl DurableObject for CodingAgent {
+        fn new(state: State, env: Env) -> Self {
+            Self { state, env }
+        }
+
+        async fn fetch(&self, req: Request) -> Result<Response> {
+            route(&self.state, &self.env, req).await
+        }
+    }
+}
+pub use do_class::CodingAgent;
+
+/// The store over this DO's database.
+fn store_of(state: &State) -> Result<DoStore> {
+    DoStore::new(state.storage().sql()).map_err(|err| Error::from(err.to_string()))
+}
+
+/// GET /session: the checkpointed record, as the client stored it.
+async fn snapshot(state: &State) -> Result<Response> {
+    let store = store_of(state)?;
+    let key = session_key(&state.id().to_string());
+    let record = store
+        .load_checked(&key)
+        .map_err(|err| Error::from(err.to_string()))?;
+    match record {
+        Some(record) => Response::from_json(&serde_json::json!({
+            "revision": record.revision,
+            "record": serde_json::to_value(&record)
+                .map_err(|err| Error::from(err.to_string()))?,
+        })),
+        None => Response::from_json(&serde_json::json!({ "revision": null })),
+    }
+}
+
+/// POST /config: installs the session's [`SessionConfig`].
+async fn set_config(mut req: Request, state: &State) -> Result<Response> {
+    let config: SessionConfig = req.json().await?;
+    let store = store_of(state)?;
+    let value = serde_json::to_string(&config).map_err(|err| Error::from(err.to_string()))?;
+    store
+        .set_meta("config", &value)
+        .map_err(|err| Error::from(err.to_string()))?;
+    Response::from_json(&serde_json::json!({ "ok": true }))
+}
+
+/// The stored config, defaults filling what was never set.
+fn load_config(env: &Env, store: &DoStore) -> Result<SessionConfig> {
+    let stored = store
+        .meta("config")
+        .map_err(|err| Error::from(err.to_string()))?;
+    match stored {
+        Some(value) => serde_json::from_str(&value)
+            .map_err(|err| Error::from(format!("stored config is malformed: {err}"))),
+        None => {
+            let mut config = SessionConfig::default();
+            if let Ok(model) = env.var("RHEMA_MODEL") {
+                config.model = model.to_string();
+            }
+            Ok(config)
+        }
+    }
+}
+
+/// One plain var of the deployment env, absent when unset — the
+/// `RHEMA_MODEL` pattern; secrets go through `env.secret`.
+fn env_var(env: &Env, name: &str) -> Option<String> {
+    env.var(name).ok().map(|value| value.to_string())
+}
+
+/// GET /ws: accept the socket and hand it to the pump. The pump and
+/// the turns it spawns run for the life of the connection — the
+/// client's live request.
+async fn chat_socket(state: &State, env: &Env) -> Result<Response> {
+    let pair = WebSocketPair::new()?;
+    state.accept_web_socket(&pair.server);
+
+    let store = store_of(state)?;
+    let config = load_config(env, &store)?;
+    let session_id = state.id().to_string();
+    // Secrets ride wrangler secrets, never DO storage.
+    let api_key = env
+        .secret("ANTHROPIC_API_KEY")
+        .map_err(|_| Error::from("the ANTHROPIC_API_KEY secret is not set"))?
+        .to_string();
+    let sandbox_token = env
+        .secret("SANDBOX_TOKEN")
+        .map(|secret| secret.to_string())
+        .unwrap_or_default();
+    // The sandbox backend rides the deployment env (spec: backend
+    // selection rides deployment env), read once per connection here
+    // where every other piece of env is. Under `http` the shim's DO
+    // binding is never resolved — no namespace call happens — so the
+    // binding is absent from deployments that do not need it.
+    let sandbox = match select_backend(&SandboxEnv {
+        backend: env_var(env, BACKEND_VAR),
+        url: env_var(env, URL_VAR),
+    })
+    .map_err(|err| Error::from(err.to_string()))?
+    {
+        SandboxBackend::Cloudflare => SandboxSource::Cloudflare(env.durable_object("SANDBOX")?),
+        SandboxBackend::Http(base) => SandboxSource::Http(base),
+    };
+
+    let live = Arc::new(Live {
+        socket: pair.server.clone(),
+        pending: Mutex::new(HashMap::new()),
+        cancelled: Arc::new(AtomicBool::new(false)),
+        busy: AtomicBool::new(false),
+    });
+    let channel = Arc::new(WsChannel::new(live.clone(), APPROVAL_TIMEOUT));
+
+    wasm_bindgen_futures::spawn_local(pump(
+        pair.server,
+        live,
+        channel,
+        store,
+        config,
+        api_key,
+        sandbox,
+        sandbox_token,
+        session_id,
+    ));
+    Response::from_websocket(pair.client)
+}
+
+/// The socket pump: client messages in, turns spawned, approvals
+/// routed. Runs for the connection's life.
+async fn pump(
+    server: worker::WebSocket,
+    live: Arc<Live>,
+    channel: Arc<WsChannel>,
+    store: DoStore,
+    config: SessionConfig,
+    api_key: String,
+    sandbox: SandboxSource,
+    sandbox_token: String,
+    session_id: String,
+) {
+    let mut events = match server.events() {
+        Ok(events) => events,
+        Err(_) => return,
+    };
+    while let Some(Ok(event)) = events.next().await {
+        match event {
+            worker::WebsocketEvent::Message(message) => {
+                let Some(text) = message.text() else {
+                    continue;
+                };
+                match serde_json::from_str::<ClientMessage>(&text) {
+                    Ok(ClientMessage::Chat { text }) => {
+                        // One turn at a time.
+                        if live.busy.swap(true, Ordering::SeqCst) {
+                            live.send(&ServerMessage::Error {
+                                message: "a turn is already running".into(),
+                            });
+                            continue;
+                        }
+                        live.cancelled.store(false, Ordering::SeqCst);
+                        wasm_bindgen_futures::spawn_local(run_turn(
+                            live.clone(),
+                            channel.clone(),
+                            store.clone(),
+                            config.clone(),
+                            api_key.clone(),
+                            sandbox.clone(),
+                            sandbox_token.clone(),
+                            session_id.clone(),
+                            text,
+                        ));
+                    }
+                    Ok(ClientMessage::Approve { id, approve }) => {
+                        channel.answer(&id, approve);
+                    }
+                    Ok(ClientMessage::Stop) => {
+                        live.cancelled.store(true, Ordering::SeqCst);
+                    }
+                    Err(err) => {
+                        live.send(&ServerMessage::Error {
+                            message: format!("unrecognized message: {err}"),
+                        });
+                    }
+                }
+            }
+            worker::WebsocketEvent::Close(_) => {
+                // The client is gone: the turn unwinds at its next
+                // check, and pending approvals deny by absence.
+                live.cancelled.store(true, Ordering::SeqCst);
+                channel.abandon_all();
+                break;
+            }
+        }
+    }
+}
+
+/// One turn: assemble the agent from durable state, run it, checkpoint
+/// around it. Every piece comes from the DO's storage or its bindings;
+/// nothing reads a filesystem.
+#[expect(
+    clippy::too_many_arguments,
+    reason = "the turn's inputs are the DO's \
+    durable pieces, each from a different source; a struct would only move the list"
+)]
+async fn run_turn(
+    live: Arc<Live>,
+    channel: Arc<WsChannel>,
+    store: DoStore,
+    config: SessionConfig,
+    api_key: String,
+    sandbox: SandboxSource,
+    sandbox_token: String,
+    session_id: String,
+    input: String,
+) {
+    let key = session_key(&session_id);
+    // 1. The checkpointed session, if any — the engine's revive
+    //    machinery closes committed-but-unanswered calls on resume.
+    let checkpointed = store
+        .load_checked(&key)
+        .map_err(|err| format!("checkpoint unreadable: {err}"));
+    let (revision, session) = match checkpointed {
+        Ok(Some(record)) => (
+            record.revision,
+            record
+                .session
+                .map(|wrapped| wrapped.core)
+                .unwrap_or_default(),
+        ),
+        Ok(None) => (0, Session::default()),
+        Err(message) => {
+            live.busy.store(false, Ordering::SeqCst);
+            live.send(&ServerMessage::Error { message });
+            return;
+        }
+    };
+
+    // 2. The provider: the Anthropic dialect over host fetch, with the
+    //    retry ladder paused by workerd's timers. The short one on both
+    //    classes: a worker's request outlives no five-minute outage.
+    let provider: Arc<dyn ModelProvider> = Arc::new(
+        Retrying::with_ladders(
+            Arc::new(AnthropicProvider::new_with_transport(
+                api_key,
+                Arc::new(FetchTransport::new()),
+            )),
+            vec![Duration::from_secs(1), Duration::from_secs(2)],
+            vec![Duration::from_secs(1), Duration::from_secs(2)],
+        )
+        .with_sleeper(Arc::new(DelaySleeper)),
+    );
+
+    // 3. The sandbox of the same session name, seeded on first use —
+    //    the shim DO wrapper or the shared HTTP client, per the
+    //    backend the deployment selected.
+    let sandbox = sandbox.for_session(&session_id, &sandbox_token);
+    if let Err(err) = ensure_seeded(&store, sandbox.as_ref(), &config).await {
+        live.busy.store(false, Ordering::SeqCst);
+        live.send(&ServerMessage::Error {
+            message: err.to_string(),
+        });
+        return;
+    }
+
+    // 4. The agent: the ordinary assembly, resumed from the checkpoint.
+    let skills = config.skills();
+    let assembly = AgentAssembly {
+        model: config.model.clone(),
+        base_system: config.base_system.clone(),
+        skills,
+        max_turn_steps: config.max_turn_steps,
+        thinking: rhema_core::request::ThinkingEffort::Default,
+        workspace: config.workspace.clone(),
+        transcript: Some(rhema_transcript::TranscriptSettings::default()),
+        approval: config.approval.clone(),
+        channel: Some(channel),
+    };
+    let shell = Arc::new(ShellSandbox::new(sandbox));
+    let mut agent = match assembly.assemble_from_session(provider, shell, session) {
+        Ok(agent) => agent,
+        Err(err) => {
+            live.busy.store(false, Ordering::SeqCst);
+            live.send(&ServerMessage::Error {
+                message: err.to_string(),
+            });
+            return;
+        }
+    };
+
+    // 5. The commit hook: the checkpoint before every side effect. The
+    //    revision rides a cell the hook and this function share.
+    let revision = Arc::new(Mutex::new(revision));
+    let hook = {
+        let store = store.clone();
+        let key = key.clone();
+        let revision = revision.clone();
+        Box::new(move |session: &Session| {
+            checkpoint(&store, &key, &revision, session)
+                .map_err(|err| rhema_core::CommitError(err.to_string()))
+        }) as rhema_core::CommitHook
+    };
+    agent.set_commit_hook(Some(hook));
+
+    // 6. The turn. The turn-level approval stays deny: the gate in
+    //    front of the shell is the only path a call takes, and the
+    //    fail-safe default stands behind it.
+    let mut sink = WsSink { live: live.clone() };
+    let result = agent
+        .turn_with_cancel(input, &mut sink, Arc::new(DenyAll), live.cancelled.clone())
+        .await;
+
+    // 7. The post-turn checkpoint and the close. A failed checkpoint is
+    //    reported to the client AND logged: the reply still leaves, but
+    //    the history only survives an eviction if the write did.
+    match result {
+        Ok(output) => {
+            if let Err(err) = checkpoint(&store, &key, &revision, agent.session()) {
+                worker::console_error!("checkpoint after a completed turn failed for {key}: {err}");
+                live.send(&ServerMessage::Error {
+                    message: format!("post-turn checkpoint failed: {err}"),
+                });
+            }
+            live.send(&ServerMessage::turn_complete(
+                output.stop_reason,
+                output.usage,
+                agent.session().cost_usd,
+            ));
+        }
+        Err(err) => {
+            // The history is whole; checkpoint so a follow-up continues
+            // instead of replaying. Losing this write to a failed save
+            // is logged loudly: the next session would start from an
+            // older revision with no other signal.
+            if let Err(state) = checkpoint(&store, &key, &revision, agent.session()) {
+                worker::console_error!("checkpoint on a failed turn lost for {key}: {state}");
+            }
+            live.send(&ServerMessage::turn_failed(err.to_string()));
+        }
+    }
+    live.busy.store(false, Ordering::SeqCst);
+}
+
+/// Why making a turn's durable pieces failed: the DO's storage refused
+/// a checkpoint or meta write, or the sandbox refused the workspace
+/// seed. One type so every turn-loop failure reports the same way.
+#[derive(Debug, thiserror::Error)]
+pub enum TurnStateError {
+    /// The DO's SQLite storage layer refused a read or write.
+    #[error("session store: {0}")]
+    Store(#[from] StoreError),
+    /// The sandbox refused the workspace seed (a git clone that would
+    /// not clone, a sandbox that is not up).
+    #[error("workspace seed: {0}")]
+    Seed(#[from] SandboxError),
+    /// One skill's `SKILL.md` write failed; the name says which.
+    #[error("seeding skill {name}: {source}")]
+    Skill {
+        /// The skill whose body could not be written.
+        name: String,
+        /// The sandbox's refusal.
+        source: SandboxError,
+    },
+}
+
+/// Seeds the workspace the first time it is used — a git clone when the
+/// session carries one — and records that in the meta table, so a
+/// resumed DO does not re-clone over a workspace the client has since
+/// changed. A full container eviction loses the files; the next
+/// session re-seeds by clearing the marker (SPEC.md).
+async fn ensure_seeded(
+    store: &DoStore,
+    sandbox: &dyn Sandbox,
+    config: &SessionConfig,
+) -> Result<(), TurnStateError> {
+    match store.meta("workspace:seeded") {
+        Ok(Some(_)) => return Ok(()),
+        Ok(None) => {}
+        Err(err) => return Err(err.into()),
+    }
+    if let Some(url) = &config.seed_git {
+        sandbox
+            .seed(&WorkspaceSeed::GitClone(url.clone()))
+            .await
+            .map_err(TurnStateError::from)?;
+    }
+    for skill in &config.skills {
+        let path = rhema_core::workspace::skills_dir(std::path::Path::new(&config.workspace))
+            .join(&skill.name)
+            .join("SKILL.md");
+        let path = path.display().to_string();
+        sandbox
+            .write_file(&path, skill.body.as_bytes())
+            .await
+            .map_err(|err| TurnStateError::Skill {
+                name: skill.name.clone(),
+                source: err,
+            })?;
+    }
+    store
+        .set_meta("workspace:seeded", "1")
+        .map_err(TurnStateError::from)
+}
+
+/// Persists `session` under the key at `current + 1`. Called from the
+/// commit hook (sync, inside the turn) and after the turn. The engine's
+/// session rides in the record's RSD wrapper — the same shape the
+/// dispatcher checkpoints — so the row stays one contract's shape.
+fn checkpoint(
+    store: &DoStore,
+    key: &ResourceKey,
+    current: &Mutex<u64>,
+    session: &Session,
+) -> Result<(), TurnStateError> {
+    let mut current = current
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let mut record = storage::fresh_record(key.clone());
+    record.revision = *current + 1;
+    let mut wrapped = rhema_rsd::Session::fresh(key.clone(), storage::now_millis());
+    wrapped.core = session.clone();
+    record.session = Some(wrapped);
+    store.save(record, Some(*current))?;
+    *current += 1;
+    Ok(())
+}

crates/rhema-cloudflare-code/src/worker/sandbox.rs

@@ -0,0 +1,236 @@
+//! The sandbox clients of one session and the deployment's choice
+//! between them: [`SandboxDo`] over the session's SandboxShim Durable
+//! Object, and — under `SANDBOX_BACKEND=http` — the shared
+//! [`HttpSandbox`] over host fetch.
+//!
+//! The shim (a small JS DO, `deploy/cloudflare/sandbox-shim.mjs`) owns
+//! the container binding — the container runtime is a JS-only API — and
+//! speaks plain JSON over its DO fetch: exec, seed, get, put. The two
+//! DOs of one session share a name, so binding by the session's id
+//! reaches the container whose lifetime matches the session's
+//! checkpoints. A shared secret guards the shim's HTTP surface. Those
+//! routes and shapes are the sandbox wire protocol, pinned in
+//! `rhema-sandbox::protocol`; the `http` backend speaks the same
+//! protocol to a `rhema-sandboxd` on any machine, where the DO binding
+//! is never touched at all.
+#![forbid(unsafe_code)]
+
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+
+use async_trait::async_trait;
+use base64::Engine;
+use worker::{Headers, Method, ObjectNamespace, Request, RequestInit};
+
+use rhema_cloud_agent::{ExecOutcome, ExecRequest, Sandbox, SandboxError, WorkspaceSeed};
+use rhema_sandbox::protocol::{
+    Ack, EXEC, ExecCall, ExecReply, FileReply, GET, PUT, PutFile, SEED, SeedRequest, TOKEN_HEADER,
+    stream_reply,
+};
+use rhema_sandbox_http::HttpSandbox;
+
+use super::sandbox_http::FetchHttpTransport;
+
+/// The sandbox a turn's `bash` runs in: the deployment's backend choice
+/// ([`crate::backend::select_backend`]), resolved once per connection
+/// from its env and threaded to every turn. Under
+/// [`SandboxSource::Cloudflare`] the shim's DO binding is the only
+/// sandbox thing touched; under [`SandboxSource::Http`] no namespace
+/// call happens at all — that is the point of the backend.
+#[derive(Clone)]
+pub(crate) enum SandboxSource {
+    /// The session's SandboxShim DO binding.
+    Cloudflare(ObjectNamespace),
+    /// The protocol daemon's base URL; the session name rides the path.
+    Http(String),
+}
+
+impl SandboxSource {
+    /// The session's sandbox client: the DO wrapper, or the shared HTTP
+    /// client over host fetch — the one seam either way, so the turn's
+    /// shell and seeding cannot tell the backends apart.
+    pub(crate) fn for_session(&self, name: &str, token: &str) -> Arc<dyn Sandbox> {
+        match self {
+            SandboxSource::Cloudflare(namespace) => {
+                Arc::new(SandboxDo::new(namespace.clone(), name, token))
+            }
+            SandboxSource::Http(base) => Arc::new(HttpSandbox::new_with_transport(
+                base.clone(),
+                name,
+                token,
+                Arc::new(FetchHttpTransport),
+            )),
+        }
+    }
+}
+
+/// The sandbox of one session, over its DO binding.
+pub struct SandboxDo {
+    namespace: ObjectNamespace,
+    /// The DO name — both DOs of a session derive from the same string.
+    name: String,
+    /// The shared secret the shim checks.
+    token: String,
+}
+
+impl SandboxDo {
+    /// A client for the sandbox DO named `name` in `namespace`.
+    pub fn new(
+        namespace: ObjectNamespace,
+        name: impl Into<String>,
+        token: impl Into<String>,
+    ) -> Self {
+        Self {
+            namespace,
+            name: name.into(),
+            token: token.into(),
+        }
+    }
+
+    /// One JSON POST to the shim: the whole exchange — fetch, status,
+    /// body text — runs inside one local task, because every piece is
+    /// JS-backed and `!Send`; what crosses back is plain data.
+    async fn post<T: serde::de::DeserializeOwned>(
+        &self,
+        route: &str,
+        body: &impl serde::Serialize,
+    ) -> Result<T, SandboxError> {
+        let id = self
+            .namespace
+            .id_from_name(&self.name)
+            .map_err(shim_error)?;
+        let stub = id.get_stub().map_err(shim_error)?;
+        let mut init = RequestInit::new();
+        init.method = Method::Post;
+        init.headers = Headers::new();
+        init.headers
+            .set(TOKEN_HEADER, &self.token)
+            .map_err(shim_error)?;
+        init.headers
+            .set("content-type", "application/json")
+            .map_err(shim_error)?;
+        let payload = serde_json::to_vec(body).map_err(|err| {
+            SandboxError::Unavailable(format!("request does not serialize: {err}"))
+        })?;
+        init.body = Some(worker::js_sys::Uint8Array::from(payload.as_slice()).into());
+        let request = Request::new_with_init(&format!("https://sandbox.shim{route}"), &init)
+            .map_err(shim_error)?;
+        let (status_slot, status) = futures::channel::oneshot::channel();
+        let (body_slot, body) = futures::channel::oneshot::channel::<String>();
+        wasm_bindgen_futures::spawn_local(async move {
+            let outcome: worker::Result<(u16, String)> = async {
+                let mut response = stub.fetch_with_request(request).await?;
+                let status = response.status_code();
+                let text = response.text().await?;
+                worker::Result::Ok((status, text))
+            }
+            .await;
+            match outcome {
+                Ok((status, text)) => {
+                    let _ = status_slot.send(status);
+                    let _ = body_slot.send(text);
+                }
+                Err(err) => {
+                    let _ = status_slot.send(0);
+                    let _ = body_slot.send(err.to_string());
+                }
+            }
+        });
+        let status = status
+            .await
+            .map_err(|_| shim_error(worker::Error::from("the shim call ended without a status")))?;
+        let text = match body.await {
+            Ok(text) => text,
+            Err(_) => {
+                return Err(shim_error(worker::Error::from(
+                    "the shim call ended without a body",
+                )));
+            }
+        };
+        if !(200..300).contains(&status) {
+            return Err(SandboxError::Unavailable(format!(
+                "the sandbox shim answered (HTTP {status}): {text}"
+            )));
+        }
+        serde_json::from_str(&text)
+            .map_err(|err| SandboxError::Unavailable(format!("malformed shim reply: {err}")))
+    }
+}
+
+/// Maps a transport failure below the seam: unreachable, not failed.
+fn shim_error(err: worker::Error) -> SandboxError {
+    SandboxError::Unavailable(err.to_string())
+}
+
+/// The ack's `error` string as a sandbox result: absent and null both
+/// mean success.
+fn ack_result(ack: Ack) -> Result<(), SandboxError> {
+    match ack.error {
+        Some(message) => Err(SandboxError::Failed(message)),
+        None => Ok(()),
+    }
+}
+
+#[async_trait]
+impl Sandbox for SandboxDo {
+    async fn exec(
+        &self,
+        request: ExecRequest,
+        cancelled: &AtomicBool,
+        progress: &mut (dyn for<'a> FnMut(&'a str) + Send),
+    ) -> Result<ExecOutcome, SandboxError> {
+        if cancelled.load(std::sync::atomic::Ordering::Relaxed) {
+            return Err(SandboxError::Failed("cancelled".into()));
+        }
+        let call = ExecCall {
+            command: request.command,
+            cwd: Some(request.cwd),
+            timeout_secs: request.timeout.as_secs(),
+        };
+        let reply: ExecReply = self.post(EXEC, &call).await?;
+        // No cancel check here: the command completed remotely and its
+        // result is real — the model sees the truth, whatever the turn
+        // decided meanwhile.
+        if reply.timed_out {
+            return Err(SandboxError::Timeout(request.timeout.as_secs()));
+        }
+        stream_reply(&reply, progress);
+        Ok(ExecOutcome {
+            stdout: reply.stdout.into_bytes(),
+            stderr: reply.stderr.into_bytes(),
+            code: reply.exit_code,
+        })
+    }
+
+    async fn seed(&self, seed: &WorkspaceSeed) -> Result<(), SandboxError> {
+        let body = match seed {
+            WorkspaceSeed::GitClone(url) => SeedRequest::Git { git: url.clone() },
+            WorkspaceSeed::Tarball {
+                name,
+                tar_gz_base64,
+            } => SeedRequest::Tarball {
+                tarball_base64: tar_gz_base64.clone(),
+                tarball_name: Some(name.clone()),
+            },
+        };
+        ack_result(self.post(SEED, &body).await?)
+    }
+
+    async fn read_file(&self, path: &str) -> Result<Vec<u8>, SandboxError> {
+        let reply: FileReply = self.post(GET, &serde_json::json!({ "path": path })).await?;
+        if let Some(error) = reply.error {
+            return Err(SandboxError::Failed(error));
+        }
+        base64::engine::general_purpose::STANDARD
+            .decode(reply.content_base64.as_deref().unwrap_or_default())
+            .map_err(|err| SandboxError::Failed(format!("malformed file reply: {err}")))
+    }
+
+    async fn write_file(&self, path: &str, contents: &[u8]) -> Result<(), SandboxError> {
+        let put = PutFile {
+            path: path.to_owned(),
+            content_base64: base64::engine::general_purpose::STANDARD.encode(contents),
+        };
+        ack_result(self.post(PUT, &put).await?)
+    }
+}

crates/rhema-cloudflare-code/src/worker/sandbox_http.rs

@@ -0,0 +1,80 @@
+//! The host-fetch transport under the `http` sandbox backend: the
+//! [`HttpTransport`] the DO hands [`HttpSandbox`] when the deployment
+//! selects a protocol daemon over the shim.
+#![forbid(unsafe_code)]
+
+use async_trait::async_trait;
+use worker::{Fetch, Headers, Method, Request, RequestInit};
+
+use rhema_sandbox::SandboxError;
+use rhema_sandbox::protocol::TOKEN_HEADER;
+use rhema_sandbox_http::HttpTransport;
+
+/// [`HttpTransport`] over workerd's host fetch. Stateless: workerd owns
+/// the socket and the connection pool. The exchange runs inside one
+/// local task — the fetch and its body text are JS-backed and `!Send`,
+/// so no JS handle may cross an await the seam's future owns — and only
+/// the plain status-and-bytes pair crosses back, `SandboxDo::post`'s
+/// spawn_local bridge.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct FetchHttpTransport;
+
+#[async_trait]
+impl HttpTransport for FetchHttpTransport {
+    async fn post(
+        &self,
+        url: &str,
+        token: &str,
+        body: Vec<u8>,
+    ) -> Result<(u16, Vec<u8>), SandboxError> {
+        let mut init = RequestInit::new();
+        init.method = Method::Post;
+        init.headers = Headers::new();
+        init.headers
+            .set(TOKEN_HEADER, token)
+            .map_err(|err| SandboxError::Unavailable(err.to_string()))?;
+        init.headers
+            .set("content-type", "application/json")
+            .map_err(|err| SandboxError::Unavailable(err.to_string()))?;
+        // A body rides as bytes; the JS side wants a typed array.
+        init.body = Some(worker::js_sys::Uint8Array::from(body.as_slice()).into());
+        let request = Request::new_with_init(url, &init)
+            .map_err(|err| SandboxError::Unavailable(err.to_string()))?;
+        let (status_slot, status) = futures::channel::oneshot::channel();
+        let (body_slot, body) = futures::channel::oneshot::channel::<String>();
+        wasm_bindgen_futures::spawn_local(async move {
+            let outcome: worker::Result<(u16, String)> = async {
+                let mut response = Fetch::Request(request).send().await?;
+                let status = response.status_code();
+                let text = response.text().await?;
+                worker::Result::Ok((status, text))
+            }
+            .await;
+            match outcome {
+                Ok((status, text)) => {
+                    let _ = status_slot.send(status);
+                    let _ = body_slot.send(text);
+                }
+                Err(err) => {
+                    // Status 0 marks "no HTTP answer": a real reply
+                    // always carries its code, so the pair stays
+                    // unambiguous.
+                    let _ = status_slot.send(0);
+                    let _ = body_slot.send(err.to_string());
+                }
+            }
+        });
+        let status = status.await.map_err(|_| {
+            SandboxError::Unavailable("the fetch task ended without a status".into())
+        })?;
+        let text = body
+            .await
+            .map_err(|_| SandboxError::Unavailable("the fetch task ended without a body".into()))?;
+        if status == 0 {
+            // The fetch itself failed; the body slot carries workerd's
+            // error text.
+            return Err(SandboxError::Unavailable(text));
+        }
+        Ok((status, text.into_bytes()))
+    }
+}

crates/rhema-cloudflare-code/src/worker/sleeper.rs

@@ -0,0 +1,31 @@
+//! The retry pause over workerd's timers.
+//!
+//! tokio's time driver does not run inside workerd: there is no
+//! Rust-side timer to fire. [`DelaySleeper`] implements the provider
+//! crate's [`Sleeper`] over `worker::Delay`, which schedules a host
+//! `setTimeout` — the same backoff ladder, a pause the runtime can
+//! actually deliver.
+#![forbid(unsafe_code)]
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use rhema_provider::Sleeper;
+
+/// The host-timer pause.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct DelaySleeper;
+
+#[async_trait]
+impl Sleeper for DelaySleeper {
+    async fn sleep(&self, duration: Duration) {
+        // The `!Send` timer future runs on the local task queue; this
+        // future only waits the plain one-shot back.
+        let (done, fired) = futures::channel::oneshot::channel();
+        wasm_bindgen_futures::spawn_local(async move {
+            worker::Delay::from(duration).await;
+            let _ = done.send(());
+        });
+        let _ = fired.await;
+    }
+}

crates/rhema-cloudflare-code/src/worker/storage.rs

@@ -0,0 +1,225 @@
+//! The DO's SQLite storage as the persistence layer.
+//!
+//! Sessions checkpoint as [`KeyRecord`]s through the
+//! [`rhema_rsd::Store`] contract — one row per session key, the
+//! revision column the compare-and-swap token — so the DO gets the
+//! same revision-guarded persistence the dispatcher spec'd, not an
+//! ad-hoc table. The workspace's seeded flag lives beside them in a
+//! meta table. One row, one writer: the DO instance per session id is
+//! the single writer, and the revision check still holds the line
+//! against a split brain during eviction. `SqlStorage` calls are
+//! synchronous (no await points), so a read-modify-write between two
+//! of them cannot interleave with another task.
+#![forbid(unsafe_code)]
+
+use rhema_rsd::{KeyRecord, ResourceKey, Revision, Store, StoreError};
+use worker::SqlStorage;
+
+/// The storage-backed store. Cloneable — the inner handle is a JS
+/// object the runtime keeps alive — and `Send + Sync` by the runtime's
+/// own guarantee for storage handles.
+#[derive(Clone)]
+pub struct DoStore {
+    sql: SqlStorage,
+}
+
+/// The one session key this DO persists.
+pub fn session_key(session_id: &str) -> ResourceKey {
+    ResourceKey::new(format!("coding-agent/{session_id}"))
+}
+
+/// Milliseconds since the Unix epoch, from the host clock — the
+/// timestamps a fresh [`KeyRecord`] carries.
+pub fn now_millis() -> u64 {
+    worker::js_sys::Date::now() as u64
+}
+
+/// The record a session starts from: fresh, with the session absent
+/// until the first checkpoint lands.
+pub fn fresh_record(key: ResourceKey) -> KeyRecord {
+    KeyRecord::fresh(key, now_millis())
+}
+
+impl DoStore {
+    /// Opens (and creates) the schema on the DO's database.
+    pub fn new(sql: SqlStorage) -> Result<Self, StoreError> {
+        let store = Self { sql };
+        store.run(
+            "CREATE TABLE IF NOT EXISTS records (
+                key TEXT PRIMARY KEY,
+                revision INTEGER NOT NULL,
+                data BLOB NOT NULL
+            )",
+            None,
+        )?;
+        store.run(
+            "CREATE TABLE IF NOT EXISTS meta (
+                key TEXT PRIMARY KEY,
+                value TEXT NOT NULL
+            )",
+            None,
+        )?;
+        Ok(store)
+    }
+
+    /// Runs one statement, mapping worker errors below the seam.
+    fn run(
+        &self,
+        query: &str,
+        bindings: Option<Vec<worker::SqlStorageValue>>,
+    ) -> Result<worker::SqlCursor, StoreError> {
+        self.sql
+            .exec(query, bindings)
+            .map_err(|err| StoreError::Backend(err.to_string()))
+    }
+
+    /// The stored value of a meta key, if any.
+    pub fn meta(&self, key: &str) -> Result<Option<String>, StoreError> {
+        let cursor = self.run(
+            "SELECT value FROM meta WHERE key = ?1",
+            Some(vec![worker::SqlStorageValue::String(key.to_owned())]),
+        )?;
+        Ok(cursor
+            .next::<MetaRow>()
+            .filter_map(Result::ok)
+            .map(|row| row.value)
+            .next())
+    }
+
+    /// Upserts a meta key.
+    pub fn set_meta(&self, key: &str, value: &str) -> Result<(), StoreError> {
+        self.run(
+            "INSERT INTO meta (key, value) VALUES (?1, ?2)
+             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+            Some(vec![
+                worker::SqlStorageValue::String(key.to_owned()),
+                worker::SqlStorageValue::String(value.to_owned()),
+            ]),
+        )?;
+        Ok(())
+    }
+
+    /// The checkpointed record for a key, with its revision.
+    pub fn load_checked(&self, key: &ResourceKey) -> Result<Option<KeyRecord>, StoreError> {
+        let cursor = self.run(
+            "SELECT data FROM records WHERE key = ?1",
+            Some(vec![worker::SqlStorageValue::String(
+                key.as_str().to_owned(),
+            )]),
+        )?;
+        let row = cursor
+            .next::<DataRow>()
+            .filter_map(Result::ok)
+            .map(|row| row.data)
+            .next();
+        match row {
+            Some(data) => Ok(Some(serde_json::from_str(&data)?)),
+            None => Ok(None),
+        }
+    }
+
+    /// The stored revision of a key, for the conflict diagnostics.
+    fn stored_revision(&self, key: &ResourceKey) -> Result<Option<Revision>, StoreError> {
+        let cursor = self.run(
+            "SELECT revision FROM records WHERE key = ?1",
+            Some(vec![worker::SqlStorageValue::String(
+                key.as_str().to_owned(),
+            )]),
+        )?;
+        Ok(cursor
+            .next::<RevisionRow>()
+            .filter_map(Result::ok)
+            .map(|row| row.revision)
+            .next())
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct DataRow {
+    data: String,
+}
+
+#[derive(serde::Deserialize)]
+struct RevisionRow {
+    revision: Revision,
+}
+
+#[derive(serde::Deserialize)]
+struct MetaRow {
+    value: String,
+}
+
+impl Store for DoStore {
+    fn load(&self, key: &ResourceKey) -> Result<Option<KeyRecord>, StoreError> {
+        self.load_checked(key)
+    }
+
+    fn save(
+        &self,
+        record: KeyRecord,
+        expected_revision: Option<Revision>,
+    ) -> Result<(), StoreError> {
+        let data = serde_json::to_string(&record)?;
+        let written = match expected_revision {
+            // The key must not exist yet; a conflict is a rejection.
+            None => {
+                let cursor = self.run(
+                    "INSERT INTO records (key, revision, data) VALUES (?1, ?2, ?3)
+                     ON CONFLICT(key) DO NOTHING",
+                    Some(vec![
+                        worker::SqlStorageValue::String(record.key.as_str().to_owned()),
+                        worker::SqlStorageValue::Integer(record.revision as i64),
+                        worker::SqlStorageValue::String(data),
+                    ]),
+                )?;
+                cursor.rows_written()
+            }
+            Some(expected) => {
+                let cursor = self.run(
+                    "UPDATE records SET revision = ?2, data = ?3
+                     WHERE key = ?1 AND revision = ?4",
+                    Some(vec![
+                        worker::SqlStorageValue::String(record.key.as_str().to_owned()),
+                        worker::SqlStorageValue::Integer(record.revision as i64),
+                        worker::SqlStorageValue::String(data),
+                        worker::SqlStorageValue::Integer(expected as i64),
+                    ]),
+                )?;
+                cursor.rows_written()
+            }
+        };
+        if written == 0 {
+            let actual = self.stored_revision(&record.key)?.unwrap_or(0);
+            return Err(StoreError::RevisionMismatch {
+                expected: expected_revision.unwrap_or(0),
+                actual,
+            });
+        }
+        Ok(())
+    }
+
+    fn delete(&self, key: &ResourceKey) -> Result<bool, StoreError> {
+        let cursor = self.run(
+            "DELETE FROM records WHERE key = ?1",
+            Some(vec![worker::SqlStorageValue::String(
+                key.as_str().to_owned(),
+            )]),
+        )?;
+        // `key` is the table's primary key: at most one row per delete.
+        Ok(cursor.rows_written() > 0)
+    }
+
+    fn keys(&self) -> Result<Vec<ResourceKey>, StoreError> {
+        let cursor = self.run("SELECT key FROM records ORDER BY key", None)?;
+        Ok(cursor
+            .next::<KeyRow>()
+            .filter_map(Result::ok)
+            .map(|row| ResourceKey::new(row.key))
+            .collect())
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct KeyRow {
+    key: String,
+}

crates/rhema-cloudflare-code/src/worker/transport.rs

@@ -0,0 +1,110 @@
+//! The host-fetch transport: the [`Transport`] the DO speaks the
+//! Anthropic Messages API through, and the bridge that keeps the seam
+//! `Send`.
+//!
+//! workerd gives wasm no sockets; every request rides the host's
+//! `fetch`. The response body arrives as a JS-backed stream — raw
+//! handles into the JS heap, `!Send` by soundness — so the bridge runs
+//! a local task that pumps plain `Vec<u8>` chunks into a channel whose
+//! receiver is plain data and therefore `Send`. The JS handle never
+//! leaves the bridge; the provider adapters above the seam never know.
+#![forbid(unsafe_code)]
+
+use async_trait::async_trait;
+use futures::{SinkExt, StreamExt};
+use rhema_provider::{HttpRequest, HttpResponse, Transport, TransportError};
+use worker::{Fetch, Method, Request, RequestInit};
+
+/// The global-`fetch` transport. Stateless: workerd owns the socket and
+/// the connection pool.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct FetchTransport;
+
+impl FetchTransport {
+    /// The transport.
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+/// The HTTP method of a request, by name — the seam carries the method
+/// as text.
+fn method(name: &str) -> Method {
+    match name {
+        "POST" => Method::Post,
+        "PUT" => Method::Put,
+        "DELETE" => Method::Delete,
+        "HEAD" => Method::Head,
+        "PATCH" => Method::Patch,
+        _ => Method::Get,
+    }
+}
+
+#[async_trait]
+impl Transport for FetchTransport {
+    async fn send(&self, request: HttpRequest) -> Result<HttpResponse, TransportError> {
+        let mut init = RequestInit::new();
+        init.method = method(&request.method);
+        init.headers = worker::Headers::new();
+        for (name, value) in &request.headers {
+            init.headers
+                .set(name, value)
+                .map_err(|err| TransportError::Other(err.to_string()))?;
+        }
+        if let Some(body) = &request.body {
+            // A body rides as bytes; the JS side wants a typed array.
+            init.body = Some(worker::js_sys::Uint8Array::from(body.as_slice()).into());
+        }
+        let fetch_request = Request::new_with_init(&request.url, &init)
+            .map_err(|err| TransportError::Other(err.to_string()))?;
+        // `HttpRequest::timeout` is not carried: workerd gives wasm no
+        // Rust-side timers to enforce it with, and the cloud path sends
+        // none — a model generation runs unbounded, and workerd bounds
+        // the fetch itself.
+        //
+        // The whole exchange runs inside one local task — the fetch
+        // future and the body stream are JS-backed and `!Send` — which
+        // reports the status up front and pumps plain chunks into a
+        // channel the seam's receiver can own.
+        let (chunks, receiver) = futures::channel::mpsc::unbounded();
+        let (head_slot, head) = futures::channel::oneshot::channel();
+        wasm_bindgen_futures::spawn_local(async move {
+            let mut sink = chunks;
+            let failure = |err: String| TransportError::Other(err);
+            let mut response = match Fetch::Request(fetch_request).send().await {
+                Ok(response) => response,
+                Err(err) => {
+                    let _ = head_slot.send(None);
+                    let _ = sink.send(Err(failure(err.to_string()))).await;
+                    return;
+                }
+            };
+            let headers: Vec<(String, String)> = response.headers().entries().collect();
+            let _ = head_slot.send(Some((response.status_code(), headers)));
+            let mut body = match response.stream() {
+                Ok(body) => body,
+                Err(err) => {
+                    let _ = sink.send(Err(failure(err.to_string()))).await;
+                    return;
+                }
+            };
+            while let Some(item) = body.next().await {
+                let chunk = item.map_err(|err| failure(err.to_string()));
+                if sink.send(chunk).await.is_err() {
+                    break; // the reader is gone; stop pumping
+                }
+            }
+        });
+        let (status, headers) = head
+            .await
+            .map_err(|_| TransportError::Other("the fetch task ended before its status".into()))?
+            .ok_or_else(|| {
+                TransportError::Other("the fetch failed before a response head".into())
+            })?;
+        Ok(HttpResponse {
+            status,
+            headers,
+            body: Box::new(receiver),
+        })
+    }
+}

crates/rhema-sandbox-server/Cargo.toml

@@ -0,0 +1,57 @@
+[package]
+name = "rhema-sandbox-server"
+description = "`rhema-sandboxd`: the rhema sandbox wire protocol served from a host, one root directory per session"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+authors.workspace = true
+
+[dependencies]
+# The seam, the protocol, and `FsRoot` (feature `spawn`)
+rhema-sandbox = { version = "0.2.6", features = ["spawn"] }
+# The exec and seed routes run on the local backend, one `LocalSandbox`
+# per request
+rhema-sandbox-local = { version = "0.2.6" }
+# The daemon's HTTP surface: axum for the routes
+axum = { workspace = true }
+# The binary's args; the version pin mirrors the rest of the workspace's
+clap = { version = "4.5", features = ["derive"] }
+# File contents cross the put/get routes base64
+base64 = "0.22"
+# The error replies and the parsed request bodies
+serde = { workspace = true }
+serde_json = { workspace = true }
+# Serving (rt-multi-thread, sockets) and the exec route's child
+tokio = { workspace = true, features = [
+    "rt-multi-thread",
+    "net",
+    "macros",
+    "process",
+    "io-util",
+    "time",
+    "fs",
+] }
+
+[dev-dependencies]
+# The loopback test drives the daemon with a raw HTTP/1.1 client
+tokio = { workspace = true }
+# The loopback test drives the real client over the real wire
+rhema-sandbox-http = { version = "0.2.6" }
+# The daemon roots in throwaway directories
+tempfile = "3"
+# `probe_healthz` speaks raw HTTP to `/healthz` — outside the protocol
+# client — so the token middleware is tested on the wire itself; the
+# pin and features mirror rhema-sandbox-http's native transport
+reqwest = { version = "0.12", default-features = false, features = [
+    "json",
+    "rustls-tls",
+] }
+
+[[bin]]
+name = "rhema-sandboxd"
+path = "src/bin/rhema-sandboxd.rs"
+
+[lints]
+workspace = true

crates/rhema-sandbox-server/src/bin/rhema-sandboxd.rs

@@ -0,0 +1,16 @@
+//! Binary wiring only: parse the args, bind, serve. The daemon, its
+//! routes, and the run loop live in `rhema_sandbox_server`.
+
+#![forbid(unsafe_code)]
+
+use clap::Parser;
+use rhema_sandbox_server::{Args, run};
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+    if let Err(err) = run(args).await {
+        eprintln!("error: {err}");
+        std::process::exit(1);
+    }
+}

crates/rhema-sandbox-server/src/lib.rs

@@ -0,0 +1,696 @@
+//! The daemon half of the protocol: `rhema-sandboxd` served from a host.
+//!
+//! One [`FsRoot`] per session name under `--base`, created on first
+//! use; exec through a [`LocalSandbox`] per request (its `timeout`
+//! prefix is the kill, its stream pump the cap); files and seeds
+//! straight on the root. The token is checked on every route,
+//! `/healthz` included. The daemon adds no caps of its own: the
+//! streams arrive tail-capped because [`LocalSandbox`] caps them
+//! before the reply is built — the spec's "sandbox side caps streams"
+//! clause, inherited rather than restated.
+#![forbid(unsafe_code)]
+
+use std::collections::BTreeMap;
+use std::path::PathBuf;
+use std::sync::atomic::AtomicBool;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use axum::extract::{Path, Request, State};
+use axum::http::StatusCode;
+use axum::middleware::Next;
+use axum::response::{IntoResponse, Response};
+use axum::routing::{get, post};
+use axum::{Json, Router};
+use base64::Engine;
+use clap::Parser;
+
+use rhema_sandbox::fsroot::FsRoot;
+use rhema_sandbox::protocol::{
+    Ack, EXEC, ExecCall, ExecReply, FileReply, GET, PUT, PutFile, SEED, SeedRequest, TOKEN_HEADER,
+};
+use rhema_sandbox::{ExecRequest, Sandbox, SandboxError, WorkspaceSeed};
+use rhema_sandbox_local::LocalSandbox;
+
+/// `rhema-sandboxd`'s arguments: where sessions live, the shared
+/// secret, and the address to serve on.
+#[derive(Debug, Parser)]
+#[command(
+    name = "rhema-sandboxd",
+    version,
+    about = "The sandbox protocol served from a host"
+)]
+pub struct Args {
+    /// The root directory every session lives under, one subdirectory
+    /// per session name.
+    #[arg(long)]
+    pub base: PathBuf,
+
+    /// The shared secret; a wrong or missing one is a 401 on every
+    /// route, `/healthz` included.
+    #[arg(long)]
+    pub token: String,
+
+    /// The address to bind; loopback unless the daemon is meant to be
+    /// reached from elsewhere.
+    #[arg(long, default_value = "127.0.0.1")]
+    pub host: String,
+
+    /// The port to bind; `0` lets the OS pick, and the bound address
+    /// is printed on startup so scripts can wait for it.
+    #[arg(long, default_value_t = 0)]
+    pub port: u16,
+}
+
+/// The daemon state: the base directory, the shared secret, and the
+/// one root per session name.
+pub struct Daemon {
+    /// The directory every session root is created under; the session
+    /// name is validated to a single component before the join, so
+    /// nothing climbs out of here.
+    base: PathBuf,
+    /// The shared secret checked on every route.
+    token: String,
+    /// The sessions seen so far; the first request for a name stands
+    /// its root up. A plain mutex, not a concurrent map: the lock is
+    /// never held across an await.
+    sessions: Mutex<BTreeMap<String, Arc<FsRoot>>>,
+}
+
+impl Daemon {
+    /// A daemon serving `base` with `token`; no session exists yet.
+    pub fn new(base: PathBuf, token: String) -> Self {
+        Self {
+            base,
+            token,
+            sessions: Mutex::new(BTreeMap::new()),
+        }
+    }
+
+    /// The session's root, created under the base on first use. The
+    /// first request for a name pays the `create_dir_all` and the
+    /// canonicalize; every later one reuses the same root, so a
+    /// session's files stay put across requests.
+    fn session_root(&self, session: &str) -> Result<Arc<FsRoot>, SandboxError> {
+        let mut sessions = self.sessions.lock().unwrap_or_else(PoisonError::into_inner);
+        if let Some(root) = sessions.get(session) {
+            return Ok(Arc::clone(root));
+        }
+        let root = Arc::new(FsRoot::new(self.base.join(session))?);
+        sessions.insert(session.to_owned(), Arc::clone(&root));
+        Ok(root)
+    }
+}
+
+/// The session path segment as a single safe component: no emptiness,
+/// no traversal, no separator. The name becomes a directory under the
+/// base, and the segments are remote-influenced — the same guard
+/// [`FsRoot`] keeps inside the workspace, restated one level up, where
+/// the daemon decides where a session's root even sits.
+fn session_name(session: &str) -> Result<(), SandboxError> {
+    let safe = !session.is_empty()
+        && session != "."
+        && session != ".."
+        && !session.contains('/')
+        && !session.contains('\\')
+        && !session.contains('\u{0}');
+    if safe {
+        Ok(())
+    } else {
+        Err(SandboxError::Failed(format!(
+            "the session name must be one safe path component: {session:?}"
+        )))
+    }
+}
+
+/// The daemon's routes: `/healthz` and the four protocol routes, one
+/// session segment ahead of each — the paths are [`rhema_sandbox::protocol`]'s
+/// constants, the token middleware layered over the lot.
+pub fn router(daemon: Arc<Daemon>) -> Router {
+    Router::new()
+        .route("/healthz", get(healthz))
+        .route(&format!("/{{session}}{EXEC}"), post(exec))
+        .route(&format!("/{{session}}{PUT}"), post(put))
+        .route(&format!("/{{session}}{GET}"), post(get_file))
+        .route(&format!("/{{session}}{SEED}"), post(seed))
+        .route_layer(axum::middleware::from_fn_with_state(
+            Arc::clone(&daemon),
+            require_token,
+        ))
+        .with_state(daemon)
+}
+
+/// Binds `args.host:args.port`, prints the bound address (the OS picks
+/// one when `--port` is `0`), and serves until killed.
+pub async fn run(args: Args) -> std::io::Result<()> {
+    let listener = tokio::net::TcpListener::bind((args.host.as_str(), args.port)).await?;
+    let addr = listener.local_addr()?;
+    let daemon = Arc::new(Daemon::new(args.base, args.token));
+    println!("rhema-sandboxd listening on {addr}");
+    axum::serve(listener, router(daemon)).await
+}
+
+/// Byte-length-safe constant-time equality over the two strings' UTF-8
+/// bytes: every compared byte is XOR-folded into one difference word
+/// along with the length difference, so the answer takes the same work
+/// whether the mismatch is in the first byte or the last. Not a MAC —
+/// just enough that the bearer check does not leak the token a timing
+/// probe at a time.
+fn tokens_match(presented: &str, expected: &str) -> bool {
+    let (presented, expected) = (presented.as_bytes(), expected.as_bytes());
+    let mut diff = presented.len() ^ expected.len();
+    for (a, b) in presented.iter().zip(expected) {
+        diff |= usize::from(a ^ b);
+    }
+    diff == 0
+}
+
+/// The shared secret on every route: a wrong or missing token is a 401
+/// before the route runs.
+async fn require_token(
+    State(daemon): State<Arc<Daemon>>,
+    request: Request,
+    next: Next,
+) -> Response {
+    let good = request
+        .headers()
+        .get(TOKEN_HEADER)
+        .and_then(|value| value.to_str().ok())
+        .is_some_and(|candidate| tokens_match(candidate, &daemon.token));
+    if !good {
+        return (
+            StatusCode::UNAUTHORIZED,
+            Json(serde_json::json!({ "error": "forbidden" })),
+        )
+            .into_response();
+    }
+    next.run(request).await
+}
+
+/// `GET /healthz`: the shape clients probe.
+async fn healthz() -> impl IntoResponse {
+    Json(serde_json::json!({ "ok": true }))
+}
+
+/// `POST /{session}/exec`: one command through a fresh
+/// [`LocalSandbox`] over the session's root.
+async fn exec(
+    Path(session): Path<String>,
+    State(daemon): State<Arc<Daemon>>,
+    Json(call): Json<ExecCall>,
+) -> Response {
+    if let Err(err) = session_name(&session) {
+        return error_response(err);
+    }
+    let root = match daemon.session_root(&session) {
+        Ok(root) => root,
+        Err(err) => return error_response(err),
+    };
+    let sandbox = LocalSandbox::new((*root).clone());
+    let request = ExecRequest {
+        command: call.command,
+        cwd: call.cwd.unwrap_or_default(),
+        timeout: Duration::from_secs(call.timeout_secs),
+    };
+    // No streaming here: the reply carries the completed output, and
+    // the caps it arrives under are LocalSandbox's, applied before
+    // this builds the reply.
+    match sandbox
+        .exec(request, &AtomicBool::new(false), &mut |_| {})
+        .await
+    {
+        Ok(outcome) => Json(ExecReply {
+            exit_code: outcome.code,
+            stdout: String::from_utf8_lossy(&outcome.stdout).into_owned(),
+            stderr: String::from_utf8_lossy(&outcome.stderr).into_owned(),
+            timed_out: false,
+        })
+        .into_response(),
+        // A timeout crosses as the reply's `timed_out`, never as an
+        // error status: the client maps that field to its timeout
+        // error, and an error status would surface as unavailability.
+        // 137 is the kill code the protocol's shape test pins.
+        Err(SandboxError::Timeout(_)) => Json(ExecReply {
+            exit_code: Some(137),
+            stdout: String::new(),
+            stderr: String::new(),
+            timed_out: true,
+        })
+        .into_response(),
+        Err(err) => error_response(err),
+    }
+}
+
+/// `POST /{session}/put`: one file straight onto the session's root.
+async fn put(
+    Path(session): Path<String>,
+    State(daemon): State<Arc<Daemon>>,
+    Json(put): Json<PutFile>,
+) -> Response {
+    match write_file(&daemon, &session, &put).await {
+        Ok(ack) => Json(ack).into_response(),
+        Err(err) => error_response(err),
+    }
+}
+
+async fn write_file(daemon: &Daemon, session: &str, put: &PutFile) -> Result<Ack, SandboxError> {
+    session_name(session)?;
+    let mapped = daemon.session_root(session)?.map(&put.path)?;
+    let contents = base64::engine::general_purpose::STANDARD
+        .decode(&put.content_base64)
+        .map_err(|err| SandboxError::Failed(format!("the put body is not base64: {err}")))?;
+    // The shim's put creates the parent directories, and the skills
+    // seeding writes two levels deep — the local route keeps that
+    // contract.
+    if let Some(parent) = mapped.parent() {
+        tokio::fs::create_dir_all(parent)
+            .await
+            .map_err(|err| SandboxError::Failed(format!("{}: {err}", put.path)))?;
+    }
+    tokio::fs::write(&mapped, contents)
+        .await
+        .map_err(|err| SandboxError::Failed(format!("{}: {err}", put.path)))?;
+    Ok(Ack { error: None })
+}
+
+/// The request body of [`GET`]: `{ "path": ... }` — the one shape
+/// `protocol.rs` pins in prose only, so the daemon keeps a private
+/// type for it rather than parsing loose JSON.
+#[derive(Debug, serde::Deserialize)]
+struct GetFile {
+    /// The sandbox-absolute POSIX path to read.
+    path: String,
+}
+
+/// `POST /{session}/get`: one file read straight off the session's
+/// root; a missing file is a 404 carrying the error.
+async fn get_file(
+    Path(session): Path<String>,
+    State(daemon): State<Arc<Daemon>>,
+    Json(body): Json<GetFile>,
+) -> Response {
+    if let Err(err) = session_name(&session) {
+        return error_response(err);
+    }
+    let mapped = match daemon
+        .session_root(&session)
+        .and_then(|r| r.map(&body.path))
+    {
+        Ok(mapped) => mapped,
+        Err(err) => return error_response(err),
+    };
+    match tokio::fs::read(&mapped).await {
+        Ok(contents) => Json(FileReply {
+            content_base64: Some(base64::engine::general_purpose::STANDARD.encode(contents)),
+            error: None,
+        })
+        .into_response(),
+        Err(err) => {
+            let status = if err.kind() == std::io::ErrorKind::NotFound {
+                StatusCode::NOT_FOUND
+            } else {
+                StatusCode::INTERNAL_SERVER_ERROR
+            };
+            (
+                status,
+                Json(FileReply {
+                    content_base64: None,
+                    error: Some(format!("{}: {err}", body.path)),
+                }),
+            )
+                .into_response()
+        }
+    }
+}
+
+/// `POST /{session}/seed`: the [`WorkspaceSeed`] ingress, through
+/// [`LocalSandbox::seed`] — the clone and the unpack run under the
+/// same budget and caps as every other command.
+async fn seed(
+    Path(session): Path<String>,
+    State(daemon): State<Arc<Daemon>>,
+    Json(request): Json<SeedRequest>,
+) -> Response {
+    if let Err(err) = session_name(&session) {
+        return error_response(err);
+    }
+    let root = match daemon.session_root(&session) {
+        Ok(root) => root,
+        Err(err) => return error_response(err),
+    };
+    let seed = match request {
+        SeedRequest::Git { git } => WorkspaceSeed::GitClone(git),
+        // The wire's name is diagnostics only; the unpack is at the
+        // root whatever it says.
+        SeedRequest::Tarball {
+            tarball_base64,
+            tarball_name,
+        } => WorkspaceSeed::Tarball {
+            name: tarball_name.unwrap_or_default(),
+            tar_gz_base64: tarball_base64,
+        },
+    };
+    match LocalSandbox::new((*root).clone()).seed(&seed).await {
+        Ok(()) => Json(Ack { error: None }).into_response(),
+        Err(err) => error_response(err),
+    }
+}
+
+/// One handler failure as the protocol's error convention: the status
+/// carries the class (4xx the caller's input, 5xx the daemon's own
+/// trouble), the body carries the message.
+fn error_response(err: SandboxError) -> Response {
+    let status = match &err {
+        SandboxError::Unavailable(_) => StatusCode::SERVICE_UNAVAILABLE,
+        SandboxError::Timeout(_) => StatusCode::GATEWAY_TIMEOUT,
+        SandboxError::Failed(_) => StatusCode::BAD_REQUEST,
+    };
+    (
+        status,
+        Json(serde_json::json!({ "error": err.to_string() })),
+    )
+        .into_response()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn session_names_must_be_single_safe_components() {
+        for good in ["sess-7", "workspace_one", "a", "Up1"] {
+            assert!(session_name(good).is_ok(), "{good:?} is a safe name");
+        }
+        for bad in ["", ".", "..", "a/b", "../etc", "a\\b", "a\u{0}b", "/"] {
+            let error = session_name(bad).expect_err("the name is refused");
+            assert!(
+                matches!(error, SandboxError::Failed(_)),
+                "the refusal is in band: {error:?}"
+            );
+        }
+    }
+}
+
+#[cfg(all(test, unix))]
+mod loopback_tests {
+    //! The loopback contract (spec §6): the daemon in-process on an
+    //! ephemeral port, driven by the real client — the two sides
+    //! exercised together, not asserted at each other.
+
+    use super::*;
+    use rhema_sandbox::ExecOutcome;
+    use rhema_sandbox_http::HttpSandbox;
+
+    /// The daemon's token in these tests.
+    const TOKEN: &str = "sekrit";
+
+    /// The daemon on an ephemeral loopback port over a fresh base; the
+    /// guard comes back so the base outlives the test.
+    /// True when the GNU `timeout` binary answers: the loopback wraps
+    /// the local executor, whose budgets rely on it. macOS ships none,
+    /// and the tests skip rather than fail where it cannot run.
+    fn timeout_available() -> bool {
+        std::process::Command::new("timeout")
+            .arg("1")
+            .arg("true")
+            .stdout(std::process::Stdio::null())
+            .stderr(std::process::Stdio::null())
+            .status()
+            .map(|status| status.success())
+            .unwrap_or(false)
+    }
+
+    async fn spawn_daemon() -> (String, tempfile::TempDir) {
+        let base = tempfile::tempdir().expect("the base is made");
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("the loopback port binds");
+        let addr = listener.local_addr().expect("the address reads");
+        let daemon = Arc::new(Daemon::new(base.path().to_path_buf(), TOKEN.to_owned()));
+        tokio::spawn(async move {
+            axum::serve(listener, router(daemon))
+                .await
+                .expect("the server serves");
+        });
+        (format!("http://{addr}"), base)
+    }
+
+    /// A client for session `sess-7` holding the right token.
+    fn sandbox(base: &str) -> HttpSandbox {
+        HttpSandbox::new(base.to_owned(), "sess-7", TOKEN)
+    }
+
+    /// One command at the root, budgeted by the caller.
+    fn exec_request(command: &str, timeout: Duration) -> ExecRequest {
+        ExecRequest {
+            command: command.to_owned(),
+            cwd: "/".into(),
+            timeout,
+        }
+    }
+
+    /// One exec through the client with nothing cancelled and nothing
+    /// watching the progress — the shape most tests want.
+    async fn run(sandbox: &HttpSandbox, request: ExecRequest) -> Result<ExecOutcome, SandboxError> {
+        let cancelled = AtomicBool::new(false);
+        sandbox.exec(request, &cancelled, &mut |_| {}).await
+    }
+
+    /// The healthz route probed directly, with or without a token; the
+    /// status and the decoded body come back.
+    async fn probe_healthz(base: &str, token: Option<&str>) -> (u16, serde_json::Value) {
+        let http = reqwest::Client::new();
+        let mut request = http.get(format!("{base}/healthz"));
+        if let Some(token) = token {
+            request = request.header(TOKEN_HEADER, token);
+        }
+        let response = request.send().await.expect("the probe answers");
+        let status = response.status().as_u16();
+        (status, response.json().await.expect("the body parses"))
+    }
+
+    /// Runs one git command in `dir`; the global and system config are
+    /// blanked so the fixture does not depend on the host's.
+    fn git(dir: &std::path::Path, args: &[&str]) {
+        let output = std::process::Command::new("git")
+            .args(args)
+            .current_dir(dir)
+            .env("GIT_CONFIG_GLOBAL", "/dev/null")
+            .env("GIT_CONFIG_SYSTEM", "/dev/null")
+            .output()
+            .expect("git runs");
+        assert!(
+            output.status.success(),
+            "git {args:?} failed: {}",
+            String::from_utf8_lossy(&output.stderr),
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn exec_reports_success_nonzero_and_the_timeout_kill() {
+        if !timeout_available() {
+            println!(
+                "skipping: no GNU `timeout` on PATH; the loopback's local executor budgets with it"
+            );
+            return;
+        }
+        let (base, _dir) = spawn_daemon().await;
+        let sbx = sandbox(&base);
+
+        // Success: the stdout and the exit code cross as sent.
+        let outcome = run(&sbx, exec_request("echo hi", Duration::from_secs(30)))
+            .await
+            .expect("the command runs");
+        assert_eq!(outcome.code, Some(0));
+        assert_eq!(outcome.stdout, b"hi\n");
+
+        // Nonzero: the real code crosses.
+        let outcome = run(&sbx, exec_request("exit 3", Duration::from_secs(30)))
+            .await
+            .expect("the command runs");
+        assert_eq!(outcome.code, Some(3));
+
+        // Timeout: the daemon's `timeout` prefix kills at the budget,
+        // the reply's `timed_out` crosses, and the client maps it to
+        // its own timeout error.
+        let error = run(&sbx, exec_request("sleep 5", Duration::from_secs(1)))
+            .await
+            .expect_err("the budget kills the command");
+        assert_eq!(error, SandboxError::Timeout(1));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn put_then_get_round_trips_and_a_missing_get_fails() {
+        let (base, _dir) = spawn_daemon().await;
+        let sbx = sandbox(&base);
+
+        sbx.write_file("/workspace/deep/note.md", b"# note")
+            .await
+            .expect("the write lands");
+        let read = sbx
+            .read_file("/workspace/deep/note.md")
+            .await
+            .expect("the read lands");
+        assert_eq!(read, b"# note");
+
+        // A missing file is a 404 from the daemon; the client reads
+        // every non-2xx as unavailability carrying the status.
+        let error = sbx
+            .read_file("/workspace/absent")
+            .await
+            .expect_err("the read fails");
+        match &error {
+            SandboxError::Unavailable(message) => {
+                assert!(message.contains("404"), "the miss is a 404: {message}");
+            }
+            other => panic!("a missing read fails in band, not {other:?}"),
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn seeding_lands_from_git_and_from_a_tarball() {
+        if !timeout_available() {
+            println!(
+                "skipping: no GNU `timeout` on PATH; the loopback's local executor budgets with it"
+            );
+            return;
+        }
+        let (base, _dir) = spawn_daemon().await;
+
+        // Git: a one-commit repository behind a file:// URL.
+        let repo = tempfile::tempdir().expect("tempdir");
+        git(repo.path(), &["init", "-q"]);
+        std::fs::write(repo.path().join("seeded.txt"), b"from git")
+            .expect("the fixture file is written");
+        git(repo.path(), &["add", "."]);
+        git(
+            repo.path(),
+            &[
+                "-c",
+                "user.email=test@example.com",
+                "-c",
+                "user.name=test",
+                "commit",
+                "-q",
+                "-m",
+                "seed",
+            ],
+        );
+        let git_session = HttpSandbox::new(base.clone(), "git-seed", TOKEN);
+        git_session
+            .seed(&WorkspaceSeed::GitClone(format!(
+                "file://{}",
+                repo.path().display()
+            )))
+            .await
+            .expect("the clone lands");
+        let outcome = run(&git_session, exec_request("ls", Duration::from_secs(30)))
+            .await
+            .expect("the listing runs");
+        let listing = String::from_utf8(outcome.stdout).expect("stdout is text");
+        assert!(
+            listing.contains("seeded.txt"),
+            "the clone landed: {listing}"
+        );
+
+        // Tarball: one file, packed and base64'd, unpacked at the root.
+        let payload = tempfile::tempdir().expect("tempdir");
+        std::fs::write(payload.path().join("packed.txt"), b"from tar")
+            .expect("the payload file is written");
+        let archive = payload.path().join("packed.tar.gz");
+        let status = std::process::Command::new("tar")
+            .arg("czf")
+            .arg(&archive)
+            .arg("-C")
+            .arg(payload.path())
+            .arg("packed.txt")
+            .status()
+            .expect("tar runs");
+        assert!(status.success(), "the fixture tarball is built");
+        let tar_session = HttpSandbox::new(base.clone(), "tar-seed", TOKEN);
+        tar_session
+            .seed(&WorkspaceSeed::Tarball {
+                name: "packed.tar.gz".into(),
+                tar_gz_base64: base64::engine::general_purpose::STANDARD
+                    .encode(std::fs::read(&archive).expect("the tarball is read")),
+            })
+            .await
+            .expect("the unpack lands");
+        let outcome = run(&tar_session, exec_request("ls", Duration::from_secs(30)))
+            .await
+            .expect("the listing runs");
+        let listing = String::from_utf8(outcome.stdout).expect("stdout is text");
+        assert!(
+            listing.contains("packed.txt"),
+            "the unpack landed: {listing}"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn a_wrong_token_is_refused_on_exec_and_healthz() {
+        let (base, _dir) = spawn_daemon().await;
+
+        // Exec: the real client with the wrong token — a non-2xx reads
+        // as unavailability carrying the status.
+        let intruder = HttpSandbox::new(base.clone(), "sess-7", "wrong");
+        let error = run(&intruder, exec_request("true", Duration::from_secs(30)))
+            .await
+            .expect_err("the refusal surfaces");
+        match &error {
+            SandboxError::Unavailable(message) => {
+                assert!(message.contains("401"), "the status crosses: {message}");
+            }
+            other => panic!("a refused token is unavailable, not {other:?}"),
+        }
+
+        // healthz: wrong and missing are both a 401 with the pinned
+        // body; the right token is answered.
+        let (status, body) = probe_healthz(&base, Some("wrong")).await;
+        assert_eq!(status, 401);
+        assert_eq!(body, serde_json::json!({ "error": "forbidden" }));
+        let (status, body) = probe_healthz(&base, None).await;
+        assert_eq!(status, 401);
+        assert_eq!(body, serde_json::json!({ "error": "forbidden" }));
+        let (status, body) = probe_healthz(&base, Some(TOKEN)).await;
+        assert_eq!(status, 200);
+        assert_eq!(body, serde_json::json!({ "ok": true }));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn the_stream_cap_keeps_the_tail() {
+        if !timeout_available() {
+            println!(
+                "skipping: no GNU `timeout` on PATH; the loopback's local executor budgets with it"
+            );
+            return;
+        }
+        let (base, _dir) = spawn_daemon().await;
+        let sbx = sandbox(&base);
+
+        // 200 KB of stdout crosses capped: at most the shim's 50 KiB,
+        // because LocalSandbox caps before the daemon builds the reply.
+        let outcome = run(
+            &sbx,
+            exec_request("yes | head -c 200000", Duration::from_secs(30)),
+        )
+        .await
+        .expect("the command runs");
+        assert!(
+            outcome.stdout.len() <= 50 * 1024,
+            "the stream is capped: {}",
+            outcome.stdout.len()
+        );
+
+        // The tail is kept, and distinct lines make it visible: the
+        // last line of `seq`'s output, not the first.
+        let outcome = run(&sbx, exec_request("seq 1 200000", Duration::from_secs(30)))
+            .await
+            .expect("the command runs");
+        assert!(outcome.stdout.len() <= 50 * 1024);
+        assert!(
+            outcome.stdout.ends_with(b"200000\n"),
+            "the last bytes survive the cap: {:?}",
+            String::from_utf8_lossy(&outcome.stdout),
+        );
+    }
+}