From 5e1cda046bf082691e0bb8baec2586bc3f84017b Mon Sep 17 00:00:00 2001 From: Nishad Date: Wed, 19 Aug 2026 17:20:08 -0700 Subject: [PATCH] Preserve run context for detached sub-runs --- README.md | 8 +- crates/caos/src/bin/caos.rs | 23 ++-- crates/caos/src/lib.rs | 153 ++++++++---------------- crates/server/src/compute.rs | 57 ++++++++- crates/server/src/main.rs | 16 ++- crates/server/src/runner.rs | 83 +++++++++++-- crates/worker-common/src/lib.rs | 2 +- design/chat.md | 2 +- design/map-then.md | 11 +- design/runner-protocol.md | 14 +++ std/llm-step/src/async_work.rs | 6 +- tests/secrets/cli.sh | 66 ++++++++++ tests/{run-async => sub-run}/DEPS | 0 tests/{run-async => sub-run}/cli.sh | 14 +-- tests/{run-async => sub-run}/delayed.sh | 0 tests/{run-async => sub-run}/launch.sh | 2 +- tests/{run-async => sub-run}/self.sh | 4 +- 17 files changed, 309 insertions(+), 152 deletions(-) rename tests/{run-async => sub-run}/DEPS (100%) rename tests/{run-async => sub-run}/cli.sh (81%) rename tests/{run-async => sub-run}/delayed.sh (100%) rename tests/{run-async => sub-run}/launch.sh (74%) rename tests/{run-async => sub-run}/self.sh (76%) diff --git a/README.md b/README.md index e824d7d9..551bbc86 100644 --- a/README.md +++ b/README.md @@ -153,9 +153,10 @@ It serves requests **concurrently — one thread per request** — so a worker c fetch objects while its own `/run` is in flight, and several top-level runs can proceed at once. Ordinary dependent sub-computations use a **map-then continuation**: the worker records the continuation as its result and finishes -its job before the server resolves it (see [compute](#compute)). `run-async` is -the deliberate exception: it starts a detached `/run` request from a worker and -returns the request hash immediately, without waiting for that subrequest. +its job before the server resolves it (see [compute](#compute)). `sub-run` is +the deliberate exception: it starts detached work from a worker and returns the +request hash immediately. The server recovers the launching job's run stack and +secret store from its job nonce; neither is sent back into the worker. Capacity lives runner-side: the set of hanging `/runner/poll`s *is* the pool. | Request | Behaviour | @@ -163,6 +164,7 @@ Capacity lives runner-side: the set of hanging `/runner/poll`s *is* the pool. | `GET /object/` | Return the serialized object (` \0`, the bytes git hashes). `400` if malformed, `404` if absent. | | `POST /object/` | Store the serialized object in the body, return its git hash. Content-addressed, so idempotent. | | `GET /run?req=&trace=` | Run the ArgTree `` (`req` is the query param's historical name; its value is the ArgTree hash) and return `" "` (the fully-resolved result), optionally emitting trace events. See [compute](#compute). | +| `POST /sub-run` | Start one exact request without waiting, inheriting the in-flight launching job's server-side run context. | | `GET /trace//stream` | Stream one live trace as Chrome `B`/`E` events in JSONL. | | `POST /runner/poll` | A runner's hanging request for work, carrying its required args (name → oid). Answered with a job, `idle` (TTL expired), or `exit` (eviction). See `design/runner-protocol.md`. | | `POST /runner/result` | A runner posting a job's outcome, keyed by (req, nonce) — first post per nonce wins. | diff --git a/crates/caos/src/bin/caos.rs b/crates/caos/src/bin/caos.rs index 0fc61ba2..e0f1a00e 100644 --- a/crates/caos/src/bin/caos.rs +++ b/crates/caos/src/bin/caos.rs @@ -7,13 +7,13 @@ //! post the kind + hash recorded at `/cas/out` back to the server), then //! long-polls for more work for its image until an idle TTL passes (see //! `design/runner-protocol.md`). It normally records continuations that the -//! server resolves after the worker's job finishes; `run-async` is the -//! deliberate exception that starts a detached top-level computation. The +//! server resolves after the worker's job finishes; `sub-run` starts detached +//! work while retaining the current server-side run context. The //! shared command logic lives in the `caos` library; this binary is the worker's //! CLI surface plus the privileged runner. //! //! Subcommands: `get-hash`, `get`, `put`, `put-commit`, `hash`, `forward`, `map-then`, -//! `run-then`, `run-request-then`, `run-async`, `prepare-request`, `curry`, and `runner`. +//! `run-then`, `run-request-then`, `sub-run`, `prepare-request`, `curry`, and `runner`. //! (Image import and ref resolution are user-facing only — see `caos-cli`.) use std::os::unix::fs::PermissionsExt; @@ -104,15 +104,15 @@ fn run(args: &[String]) -> Result<(), String> { [request, kvs @ ..] => caos::caos_run_request_then(&http()?, request, kvs), _ => Err(usage(args)), }, - // Send an ordinary /run request for an already-stored ArgTree without - // waiting for its result. - Some("run-async") => match &args[2..] { - [arg_tree] => caos::caos_run_async(&http()?, arg_tree), + // Start an already-stored ArgTree without waiting, preserving this + // job's server-side run stack and secret store. + Some("sub-run") => match &args[2..] { + [arg_tree] => caos::caos_sub_run(&http()?, arg_tree), _ => Err(usage(args)), }, // `prepare-request --base:= [...]` — construct and store the // exact flat runnable ArgTree without executing it. This is the durable - // identity accepted by run-async. + // identity accepted by sub-run. Some("prepare-request") => caos::caos_prepare_request(&http()?, &args[2..]), // `curry [--unbind= ...] --base:= [--name=value | --name:@=path ...]` — // bind args to the `--base` ArgTree (a bare image, a curry node, or a flat @@ -237,7 +237,10 @@ fn run_runner_job( if image_oid.is_none() { *image_oid = caos::read_hash(&cas.join("args").join("base")).ok(); } - let envs = [(caos::SALT_ENV, salt.as_str())]; + let envs = [ + (caos::SALT_ENV, salt.as_str()), + (caos::JOB_NONCE_ENV, job.nonce.as_str()), + ]; // Drop the granted secrets at `/secret/` just before the worker runs // (design/secrets.md). `write_secrets` wipes any prior job's `/secret` // first, so a warm runner never leaks a secret into a later job that wasn't @@ -480,7 +483,7 @@ fn usage(args: &[String]) -> String { {prog} map-then [--map:=] [--then:=]\n \ {prog} run-then --run:= [--then:=] [--catch]\n \ {prog} run-request-then [--then:=] [--catch]\n \ - {prog} run-async \n \ + {prog} sub-run \n \ {prog} prepare-request --base:= [--name=value | --name:@=path ...]\n \ {prog} curry [--unbind= ...] --base:= [--name=value | --name:@=path ...]\n \ (an image is :@=, :docker= or :hash=)\n \ diff --git a/crates/caos/src/lib.rs b/crates/caos/src/lib.rs index 055d50ef..ab26ef55 100644 --- a/crates/caos/src/lib.rs +++ b/crates/caos/src/lib.rs @@ -4,7 +4,7 @@ //! It talks to the server over HTTP (`/object`) and runs the container //! `runner` (jobs arrive by long-poll; see `design/runner-protocol.md`). It //! normally records continuations for the server to resolve after the job; -//! `run-async` is the one command that directly dispatches `/run`. +//! `sub-run` starts detached work inside the current server-side run context. //! //! Everything that doesn't depend on *how* objects move — the object model, //! currying, args-tree assembly, CAS materialization, image import — lives here, @@ -21,13 +21,12 @@ use std::ffi::OsStr; use std::fs::OpenOptions; use std::io::{IsTerminal, Read, Write}; -use std::net::{Shutdown, TcpStream, ToSocketAddrs}; use std::os::unix::ffi::{OsStrExt, OsStringExt}; use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::OnceLock; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{SystemTime, UNIX_EPOCH}; use gix::objs::WriteTo; @@ -191,6 +190,11 @@ const SECRETS_HEADER: &str = "X-Caos-Secrets"; /// runs without ever touching Redis. pub const SALT_ENV: &str = "CAOS_SALT"; +/// The current runner job's short-lived capability. A worker presents it to +/// `POST /sub-run`; the server accepts it only while that exact job is in +/// flight, and uses it to recover the job's server-held run context. +pub const JOB_NONCE_ENV: &str = "CAOS_JOB_NONCE"; + /// Image-ref scheme marking an ordinary docker reference (vs. a git-image hash). pub const DOCKER_SCHEME: &str = "docker://"; @@ -3055,8 +3059,8 @@ mod git_ref_tests { /// Resolve curry layers, build the args tree, bundle + push the request, and run /// it — the CLI's blocking run. Ordinary worker sub-runs are continuations the -/// server resolves; deliberately independent work may start a new top-level run -/// with `run-async`. Returns the server's +/// server resolves; detached worker work uses `sub-run` to retain the current +/// server-side run context. Returns the server's /// `(kind, result-hash)`. `cas` is `None` here: every path arg is a host path to /// ingest. fn run_request( @@ -3115,7 +3119,7 @@ pub fn prepare_client_request( /// /// Unlike [`caos_curry`], the result is not a partial curry node. It is the same /// request `run_request` would send to `/run`, so it can be recorded durably -/// and later handed unchanged to `run-async` or `run-request-then`. +/// and later handed unchanged to `sub-run` or `run-request-then`. pub fn caos_prepare_request(t: &dyn Transport, kvs: &[String]) -> Result<(), String> { let cas = cas_dir(); let (bty, bval, kvs) = split_base_arg("prepare-request", kvs)?; @@ -3319,45 +3323,41 @@ pub fn caos_run_request_then( ) } -/// Start an already-stored ArgTree through the ordinary `/run` endpoint without -/// waiting for its result. The server handles each HTTP request on its own -/// thread, so closing our side after sending the request does not cancel it. -/// -/// This starts a new top-level run: it does not inherit the caller's run stack, -/// secrets, credentials, model settings, or other ephemeral execution context. -/// Callers must use it only for independent work whose complete context is in -/// the ArgTree. -pub fn caos_run_async(t: &dyn Transport, arg_tree: &str) -> Result<(), String> { +/// Start an already-stored ArgTree in the current job's server-side run context +/// without waiting for its result. The job nonce identifies that context; the +/// worker never receives the carried stack or secret store. +pub fn caos_sub_run(t: &dyn Transport, arg_tree: &str) -> Result<(), String> { if !is_hex_hash(arg_tree) || arg_tree.bytes().any(|byte| byte.is_ascii_uppercase()) { return Err(format!( - "run-async needs a lowercase 40-character ArgTree hash, got {arg_tree:?}" + "sub-run needs a lowercase 40-character ArgTree hash, got {arg_tree:?}" )); } if !t.has_object(arg_tree)? { return Err(format!( - "run-async needs an already-stored ArgTree, and {arg_tree} is absent" + "sub-run needs an already-stored ArgTree, and {arg_tree} is absent" )); } let (kind, content) = t.get_object(arg_tree)?; if kind != "tree" { return Err(format!( - "run-async needs an ArgTree, but {arg_tree} is a {kind}" + "sub-run needs an ArgTree, but {arg_tree} is a {kind}" )); } let tree = gix::objs::TreeRef::from_bytes(&content, gix::hash::Kind::Sha1) - .map_err(|error| format!("run-async ArgTree {arg_tree} is malformed: {error}"))?; + .map_err(|error| format!("sub-run ArgTree {arg_tree} is malformed: {error}"))?; if !tree .entries .iter() .any(|entry| entry.filename.to_vec().as_slice() == b"base") { return Err(format!( - "run-async needs a runnable ArgTree, but {arg_tree} has no 'base' entry" + "sub-run needs a runnable ArgTree, but {arg_tree} has no 'base' entry" )); } t.ensure_pushed(arg_tree)?; - let url = run_url(&t.server_url()?, arg_tree, None); - dispatch_compute_url(&url)?; + let nonce = std::env::var(JOB_NONCE_ENV) + .map_err(|_| "sub-run is available only inside a running worker".to_string())?; + request_sub_run(&t.server_url()?, arg_tree, &nonce)?; println!("request {arg_tree}"); Ok(()) } @@ -4425,83 +4425,30 @@ fn run_url(base: &str, arg_tree: &str, trace_id: Option<&str>) -> String { url } -/// Send a plain-HTTP compute request and deliberately do not read its response. -/// This is the small worker-side primitive behind `caos run-async`: `/run` -/// already outlives a disconnected caller, while the conversation's durable -/// `pending` entry makes a dispatch that dies before reaching the server safe to -/// retry. Waiting for response headers here would wait for the run itself. -fn dispatch_compute_url(url: &str) -> Result<(), String> { - let rest = url - .strip_prefix("http://") - .ok_or_else(|| format!("run-async only supports plain HTTP server URLs: {url}"))?; - let (authority, path) = match rest.split_once('/') { - Some((authority, path)) => (authority, format!("/{path}")), - None => (rest, "/".to_string()), - }; - if authority.is_empty() { - return Err(format!("run-async server URL has no host: {url}")); - } - - let (host, port) = if let Some(bracketed) = authority.strip_prefix('[') { - let close = bracketed - .find(']') - .ok_or_else(|| format!("invalid IPv6 server URL: {url}"))?; - let host = &bracketed[..close]; - let suffix = &bracketed[close + 1..]; - let port = match suffix.strip_prefix(':') { - Some(port) => port - .parse::() - .map_err(|_| format!("invalid port in server URL: {url}"))?, - None if suffix.is_empty() => 80, - None => return Err(format!("invalid server URL: {url}")), - }; - (host, port) - } else { - match authority.rsplit_once(':') { - Some((host, port)) if !host.contains(':') => ( - host, - port.parse::() - .map_err(|_| format!("invalid port in server URL: {url}"))?, - ), - _ => (authority, 80), - } - }; - - const DISPATCH_TIMEOUT: Duration = Duration::from_secs(5); - let addresses = (host, port) - .to_socket_addrs() - .map_err(|error| format!("resolving {authority}: {error}"))?; - let mut last_error = None; - let mut stream = None; - for address in addresses { - match TcpStream::connect_timeout(&address, DISPATCH_TIMEOUT) { - Ok(connected) => { - stream = Some(connected); - break; - } - Err(error) => last_error = Some(error), - } +/// Ask the server to start `arg_tree` with the current in-flight job's +/// un-hashed context. The response acknowledges admission only; the sub-run +/// continues on a server thread after this call returns. +fn request_sub_run(base: &str, arg_tree: &str, nonce: &str) -> Result<(), String> { + let url = format!("{}/sub-run", base.trim_end_matches('/')); + let body = serde_json::json!({"req": arg_tree, "nonce": nonce}).to_string(); + let response = minreq::post(&url) + .with_header(caos_world::WORLD_HEADER, caos_world::WORLD) + .with_header("content-type", "application/json") + .with_timeout(5) + .with_body(body) + .send() + .map_err(|error| format!("POST {url}: {error}"))?; + if !(200..300).contains(&response.status_code) { + let detail = response.as_str().unwrap_or("").trim(); + return Err(if detail.is_empty() { + format!("POST {url}: server returned {}", response.status_code) + } else { + format!( + "POST {url}: server returned {}: {detail}", + response.status_code + ) + }); } - let mut stream = stream.ok_or_else(|| match last_error { - Some(error) => format!("connecting to {authority}: {error}"), - None => format!("resolving {authority}: no socket addresses"), - })?; - stream - .set_write_timeout(Some(DISPATCH_TIMEOUT)) - .map_err(|error| format!("setting write timeout for {authority}: {error}"))?; - write!( - stream, - "GET {path} HTTP/1.1\r\nHost: {authority}\r\n{}: {}\r\nConnection: close\r\n\r\n", - caos_world::WORLD_HEADER, - caos_world::WORLD, - ) - .map_err(|error| format!("sending GET {url}: {error}"))?; - stream - .flush() - .map_err(|error| format!("sending GET {url}: {error}"))?; - stream - .shutdown(Shutdown::Write) - .map_err(|error| format!("finishing GET {url}: {error}"))?; Ok(()) } @@ -4683,29 +4630,29 @@ mod git_transport_tests { } #[test] - fn run_async_rejects_noncanonical_and_nonrunnable_requests_before_dispatch() { + fn sub_run_rejects_noncanonical_and_nonrunnable_requests_before_dispatch() { let request = "a".repeat(40); let missing = ObjectTransport { object: None }; - assert!(caos_run_async(&missing, &request) + assert!(caos_sub_run(&missing, &request) .unwrap_err() .contains("already-stored ArgTree")); let blob = ObjectTransport { object: Some(("blob", b"not a request".to_vec())), }; - assert!(caos_run_async(&blob, &request) + assert!(caos_sub_run(&blob, &request) .unwrap_err() .contains("is a blob")); let curry_or_plain_tree = ObjectTransport { object: Some(("tree", Vec::new())), }; - assert!(caos_run_async(&curry_or_plain_tree, &request) + assert!(caos_sub_run(&curry_or_plain_tree, &request) .unwrap_err() .contains("has no 'base' entry")); let uppercase = request.to_ascii_uppercase(); - assert!(caos_run_async(&missing, &uppercase) + assert!(caos_sub_run(&missing, &uppercase) .unwrap_err() .contains("lowercase 40-character")); } diff --git a/crates/server/src/compute.rs b/crates/server/src/compute.rs index cb02bb14..7880e0a9 100644 --- a/crates/server/src/compute.rs +++ b/crates/server/src/compute.rs @@ -108,10 +108,9 @@ struct WorkRequest<'a> { /// and the rendezvous id: an external run also pins /// `refs/caos/res/` at the result, so a client can fetch it by ref. /// Most worker sub-runs are promise resolutions the server performs itself -/// ([`run_work_request`] recursion); `caos run-async` is the one worker command -/// that sends this same endpoint and disconnects without waiting for the result. -/// That dispatch is deliberately a new top-level run: it has an empty run stack, -/// so stack-based cycle detection does not cross the detached-work boundary. +/// ([`run_work_request`] recursion). Detached worker work enters through +/// `POST /sub-run`, which starts this same pipeline with the launching job's +/// existing stack and secret store. pub(crate) fn run( config: &Config, query: &str, @@ -351,7 +350,15 @@ fn run_dispatch( // cache key — and the container runner drops them at `/secret/`. // Read from `arg_entries` before dispatch takes ownership of it. let granted = crate::secrets::grant(secrets, &arg_entries); - crate::runner::dispatch(arg_tree, arg_entries, &image_ref, seeded, granted).map_err(fail)? + crate::runner::dispatch( + arg_tree, + arg_entries, + &image_ref, + seeded, + granted, + |sub_request| start_sub_run(config, sub_request, &child_stack, trace_id, secrets), + ) + .map_err(fail)? }; if result_hash(&result).is_empty() { @@ -387,6 +394,46 @@ fn run_dispatch( Ok(result) } +/// Admit an exact detached child request while the parent job is still in +/// flight. Ownership crosses the thread boundary here: the server clones the +/// current unhashed context, while the worker sends only the child hash and its +/// job-scoped nonce. Validation happens before acknowledgement so a missing or +/// malformed request is not reported as queued. +fn start_sub_run( + config: &Config, + arg_tree: &str, + stack: &[String], + trace_id: Option<&str>, + secrets: &[crate::secrets::Grant], +) -> Result<(), HttpError> { + let (image, _) = read_arg_tree(config, arg_tree)?; + if image.is_empty() { + return Err(HttpError::new(400, "sub-run request has empty image")); + } + + let config = config.clone(); + let arg_tree = arg_tree.to_string(); + let stack = stack.to_vec(); + let trace_id = trace_id.map(str::to_string); + let secrets = secrets.to_vec(); + std::thread::spawn(move || { + let result = run_work_request( + &config, + &WorkRequest { + arg_tree: &arg_tree, + stack: &stack, + trace_id: trace_id.as_deref(), + secrets: &secrets, + }, + ); + match result { + Ok(result) => eprintln!("sub-run completed: arg_tree={arg_tree} -> {result}"), + Err(error) => eprintln!("sub-run failed: arg_tree={arg_tree}: {}", error.message()), + } + }); + Ok(()) +} + // ---- Single-flight ----------------------------------------------------------- /// A run's outcome in plain data, so it can be sent to every parked waiter. diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index b9068365..f662e1a8 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -12,6 +12,8 @@ //! query param's historical name; its value is the ArgTree hash) and return //! the hash of its result, optionally emitting this invocation to an open //! trace stream. +//! * `POST /sub-run` — admit an exact detached child under an in-flight job's +//! existing server-side run stack and secret store. //! * `GET /trace//stream` — follow one live invocation as chunked NDJSON. //! //! The server runs no workers itself. Dispatch is pull-based (see @@ -74,7 +76,10 @@ const DEFAULT_REGISTRY_PULL_HOST: &str = "localhost:5000"; /// Redis (host:port) used to cache results. Override with `CAOS_REDIS_ADDR`. const DEFAULT_REDIS_ADDR: &str = "caos-redis:6379"; -/// Runtime configuration, read once from the environment at startup. +/// Runtime configuration, read once from the environment at startup. Cloning +/// is cheap and lets admitted sub-runs outlive the request thread that launched +/// them while sharing the same repository and trace hub handles. +#[derive(Clone)] struct Config { registry_push_url: String, registry_pull_host: String, @@ -526,8 +531,8 @@ fn handle(config: Arc, mut request: Request) -> std::io::Result<()> { } /// Match the request to a handler and produce the response body. Serves the -/// storage endpoints (`/object*`), compute (`/run`), and the runner protocol -/// (`/runner/poll`, `/runner/result`). +/// storage endpoints (`/object*`), compute (`/run`, `/sub-run`), and the runner +/// protocol (`/runner/poll`, `/runner/result`). fn route(config: &Arc, request: &mut Request) -> Result, HttpError> { let url = request.url().to_string(); let (path, query) = match url.split_once('?') { @@ -560,6 +565,11 @@ fn route(config: &Arc, request: &mut Request) -> Result, HttpErr request.as_reader().read_to_end(&mut body)?; storage::post_object(config, &body) } + Method::Post if path == "/sub-run" => { + let mut body = String::new(); + request.as_reader().read_to_string(&mut body)?; + runner::sub_run(&body) + } Method::Post if path == "/runner/poll" || path == "/runner/result" => { let authorization = request .headers() diff --git a/crates/server/src/runner.rs b/crates/server/src/runner.rs index e5762698..5f67a46c 100644 --- a/crates/server/src/runner.rs +++ b/crates/server/src/runner.rs @@ -100,6 +100,17 @@ enum Outcome { Failed(String), } +/// Messages delivered to the compute thread that owns an in-flight job. A +/// sub-run request is handled there because that thread owns the unhashed run +/// context; the rendezvous table retains only this channel, never the context. +enum DispatchEvent { + Outcome(Outcome), + SubRun { + arg_tree: String, + reply: mpsc::Sender>, + }, +} + /// A hanging `POST /runner/poll`, parked until matched, kicked, or expired. struct ParkedPoll { /// Monotone arrival id — ties between equally specific polls go to the @@ -147,7 +158,7 @@ struct Job { nonce: String, phase: Phase, enqueued: Instant, - outcome: mpsc::Sender, + events: mpsc::Sender, } /// The rendezvous state: parked polls and dispatched jobs, one lock. @@ -313,8 +324,9 @@ pub(crate) fn dispatch( image_ref: &str, seeded: bool, secrets: Vec<(String, String)>, + mut start_sub_run: impl FnMut(&str) -> Result<(), HttpError>, ) -> Result { - let (outcome_tx, outcome_rx) = mpsc::channel(); + let (event_tx, event_rx) = mpsc::channel(); let id = { let mut st = lock(); let id = st.next_id; @@ -350,7 +362,7 @@ pub(crate) fn dispatch( defer_generic_until, }, enqueued: Instant::now(), - outcome: outcome_tx, + events: event_tx, }, ); offer_job(&mut st, id); @@ -391,9 +403,16 @@ pub(crate) fn dispatch( Some(at) => wait.min(at.saturating_duration_since(Instant::now())), None => wait, }; - match outcome_rx.recv_timeout(wait.max(Duration::from_millis(10))) { - Ok(Outcome::Done(result)) => return Ok(result), - Ok(Outcome::Failed(message)) => return Err(HttpError::new(500, message)), + match event_rx.recv_timeout(wait.max(Duration::from_millis(10))) { + Ok(DispatchEvent::Outcome(Outcome::Done(result))) => return Ok(result), + Ok(DispatchEvent::Outcome(Outcome::Failed(message))) => { + return Err(HttpError::new(500, message)) + } + Ok(DispatchEvent::SubRun { arg_tree, reply }) => { + let outcome = start_sub_run(&arg_tree) + .map_err(|error| (error.status(), error.message().to_string())); + let _ = reply.send(outcome); + } Err(mpsc::RecvTimeoutError::Timeout) => { let mut st = lock(); // A bool, not a borrow of the job: the grace block below needs @@ -487,6 +506,56 @@ pub(crate) fn dispatch( } } +/// `POST /sub-run` — ask the compute thread that owns an in-flight job to +/// start one exact child request. The nonce is the entire authority: it is +/// unpredictable, scoped to one claimed job, and removed with that job. +pub(crate) fn sub_run(body: &str) -> Result, HttpError> { + let value: serde_json::Value = serde_json::from_str(body) + .map_err(|error| HttpError::new(400, format!("invalid sub-run json: {error}")))?; + let arg_tree = value["req"].as_str().unwrap_or_default(); + let nonce = value["nonce"].as_str().unwrap_or_default(); + if !valid_hex(arg_tree, 40) { + return Err(HttpError::new( + 400, + "sub-run needs a lowercase request hash", + )); + } + if !valid_hex(nonce, 32) { + return Err(HttpError::new(400, "sub-run needs a lowercase job nonce")); + } + + let events = { + let st = lock(); + let Some(id) = st.by_nonce.get(nonce) else { + return Err(HttpError::new(410, "unknown or consumed job nonce")); + }; + let job = &st.jobs[id]; + if !matches!(job.phase, Phase::Inflight) { + return Err(HttpError::new(409, "job is not in flight")); + } + job.events.clone() + }; + let (reply_tx, reply_rx) = mpsc::channel(); + events + .send(DispatchEvent::SubRun { + arg_tree: arg_tree.to_string(), + reply: reply_tx, + }) + .map_err(|_| HttpError::new(410, "job finished before sub-run admission"))?; + match reply_rx.recv() { + Ok(Ok(())) => Ok(b"{}".to_vec()), + Ok(Err((status, message))) => Err(HttpError::new(status, message)), + Err(_) => Err(HttpError::new(410, "job finished before sub-run admission")), + } +} + +fn valid_hex(value: &str, len: usize) -> bool { + value.len() == len + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')) +} + /// Remove job `id` (and its nonce mapping), returning it. fn remove_job(st: &mut State, id: u64) -> Job { let job = st.jobs.remove(&id).expect("job present under lock"); @@ -717,7 +786,7 @@ pub(crate) fn result(authorization: Option<&str>, body: &str) -> Result, }; Outcome::Failed(message) }; - let _ = job.outcome.send(outcome); + let _ = job.events.send(DispatchEvent::Outcome(outcome)); Ok(b"{}".to_vec()) } diff --git a/crates/worker-common/src/lib.rs b/crates/worker-common/src/lib.rs index ba072be2..9a633f21 100644 --- a/crates/worker-common/src/lib.rs +++ b/crates/worker-common/src/lib.rs @@ -163,7 +163,7 @@ pub fn caos_curry(base: Arg, args: &[(&str, Arg)]) -> Result { /// Construct the exact flat runnable ArgTree for `base` plus `args`, without /// executing it. The returned hash is suitable for durable recording and later -/// `run-async`/`run-request-then` calls; unlike [`caos_curry`], it is not a +/// `sub-run`/`run-request-then` calls; unlike [`caos_curry`], it is not a /// partial curry node. pub fn prepare_request(base: Arg, args: &[(&str, Arg)]) -> Result { let mut argv = vec!["prepare-request".to_string(), base.token("base")]; diff --git a/design/chat.md b/design/chat.md index 31a38e61..a96651ff 100644 --- a/design/chat.md +++ b/design/chat.md @@ -277,7 +277,7 @@ task ID = hash(Q) `Q` contains both the work and its destination, and is the task identity. When no state exists for `Q`, `llm-step` first appends -`{"async":{"task":"","status":"pending"}}`, then calls `caos run-async Q` +`{"async":{"task":"","status":"pending"}}`, then calls `caos sub-run Q` without waiting. Repeating the same tool request refolds `Q` and does not reset a terminal task to pending. diff --git a/design/map-then.md b/design/map-then.md index e4a953ec..7303e766 100644 --- a/design/map-then.md +++ b/design/map-then.md @@ -189,12 +189,11 @@ It is an internal argument of the server's run pipeline: promise sub-runs carry listing the cycle, exactly as before. An HTTP `/run` is always top-level (empty stack). -`caos run-async` is the deliberate exception to workers normally describing -sub-runs as continuations: it sends a new HTTP `/run` for detached, independent -work. The new run does not inherit the caller's stack, so cycle detection does -not cross that boundary. A request must therefore not use `run-async` as a -recursive edge; independently dispatched work must be acyclic without relying -on ancestors from the dispatching run. +`caos sub-run` is the deliberate exception to workers normally describing +sub-runs as continuations. It asks the server to start detached work under the +launching job's existing run context. The child therefore sees the same +ancestor stack as a `run-then` or `map-then` child, and recursive edges are +detected even though nobody waits for the detached result. ## Parallelism diff --git a/design/runner-protocol.md b/design/runner-protocol.md index b926a617..0327418b 100644 --- a/design/runner-protocol.md +++ b/design/runner-protocol.md @@ -90,6 +90,20 @@ other scheduling metadata: a runner learns the oids for its next `required` set from its own materialization of the job's args (see "container runner" below). +### `POST /sub-run` + +An in-flight worker may start one already-prepared request without waiting: + +```json +{ "req": "", "nonce": "" } +``` + +The nonce selects only the server-side context of the currently claimed job. +The compute thread starts the child with `parent stack + parent request`, the +same trace id, and the same secret store. None of those values enter this body +or the runner payload. The nonce stops authorizing sub-runs as soon as the +parent job reports its result. + ### `POST /runner/result` ```json diff --git a/std/llm-step/src/async_work.rs b/std/llm-step/src/async_work.rs index 3deb4ed5..eff0911b 100644 --- a/std/llm-step/src/async_work.rs +++ b/std/llm-step/src/async_work.rs @@ -335,7 +335,7 @@ fn validate_subrequest(subrequest: &str) -> Result<(), String> { fn dispatch(task: &str) -> Result<(), String> { let output = Command::new("caos") - .args(["run-async", task]) + .args(["sub-run", task]) .output() .map_err(|error| format!("launching async task {task}: {error}"))?; if !output.status.success() { @@ -346,11 +346,11 @@ fn dispatch(task: &str) -> Result<(), String> { )); } let reply = String::from_utf8(output.stdout) - .map_err(|error| format!("run-async reply for {task} is not UTF-8: {error}"))?; + .map_err(|error| format!("sub-run reply for {task} is not UTF-8: {error}"))?; let expected = format!("request {task}"); if reply.trim() != expected { return Err(format!( - "run-async dispatched the wrong task: expected {expected:?}, got {:?}", + "sub-run dispatched the wrong task: expected {expected:?}, got {:?}", reply.trim() )); } diff --git a/tests/secrets/cli.sh b/tests/secrets/cli.sh index 3735827f..8a86b90a 100644 --- a/tests/secrets/cli.sh +++ b/tests/secrets/cli.sh @@ -11,6 +11,9 @@ set -euo pipefail fail() { echo "FAIL: $*" >&2; exit 1; } commit() { git add -A && git -c user.email=test@caos -c user.name=caos commit -qm "$1"; } +object_status() { # + curl -sS -o /dev/null -w '%{http_code}' -I "$CAOS_SERVER_URL/object/$1" +} # --- fixtures (committed) ---------------------------------------------------- # A worker that reads the granted secret and reports a leak-free verdict. @@ -68,6 +71,34 @@ cat > mytool/.caos-expr <<'EOF' curry --base:@=bash --worker1:@=run.sh EOF +# A child-only reader used to prove that sub-run preserves the whole carried +# store, not merely the values granted to its launching worker. +mkdir -p subtool +cp -r DEEP-DEPS/bash subtool/bash +cat > subtool/run.sh <<'EOF' +#!/bin/bash +set -euo pipefail +caos get /cas/args/marker +if [ -r /secret/deploytok ] && [ "$(cat /secret/deploytok)" = "DEPLOY-xyz-789" ]; then + verdict=sub-run-secret-ok +else + verdict=sub-run-secret-missing +fi +printf '%s:%s\n' "$verdict" "$(cat /cas/args/marker)" > /tmp/sub-run-result +caos put /tmp/sub-run-result /cas/out +EOF +cat > subtool/.caos-expr <<'EOF' +curry --base:@=bash --worker1:@=run.sh +EOF + +cat > launch-sub-run.sh <<'EOF' +#!/bin/bash +set -euo pipefail +caos get /cas/args/request +caos sub-run "$(cat /cas/args/request)" > /tmp/sub-run-admission +caos put /tmp/sub-run-admission /cas/out +EOF + # An EMBEDDER: it binds mytool as a `:@=` arg rather than running it. Since a # `:@=` target carrying a `.caos-expr` is evaluated (design/caos-expr.md), the # arg binds mytool's MARKED curry — so the embedder's own arg tree turns over @@ -110,6 +141,7 @@ cat > .caos-secrets/deploytok <<'EOF' value=DEPLOY-xyz-789 entropy=aa11bb22cc33dd44ee55ff6677889900 reader=mytool +reader=subtool EOF echo "== a granted secret is injected; a non-matching one is not ==" >&2 @@ -146,6 +178,40 @@ verdict=$(cat tgot/verdict) || fail "current-tree grant verdict: $verdict (expected 'deploytok-ok')" echo " ok: a reader naming a repo path grants the secret" >&2 +echo "== sub-run preserves the server-held secret store ==" >&2 +# The launcher is an ordinary bash worker and cannot read deploytok. Its child +# is the independently authorized subtool request; the child succeeds only if +# the server carries the original store through the detached edge. +subtool=$("$CAOS_CLI" eval-path subtool) || fail "eval-path subtool failed: $subtool" +marker="detached-$(date +%s%N)-$$-$RANDOM" +expected="sub-run-secret-ok:$marker" +expected_oid=$(printf '%s\n' "$expected" | git hash-object --stdin) +request=$("$CAOS_CLI" prepare-request --base:hash="${subtool##* }" --marker="$marker") \ + || fail "preparing subtool request failed" +launcher=$("$CAOS_CLI" curry --base:@=DEEP-DEPS/bash \ + --worker1:@=launch-sub-run.sh --request="$request") \ + || fail "currying sub-run launcher failed" +admitted=$("$CAOS_CLI" run --base:hash="$launcher") || fail "sub-run launcher failed" +[ "$admitted" = "request $request" ] \ + || fail "sub-run launcher admitted the wrong request: $admitted" + +complete=0 +for _ in $(seq 1 150); do + status=$(object_status "$expected_oid") || fail "server unreachable while waiting for sub-run" + case "$status" in + 200) complete=1; break ;; + 404) ;; + *) fail "server returned HTTP $status while waiting for sub-run" ;; + esac + sleep 0.1 +done +[ "$complete" -eq 1 ] || fail "sub-run never produced its secret-backed result" +"$CAOS_CLI" run sub-run-result --base:hash="$request" >/dev/null \ + || fail "reading completed sub-run failed" +[ "$(cat sub-run-result)" = "$expected" ] \ + || fail "sub-run result was wrong: $(cat sub-run-result)" +echo " ok: a child received its reader-matched secret through server context" >&2 + echo "== eval-path folds secret-hash, so a worker's callers become per-user ==" >&2 # mytool matches the deploytok reader, so eval-path marks its returned arg tree. # With the secret removed it matches nothing, so the result differs — which is diff --git a/tests/run-async/DEPS b/tests/sub-run/DEPS similarity index 100% rename from tests/run-async/DEPS rename to tests/sub-run/DEPS diff --git a/tests/run-async/cli.sh b/tests/sub-run/cli.sh similarity index 81% rename from tests/run-async/cli.sh rename to tests/sub-run/cli.sh index fc2b5d73..282ff08f 100755 --- a/tests/run-async/cli.sh +++ b/tests/sub-run/cli.sh @@ -2,9 +2,9 @@ # Runs cwd'd into a client repo with this test tree at ./test and $CAOS_CLI # set, INSIDE a test stack (tests/lib/run-test.sh). # -# The worker schedules its own in-flight ArgTree. If run-async waited for the -# result, the request would deadlock on itself. The eventual result below proves -# that the ordinary /run request continues after the worker disconnects. +# The worker schedules its own in-flight ArgTree. If sub-run waited for the +# result, the request would deadlock on itself. The server instead admits it +# under the current stack, where the recursive edge fails independently. set -euo pipefail fail() { echo "FAIL: $*" >&2; exit 1; } @@ -12,7 +12,7 @@ object_status() { # curl -sS -o /dev/null -w '%{http_code}' -I "$CAOS_SERVER_URL/object/$1" } -echo "== run-async dispatches an in-flight request without waiting ==" >&2 +echo "== sub-run dispatches an in-flight request without waiting ==" >&2 self=$("$CAOS_CLI" curry --base:@=DEEP-DEPS/bash --worker1:@=test/self.sh) reply=$("$CAOS_CLI" run --base:hash="$self") @@ -38,7 +38,7 @@ dispatched=$("$CAOS_CLI" run --base:hash="$launcher") [ "$dispatched" = "request $q" ] \ || fail "launcher dispatched the wrong request: $dispatched" -# There is no blocking /run Q here: the launcher container is already gone. +# There is no blocking caller for Q here: the launcher container is already gone. # The result content is unique to this run, so only the disconnected request # can make its known object id addressable through the core object API. complete=0 @@ -50,7 +50,7 @@ for _ in $(seq 1 150); do case "$status" in 200) complete=1; break ;; 404) ;; - *) infra "server returned HTTP $status while waiting for dispatched request $q" ;; + *) fail "server returned HTTP $status while waiting for dispatched request $q" ;; esac sleep 0.1 done @@ -61,4 +61,4 @@ done || fail "background result contents were wrong: $(cat actual)" echo " ok: result became addressable with no foreground Q caller" >&2 -echo "run-async: ALL PASS" >&2 +echo "sub-run: ALL PASS" >&2 diff --git a/tests/run-async/delayed.sh b/tests/sub-run/delayed.sh similarity index 100% rename from tests/run-async/delayed.sh rename to tests/sub-run/delayed.sh diff --git a/tests/run-async/launch.sh b/tests/sub-run/launch.sh similarity index 74% rename from tests/run-async/launch.sh rename to tests/sub-run/launch.sh index 4be83a03..c50898cd 100755 --- a/tests/run-async/launch.sh +++ b/tests/sub-run/launch.sh @@ -2,6 +2,6 @@ set -euo pipefail caos get /cas/args/request -reply=$(caos run-async "$(< /cas/args/request)") +reply=$(caos sub-run "$(< /cas/args/request)") printf '%s\n' "$reply" > /tmp/dispatched caos put /tmp/dispatched /cas/out diff --git a/tests/run-async/self.sh b/tests/sub-run/self.sh similarity index 76% rename from tests/run-async/self.sh rename to tests/sub-run/self.sh index c9031a79..2df3b609 100755 --- a/tests/run-async/self.sh +++ b/tests/sub-run/self.sh @@ -4,10 +4,10 @@ set -euo pipefail q=$(caos hash /cas/args) # A blocking implementation would wait on this worker's own result. Bound the # regression so a broken test fails promptly instead of hanging the suite. -reply=$(timeout 5 caos run-async "$q") +reply=$(timeout 5 caos sub-run "$q") [ "$reply" = "request $q" ] || { - echo "run-async returned $reply, expected request $q" >&2 + echo "sub-run returned $reply, expected request $q" >&2 exit 1 }