diff --git a/Cargo.lock b/Cargo.lock index 0a3ab56..e8fbf15 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1191,6 +1191,7 @@ dependencies = [ "quote", "rmp-serde", "rust_decimal", + "rustix", "serde", "serde_json", "sha2", diff --git a/crates/mind-body/Cargo.toml b/crates/mind-body/Cargo.toml index 88f1ccd..543cd96 100644 --- a/crates/mind-body/Cargo.toml +++ b/crates/mind-body/Cargo.toml @@ -25,6 +25,7 @@ publish.workspace = true # MessagePack codec (`chat_client`); `sha2` derives the mouth's connection id from # the chat credential. The kit pins its own CultLib revision, so the graph holds # two until the workspace pin and the kit's meet. +# `rmp-serde` is also the lease wire (`lease_client`): the connector's MessagePack frames. # # `serde_json` reads Claude Code's stream-json, a foreign owner's wire (`self_run`); # `uuid`'s v5 makes a Self run's session id derivable from the run's id. @@ -46,6 +47,10 @@ uuid = { version = "1", default-features = false, features = ["v5"] } [target.'cfg(unix)'.dependencies] libc = "0.2" +# The lease socket: SCM_RIGHTS receive with MSG_CMSG_CLOEXEC and the descriptor +# duplicates, as safe calls (the library forbids unsafe code). Already in the +# graph through tempfile; this adds its `net` feature only. +rustix = { version = "1", default-features = false, features = ["std", "net"] } # `syn`, `quote` and `proc-macro2` read the library's own source for # tests/public_surface.rs, which pins its public surface. All three are already diff --git a/crates/mind-body/src/bin/mind-self-run.rs b/crates/mind-body/src/bin/mind-self-run.rs index 04bc20d..52322cf 100644 --- a/crates/mind-body/src/bin/mind-self-run.rs +++ b/crates/mind-body/src/bin/mind-self-run.rs @@ -3,18 +3,25 @@ //! `mind-self@`. This reads the process's arguments and environment, //! the things the library never reads, and prints the request lines and the //! outcome. It exits 0 when the child exited 0, 1 when it exited otherwise or by -//! a signal, 3 when she has no single open Self run (no child started), and 2 -//! when the mind cannot be reached or the arguments are wrong. Without -//! `--endpoint` the daemon is `HUGINN_ENDPOINT`. +//! a signal, 3 when she has no single open Self run (no child started), 4 when +//! the connector refused the run's credential lease (the run is closed, no child +//! started), and 2 when the mind or the connector cannot be reached or the +//! arguments are wrong. Without `--endpoint` the daemon is `HUGINN_ENDPOINT`. +//! `MIND_SELF_RUN_CAP_S` is the unit's `RuntimeMaxSec` in seconds, the one length of a run. +//! `RUNTIME_DIRECTORY` (the unit's `RuntimeDirectory=`) holds each child's fresh +//! configuration; `MIND_SELF_CONFIG_SEED` optionally names a root-owned directory copied into it. +//! The connector's lease socket is `CLAUDE_LEASE_SOCKET`: the token itself is +//! never in this process's environment, only the descriptor the lease grants. use std::time::Duration; -use anyhow::Result; +use anyhow::{Result, anyhow}; use eureka_state::HuginnClient; use mind_body::cli::instance_and_endpoint; use mind_body::launch::SystemClock; +use mind_body::lease_client::{LEASE_SOCKET_VARIABLE, SocketLender}; use mind_body::queue::HuginnMind; -use mind_body::self_run::{ClaudeSpawner, carry}; +use mind_body::self_run::{ClaudeSpawner, RUN_CAP_VARIABLE, carry}; /// How long one call to the mind may take. const TIMEOUT: Duration = Duration::from_secs(15); @@ -23,12 +30,18 @@ const USAGE: &str = "usage: mind-self-run --instance [--endpoint rudp://< // The binary is the one place that reads the process: its arguments and the // daemon's address. The library reads neither (clippy.toml, `#![forbid]` in lib.rs). -#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and the daemon's address, and hands them to the library")] +#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")] fn run() -> Result { let args: Vec = std::env::args().skip(1).collect(); let (instance, address) = instance_and_endpoint(&args, std::env::var("HUGINN_ENDPOINT").ok(), USAGE)?; + let socket = std::env::var_os(LEASE_SOCKET_VARIABLE).ok_or_else(|| anyhow!("{LEASE_SOCKET_VARIABLE} is not set"))?; + let run_cap_s: u64 = std::env::var(RUN_CAP_VARIABLE) + .ok() + .and_then(|seconds| seconds.parse().ok()) + .filter(|seconds| *seconds > 0) + .ok_or_else(|| anyhow!("{RUN_CAP_VARIABLE} is not set to the unit's RuntimeMaxSec in seconds"))?; let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT)); - let outcome = carry(&mind, &SystemClock, &ClaudeSpawner, &mut std::io::stdout())?; + let outcome = carry(&mind, &SystemClock, &SocketLender::new(socket), &ClaudeSpawner::new(std::env::vars_os())?, run_cap_s, &mut std::io::stdout())?; println!("{outcome}"); Ok(outcome.exit_status()) } diff --git a/crates/mind-body/src/launch.rs b/crates/mind-body/src/launch.rs index 7808def..d8a5ec5 100644 --- a/crates/mind-body/src/launch.rs +++ b/crates/mind-body/src/launch.rs @@ -275,9 +275,18 @@ fn past_grace(admitted_at: &str, now: DateTime) -> bool { } /// The resolution that closes a dead run of hers: Recorded, not Withdrawn, so -/// a unit that keeps dying counts as an empty run and trips the breaker. Its -/// sequence is the run's next, counting a closure that was itself withdrawn. +/// a unit that keeps dying counts as an empty run and trips the breaker. fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime) -> Result { + closure(mind, run, now, ResolutionOutcome::Recorded { reason: "unit not active".into() }, "The run's unit was not active when its turn's slot was next wanted.") +} + +/// The closing resolution of `run` with `outcome`. Its sequence is the run's +/// next, counting a closure that was itself withdrawn. The one builder of a +/// closure for a run that ended without its own flush: a dead unit +/// (`close_dead`, Recorded: it counts as an empty run) and a Self run whose +/// credential lease was refused (`self_run`, Recorded for a refusal, Withdrawn +/// for a misconfigured length: no run happened and none is counted). +pub(crate) fn closure(mind: &dyn MindPort, run: &PipelineRef, now: DateTime, outcome: ResolutionOutcome, rationale: &str) -> Result { let closures = Selection { cites: Some(citing(run, Some("subject"))), ..of_kind(PipelineKind::Resolution) }; let latest = headers(mind, &closures)? .iter() @@ -290,8 +299,8 @@ fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime) -> Res Ok(PipelineDocument::Resolution(PipelineResolution { subject: run.clone(), sequence: latest + 1, - outcome: ResolutionOutcome::Recorded { reason: "unit not active".into() }, - rationale: "The run's unit was not active when its turn's slot was next wanted.".into(), + outcome, + rationale: rationale.into(), resolved_on: date(now), })) } diff --git a/crates/mind-body/src/lease_client.rs b/crates/mind-body/src/lease_client.rs new file mode 100644 index 0000000..c220cd7 --- /dev/null +++ b/crates/mind-body/src/lease_client.rs @@ -0,0 +1,379 @@ +//! The Self run's credential lease, from the asking side (rulings +//! `eureka-body:ruling:self-runs-through-connector` and +//! `eureka-body:ruling:self-run-credential-lease`). +//! +//! The Claude connector is the only holder of the subscription token and the +//! one owner of Self-run admission, reservation and the spend ledger. A Self run +//! asks it over a unix socket for a lease. The connector reserves the run's +//! maximum cost, then answers with one frame and, with it, one descriptor: the +//! read end of a pipe that holds the token, its write end already closed. The +//! run settles with what it used, per model. A run that never +//! settles (killed by the brake, a crash, a dropped connection) is billed the +//! whole reservation by the connector, which is why [`Lease`] settles only by +//! being told to, and a dropped one just closes the connection. +//! +//! The token is never read here. It travels as a descriptor from the socket to +//! the child's descriptor table and is not in an environment, an argument, a +//! file, a log or any value this module can print. +//! +//! Wire: a `u32` big-endian length, then a MessagePack named map `{kind, value}` +//! of at most [`MAX_LEASE_FRAME_BYTES`]. The connector owns the format; the +//! golden frames under `lease_client/lease_v1` are copies of its fixtures. + +use std::path::PathBuf; + +use anyhow::{Result, anyhow, bail}; +use serde::{Deserialize, Serialize}; + +/// The variable the Self unit's environment names the connector's lease socket +/// in. The library reads no variable: the binary reads it and builds a +/// [`SocketLender`]. +pub const LEASE_SOCKET_VARIABLE: &str = "CLAUDE_LEASE_SOCKET"; + +/// The largest frame either side sends or accepts, in bytes. +pub const MAX_LEASE_FRAME_BYTES: usize = 64 * 1024; + +/// What a run asks for: a lease until `expires_at_unix_ms` (the connector bills a +/// lease that outlives it). The connector, not the caller, sets the reservation +/// and binds the lease to the requesting unit's invocation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LeaseRequest { + pub expires_at_unix_ms: u64, +} + +/// Why the connector would not lend, as it names it. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum LeaseRefusalKind { + SpendCeiling, + CostBound, + UnknownCaller, + Expired, + Busy, +} + +/// A refusal: the kind is the typed answer, the detail the connector's words +/// (never the token, and not echoed by the actuator). +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LeaseRefused { + pub kind: LeaseRefusalKind, + pub detail: String, +} + +/// Tokens a run used, as the connector prices them. Cache reads and writes are +/// counted apart from plain input; `output_tokens` already includes reasoning. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct TokenUsage { + pub input_tokens: u64, + pub cache_read_tokens: u64, + pub cache_write_tokens: u64, + pub output_tokens: u64, + pub reasoning_tokens: Option, +} + +/// The tokens one model used in a run. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ModelUsage { + pub model: String, + pub usage: TokenUsage, +} + +/// What a run tells the connector when it ends: its usage, per model. The +/// connector trusts only its own observed plan quota. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Settlement { + pub usage: Vec, +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", content = "value")] +pub(crate) enum LeaseMessage { + #[serde(rename = "claude_lease_request")] + Request(LeaseRequest), + /// Travels with exactly one SCM_RIGHTS descriptor. + #[serde(rename = "claude_lease_grant")] + Grant(LeaseGranted), + #[serde(rename = "claude_lease_refusal")] + Refusal(LeaseRefused), + #[serde(rename = "claude_lease_settle")] + Settle(LeaseSettle), + #[serde(rename = "claude_lease_settled")] + Settled(LeaseSettled), +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct LeaseGranted { + pub(crate) lease_id: String, + pub(crate) reserved_micro_usd: u64, +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct LeaseSettle { + pub(crate) lease_id: String, + pub(crate) usage: Vec, +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct LeaseSettled { + pub(crate) billed_micro_usd: u64, + pub(crate) overrun_micro_usd: u64, +} + +/// A frame: the length, then the named-map MessagePack of `message`. +pub(crate) fn encode_frame(message: &LeaseMessage) -> Result> { + let body = rmp_serde::to_vec_named(message)?; + if body.len() > MAX_LEASE_FRAME_BYTES { + bail!("a lease frame is over the bound"); + } + let mut frame = Vec::with_capacity(4 + body.len()); + frame.extend_from_slice(&u32::try_from(body.len())?.to_be_bytes()); + frame.extend_from_slice(&body); + Ok(frame) +} + +#[cfg(test)] +/// The message in one whole frame, or none when the bytes are not exactly one +/// well-formed message within the bound. +pub(crate) fn decode_frame(frame: &[u8]) -> Option { + let (length, body) = frame.split_first_chunk::<4>()?; + let length = usize::try_from(u32::from_be_bytes(*length)).ok()?; + if length == 0 || length > MAX_LEASE_FRAME_BYTES || body.len() != length { + return None; + } + rmp_serde::from_slice(body).ok() +} + +#[cfg(unix)] +type Held = std::os::fd::OwnedFd; +#[cfg(not(unix))] +type Held = (); +#[cfg(unix)] +type Placed = std::os::fd::OwnedFd; +#[cfg(not(unix))] +type Placed = std::convert::Infallible; + +/// The token as the descriptor the connector sent. It is close-on-exec, so no +/// process the actuator starts inherits it by accident; the only way to hand it +/// to a child is [`TokenFd::inheritable`]. +pub struct TokenFd(Held); + +/// A duplicate of the token descriptor that survives exec, at a number above +/// stdio. It exists to be named to the child and dropped right after the spawn. +pub struct Inheritable(Placed); + +impl std::fmt::Debug for TokenFd { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("TokenFd") + } +} + +#[cfg(unix)] +impl TokenFd { + pub(crate) fn from_owned(fd: std::os::fd::OwnedFd) -> Self { + Self(fd) + } + + /// The duplicate to inherit: lowest free number from 3 up (never a stdio + /// number, even if stdio were closed), close-on-exec cleared. + pub fn inheritable(&self) -> Result { + use rustix::io::{FdFlags, fcntl_dupfd_cloexec, fcntl_setfd}; + let placed = fcntl_dupfd_cloexec(&self.0, 3)?; + fcntl_setfd(&placed, FdFlags::empty())?; + Ok(Inheritable(placed)) + } +} + +#[cfg(not(unix))] +impl TokenFd { + pub fn inheritable(&self) -> Result { + bail!("the credential lease needs a unix host") + } +} + +impl Inheritable { + /// The descriptor number the child is told to read the token from. + #[cfg(unix)] + pub fn number(&self) -> i32 { + std::os::fd::AsRawFd::as_raw_fd(&self.0) + } + + #[cfg(not(unix))] + pub fn number(&self) -> i32 { + match self.0 {} + } +} + +/// A held lease: the token, and the means to settle it exactly once. +pub trait Lease { + fn token(&self) -> &TokenFd; + /// Tells the connector what the run used, and answers what it billed in + /// micro-USD. A lease dropped without this is billed its whole reservation. + fn settle(self: Box, settlement: Settlement) -> Result; +} + +/// The connector's answer to a request. +pub enum Lent { + Granted(Box), + Refused(LeaseRefused), +} + +/// Where leases come from. A transport fault is an error; a refusal is an answer. +pub trait Lender { + fn lend(&self, request: &LeaseRequest) -> Result; +} + +/// The connector's lease socket. +pub struct SocketLender { + path: PathBuf, +} + +impl SocketLender { + pub fn new(path: impl Into) -> Self { + Self { path: path.into() } + } +} + +#[cfg(not(unix))] +impl Lender for SocketLender { + fn lend(&self, _request: &LeaseRequest) -> Result { + let _ = &self.path; + Err(anyhow!("the credential lease needs a unix host")) + } +} + +#[cfg(unix)] +impl Lender for SocketLender { + fn lend(&self, request: &LeaseRequest) -> Result { + socket::lend(&self.path, request) + } +} + +#[cfg(unix)] +mod socket { + use std::io::{IoSliceMut, Read, Write}; + use std::mem::MaybeUninit; + use std::os::fd::OwnedFd; + use std::os::unix::net::UnixStream; + use std::path::Path; + use std::time::Duration; + + use anyhow::{Context, Result, anyhow, bail}; + use rustix::io::Errno; + use rustix::net::{RecvAncillaryBuffer, RecvAncillaryMessage, RecvFlags, ReturnFlags, recvmsg}; + + use super::{LeaseMessage, LeaseRequest, Lease, Lent, Settlement, TokenFd, decode_frame_body, encode_frame}; + use super::{LeaseSettle, MAX_LEASE_FRAME_BYTES}; + + /// How long the connector may take over any one frame. + const IO_TIMEOUT: Duration = Duration::from_secs(30); + + pub(super) fn lend(path: &Path, request: &LeaseRequest) -> Result { + let mut stream = UnixStream::connect(path).context("the lease socket could not be reached")?; + stream.set_read_timeout(Some(IO_TIMEOUT))?; + stream.set_write_timeout(Some(IO_TIMEOUT))?; + stream.write_all(&encode_frame(&LeaseMessage::Request(request.clone()))?)?; + let (message, mut fds) = receive(&stream)?; + match message { + LeaseMessage::Grant(grant) => { + let (Some(fd), None) = (fds.pop(), fds.pop()) else { bail!("a lease grant carries exactly one descriptor") }; + if grant.lease_id.is_empty() { + bail!("a lease grant names no lease"); + } + Ok(Lent::Granted(Box::new(Granted { token: TokenFd::from_owned(fd), lease_id: grant.lease_id, stream }))) + } + LeaseMessage::Refusal(refused) if fds.is_empty() => Ok(Lent::Refused(refused)), + _ => Err(anyhow!("the connector answered a lease request with something else")), + } + } + + /// One frame and any descriptors that arrived with it. The descriptors + /// travel with the frame's first byte, so they come with the length; every + /// receive asks for close-on-exec. + fn receive(stream: &UnixStream) -> Result<(LeaseMessage, Vec)> { + let mut fds = Vec::new(); + let mut length = [0_u8; 4]; + let mut filled = 0; + while filled < length.len() { + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(1))]; + let mut control = RecvAncillaryBuffer::new(&mut space); + let mut vector = [IoSliceMut::new(&mut length[filled..])]; + let got = match recvmsg(stream, &mut vector, &mut control, RecvFlags::CMSG_CLOEXEC) { + Ok(got) => got, + Err(Errno::INTR) => continue, + Err(error) => return Err(error.into()), + }; + for message in control.drain() { + if let RecvAncillaryMessage::ScmRights(received) = message { + fds.extend(received); + } + } + if got.flags.contains(ReturnFlags::CTRUNC) { + bail!("descriptors were dropped from a lease frame"); + } + if got.bytes == 0 { + bail!("the lease socket closed"); + } + filled += got.bytes; + } + let size = usize::try_from(u32::from_be_bytes(length))?; + if size == 0 || size > MAX_LEASE_FRAME_BYTES { + bail!("a lease frame is outside its bound"); + } + let mut body = vec![0_u8; size]; + let mut reader = stream; + reader.read_exact(&mut body)?; + Ok((decode_frame_body(&body)?, fds)) + } + + /// The lease on the wire: its token, and the connection a settle goes down. + /// Dropping it closes the connection, which the connector bills in full. + struct Granted { + token: TokenFd, + lease_id: String, + stream: UnixStream, + } + + impl Lease for Granted { + fn token(&self) -> &TokenFd { + &self.token + } + + fn settle(self: Box, settlement: Settlement) -> Result { + let mut stream = self.stream; + let frame = encode_frame(&LeaseMessage::Settle(LeaseSettle { + lease_id: self.lease_id, + usage: settlement.usage, + }))?; + stream.write_all(&frame)?; + let (message, fds) = receive(&stream)?; + match message { + LeaseMessage::Settled(settled) if fds.is_empty() => Ok(settled.billed_micro_usd), + _ => Err(anyhow!("the connector answered a settle with something else")), + } + } + } +} + +/// One message from a frame body already read within the bound. +fn decode_frame_body(body: &[u8]) -> Result { + rmp_serde::from_slice(body).map_err(|_| anyhow!("a lease frame is not a lease message")) +} + +#[cfg(test)] +mod tests; + +/// A token descriptor holding `text`, as the connector's pipe does: written +/// whole, write end closed. +#[cfg(test)] +pub(crate) fn token_holding(text: &str) -> TokenFd { + #[cfg(unix)] + { + let (read, mut write) = std::io::pipe().unwrap(); + std::io::Write::write_all(&mut write, text.as_bytes()).unwrap(); + TokenFd::from_owned(read.into()) + } + #[cfg(not(unix))] + { + let _ = text; + TokenFd(()) + } +} diff --git a/crates/mind-body/src/lease_client/lease_v1/grant.bin b/crates/mind-body/src/lease_client/lease_v1/grant.bin new file mode 100644 index 0000000..554e3a6 Binary files /dev/null and b/crates/mind-body/src/lease_client/lease_v1/grant.bin differ diff --git a/crates/mind-body/src/lease_client/lease_v1/refusal.bin b/crates/mind-body/src/lease_client/lease_v1/refusal.bin new file mode 100644 index 0000000..22a4a25 Binary files /dev/null and b/crates/mind-body/src/lease_client/lease_v1/refusal.bin differ diff --git a/crates/mind-body/src/lease_client/lease_v1/request.bin b/crates/mind-body/src/lease_client/lease_v1/request.bin new file mode 100644 index 0000000..fddaa37 Binary files /dev/null and b/crates/mind-body/src/lease_client/lease_v1/request.bin differ diff --git a/crates/mind-body/src/lease_client/lease_v1/settle.bin b/crates/mind-body/src/lease_client/lease_v1/settle.bin new file mode 100644 index 0000000..b7498de Binary files /dev/null and b/crates/mind-body/src/lease_client/lease_v1/settle.bin differ diff --git a/crates/mind-body/src/lease_client/lease_v1/settled.bin b/crates/mind-body/src/lease_client/lease_v1/settled.bin new file mode 100644 index 0000000..710d4b8 Binary files /dev/null and b/crates/mind-body/src/lease_client/lease_v1/settled.bin differ diff --git a/crates/mind-body/src/lease_client/tests.rs b/crates/mind-body/src/lease_client/tests.rs new file mode 100644 index 0000000..12ac610 --- /dev/null +++ b/crates/mind-body/src/lease_client/tests.rs @@ -0,0 +1,269 @@ +//! The wire against the connector's golden frames, and the client against a +//! fake connector that speaks them over a real unix socket with real SCM_RIGHTS. + +use super::*; + +const REQUEST: &[u8] = include_bytes!("lease_v1/request.bin"); +const GRANT: &[u8] = include_bytes!("lease_v1/grant.bin"); +const REFUSAL: &[u8] = include_bytes!("lease_v1/refusal.bin"); +const SETTLE: &[u8] = include_bytes!("lease_v1/settle.bin"); +const SETTLED: &[u8] = include_bytes!("lease_v1/settled.bin"); + +const RESET: u64 = 1_793_307_000_000; + +fn request() -> LeaseRequest { + LeaseRequest { expires_at_unix_ms: RESET } +} + +const LEASE_ID: &str = "lease:0123456789abcdef0123456789abcdef"; + +fn settlement() -> Settlement { + Settlement { + usage: vec![ + ModelUsage { model: "claude-opus-5-1".into(), usage: TokenUsage { input_tokens: 1200, cache_read_tokens: 30_000, cache_write_tokens: 500, output_tokens: 800, reasoning_tokens: Some(100) } }, + ModelUsage { model: "claude-haiku-5-1".into(), usage: TokenUsage { input_tokens: 400, cache_read_tokens: 0, cache_write_tokens: 0, output_tokens: 60, reasoning_tokens: None } }, + ], + } +} + +fn settle_message() -> LeaseMessage { + LeaseMessage::Settle(LeaseSettle { lease_id: LEASE_ID.into(), usage: settlement().usage }) +} + +#[test] +fn lease_fixtures_decode() { + let refusal = LeaseRefused { kind: LeaseRefusalKind::SpendCeiling, detail: "the window cannot hold this reservation".into() }; + let cases: Vec<(&str, &[u8], LeaseMessage)> = vec![ + ("request", REQUEST, LeaseMessage::Request(request())), + ("grant", GRANT, LeaseMessage::Grant(LeaseGranted { lease_id: LEASE_ID.into(), reserved_micro_usd: 600_000 })), + ("refusal", REFUSAL, LeaseMessage::Refusal(refusal)), + ("settle", SETTLE, settle_message()), + ("settled", SETTLED, LeaseMessage::Settled(LeaseSettled { billed_micro_usd: 123_456, overrun_micro_usd: 23_456 })), + ]; + for (name, bytes, message) in cases { + assert_eq!(decode_frame(bytes).as_ref(), Some(&message), "{name} decodes"); + assert_eq!(encode_frame(&message).unwrap(), bytes, "{name} re-encodes to the connector's bytes"); + } +} + +#[test] +fn a_frame_is_one_whole_message_within_the_bound() { + let mut trailing = GRANT.to_vec(); + trailing.push(0); + assert!(decode_frame(&trailing).is_none()); + assert!(decode_frame(&GRANT[..GRANT.len() - 1]).is_none()); + assert!(decode_frame(&[0, 0, 0, 0]).is_none()); + assert!(decode_frame(&[0, 0]).is_none()); + let mut oversize = vec![0_u8; 4 + MAX_LEASE_FRAME_BYTES + 1]; + oversize[..4].copy_from_slice(&u32::try_from(MAX_LEASE_FRAME_BYTES + 1).unwrap().to_be_bytes()); + assert!(decode_frame(&oversize).is_none()); + let too_big = LeaseMessage::Refusal(LeaseRefused { kind: LeaseRefusalKind::Busy, detail: "x".repeat(MAX_LEASE_FRAME_BYTES) }); + assert!(encode_frame(&too_big).is_err(), "the client never sends a frame the connector would refuse"); +} + +#[test] +fn the_frame_bound_is_the_connectors_65536_bytes() { + assert_eq!(MAX_LEASE_FRAME_BYTES, 65_536); +} + +/// A refusal whose MessagePack body is exactly `body_len` bytes. +fn refusal_with_body_of(body_len: usize) -> LeaseMessage { + let refusal = |filler: usize| LeaseMessage::Refusal(LeaseRefused { kind: LeaseRefusalKind::Busy, detail: "x".repeat(filler) }); + let overhead = rmp_serde::to_vec_named(&refusal(0)).unwrap().len(); + // The string header grows with the filler, so walk to the length instead of computing it. + let message = (body_len.saturating_sub(overhead + 8)..body_len).map(refusal).find(|message| rmp_serde::to_vec_named(message).unwrap().len() == body_len); + message.expect("a refusal of exactly that length exists") +} + +#[test] +fn a_frame_of_exactly_the_bound_is_sent_and_one_byte_more_is_not() { + let exact = encode_frame(&refusal_with_body_of(MAX_LEASE_FRAME_BYTES)).unwrap(); + assert_eq!(exact.len(), 4 + MAX_LEASE_FRAME_BYTES); + assert_eq!(decode_frame(&exact).map(|_| ()), Some(()), "and it reads back"); + assert!(encode_frame(&refusal_with_body_of(MAX_LEASE_FRAME_BYTES + 1)).is_err()); +} + +#[cfg(unix)] +mod socket_wire { + use std::io::{IoSlice, Read, Write}; + use std::mem::MaybeUninit; + use std::os::fd::{AsFd, BorrowedFd}; + use std::os::unix::net::{UnixListener, UnixStream}; + use std::thread::JoinHandle; + + use rustix::io::{FdFlags, fcntl_getfd}; + use rustix::net::{SendAncillaryBuffer, SendAncillaryMessage, SendFlags, sendmsg}; + use tempfile::TempDir; + + use super::*; + + /// A fake connector: it accepts one connection and runs `script` on it. + fn connector(script: impl FnOnce(UnixStream) -> T + Send + 'static) -> (TempDir, SocketLender, JoinHandle) { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("lease.sock"); + let listener = UnixListener::bind(&path).unwrap(); + let served = std::thread::spawn(move || script(listener.accept().unwrap().0)); + (dir, SocketLender::new(path), served) + } + + fn read_frame(stream: &mut UnixStream) -> Vec { + let mut length = [0_u8; 4]; + stream.read_exact(&mut length).unwrap(); + let mut frame = length.to_vec(); + frame.resize(4 + usize::try_from(u32::from_be_bytes(length)).unwrap(), 0); + stream.read_exact(&mut frame[4..]).unwrap(); + frame + } + + fn send_with(stream: &UnixStream, frame: &[u8], fds: &[BorrowedFd<'_>]) { + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(2))]; + let mut control = SendAncillaryBuffer::new(&mut space); + if !fds.is_empty() { + assert!(control.push(SendAncillaryMessage::ScmRights(fds))); + } + sendmsg(stream, &[IoSlice::new(frame)], &mut control, SendFlags::empty()).unwrap(); + } + + /// The read end of a pipe holding `text`, write end closed, as the connector lends it. + fn pipe_holding(text: &str) -> std::os::fd::OwnedFd { + let (read, mut write) = std::io::pipe().unwrap(); + write.write_all(text.as_bytes()).unwrap(); + read.into() + } + + #[test] + fn a_lease_round_trips_with_the_token_on_a_close_on_exec_descriptor_and_the_settle_in_the_connectors_bytes() { + let (_dir, lender, served) = connector(|mut stream| { + assert_eq!(read_frame(&mut stream), REQUEST, "the request is the connector's golden bytes"); + send_with(&stream, GRANT, &[pipe_holding("tok-canary-1").as_fd()]); + assert_eq!(read_frame(&mut stream), SETTLE, "the settle is the connector's golden bytes"); + stream.write_all(SETTLED).unwrap(); + }); + let Lent::Granted(lease) = lender.lend(&request()).unwrap() else { panic!("refused") }; + // Close-on-exec: nothing the actuator starts inherits the lease's own descriptor. + assert!(fcntl_getfd(&lease.token().0).unwrap().contains(FdFlags::CLOEXEC)); + // The duplicate for the child survives exec, sits above stdio and holds the token once. + let inheritable = lease.token().inheritable().unwrap(); + assert!(inheritable.number() >= 3 && !fcntl_getfd(&inheritable.0).unwrap().contains(FdFlags::CLOEXEC)); + let mut text = String::new(); + std::fs::File::from(rustix::io::dup(&inheritable.0).unwrap()).read_to_string(&mut text).unwrap(); + assert_eq!(text, "tok-canary-1"); + assert_eq!(lease.settle(settlement()).unwrap(), 123_456); + served.join().unwrap(); + } + + #[test] + fn a_refusal_is_an_answer_with_its_kind_and_no_descriptor() { + let (_dir, lender, served) = connector(|mut stream| { + read_frame(&mut stream); + stream.write_all(REFUSAL).unwrap(); + }); + let Lent::Refused(refused) = lender.lend(&request()).unwrap() else { panic!("granted") }; + assert_eq!(refused.kind, LeaseRefusalKind::SpendCeiling); + served.join().unwrap(); + } + + /// What a connector sends for a frame length: the length, then `body`. + fn framed(length: u32, body: &[u8]) -> Vec { + let mut frame = length.to_be_bytes().to_vec(); + frame.extend_from_slice(body); + frame + } + + #[test] + fn the_receive_bound_refuses_zero_and_one_past_and_takes_exactly_the_bound() { + let outside = |frame: Vec| { + let (_dir, lender, served) = connector(move |mut stream| { + read_frame(&mut stream); + // Enough body that only the length can be what is wrong with the frame. + // The client may refuse on the length and hang up before the body is written: that is no failure of the connector. + let _ = stream.write_all(&frame); + }); + let error = lender.lend(&request()).err().map(|error| format!("{error:#}")); + served.join().unwrap(); + error + }; + let past = u32::try_from(MAX_LEASE_FRAME_BYTES + 1).unwrap(); + assert!(outside(framed(0, &[0x80; 8])).unwrap().contains("outside its bound")); + assert!(outside(framed(past, &vec![0x80; MAX_LEASE_FRAME_BYTES + 1])).unwrap().contains("outside its bound")); + + let exact = encode_frame(&refusal_with_body_of(MAX_LEASE_FRAME_BYTES)).unwrap(); + let (_dir, lender, served) = connector(move |mut stream| { + read_frame(&mut stream); + stream.write_all(&exact).unwrap(); + }); + let Lent::Refused(refused) = lender.lend(&request()).unwrap() else { panic!("granted") }; + assert_eq!(refused.kind, LeaseRefusalKind::Busy); + served.join().unwrap(); + } + + #[test] + fn a_grant_without_exactly_one_descriptor_is_an_error() { + for fds in [0, 2] { + let (_dir, lender, served) = connector(move |mut stream| { + read_frame(&mut stream); + let held: Vec<_> = (0..fds).map(|_| pipe_holding("tok")).collect(); + let borrowed: Vec<_> = held.iter().map(AsFd::as_fd).collect(); + send_with(&stream, GRANT, &borrowed); + }); + assert!(lender.lend(&request()).is_err(), "{fds} descriptors"); + served.join().unwrap(); + } + } + + #[test] + fn a_lease_dropped_without_settling_sends_no_settle_and_closes_the_connection() { + // A run killed by the brake never settles: the connector sees the connection end with no settle, + // which it bills as the whole reservation. + let (_dir, lender, served) = connector(|mut stream| { + read_frame(&mut stream); + send_with(&stream, GRANT, &[pipe_holding("tok-canary-2").as_fd()]); + let mut rest = Vec::new(); + stream.read_to_end(&mut rest).unwrap(); + rest + }); + let Lent::Granted(lease) = lender.lend(&request()).unwrap() else { panic!("refused") }; + drop(lease); + assert!(served.join().unwrap().is_empty(), "nothing was sent after the request"); + } + + #[test] + fn a_connector_that_is_not_there_or_answers_wrongly_is_an_error_not_a_refusal() { + let dir = TempDir::new().unwrap(); + assert!(SocketLender::new(dir.path().join("none.sock")).lend(&request()).is_err()); + let (_dir, lender, served) = connector(|mut stream| { + read_frame(&mut stream); + stream.write_all(SETTLED).unwrap(); + }); + assert!(lender.lend(&request()).is_err()); + served.join().unwrap(); + let (_dir, lender, served) = connector(|mut stream| { + read_frame(&mut stream); + // A frame length past the bound. + stream.write_all(&u32::MAX.to_be_bytes()).unwrap(); + }); + assert!(lender.lend(&request()).is_err()); + served.join().unwrap(); + } + + #[test] + fn a_refusal_or_settled_answer_that_carries_a_descriptor_is_an_error() { + // A refusal lends nothing: a descriptor with one is a connector fault, and is closed, not used. + let (_dir, lender, served) = connector(|mut stream| { + read_frame(&mut stream); + send_with(&stream, REFUSAL, &[pipe_holding("tok-canary-3").as_fd()]); + }); + assert!(lender.lend(&request()).is_err()); + served.join().unwrap(); + + let (_dir, lender, served) = connector(|mut stream| { + read_frame(&mut stream); + send_with(&stream, GRANT, &[pipe_holding("tok-canary-4").as_fd()]); + read_frame(&mut stream); + send_with(&stream, SETTLED, &[pipe_holding("tok-canary-5").as_fd()]); + }); + let Lent::Granted(lease) = lender.lend(&request()).unwrap() else { panic!("refused") }; + assert!(lease.settle(settlement()).is_err()); + served.join().unwrap(); + } +} diff --git a/crates/mind-body/src/lib.rs b/crates/mind-body/src/lib.rs index a415af9..0a31c56 100644 --- a/crates/mind-body/src/lib.rs +++ b/crates/mind-body/src/lib.rs @@ -16,7 +16,8 @@ //! the chat wire and port, `chat_rudp` its RUDP transport to Bifrost's mouth, and //! `connector` the model transport over ModelConnector's kit client). //! -//! `self_run` carries one Self run inside `mind-self@`: the +//! `lease_client` asks the Claude connector for a Self run's credential and settles +//! what the run used. `self_run` carries one Self run inside `mind-self@`: the //! `mind-self-run` binary runs `claude` over stream-json for her one open Self //! run, asks for the flush at the context ceiling and stops a child that outlives it. @@ -34,6 +35,7 @@ pub mod cli; pub mod connector; pub mod control; pub mod launch; +pub mod lease_client; pub mod persona; pub mod queue; pub mod self_run; diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 220c262..4afe8fc 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -17,27 +17,43 @@ //! ruling `eureka-body:ruling:self-run-decider`). `carry` is the loop that turns //! the operating system's facts into events and executes the actions. //! -//! The actuator admits nothing, reads neither the brake nor the environment and -//! stops no unit: the gate and the brake hold own those. It never echoes an -//! argument, an endpoint or a stream line. Stream-json is Claude Code's wire, a +//! The child's credential is a lease, not a variable (rulings +//! `eureka-body:ruling:self-runs-through-connector` and +//! `eureka-body:ruling:self-run-credential-lease`): before it starts `claude`, +//! `carry` asks the connector for a lease (`lease_client`) until the unit's +//! runtime cap, given to it in seconds (the connector sets the reservation and binds the lease to this +//! unit's invocation), and the granted descriptor is the only place the +//! token reaches the child. A refusal closes the run Recorded and starts nothing. +//! When the child has ended `carry` settles with the usage, per model, the +//! stream reported; a run that ends without settling (killed by the brake, a +//! crash) is billed its reservation by the connector. +//! +//! The actuator admits only the closure of a run whose lease was refused. It +//! reads neither the brake nor the environment and stops no unit: the gate and +//! the brake hold own those. It never echoes an argument, an endpoint, a +//! connector's refusal text or a stream line. Stream-json is Claude Code's wire, a //! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing //! parsed is stored. +use std::collections::{BTreeMap, BTreeSet}; +use std::ffi::{OsStr, OsString}; use std::fmt; use std::io::{BufRead, BufReader, Write}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::process::{ChildStdin, Command, Stdio}; use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; use std::time::Duration as StdDuration; use anyhow::{Context, Result, anyhow, bail}; +use chrono::{DateTime, Utc}; use cultnet_rs::Selection; -use eureka_pipeline::{PipelineDocument, PipelineRef, RunTurn, Slug}; -use huginn_mind::PipelinePageItems; +use eureka_pipeline::{PipelineDocument, PipelineRef, ResolutionOutcome, RunTurn, Slug}; +use huginn_mind::{PipelineAdmissionOutcome, PipelinePageItems}; use serde_json::{Value, json}; use uuid::Uuid; -use crate::launch::{Clock, holders}; +use crate::launch::{Clock, closure, holders}; +use crate::lease_client::{Lender, LeaseRefusalKind, LeaseRequest, Lent, ModelUsage, Settlement, TokenFd, TokenUsage}; use crate::queue::MindPort; use decide::{Action, Event, State, decide, next_deadline}; @@ -56,6 +72,22 @@ pub const ESCALATE_AFTER_S: i64 = 120; pub const EXIT_DRAIN_S: i64 = 2; /// A backstop `claude` enforces; the ceiling is reached first. pub const MAX_TURNS: u32 = 500; +/// Names the variable the Self unit sets to its own `RuntimeMaxSec`, in seconds: +/// the one length of a run. The unit owns the number; the lease asked for and +/// the connector's `--lease-max-ms` follow it. +pub const RUN_CAP_VARIABLE: &str = "MIND_SELF_RUN_CAP_S"; +/// Names the directory `claude` keeps its configuration, settings, hooks and skills in. +const CONFIG_DIR_VARIABLE: &str = "CLAUDE_CONFIG_DIR"; +/// The unit's per-invocation scratch directory (systemd's `RuntimeDirectory=`, +/// removed when the unit stops): where each child's configuration is made. +pub const SCRATCH_VARIABLE: &str = "RUNTIME_DIRECTORY"; +/// Optional: a directory root owns, copied into each child's fresh configuration +/// (its skills and settings). The child may change its copy; no run sees another's. +pub const SEED_VARIABLE: &str = "MIND_SELF_CONFIG_SEED"; +/// Names the descriptor the child reads its credential from. +const TOKEN_FD_VARIABLE: &str = "CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR"; +/// Tells `claude` to scrub credentials from the environment of what it starts. +const ENV_SCRUB_VARIABLE: &str = "CLAUDE_CODE_SUBPROCESS_ENV_SCRUB"; /// The name the start message tells her to admit as. pub const AGENT: &str = "mind-self-run"; /// The namespace of the session id: the run id is the name in it. @@ -82,7 +114,10 @@ pub fn session_id(run: &PipelineRef) -> Uuid { } /// The argument list of the child, and nothing else: no resume, continue or -/// fork, no model, no permission mode and no settings belong to this actuator. +/// fork, no model, no permission mode and no settings of its own belong to this +/// actuator. Only the user's settings load (the fresh configuration directory +/// the seed fills); the working directory's project and local settings, which a +/// run could have written, do not. /// `budget_usd` is the run's, as stored. pub fn claude_args(session: Uuid, budget_usd: &str) -> Vec { [ @@ -100,6 +135,8 @@ pub fn claude_args(session: Uuid, budget_usd: &str) -> Vec { &MAX_TURNS.to_string(), "--permission-prompts", "none", + "--setting-sources", + "user", ] .map(String::from) .to_vec() @@ -178,7 +215,8 @@ impl Stop { } pub trait Spawner { - fn spawn(&self, args: &[String]) -> Result>; + /// Starts the child with `token` as its credential descriptor. + fn spawn(&self, args: &[String], token: &TokenFd) -> Result>; } /// Why a run ended. @@ -199,15 +237,18 @@ pub enum Cause { pub enum SelfRunOutcome { /// She has not exactly one open Self run: nothing was started. NoRun, + /// The connector would not lend: the run is closed Recorded and nothing was started. + LeaseRefused { kind: LeaseRefusalKind }, Ended { cause: Cause, code: Option, requests: u64, prefix: u64 }, } impl SelfRunOutcome { - /// 0 when the child exited 0, 3 when no child was started, else 1 (a - /// child killed by a signal has no code). + /// 0 when the child exited 0, 3 when no child was started, 4 when the + /// connector refused the lease, else 1 (a child killed by a signal has no code). pub fn exit_status(&self) -> i32 { match self { Self::NoRun => 3, + Self::LeaseRefused { .. } => 4, Self::Ended { code: Some(0), .. } => 0, Self::Ended { .. } => 1, } @@ -218,6 +259,7 @@ impl fmt::Display for SelfRunOutcome { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::NoRun => write!(f, "no run: she has no single open Self run"), + Self::LeaseRefused { kind } => write!(f, "no run: the connector refused the lease ({kind:?})"), Self::Ended { cause, code, requests, prefix } => { let cause = match cause { Cause::Own => "own", @@ -249,26 +291,177 @@ fn her_run(mind: &dyn MindPort) -> Result> { } } -/// Carries her one open Self run: one `claude` child from the start message to -/// its end. The request lines go to `out`. A mind that cannot be reached or a -/// child that cannot be started is an error; every other end is an outcome. -/// This is only the loop: facts in, one clock read per fact, actions out. -pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, spawner: &dyn Spawner, out: &mut dyn Write) -> Result { +/// What a run used, from its stream. `claude`'s own account comes first: a +/// result record's `modelUsage` is its cumulative per-model total as of that +/// result, so the last result that has one is the run's usage up to then, plus +/// the tokens of every assistant message after it (the flush turn, or a turn the +/// child was cut off in: those requests are billed too). A stream with no such +/// result falls back to the tokens of every assistant message (once per message +/// id, subagents included), by model. +#[derive(Default)] +struct Spent { + requests: BTreeMap, + reported: Option>, + /// The message ids heard when `reported` was stated: its total already holds them. + counted: BTreeSet, +} + +/// The model of a message that names none; the connector prices it as it sees fit. +const UNNAMED_MODEL: &str = "unknown"; + +impl Spent { + fn hear(&mut self, line: &str) { + let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return }; + match event.get("type").and_then(Value::as_str) { + Some("assistant") => {} + Some("result") => return self.hear_result(&event), + _ => return, + } + let message = event.get("message"); + let Some(id) = message.and_then(|message| message.get("id")).and_then(Value::as_str) else { return }; + let model = message.and_then(|message| message.get("model")).and_then(Value::as_str).unwrap_or(UNNAMED_MODEL); + let usage = message.and_then(|message| message.get("usage")); + let field = |name: &str| usage.and_then(|usage| usage.get(name)).and_then(Value::as_u64).unwrap_or(0); + let (named, seen) = self.requests.entry(id.to_string()).or_insert_with(|| (model.to_string(), TokenUsage::default())); + *named = model.to_string(); + // Records that share an id are one request, each with the usage so far. + seen.input_tokens = seen.input_tokens.max(field("input_tokens")); + seen.cache_read_tokens = seen.cache_read_tokens.max(field("cache_read_input_tokens")); + seen.cache_write_tokens = seen.cache_write_tokens.max(field("cache_creation_input_tokens")); + seen.output_tokens = seen.output_tokens.max(field("output_tokens")); + } + + /// A result's own per-model total, when it states one. + fn hear_result(&mut self, result: &serde_json::Map) { + let Some(Value::Object(models)) = result.get("modelUsage") else { return }; + let counted = |used: &Value, name: &str| used.get(name).and_then(Value::as_u64).unwrap_or(0); + let stated: BTreeMap = models + .iter() + .map(|(model, used)| { + let usage = TokenUsage { + input_tokens: counted(used, "inputTokens"), + cache_read_tokens: counted(used, "cacheReadInputTokens"), + cache_write_tokens: counted(used, "cacheCreationInputTokens"), + output_tokens: counted(used, "outputTokens"), + reasoning_tokens: None, + }; + (model.clone(), usage) + }) + .collect(); + if !stated.is_empty() { + self.reported = Some(stated); + self.counted = self.requests.keys().cloned().collect(); + } + } + + /// The settle: the usage per model, models in name order. + fn settlement(&self) -> Settlement { + let mut by_model: BTreeMap = self.reported.clone().unwrap_or_default(); + for (id, (model, request)) in &self.requests { + if self.reported.is_some() && self.counted.contains(id) { + continue; + } + let total = by_model.entry(model.clone()).or_default(); + total.input_tokens = total.input_tokens.saturating_add(request.input_tokens); + total.cache_read_tokens = total.cache_read_tokens.saturating_add(request.cache_read_tokens); + total.cache_write_tokens = total.cache_write_tokens.saturating_add(request.cache_write_tokens); + total.output_tokens = total.output_tokens.saturating_add(request.output_tokens); + } + Settlement { usage: by_model.into_iter().map(|(model, usage)| ModelUsage { model, usage }).collect() } + } +} + +fn unix_ms(at: DateTime) -> u64 { + u64::try_from(at.timestamp_millis()).unwrap_or(0) +} + +/// Closes a run whose lease was refused: Recorded with the refusal's kind, so a +/// run that cannot be paid for counts as an empty run and trips the breaker. +fn close_refused(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, kind: LeaseRefusalKind) -> Result { + let reason = format!("the connector refused the credential lease: {kind:?}"); + let document = closure(mind, run, clock.now(), ResolutionOutcome::Recorded { reason: reason.as_str().into() }, "The Self run's credential lease was refused, so no claude was started.")?; + let label = run.id.0.rsplit(':').next().unwrap_or(&run.id.0); + match mind.admit(AGENT, label, vec![document])? { + PipelineAdmissionOutcome::Committed { .. } => Ok(SelfRunOutcome::LeaseRefused { kind }), + _ => Err(anyhow!("the run whose lease was refused could not be closed")), + } +} + +/// Closes a run whose lease was refused for its length: Withdrawn, because no +/// claude ever ran and a misconfiguration is not an empty run; the breaker +/// counts nothing and a launch that finds the unit dead has no open run to close. +fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef) -> Result<()> { + let outcome = ResolutionOutcome::Withdrawn { reason: "the connector refused the lease as too long".into() }; + let document = closure(mind, run, clock.now(), outcome, "The unit's runtime cap exceeds the connector's lease length, so no claude was started.")?; + let label = run.id.0.rsplit(':').next().unwrap_or(&run.id.0); + match mind.admit(AGENT, label, vec![document])? { + PipelineAdmissionOutcome::Committed { .. } => Ok(()), + _ => Err(anyhow!("the run whose lease was too long could not be withdrawn")), + } +} + +/// Carries her one open Self run: one `claude` child from the lease to its +/// settle. The request lines go to `out`. A mind that cannot be reached, a +/// connector that cannot be reached or a child that cannot be started is an +/// error; every other end is an outcome. The lease comes first: with none +/// granted no child exists. It is settled once, after the child is known to +/// have ended; a lease dropped any other way is billed in full by the connector. +/// `run_cap_s` is the unit's `RuntimeMaxSec`: a connector that finds the lease +/// too long is misconfigured, which is an error naming the mismatch and not a +/// refusal of the run: the run is withdrawn, so the breaker counts nothing. +pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write) -> Result { let Some((run, budget)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; - let mut child = spawner.spawn(&claude_args(session_id(&run), &budget))?; + let request = LeaseRequest { expires_at_unix_ms: unix_ms(clock.now()).saturating_add(run_cap_s.saturating_mul(1000)) }; + let lease = match lender.lend(&request)? { + Lent::Granted(lease) => lease, + Lent::Refused(refused) if refused.kind == LeaseRefusalKind::Expired => { + withdraw_misconfigured(mind, clock, &run)?; + bail!("the connector refused a {run_cap_s} s lease as too long: raise its --lease-max-ms to at least the unit's RuntimeMaxSec ({RUN_CAP_VARIABLE}) plus slack"); + } + Lent::Refused(refused) => return close_refused(mind, clock, &run, refused.kind), + }; + let child = match spawner.spawn(&claude_args(session_id(&run), &budget), lease.token()) { + Ok(child) => child, + Err(error) => { + // No child ever held the token; the empty report is the connector's cue to bill the reservation. + let _ = lease.settle(Settlement::default()); + return Err(error); + } + }; + let mut spent = Spent::default(); + let outcome = drive(mind, clock, child, &run, &mut spent, out)?; + // A child that could not even be killed may still be spending: that lease is + // dropped, and the connector bills the reservation. + if !matches!(outcome, SelfRunOutcome::Ended { cause: Cause::Unstoppable, .. }) { + // A settle that fails leaves the connector to bill the reservation; the run's outcome stands. + // Either way the journal says what the connector billed, or that it will bill the reservation. + let report = match lease.settle(spent.settlement()) { + Ok(billed) => format!("settled: billed {billed} micro-USD"), + Err(_) => "settle failed: the connector bills the reservation".to_string(), + }; + let _ = writeln!(out, "{report}"); + } + Ok(outcome) +} + +/// The loop: facts in, one clock read per fact, actions out. +fn drive(mind: &dyn MindPort, clock: &dyn Clock, mut child: Box, run: &PipelineRef, spent: &mut Spent, out: &mut dyn Write) -> Result { let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string(); - child.write_line(&user_line(&start_message(mind.instance(), &run))); + child.write_line(&user_line(&start_message(mind.instance(), run))); let mut state = State::default(); let mut now = clock.now(); loop { let wait = next_deadline(&state).map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL)); let fact = match child.next(wait) { - Read::Line(line) => match read_usage(&line) { - Usage::Request { id, prefix } => Event::Request { id, prefix }, - Usage::Result => Event::Result, - Usage::Other | Usage::Unreadable => Event::Tick, - }, + Read::Line(line) => { + spent.hear(&line); + match read_usage(&line) { + Usage::Request { id, prefix } => Event::Request { id, prefix }, + Usage::Result => Event::Result, + Usage::Other | Usage::Unreadable => Event::Tick, + } + } Read::Timeout => Event::Tick, Read::Eof => Event::StdoutEof, Read::Exited(code) => Event::ChildExited(code), @@ -293,23 +486,121 @@ pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, spawner: &dyn Spawner, out: } } -/// The real spawner: the `claude` on the unit's `PATH`, with the unit's -/// environment (the library names no variable), its stderr in the journal. -pub struct ClaudeSpawner; +/// The real spawner: the `claude` on the unit's `PATH`, its stderr in the +/// journal. Its credential is the lease's descriptor, named in the one variable +/// that says where to read it; the token is in no environment, argument or file. +/// It holds the unit's environment, which the library never reads: the binary +/// hands it over, and [`child_environment`] decides what of it the child sees. +pub struct ClaudeSpawner { + unit_environment: Vec<(OsString, OsString)>, + scratch: PathBuf, + seed: Option, +} -impl Spawner for ClaudeSpawner { - fn spawn(&self, args: &[String]) -> Result> { - spawn_program(Path::new("claude"), args) +impl ClaudeSpawner { + /// From the unit's environment: it must name [`SCRATCH_VARIABLE`], and may name [`SEED_VARIABLE`]. + pub fn new(unit_environment: impl IntoIterator) -> Result { + let unit_environment: Vec<(OsString, OsString)> = unit_environment.into_iter().collect(); + let named = |variable: &str| unit_environment.iter().find(|(name, _)| name == variable).map(|(_, value)| PathBuf::from(value)); + // systemd lists several runtime directories colon-separated; the unit names one. + let scratch = named(SCRATCH_VARIABLE).filter(|path| !path.as_os_str().is_empty()).ok_or_else(|| anyhow!("{SCRATCH_VARIABLE} is not set: the unit needs RuntimeDirectory="))?; + let seed = named(SEED_VARIABLE).filter(|path| !path.as_os_str().is_empty()); + Ok(Self { unit_environment, scratch, seed }) } } -fn spawn_program(program: &Path, args: &[String]) -> Result> { +impl Spawner for ClaudeSpawner { + fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { + let config = FreshConfig::create(&self.scratch, self.seed.as_deref())?; + spawn_program(Path::new("claude"), args, token, &self.unit_environment, config) + } +} + +/// The directory name of a child's configuration inside the scratch directory. +const CONFIG_DIR_NAME: &str = "claude-config"; + +/// One child's configuration directory: made empty (or as the seed) for it, +/// owner-only, and removed when the child's handle goes. Whatever the run writes +/// there (settings that set an environment variable, an `apiKeyHelper`, a base +/// URL, a hook) lasts only as long as that run: the next one starts from the seed. +struct FreshConfig(PathBuf); + +impl FreshConfig { + fn create(scratch: &Path, seed: Option<&Path>) -> Result { + let dir = scratch.join(CONFIG_DIR_NAME); + match std::fs::remove_dir_all(&dir) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(_) => bail!("a stale configuration directory could not be removed"), + } + let mut made = std::fs::DirBuilder::new(); + #[cfg(unix)] + std::os::unix::fs::DirBuilderExt::mode(&mut made, 0o700); + made.create(&dir).context("the configuration directory could not be made")?; + let fresh = Self(dir); + if let Some(seed) = seed { + copy_tree(seed, &fresh.0).context("the configuration seed could not be copied")?; + } + Ok(fresh) + } +} + +impl Drop for FreshConfig { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// Copies directories and regular files; a link is skipped, never followed. +fn copy_tree(from: &Path, to: &Path) -> std::io::Result<()> { + for entry in std::fs::read_dir(from)? { + let entry = entry?; + let kind = entry.file_type()?; + let target = to.join(entry.file_name()); + if kind.is_dir() { + std::fs::create_dir(&target)?; + copy_tree(&entry.path(), &target)?; + } else if kind.is_file() { + std::fs::copy(entry.path(), &target)?; + } + } + Ok(()) +} + +/// Whether the child may see a unit variable. `claude` prefers an ambient +/// credential (`ANTHROPIC_AUTH_TOKEN`, `ANTHROPIC_API_KEY`, +/// `CLAUDE_CODE_OAUTH_TOKEN`) to the lease's descriptor and takes its endpoint +/// from `ANTHROPIC_BASE_URL`, so the child gets only what a shell needs to find +/// its way; nothing credential-shaped is on the list. Its `HOME` and +/// `CLAUDE_CONFIG_DIR` are its fresh configuration directory, set by the spawn. +fn child_may_see(name: &OsStr) -> bool { + let name = name.to_string_lossy(); + matches!(name.as_ref(), "PATH" | "LANG" | "TERM") || name.starts_with("LC_") +} + +/// The unit's environment as the child sees it: the allow-list, and nothing else. +fn child_environment(unit: &[(OsString, OsString)]) -> Vec<(&OsString, &OsString)> { + unit.iter().filter(|(name, _)| child_may_see(name)).map(|(name, value)| (name, value)).collect() +} + +/// Starts `program` with a duplicate of the token descriptor open in it, and +/// closes that duplicate here the moment the child exists. The lease's own +/// descriptor is close-on-exec, so this duplicate is the child's only way to the +/// token. `claude` reads it to the end once at startup, so what its own children +/// inherit is an emptied pipe whose write end exists nowhere. The child's +/// environment is cleared and rebuilt from `unit`'s allow-listed variables. +fn spawn_program(program: &Path, args: &[String], token: &TokenFd, unit: &[(OsString, OsString)], config: FreshConfig) -> Result> { + let inheritable = token.inheritable().context("the credential descriptor could not be placed")?; let mut command = Command::new(program); command.args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit()); + command.env_clear().envs(child_environment(unit)); + command.env("HOME", &config.0).env(CONFIG_DIR_VARIABLE, &config.0); + command.env(TOKEN_FD_VARIABLE, inheritable.number().to_string()).env(ENV_SCRUB_VARIABLE, "1"); // Its own process group, so that a signal reaches what the child started too. #[cfg(unix)] std::os::unix::process::CommandExt::process_group(&mut command, 0); let mut process = command.spawn().context("claude could not be started")?; + drop(inheritable); let group = process.id(); let stdin = process.stdin.take(); let stdout = process.stdout.take().ok_or_else(|| anyhow!("claude has no stdout"))?; @@ -328,7 +619,7 @@ fn spawn_program(program: &Path, args: &[String]) -> Result> { std::thread::spawn(move || { let _ = sender.send(Read::Exited(process.wait().ok().and_then(|status| status.code()))); }); - Ok(Box::new(ClaudeProcess { stdin, facts, group })) + Ok(Box::new(ClaudeProcess { stdin, facts, group, _config: config })) } struct ClaudeProcess { @@ -336,6 +627,8 @@ struct ClaudeProcess { facts: Receiver, /// The child's pid, which is its process group's id. group: u32, + /// Its configuration, removed when the child's handle goes. + _config: FreshConfig, } impl Child for ClaudeProcess { diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 5399d34..42069df 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -13,8 +13,12 @@ use eureka_pipeline::{PipelineDocument, PipelineKind as K, PipelineRef, RunOpera use huginn_mind::{PipelineAdmissionOutcome, PipelineSelectionPage}; use super::*; +use crate::lease_client::{Lease, LeaseRefused, token_holding}; use crate::testkit::*; +/// A cap no constant of the library equals, so that a request of it came from the argument. +const TEST_CAP_S: u64 = 5_400; + const RECORDS: &str = include_str!("claude-2.1.268-assistant-usage-records.jsonl"); const PROBE: &str = include_str!("claude-2.1.268-stream-json-probe.jsonl"); @@ -69,6 +73,14 @@ enum Event { /// A signal was attempted, delivered or not. Stopped(Stop), Asked(StdDuration), + /// The connector was asked for a lease and answered one. + Lent, + /// A child was started. + Spawned, + /// The lease was settled with this. + Settled(Settlement), + /// The lease was dropped unsettled: the connector bills the reservation. + Abandoned, } #[derive(Clone)] @@ -171,12 +183,84 @@ struct Fake { } impl Spawner for Fake { - fn spawn(&self, args: &[String]) -> Result> { + fn spawn(&self, args: &[String], _token: &TokenFd) -> Result> { self.spawns.borrow_mut().push(args.to_vec()); if self.refuse { return Err(anyhow!("refused")); } - Ok(Box::new(self.child.borrow_mut().take().expect("spawned twice"))) + let child = self.child.borrow_mut().take().expect("spawned twice"); + child.log.borrow_mut().push(Event::Spawned); + Ok(Box::new(child)) + } +} + +/// What the connector answers. +#[derive(Clone, Copy)] +enum Answer { + Grant, + /// Lends, then fails the settle. + GrantSettleFails, + Refuse(LeaseRefusalKind), + Unreachable, +} + +/// The connector, as the actuator asks it: it logs into the child's log, so the +/// order of lend, spawn, the child's end and settle is one sequence. +struct FakeLender { + log: Rc>>, + asked: RefCell>, + answer: Answer, +} + +impl FakeLender { + fn new(answer: Answer) -> Self { + Self::logging(Rc::new(RefCell::new(Vec::new())), answer) + } + + fn logging(log: Rc>>, answer: Answer) -> Self { + Self { log, asked: RefCell::new(Vec::new()), answer } + } +} + +impl Lender for FakeLender { + fn lend(&self, request: &LeaseRequest) -> Result { + self.asked.borrow_mut().push(request.clone()); + match self.answer { + Answer::Grant | Answer::GrantSettleFails => { + self.log.borrow_mut().push(Event::Lent); + Ok(Lent::Granted(Box::new(FakeLease { log: self.log.clone(), token: token_holding("tok-canary"), settled: false, fails: matches!(self.answer, Answer::GrantSettleFails) }))) + } + Answer::Refuse(kind) => Ok(Lent::Refused(LeaseRefused { kind, detail: "detail-canary".into() })), + Answer::Unreachable => Err(anyhow!("down")), + } + } +} + +struct FakeLease { + log: Rc>>, + token: TokenFd, + settled: bool, + /// The settle reaches the connector's side and fails there. + fails: bool, +} + +impl Lease for FakeLease { + fn token(&self) -> &TokenFd { + &self.token + } + + fn settle(mut self: Box, settlement: Settlement) -> Result { + self.settled = true; + self.log.borrow_mut().push(Event::Settled(settlement)); + if self.fails { Err(anyhow!("detail-canary")) } else { Ok(7) } + } +} + +impl Drop for FakeLease { + fn drop(&mut self) { + if !self.settled { + self.log.borrow_mut().push(Event::Abandoned); + } } } @@ -221,8 +305,11 @@ struct Ran { outcome: Result, out: String, events: Vec, + /// Every event, the lease's included. + all: Vec, spawns: Vec>, admits: usize, + asked: Vec, } impl Ran { @@ -248,6 +335,10 @@ impl Ran { .collect() } + fn count_all(&self, wanted: &Event) -> usize { + self.all.iter().filter(|event| *event == wanted).count() + } + fn count(&self, wanted: &Event) -> usize { self.events.iter().filter(|event| *event == wanted).count() } @@ -280,11 +371,14 @@ fn carry_scripted(seen: &Seen, steps: Vec, code: Option, dies_on: Opt spawns: RefCell::new(Vec::new()), refuse: false, }; + let lender = FakeLender::logging(log.clone(), Answer::Grant); let mut out = Vec::new(); - let outcome = carry(seen, &*clock, &fake, &mut out); - let events = log.borrow().clone(); + let outcome = carry(seen, &*clock, &lender, &fake, TEST_CAP_S, &mut out); + let all = log.borrow().clone(); + // The child's own events; the lease's are in `all`, in order with them. + let events = all.iter().filter(|event| !matches!(event, Event::Lent | Event::Spawned | Event::Settled(_) | Event::Abandoned)).cloned().collect(); let spawns = fake.spawns.borrow().clone(); - Ran { outcome, out: String::from_utf8(out).unwrap(), events, spawns, admits: seen.admits.get() } + Ran { outcome, out: String::from_utf8(out).unwrap(), events, all, spawns, admits: seen.admits.get(), asked: lender.asked.borrow().clone() } } fn at_start() -> chrono::DateTime { @@ -352,7 +446,7 @@ fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_ou let mind = with_run(); let clock = Rc::new(Fixed(Cell::new(at_start()))); let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true }; - assert!(carry(&Seen::of(&mind), &*clock, &refusing, &mut Vec::new()).is_err()); + assert!(carry(&Seen::of(&mind), &*clock, &FakeLender::new(Answer::Grant), &refusing, TEST_CAP_S, &mut Vec::new()).is_err()); assert_eq!(refusing.spawns.borrow().len(), 1, "it tried exactly once"); struct Down; @@ -369,7 +463,7 @@ fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_ou } } let never = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: false }; - assert!(carry(&Down, &*clock, &never, &mut Vec::new()).is_err()); + assert!(carry(&Down, &*clock, &FakeLender::new(Answer::Grant), &never, TEST_CAP_S, &mut Vec::new()).is_err()); assert!(never.spawns.borrow().is_empty()); } @@ -396,6 +490,8 @@ fn claude_args_is_exactly_the_fixed_list_for_a_session_and_a_budget() { "500", "--permission-prompts", "none", + "--setting-sources", + "user", ] ); // Those belong to the run measure or to the unit's managed settings. @@ -491,7 +587,7 @@ fn records_that_share_a_message_id_count_once_and_subagent_records_never_move_th let ran = go(steps); assert_eq!( ran.out, - "request 1 prefix 84210\nrequest 2 prefix 100751\nrequest 3 prefix 117174\nrequest 4 prefix 118336\nrequest 5 prefix 126746\nrequest 6 prefix 127250\nrequest 7 prefix 129853\n" + "request 1 prefix 84210\nrequest 2 prefix 100751\nrequest 3 prefix 117174\nrequest 4 prefix 118336\nrequest 5 prefix 126746\nrequest 6 prefix 127250\nrequest 7 prefix 129853\nsettled: billed 7 micro-USD\n" ); let (cause, _, requests, prefix) = ran.ended(); assert_eq!((cause, requests, prefix), (Cause::Own, 7, 129_853), "the newest prefix, 129853, not a subagent's 999999"); @@ -619,7 +715,7 @@ fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_the_running_tu // The headroom is measured from the prefix the flush was asked at (400k) and counts the requests after the // first result only: 459_999 does not interrupt; 460_000 does, once, and m6 is never read. assert_eq!(jump.count(&Event::Stopped(Stop::Interrupt)), 1, "{kinds:?}"); - assert_eq!(jump.out, "request 1 prefix 290000\nrequest 2 prefix 400000\nrequest 3 prefix 470000\nrequest 4 prefix 459999\nrequest 5 prefix 460000\n"); + assert_eq!(jump.out, "request 1 prefix 290000\nrequest 2 prefix 400000\nrequest 3 prefix 470000\nrequest 4 prefix 459999\nrequest 5 prefix 460000\nsettled: billed 7 micro-USD\n"); assert_eq!(jump.ended().0, Cause::Signalled(Stop::Interrupt)); assert_eq!(flushes(&jump), 1); } @@ -651,7 +747,7 @@ fn the_message_ids_of_a_long_run_are_all_remembered() { let ran = go(steps); let (_, _, requests, prefix) = ran.ended(); assert_eq!((requests, prefix), (12, 1_000)); - assert_eq!(ran.out.lines().count(), 12); + assert_eq!(ran.out.lines().filter(|line| line.starts_with("request ")).count(), 12); } #[test] @@ -755,11 +851,12 @@ fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_p let steps: VecDeque = vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into(); let child = Scripted { steps, log: log.clone(), clock: Rc::new(Fixed(Cell::new(at_start()))), code: Some(0), dies_on: None, refuses: vec![], hangs: false, exited: false, eof: false, asked: 0 }; let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false }; - let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &fake, &mut Vec::new()).unwrap(); + let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &FakeLender::new(Answer::Grant), &fake, TEST_CAP_S, &mut Vec::new()).unwrap(); let events = log.borrow().clone(); let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32; - // One read before the first wait, then one per fact heard: never a second read inside a pass. - assert_eq!(clock.1.get(), 1 + reads, "{events:?}"); + // One read for the lease's expiry, one before the first wait, then one per fact heard: + // never a second read inside a pass. + assert_eq!(clock.1.get(), 2 + reads, "{events:?}"); // The flush was written on the first request at time 0; the very next pass sees the stepped clock. // The flush pass itself has no interrupt, whatever the step: the interrupt is on a later event. let flush_at = events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains("context ceiling"))).unwrap(); @@ -827,7 +924,7 @@ fn unreadable_and_foreign_lines_are_skipped_and_never_printed() { Out(at("m1", 10)), Out(result_line()), ]); - assert_eq!(ran.out, "request 1 prefix 10\n"); + assert_eq!(ran.out, "request 1 prefix 10\nsettled: billed 7 micro-USD\n"); let outcome = ran.outcome.as_ref().unwrap().to_string(); assert!(!ran.out.contains(canary) && !outcome.contains(canary)); assert!(!ran.written().iter().any(|text| text.contains(canary)), "nothing from the stream is written back"); @@ -848,6 +945,8 @@ fn the_actuator_admits_nothing_to_the_mind_through_a_whole_flushed_run() { #[cfg(unix)] mod process { + mod lease_child; + use std::os::unix::fs::PermissionsExt; use std::path::PathBuf; @@ -865,8 +964,25 @@ mod process { /// Spawns it, retrying the one error a freshly written executable can give /// while another test's fork still holds its write handle. fn start(program: &Path, args: &[String]) -> Box { + start_with(program, args, &token_holding("tok-canary")) + } + + /// What the unit would hold: what a shell needs, and credential-shaped variables + /// that live only here, never in the test process, so only the spawner can pass them on. + fn unit_environment() -> Vec<(std::ffi::OsString, std::ffi::OsString)> { + let mut unit: Vec<(std::ffi::OsString, std::ffi::OsString)> = vec![("PATH".into(), "/usr/local/bin:/usr/bin:/bin".into())]; + for name in CREDENTIAL_CANARIES { + unit.push((name.into(), format!("canary-{name}").into())); + } + unit + } + + const CREDENTIAL_CANARIES: [&str; 6] = ["ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_BASE_URL", "CLAUDE_CODE_API_KEY_HELPER_TTL_MS", "UNLISTED_VARIABLE"]; + + fn start_with(program: &Path, args: &[String], token: &TokenFd) -> Box { for _ in 0..50 { - match spawn_program(program, args) { + let config = FreshConfig::create(program.parent().unwrap(), None).unwrap(); + match spawn_program(program, args, token, &unit_environment(), config) { Ok(child) => return child, Err(error) if format!("{error:#}").contains("busy") => std::thread::sleep(StdDuration::from_millis(100)), Err(error) => panic!("{error:#}"), @@ -934,7 +1050,9 @@ mod process { #[test] fn a_program_that_cannot_start_is_an_error_that_names_no_argument() { let canary = "canary-argument-6b0d"; - let error = spawn_program(Path::new("/nonexistent/claude"), &[canary.to_string()]).err().unwrap(); + let scratch = tempfile::tempdir().unwrap(); + let config = FreshConfig::create(scratch.path(), None).unwrap(); + let error = spawn_program(Path::new("/nonexistent/claude"), &[canary.to_string()], &token_holding("tok-canary"), &[], config).err().unwrap(); assert!(!format!("{error:#}").contains(canary) && !format!("{error:#}").contains("nonexistent"), "{error:#}"); } @@ -961,8 +1079,8 @@ mod process { struct Spawn(PathBuf); impl Spawner for Spawn { - fn spawn(&self, args: &[String]) -> Result> { - Ok(Box::new(Prompt(start(&self.0, args)))) + fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { + Ok(Box::new(Prompt(start_with(&self.0, args, token)))) } } @@ -987,8 +1105,8 @@ mod process { struct Patient(PathBuf); impl Spawner for Patient { - fn spawn(&self, args: &[String]) -> Result> { - Ok(Box::new(Bounded(start(&self.0, args), std::time::Instant::now()))) + fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { + Ok(Box::new(Bounded(start_with(&self.0, args, token), std::time::Instant::now()))) } } @@ -1002,9 +1120,9 @@ mod process { let program = script(&dir, &body); let mind = with_run(); let mut out = Vec::new(); - let outcome = carry(&Seen::of(&mind), &SystemClock, &Spawn(program), &mut out).unwrap(); + let outcome = carry(&Seen::of(&mind), &SystemClock, &FakeLender::new(Answer::Grant), &Spawn(program), TEST_CAP_S, &mut out).unwrap(); assert_eq!(outcome, SelfRunOutcome::Ended { cause: Cause::FlushRequested, code: Some(0), requests: 2, prefix: 300_000 }); - assert_eq!(String::from_utf8(out).unwrap(), "request 1 prefix 10\nrequest 2 prefix 300000\n"); + assert_eq!(String::from_utf8(out).unwrap(), "request 1 prefix 10\nrequest 2 prefix 300000\nsettled: billed 7 micro-USD\n"); } #[test] @@ -1016,10 +1134,10 @@ mod process { let program = script(&dir, &format!("read -t 5 start\ncat <<'EOF'\n{result}\nEOF\nwhile read more; do :; done\nexit 1")); let started = std::time::Instant::now(); let mut out = Vec::new(); - let outcome = carry(&Seen::of(&with_run()), &SystemClock, &Spawn(program), &mut out).unwrap(); + let outcome = carry(&Seen::of(&with_run()), &SystemClock, &FakeLender::new(Answer::Grant), &Spawn(program), TEST_CAP_S, &mut out).unwrap(); assert_eq!(outcome, SelfRunOutcome::Ended { cause: Cause::Own, code: Some(1), requests: 0, prefix: 0 }); assert_eq!(outcome.exit_status(), 1); - assert!(out.is_empty()); + assert!(!String::from_utf8_lossy(&out).contains("request "), "no request was heard"); assert!(started.elapsed() < StdDuration::from_secs(20), "it ended promptly"); } @@ -1036,7 +1154,7 @@ mod process { let program = script(&dir, &format!("{before}cat <<'EOF'\n{result}\nEOF\nsleep 30 &\necho $! > {}\nexit 0", pids.display())); let started = std::time::Instant::now(); let mut out = Vec::new(); - let outcome = carry(&Seen::of(&with_run()), &SystemClock, &Patient(program), &mut out).unwrap(); + let outcome = carry(&Seen::of(&with_run()), &SystemClock, &FakeLender::new(Answer::Grant), &Patient(program), TEST_CAP_S, &mut out).unwrap(); let holder = std::fs::read_to_string(&pids).unwrap().trim().to_string(); let cause = if flushed { Cause::FlushRequested } else { Cause::Own }; let (requests, prefix) = if flushed { (1, CEILING_TOKENS) } else { (0, 0) }; @@ -1085,3 +1203,5 @@ mod process { } } } + +mod lease; diff --git a/crates/mind-body/src/self_run/tests/lease.rs b/crates/mind-body/src/self_run/tests/lease.rs new file mode 100644 index 0000000..112ab7f --- /dev/null +++ b/crates/mind-body/src/self_run/tests/lease.rs @@ -0,0 +1,261 @@ +//! The credential lease as `carry` uses it: asked for before claude exists, +//! refusals as typed outcomes, settled once with what the stream reported. +//! The wire itself is `lease_client`'s tests; the real child is `process::lease_child`. + +use eureka_pipeline::ResolutionOutcome; + +use super::*; + +/// A real assistant record with all four counters of its usage set. +fn billed(id: &str, input: u64, read: u64, write: u64, output: u64) -> String { + let mut event: Value = serde_json::from_str(&request(id, input, read, write)).unwrap(); + event["message"]["usage"]["output_tokens"] = json!(output); + event.to_string() +} + +fn subagent_billed(id: &str, input: u64, read: u64, write: u64, output: u64) -> String { + let mut event: Value = serde_json::from_str(&billed(id, input, read, write, output)).unwrap(); + event["parent_tool_use_id"] = json!("toolu_0001"); + event.to_string() +} + +fn never_spawns() -> Fake { + Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: false } +} + +#[test] +fn a_lease_is_asked_for_before_claude_starts_for_the_runs_budget_and_the_units_runtime_cap() { + let ran = go(vec![Out(result_line())]); + assert_eq!(ran.asked, vec![LeaseRequest { expires_at_unix_ms: u64::try_from(at_start().timestamp_millis()).unwrap() + TEST_CAP_S * 1000 }], "the connector sets the reservation: the request is only the expiry, the cap the unit gave"); + assert_eq!(ran.all.first(), Some(&Event::Lent), "nothing happens before the lease"); + assert_eq!(ran.all.get(1), Some(&Event::Spawned), "the child is the next thing"); +} + +#[test] +fn no_lease_is_asked_for_when_there_is_no_run_to_carry() { + let ran = carry_with(&Seen::of(&TestMind::seeded()), vec![], Some(0), false); + assert_eq!(ran.outcome.unwrap(), SelfRunOutcome::NoRun); + assert!(ran.asked.is_empty() && ran.all.is_empty()); +} + +#[test] +fn refused_lease_spawns_nothing() { + for kind in [LeaseRefusalKind::SpendCeiling, LeaseRefusalKind::CostBound, LeaseRefusalKind::UnknownCaller, LeaseRefusalKind::Busy] { + let mind = with_run(); + let lender = FakeLender::new(Answer::Refuse(kind)); + let never = never_spawns(); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let mut out = Vec::new(); + let outcome = carry(&mind, &*clock, &lender, &never, TEST_CAP_S, &mut out).unwrap(); + assert_eq!(outcome, SelfRunOutcome::LeaseRefused { kind }); + assert_eq!(outcome.exit_status(), 4); + assert!(never.spawns.borrow().is_empty(), "{kind:?}: the spawner is never called"); + assert!(!format!("{outcome} {}", String::from_utf8_lossy(&out)).contains("detail-canary"), "the connector's words are not echoed"); + + // The run is closed Recorded, with the kind, so a run that cannot be paid for counts as an empty one. + let closure = mind.view(K::Resolution, &format!("{INSTANCE}:resolution:run.selfrun-a.n1")).expect("the run is closed"); + let PipelineDocument::Resolution(closure) = closure.document else { panic!("not a resolution") }; + assert_eq!(closure.subject, run_ref("selfrun-a")); + assert!(matches!(&closure.outcome, ResolutionOutcome::Recorded { reason } if reason.0.contains(&format!("{kind:?}"))), "{:?}", closure.outcome); + + // Closed, so it is no longer her open run: nothing is carried or leased for it again. + let again = FakeLender::new(Answer::Grant); + assert_eq!(carry(&mind, &*clock, &again, &never, TEST_CAP_S, &mut Vec::new()).unwrap(), SelfRunOutcome::NoRun); + assert!(again.asked.borrow().is_empty()); + } +} + +#[test] +fn a_lease_refused_for_its_length_is_a_configuration_error_that_starts_nothing_and_withdraws_the_run() { + let mind = with_run(); + let lender = FakeLender::new(Answer::Refuse(LeaseRefusalKind::Expired)); + let never = never_spawns(); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let error = carry(&mind, &*clock, &lender, &never, TEST_CAP_S, &mut Vec::new()).unwrap_err(); + let message = format!("{error:#}"); + assert!(message.contains("--lease-max-ms") && message.contains("5400") && message.contains(RUN_CAP_VARIABLE), "{message}"); + assert!(!message.contains("detail-canary"), "the connector's words are not echoed"); + assert!(never.spawns.borrow().is_empty()); + // Withdrawn, never Recorded: the breaker counts no withdrawn run, so a misconfigured unit never trips it. + let closure = mind.view(K::Resolution, &format!("{INSTANCE}:resolution:run.selfrun-a.n1")).expect("the run is withdrawn"); + let PipelineDocument::Resolution(closure) = closure.document else { panic!("not a resolution") }; + assert_eq!(closure.subject, run_ref("selfrun-a")); + assert!(matches!(closure.outcome, ResolutionOutcome::Withdrawn { .. }), "{:?}", closure.outcome); + let again = FakeLender::new(Answer::Grant); + assert_eq!(carry(&mind, &*clock, &again, &never, TEST_CAP_S, &mut Vec::new()).unwrap(), SelfRunOutcome::NoRun, "no open run is left for a later launch to find dead and record"); +} + +#[test] +fn a_connector_that_cannot_be_reached_is_an_error_that_starts_nothing_and_leaves_the_run_open() { + let mind = with_run(); + let lender = FakeLender::new(Answer::Unreachable); + let never = never_spawns(); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + assert!(carry(&mind, &*clock, &lender, &never, TEST_CAP_S, &mut Vec::new()).is_err()); + assert!(never.spawns.borrow().is_empty()); + assert_eq!(mind.ids(K::Resolution), Vec::::new(), "an unreachable connector is no refusal: the run stays open"); +} + +#[test] +fn settle_sends_carry_usage() { + let haiku = |line: String| { + let mut event: Value = serde_json::from_str(&line).unwrap(); + event["message"]["model"] = json!("claude-haiku-5-1"); + event.to_string() + }; + let ran = go(vec![ + Out(billed("m1", 1, 100, 10, 5)), + // Records that share a message id are one request, at the usage it had grown to. + Out(billed("m1", 1, 100, 10, 9)), + // A subagent's request is billed too, under its own model. + Out(haiku(subagent_billed("m2", 2, 0, 0, 4))), + Out(billed("m3", 3, 0, 0, 1)), + Out("not json".to_string()), + Out(result_line()), + ]); + let used = |model: &str, input, read, write, output| ModelUsage { model: model.into(), usage: TokenUsage { input_tokens: input, cache_read_tokens: read, cache_write_tokens: write, output_tokens: output, reasoning_tokens: None } }; + let expected = Settlement { usage: vec![used("claude-haiku-5-1", 2, 0, 0, 4), used("claude-opus-5-5", 4, 100, 10, 10)] }; + let settled: Vec<&Event> = ran.all.iter().filter(|event| matches!(event, Event::Settled(_))).collect(); + assert_eq!(settled, vec![&Event::Settled(expected)], "settled exactly once, with the stream's sums per model"); + assert!(matches!(ran.all.last(), Some(Event::Settled(_))), "settled after the child ended, not before"); + assert_eq!(ran.count_all(&Event::Abandoned), 0); +} + +#[test] +fn a_run_that_reported_nothing_settles_an_empty_report_which_the_connector_bills_as_the_reservation() { + // Not free: an empty report is the connector's cue to bill the reservation, exactly as for a dropped lease. + let ran = go(vec![Out(result_line())]); + assert!(ran.all.contains(&Event::Settled(Settlement::default()))); +} + +#[test] +fn a_child_that_cannot_be_stopped_is_left_to_the_connector_to_bill_in_full() { + // Not even SIGKILL is delivered: the child may still be spending, so the lease is dropped unsettled. + let stuck = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], Some(0), None, &[Stop::Interrupt, Stop::Terminate, Stop::Kill], true); + assert_eq!(stuck.ended().0, Cause::Unstoppable); + assert_eq!(stuck.count_all(&Event::Abandoned), 1); + assert!(!stuck.all.iter().any(|event| matches!(event, Event::Settled(_)))); +} + +#[test] +fn a_run_cut_off_before_its_child_ended_is_billed_the_reservation_not_a_guess() { + // The output stream fails mid-run: `carry` unwinds with the child still alive. The lease is dropped + // unsettled, as when the brake stops the unit, and the connector bills the reservation. + struct Broken; + impl Write for Broken { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(std::io::Error::other("closed")) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let mind = with_run(); + let log = Rc::new(RefCell::new(Vec::new())); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let child = Scripted { steps: vec![Out(at("m1", 10))].into(), log: log.clone(), clock: clock.clone(), code: Some(0), dies_on: None, refuses: vec![], hangs: true, exited: false, eof: false, asked: 0 }; + let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false }; + let lender = FakeLender::logging(log.clone(), Answer::Grant); + assert!(carry(&Seen::of(&mind), &*clock, &lender, &fake, TEST_CAP_S, &mut Broken).is_err()); + assert_eq!(log.borrow().last(), Some(&Event::Abandoned)); + assert!(!log.borrow().iter().any(|event| matches!(event, Event::Settled(_)))); +} + +#[test] +fn a_child_that_cannot_start_settles_an_empty_report_which_the_connector_bills_as_the_reservation() { + let mind = with_run(); + let log = Rc::new(RefCell::new(Vec::new())); + let lender = FakeLender::logging(log.clone(), Answer::Grant); + let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true }; + let clock = Rc::new(Fixed(Cell::new(at_start()))); + assert!(carry(&Seen::of(&mind), &*clock, &lender, &refusing, TEST_CAP_S, &mut Vec::new()).is_err()); + assert_eq!(*log.borrow(), vec![Event::Lent, Event::Settled(Settlement::default())]); +} + +fn result_stating(models: Value) -> String { + let mut event: Value = serde_json::from_str(&result_line()).unwrap(); + event["modelUsage"] = models; + event.to_string() +} + +fn stated(input: u64, read: u64, write: u64, output: u64) -> Value { + json!({"inputTokens": input, "cacheReadInputTokens": read, "cacheCreationInputTokens": write, "outputTokens": output, "costUSD": 0.5, "contextWindow": 200000}) +} + +#[test] +fn claudes_own_per_model_total_in_the_result_is_what_settles_and_the_last_result_wins() { + let used = |model: &str, input, read, write, output| ModelUsage { model: model.into(), usage: TokenUsage { input_tokens: input, cache_read_tokens: read, cache_write_tokens: write, output_tokens: output, reasoning_tokens: None } }; + let ran = go(vec![ + Out(billed("m1", 1, 100, 10, 5)), + // An earlier result is superseded by the cumulative one after it (the flush turn's). + Out(result_stating(json!({"claude-opus-5-5": stated(7, 8, 9, 10)}))), + Out(billed("m2", 3, 0, 0, 1)), + Out(result_stating(json!({"claude-opus-5-5": stated(11, 12, 13, 14), "claude-haiku-5-1": stated(1, 2, 3, 4)}))), + ]); + let expected = Settlement { usage: vec![used("claude-haiku-5-1", 1, 2, 3, 4), used("claude-opus-5-5", 11, 12, 13, 14)] }; + let settled: Vec<&Event> = ran.all.iter().filter(|event| matches!(event, Event::Settled(_))).collect(); + assert_eq!(settled, vec![&Event::Settled(expected)], "the stream's own sums are not added to claude's total"); +} + +#[test] +fn a_result_that_states_no_model_usage_leaves_the_streams_sums_to_settle() { + let ran = go(vec![Out(billed("m1", 1, 100, 10, 5)), Out(result_stating(json!({}))), Out(result_stating(json!("not a map")))]); + let settled: Vec<&Event> = ran.all.iter().filter(|event| matches!(event, Event::Settled(_))).collect(); + let used = ModelUsage { model: UNNAMED_MODEL.into(), usage: TokenUsage { input_tokens: 1, cache_read_tokens: 100, cache_write_tokens: 10, output_tokens: 5, reasoning_tokens: None } }; + assert!(matches!(settled.as_slice(), [Event::Settled(Settlement { usage })] if usage.len() == 1 && usage[0].usage == used.usage), "{settled:?}"); +} + +#[test] +fn the_journal_says_what_was_billed_and_that_a_failed_settle_leaves_the_reservation_billed() { + let run = |answer: Answer| { + let mind = with_run(); + let log = Rc::new(RefCell::new(Vec::new())); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let child = Scripted { steps: vec![Out(at("m1", 10)), Out(result_line())].into(), log: log.clone(), clock: clock.clone(), code: Some(0), dies_on: None, refuses: vec![], hangs: false, exited: false, eof: false, asked: 0 }; + let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false }; + let mut out = Vec::new(); + let outcome = carry(&Seen::of(&mind), &*clock, &FakeLender::logging(log, answer), &fake, TEST_CAP_S, &mut out).unwrap(); + (outcome, String::from_utf8(out).unwrap()) + }; + let (outcome, out) = run(Answer::Grant); + assert_eq!(out, "request 1 prefix 10\nsettled: billed 7 micro-USD\n"); + let (failed, out) = run(Answer::GrantSettleFails); + assert_eq!(out, "request 1 prefix 10\nsettle failed: the connector bills the reservation\n", "the connector's words are not echoed"); + assert_eq!(failed, outcome, "the run's own outcome stands"); +} + +fn read_total(settlement: &Settlement) -> u64 { + settlement.usage.iter().map(|used| used.usage.cache_read_tokens).sum() +} + +#[test] +fn requests_after_the_last_result_are_settled_on_top_of_claudes_total() { + // The flush turn runs after the first result and is stopped before it states a result of its own. + let ran = go(vec![ + Out(billed("m1", 1, 100, 10, 5)), + Out(result_stating(json!({"claude-opus-5-5": stated(41, 812_345, 20_480, 9_123)}))), + Out(billed("m2", 1000, 900_000, 0, 500)), + Out(billed("m2", 1000, 900_000, 0, 700)), + ]); + let total = |input, read, write, output| Settlement { usage: vec![ModelUsage { model: "claude-opus-5-5".into(), usage: TokenUsage { input_tokens: input, cache_read_tokens: read, cache_write_tokens: write, output_tokens: output, reasoning_tokens: None } }] }; + let settled: Vec<&Event> = ran.all.iter().filter(|event| matches!(event, Event::Settled(_))).collect(); + assert_eq!(settled, vec![&Event::Settled(total(1_041, 1_712_345, 20_480, 9_823))], "claude's total plus the tail request once"); +} + +#[test] +fn two_results_and_a_tail_settle_the_last_total_plus_only_what_came_after_it() { + let mut spent = Spent::default(); + spent.hear(&billed("m1", 1, 100, 0, 1)); + spent.hear(&result_stating(json!({"claude-opus-5-5": stated(1, 100, 0, 1)}))); + spent.hear(&billed("m2", 2, 200, 0, 2)); + spent.hear(&result_stating(json!({"claude-opus-5-5": stated(3, 300, 0, 3)}))); + assert_eq!(read_total(&spent.settlement()), 300, "m2 is inside the second result's total"); + spent.hear(&billed("m3", 4, 400, 0, 4)); + let settled = spent.settlement(); + assert_eq!(read_total(&settled), 700); + assert_eq!(settled.usage.len(), 1, "the tail joins its model's total"); + assert_eq!(settled.usage[0].usage.input_tokens, 7); + // A later empty result neither erases the total nor folds the tail in. + spent.hear(&result_stating(json!({}))); + assert_eq!(read_total(&spent.settlement()), 700); +} diff --git a/crates/mind-body/src/self_run/tests/process/lease_child.rs b/crates/mind-body/src/self_run/tests/process/lease_child.rs new file mode 100644 index 0000000..74fb7c3 --- /dev/null +++ b/crates/mind-body/src/self_run/tests/process/lease_child.rs @@ -0,0 +1,153 @@ +//! The real child, with a fake `claude` script that reads its credential from +//! the descriptor it is told of. The token here is a canary: what these tests +//! prove is where it can and cannot be found. + +use super::*; + +/// Reads the credential from the named descriptor, then reports what it can see. +/// It prints the credential once, so that the test can tell it arrived. +const READS_ITS_CREDENTIAL: &str = r#" +fd="$CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR" +read -r -u "$fd" credential +echo "fd=$fd" +echo "scrub=$CLAUDE_CODE_SUBPROCESS_ENV_SCRUB" +echo "argv=$*" +echo "in_env=$(env | grep -c "$credential")" +echo "got=$credential" +echo "grandchild=[$(sh -c "cat <&$fd")]" +"#; + +fn field(lines: &[String], name: &str) -> String { + lines.iter().find_map(|line| line.strip_prefix(&format!("{name}="))).unwrap_or_else(|| panic!("no {name} in {lines:?}")).to_string() +} + +#[test] +fn granted_fd_reaches_child_env_var() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, READS_ITS_CREDENTIAL); + let token = token_holding("tok-canary-7f3a"); + let mut child = start_with(&program, &claude_args(session_id(&run_ref("selfrun-a")), "5"), &token); + child.close_input(); + let (lines, code) = drain(&mut *child); + assert_eq!(code, Some(0), "{lines:?}"); + let number: i32 = field(&lines, "fd").parse().unwrap(); + assert!(number >= 3, "never a stdio number: {number}"); + assert_eq!(field(&lines, "scrub"), "1"); + assert_eq!(field(&lines, "got"), "tok-canary-7f3a", "the credential came through the descriptor"); + assert_eq!(field(&lines, "in_env"), "0", "the token text is in no environment variable"); + assert!(!field(&lines, "argv").contains("tok-canary"), "nor in the arguments"); +} + +#[test] +fn the_childs_children_inherit_an_emptied_pipe_not_the_token() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, READS_ITS_CREDENTIAL); + let token = token_holding("tok-canary-7f3a"); + let mut child = start_with(&program, &[], &token); + child.close_input(); + let (lines, _) = drain(&mut *child); + assert_eq!(field(&lines, "got"), "tok-canary-7f3a"); + assert_eq!(field(&lines, "grandchild"), "[]", "a descendant that reads the descriptor after claude has gets end of file"); +} + +#[test] +fn a_credential_in_the_units_environment_never_reaches_the_child() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, "env > \"$0.env\" +echo done"); + let mut child = start_with(&program, &[], &token_holding("tok-canary-7f3a")); + child.close_input(); + drain(&mut *child); + let seen = std::fs::read_to_string(dir.path().join("claude.env")).unwrap(); + let names: Vec<&str> = seen.lines().filter_map(|line| line.split_once('=')).map(|(name, _)| name).collect(); + for canary in CREDENTIAL_CANARIES { + assert!(!names.contains(&canary), "{canary} reached the child: {names:?}"); + } + assert!(!seen.contains("canary-"), "no canary value either: {seen}"); + assert!(names.contains(&"PATH") && names.contains(&"CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR"), "{names:?}"); +} + +#[test] +fn only_the_allow_listed_names_pass_from_the_units_environment() { + let unit: Vec<(std::ffi::OsString, std::ffi::OsString)> = + ["PATH", "HOME", "LANG", "LC_ALL", "TERM", "CLAUDE_CONFIG_DIR", "ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_BASE_URL", "LCX", "XDG_DATA_HOME"].map(|name| (name.into(), "v".into())).to_vec(); + let passed: Vec = child_environment(&unit).into_iter().map(|(name, _)| name.to_string_lossy().into_owned()).collect(); + assert_eq!(passed, ["PATH", "LANG", "LC_ALL", "TERM"], "home and the configuration directory are the spawn's, never the unit's"); +} + +/// Reports where the child's configuration is, its mode, and what is in it; leaves a file and a +/// settings file behind, as a run that wanted to redirect the next one would. +const USES_ITS_CONFIGURATION: &str = r#" +echo "dir=$CLAUDE_CONFIG_DIR" +echo "home=$HOME" +echo "mode=$(stat -c %a "$CLAUDE_CONFIG_DIR")" +echo "listing=$(ls -A "$CLAUDE_CONFIG_DIR" | tr -s "[:space:]" " ")" +echo "skill=$(cat "$CLAUDE_CONFIG_DIR/skills/mind-self/SKILL.md" 2>/dev/null)" +echo '{"env":{"ANTHROPIC_BASE_URL":"http://elsewhere"},"apiKeyHelper":"steal"}' > "$CLAUDE_CONFIG_DIR/settings.json" +echo left > "$CLAUDE_CONFIG_DIR/leftover" +"#; + +fn run_once(program: &Path, scratch: &Path, seed: Option<&Path>) -> Vec { + let config = FreshConfig::create(scratch, seed).unwrap(); + let mut child = spawn_program(program, &[], &token_holding("tok-canary-7f3a"), &unit_environment(), config).unwrap(); + child.close_input(); + let (lines, code) = drain(&mut *child); + assert_eq!(code, Some(0), "{lines:?}"); + lines +} + +#[test] +fn each_run_gets_a_fresh_owner_only_configuration_that_what_it_writes_does_not_outlive() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, USES_ITS_CONFIGURATION); + let seed = tempfile::tempdir().unwrap(); + std::fs::create_dir_all(seed.path().join("skills/mind-self")).unwrap(); + std::fs::write(seed.path().join("skills/mind-self/SKILL.md"), "the brief").unwrap(); + std::os::unix::fs::symlink("/etc/passwd", seed.path().join("link")).unwrap(); + + let first = run_once(&program, dir.path(), Some(seed.path())); + assert_eq!(field(&first, "dir"), field(&first, "home")); + assert_eq!(field(&first, "dir"), dir.path().join("claude-config").to_string_lossy()); + assert_eq!(field(&first, "mode"), "700"); + assert_eq!(field(&first, "skill"), "the brief", "the seed is copied in"); + assert_eq!(field(&first, "listing").trim(), "skills", "a link in the seed is not followed or copied"); + assert!(!dir.path().join("claude-config").exists(), "removed when the child's handle went"); + + // A stale directory (a unit whose runtime directory was preserved) is wiped, not reused. + std::fs::create_dir_all(dir.path().join("claude-config")).unwrap(); + std::fs::write(dir.path().join("claude-config/settings.json"), "{\"apiKeyHelper\":\"steal\"}").unwrap(); + let second = run_once(&program, dir.path(), None); + assert_eq!(field(&second, "listing").trim(), "", "nothing of the last run, or of what was left on disk, reaches this one"); + assert_eq!(field(&second, "skill"), "", "no seed, no skill"); +} + +#[test] +fn the_spawner_needs_the_units_runtime_directory_and_reads_the_seed_from_the_units_environment() { + assert!(ClaudeSpawner::new(Vec::new()).is_err(), "no scratch directory named"); + assert!(ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "".into())]).is_err(), "an empty one is none"); + let named = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into()), ("MIND_SELF_CONFIG_SEED".into(), "/etc/seed".into())]).unwrap(); + assert_eq!((named.scratch.as_path(), named.seed.as_deref()), (Path::new("/run/mind"), Some(Path::new("/etc/seed")))); + let bare = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into())]).unwrap(); + assert_eq!(bare.seed, None); +} + +/// How many descriptors of this process are open on the pipe that `probe` is one end of. +fn open_ends_of_the_pipe_of(probe: &crate::lease_client::Inheritable) -> usize { + use std::os::unix::fs::MetadataExt; + let identity = |path: &Path| std::fs::metadata(path).ok().map(|meta| (meta.dev(), meta.ino())); + let wanted = identity(Path::new(&format!("/proc/self/fd/{}", probe.number()))).unwrap(); + std::fs::read_dir("/proc/self/fd").unwrap().filter_map(Result::ok).filter(|entry| identity(&entry.path()) == Some(wanted)).count() +} + +#[test] +fn the_parent_closes_the_duplicate_it_lent_the_child() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, "sleep 5"); + let token = token_holding("tok-canary-7f3a"); + let mut child = start_with(&program, &[], &token); + // The lease's own descriptor and this probe; the spawn's duplicate, lent to the child, is gone from the parent. + let probe = token.inheritable().unwrap(); + assert_eq!(open_ends_of_the_pipe_of(&probe), 2, "the duplicate lent to the child was closed in the parent"); + child.stop(Stop::Kill); + drain(&mut *child); +} diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index f26c48b..9cdaa83 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -4,6 +4,7 @@ mind_body: mod cli mind_body: mod connector mind_body: mod control mind_body: mod launch +mind_body: mod lease_client mind_body: mod persona mind_body: mod queue mind_body: mod self_run @@ -119,6 +120,29 @@ mind_body::launch: struct Ports< 'a > pub-fields[mind: & 'a dyn MindPort, launc mind_body::launch: struct SystemClock pub-fields[] mind_body::launch: struct SystemctlReply pub-fields[success: bool, code: Option < i32 >, stdout: String] mind_body::launch: struct SystemdLauncher pub-fields[] +mind_body::lease_client: # [derive (Clone , Copy , Debug , PartialEq , Eq , Serialize , Deserialize)] pub enum LeaseRefusalKind { SpendCeiling , CostBound , UnknownCaller , Expired , Busy , } +mind_body::lease_client: const LEASE_SOCKET_VARIABLE : & str +mind_body::lease_client: const MAX_LEASE_FRAME_BYTES : usize +mind_body::lease_client: impl Inheritable +mind_body::lease_client: impl Inheritable: fn number (& self) -> i32 +mind_body::lease_client: impl Lender for SocketLender +mind_body::lease_client: impl SocketLender +mind_body::lease_client: impl SocketLender: fn new (path : impl Into < PathBuf >) -> Self +mind_body::lease_client: impl TokenFd +mind_body::lease_client: impl TokenFd: fn inheritable (& self) -> Result < Inheritable > +mind_body::lease_client: impl std :: fmt :: Debug for TokenFd +mind_body::lease_client: pub enum Lent { Granted (Box < dyn Lease >) , Refused (LeaseRefused) , } +mind_body::lease_client: pub trait Lease { fn token (& self) -> & TokenFd ; fn settle (self : Box < Self > , settlement : Settlement) -> Result < u64 > ; } +mind_body::lease_client: pub trait Lender { fn lend (& self , request : & LeaseRequest) -> Result < Lent > ; } +mind_body::lease_client: struct Inheritable pub-fields[] +mind_body::lease_client: struct LeaseRefused # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[kind: LeaseRefusalKind, detail: String] +mind_body::lease_client: struct LeaseRequest # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[expires_at_unix_ms: u64] +mind_body::lease_client: struct ModelUsage # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[model: String, usage: TokenUsage] +mind_body::lease_client: struct Settlement # [derive (Clone , Debug , Default , PartialEq , Eq)] pub-fields[usage: Vec < ModelUsage >] +mind_body::lease_client: struct SocketLender pub-fields[] +mind_body::lease_client: struct TokenFd pub-fields[] +mind_body::lease_client: struct TokenUsage # [derive (Clone , Debug , Default , PartialEq , Eq , Serialize , Deserialize)] pub-fields[input_tokens: u64, cache_read_tokens: u64, cache_write_tokens: u64, output_tokens: u64, reasoning_tokens: Option < u64 >] +mind_body::lease_client::socket: impl Lease for Granted mind_body::persona: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Failure { Held (HeldReason) , Chat , Persona , Readout , Connector , Undecodable , RoomNotAllowed , Send , SelfRun , Ack , } mind_body::persona: # [derive (Debug , PartialEq , Eq)] pub enum Outcome { Dead (HeldReason) , NoRun (Declined) , Served { run : PipelineRef , answered : u32 , failure : Option < Failure > , self_run : Option < PipelineRef > } , } mind_body::persona: const AGENT : & str @@ -173,7 +197,7 @@ mind_body::queue: struct HuginnMind< 'a > pub-fields[] mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq , PartialOrd , Ord)] pub enum Stop { Interrupt , Terminate , Kill , } mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Cause { Own , FlushRequested , Signalled (Stop) , Unstoppable , } mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Read { Line (String) , Timeout , Eof , Exited (Option < i32 >) , } -mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum SelfRunOutcome { NoRun , Ended { cause : Cause , code : Option < i32 > , requests : u64 , prefix : u64 } , } +mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum SelfRunOutcome { NoRun , LeaseRefused { kind : LeaseRefusalKind } , Ended { cause : Cause , code : Option < i32 > , requests : u64 , prefix : u64 } , } mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Usage { Request { id : String , prefix : u64 } , Result , Other , Unreadable , } mind_body::self_run: const AGENT : & str mind_body::self_run: const CEILING_TOKENS : u64 @@ -183,20 +207,28 @@ mind_body::self_run: const FLUSH_GRACE_S : i64 mind_body::self_run: const FLUSH_HEADROOM_TOKENS : u64 mind_body::self_run: const FLUSH_MESSAGE : & str mind_body::self_run: const MAX_TURNS : u32 +mind_body::self_run: const RUN_CAP_VARIABLE : & str +mind_body::self_run: const SCRATCH_VARIABLE : & str +mind_body::self_run: const SEED_VARIABLE : & str mind_body::self_run: const SESSION_NAMESPACE : Uuid -mind_body::self_run: fn carry (mind : & dyn MindPort , clock : & dyn Clock , spawner : & dyn Spawner , out : & mut dyn Write) -> Result < SelfRunOutcome > +mind_body::self_run: fn carry (mind : & dyn MindPort , clock : & dyn Clock , lender : & dyn Lender , spawner : & dyn Spawner , run_cap_s : u64 , out : & mut dyn Write) -> Result < SelfRunOutcome > mind_body::self_run: fn claude_args (session : Uuid , budget_usd : & str) -> Vec < String > mind_body::self_run: fn read_usage (line : & str) -> Usage mind_body::self_run: fn session_id (run : & PipelineRef) -> Uuid mind_body::self_run: fn start_message (instance : & Slug , run : & PipelineRef) -> String mind_body::self_run: impl Child for ClaudeProcess +mind_body::self_run: impl ClaudeSpawner +mind_body::self_run: impl ClaudeSpawner: fn new (unit_environment : impl IntoIterator < Item = (OsString , OsString) >) -> Result < Self > +mind_body::self_run: impl Drop for FreshConfig +mind_body::self_run: impl FreshConfig mind_body::self_run: impl SelfRunOutcome mind_body::self_run: impl SelfRunOutcome: fn exit_status (& self) -> i32 mind_body::self_run: impl Spawner for ClaudeSpawner +mind_body::self_run: impl Spent mind_body::self_run: impl Stop mind_body::self_run: impl fmt :: Display for SelfRunOutcome mind_body::self_run: pub trait Child { fn write_line (& mut self , line : & str) ; fn close_input (& mut self) ; fn next (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; } -mind_body::self_run: pub trait Spawner { fn spawn (& self , args : & [String]) -> Result < Box < dyn Child > > ; } +mind_body::self_run: pub trait Spawner { fn spawn (& self , args : & [String] , token : & TokenFd) -> Result < Box < dyn Child > > ; } mind_body::self_run: struct ClaudeSpawner pub-fields[] mind_body::wake: # [derive (Clone , Debug , PartialEq , Eq)] pub enum WakeOutcome { Held { reason : HeldReason } , NotYet { until : DateTime < Utc > } , Busy { run : PipelineRef } , Launched { run : PipelineRef } , Refused (Declined) , } mind_body::wake: const AGENT : & str diff --git a/crates/mind-body/tests/self_run.rs b/crates/mind-body/tests/self_run.rs index 390b4d4..b8a0e52 100644 --- a/crates/mind-body/tests/self_run.rs +++ b/crates/mind-body/tests/self_run.rs @@ -25,6 +25,27 @@ fn all_output(output: &Output) -> String { format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)) } +#[test] +fn the_run_cap_must_be_a_positive_number_of_seconds_and_the_runtime_directory_must_be_named() { + let with_cap = |cap: Option<&str>| { + let mut command = Command::new(env!("CARGO_BIN_EXE_mind-self-run")); + command.args(["--instance", NO_SUCH_INSTANCE, "--endpoint", "rudp://127.0.0.1:9"]).env_remove("HUGINN_ENDPOINT").env_remove("RUNTIME_DIRECTORY").env("CLAUDE_LEASE_SOCKET", "/nonexistent/lease.sock"); + match cap { + Some(cap) => command.env("MIND_SELF_RUN_CAP_S", cap), + None => command.env_remove("MIND_SELF_RUN_CAP_S"), + }; + let output = command.output().unwrap(); + assert_eq!(output.status.code(), Some(2)); + all_output(&output) + }; + for bad in [None, Some("0"), Some(""), Some("abc"), Some("-5")] { + assert!(with_cap(bad).contains("MIND_SELF_RUN_CAP_S is not set to"), "{bad:?}"); + } + // A positive cap passes that check and meets the next: the unit names no scratch directory. + let message = with_cap(Some("1")); + assert!(!message.contains("MIND_SELF_RUN_CAP_S") && message.contains("RUNTIME_DIRECTORY is not set"), "{message}"); +} + #[test] fn without_an_instance_or_with_a_stray_argument_it_exits_2_with_the_usage_and_echoes_nothing() { let canary = "canary-argument-3f9a"; @@ -83,7 +104,11 @@ fn a_mind_that_cannot_be_reached_is_exit_2_before_any_child_and_names_no_value() #[cfg(unix)] mod lane { + use std::io::{IoSlice, Read, Write}; + use std::mem::MaybeUninit; use std::net::UdpSocket; + use std::os::unix::net::UnixListener; + use std::sync::{Arc, Mutex}; use std::path::Path; use std::process::{Child, Command, Output, Stdio}; use std::time::Duration; @@ -188,7 +213,7 @@ mod lane { /// The binary with `bin` first on `PATH`, until the daemon answers (exit 2 before that /// has started nothing). `SELF_RUN_ENV_CANARY` is a variable of the unit's environment. - fn ask(endpoint: &str, bin: &Path) -> Output { + fn ask(endpoint: &str, bin: &Path, socket: &Path) -> Output { let exe = Path::new(env!("CARGO_BIN_EXE_mind-self-run")); let path = format!("{}:/usr/bin:/bin", bin.display()); for _ in 0..40 { @@ -196,6 +221,9 @@ mod lane { .args(["--instance", INSTANCE, "--endpoint", endpoint]) .env("PATH", &path) .env("SELF_RUN_ENV_CANARY", ENV_CANARY) + .env("CLAUDE_LEASE_SOCKET", socket) + .env("MIND_SELF_RUN_CAP_S", "14400") + .env("RUNTIME_DIRECTORY", bin) .output() .unwrap(); if attempt.status.code() != Some(2) { @@ -206,6 +234,61 @@ mod lane { panic!("the daemon never answered"); } + const GRANT: &[u8] = include_bytes!("../src/lease_client/lease_v1/grant.bin"); + const SETTLED: &[u8] = include_bytes!("../src/lease_client/lease_v1/settled.bin"); + const LANE_TOKEN: &str = "tok-lane-canary-51c7"; + + /// The connector's lease socket: every connection gets the golden grant with a pipe holding + /// `LANE_TOKEN` attached as SCM_RIGHTS, then the golden settled answer to the settle it reads. + /// The settle frames it received are kept. + struct FakeConnector { + _dir: TempDir, + path: std::path::PathBuf, + settles: Arc>>>, + } + + fn read_frame(stream: &mut std::os::unix::net::UnixStream) -> std::io::Result> { + let mut length = [0_u8; 4]; + stream.read_exact(&mut length)?; + let mut frame = length.to_vec(); + frame.resize(4 + u32::from_be_bytes(length) as usize, 0); + stream.read_exact(&mut frame[4..])?; + Ok(frame) + } + + fn fake_connector() -> FakeConnector { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("lease.sock"); + let listener = UnixListener::bind(&path).unwrap(); + let settles = Arc::new(Mutex::new(Vec::new())); + let kept = settles.clone(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { return }; + let kept = kept.clone(); + std::thread::spawn(move || { + if read_frame(&mut stream).is_err() { + return; + } + let (read, mut write) = std::io::pipe().unwrap(); + write.write_all(LANE_TOKEN.as_bytes()).unwrap(); + drop(write); + let read: std::os::fd::OwnedFd = read.into(); + let fds = [std::os::fd::AsFd::as_fd(&read)]; + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(1))]; + let mut control = rustix::net::SendAncillaryBuffer::new(&mut space); + assert!(control.push(rustix::net::SendAncillaryMessage::ScmRights(&fds))); + rustix::net::sendmsg(&stream, &[IoSlice::new(GRANT)], &mut control, rustix::net::SendFlags::empty()).unwrap(); + if let Ok(settle) = read_frame(&mut stream) { + kept.lock().unwrap().push(settle); + let _ = stream.write_all(SETTLED); + } + }); + } + }); + FakeConnector { _dir: dir, path, settles } + } + const ENV_CANARY: &str = "canary-unit-environment-b94e"; /// Writes an executable `claude` into `bin` and returns the path of the record it keeps. @@ -237,20 +320,22 @@ mod lane { ) }; let body = format!( - "echo journal-line >&2 -echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start\" >> {record}\necho \"$flush\" >> {record}\necho \"$SELF_RUN_ENV_CANARY\" >> {record}\necho '{{\"type\":\"result\"}}'\nwhile read more; do :; done\nexit 0", + "read -r -u \"$CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR\" cred +echo journal-line >&2 +echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start\" >> {record}\necho \"$flush\" >> {record}\necho \"$SELF_RUN_ENV_CANARY\" >> {record}\necho \"$cred\" >> {record}\necho '{{\"type\":\"result\"}}'\nwhile read more; do :; done\nexit 0", first = request("msg_1", 100), second = request("msg_2", 299_999), record = bin.path().join("record").display(), ); let record = fake_claude(bin.path(), &body); - let output = ask(&endpoint, bin.path()); + let connector = fake_connector(); + let output = ask(&endpoint, bin.path(), &connector.path); drop(daemon); let stdout = String::from_utf8_lossy(&output.stdout).to_string(); assert_eq!(output.status.code(), Some(0), "{stdout}\n{}", String::from_utf8_lossy(&output.stderr)); - assert_eq!(stdout, "request 1 prefix 101\nrequest 2 prefix 300000\nended: flush-requested code 0 requests 2 prefix 300000\n"); - assert!(!all_output(&output).contains(ENV_CANARY), "the unit's environment is the child's, not the bin's output"); + assert_eq!(stdout, "request 1 prefix 101\nrequest 2 prefix 300000\nsettled: billed 123456 micro-USD\nended: flush-requested code 0 requests 2 prefix 300000\n"); + assert!(!all_output(&output).contains(ENV_CANARY), "the unit environment is not echoed"); assert!(String::from_utf8_lossy(&output.stderr).contains("journal-line"), "the child's stderr is the unit's journal"); let recorded = std::fs::read_to_string(&record).unwrap(); @@ -261,7 +346,12 @@ echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start assert!(lines.next().unwrap().contains("yggdrasil:run:selfrun-lane"), "the start message names her run"); let flush = lines.next().unwrap(); assert!(flush.contains(FLUSH_MESSAGE) && flush.contains("\"type\":\"user\""), "the flush went to the child's stdin: {flush}"); - assert_eq!(lines.next().unwrap(), ENV_CANARY, "the child inherits the unit's environment"); + assert_eq!(lines.next().unwrap(), "", "the child gets the allow-listed variables only, not the rest of the unit environment"); + assert_eq!(lines.next().unwrap(), LANE_TOKEN, "the child read its credential from the lease's descriptor"); + assert!(!all_output(&output).contains(LANE_TOKEN), "the token is never in the bin's output"); + let settles = connector.settles.lock().unwrap(); + assert_eq!(settles.len(), 1, "the run settled once"); + assert!(String::from_utf8_lossy(&settles[0]).contains("claude_lease_settle")); } /// With no Self run of hers in the mind the bin starts no child and exits 3. @@ -273,7 +363,8 @@ echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start let (daemon, _state, endpoint) = serve(documents); let bin = TempDir::new().unwrap(); let record = fake_claude(bin.path(), "exit 0"); - let output = ask(&endpoint, bin.path()); + let connector = fake_connector(); + let output = ask(&endpoint, bin.path(), &connector.path); drop(daemon); assert_eq!(output.status.code(), Some(3), "{}", all_output(&output)); assert_eq!(String::from_utf8_lossy(&output.stdout), "no run: she has no single open Self run\n"); diff --git a/docs/eureka-body-map.md b/docs/eureka-body-map.md index 3a715c1..1f705d5 100644 --- a/docs/eureka-body-map.md +++ b/docs/eureka-body-map.md @@ -516,7 +516,7 @@ pipeline kinds are in Huginn `docs/eureka-substrate-map.md`. | Identity diffs | The pair of receipts of two consecutive Persona puts whose identity fields (publicName, publicDescription, presentation, values) differ. Derived, not stored. | Composed by the Persona organ at put time and sent as one delivery request through the crossing; a failed post is retried by Bifrost's delivery journal, not by her. | Persona organ composes; Bifrost posts under the permit. No rate limit. | | Discord body surfaces | Outbound: crossing ids `epiphany.persona_discord_delivery_request.v0` and `..._permit.v0`, agent `epiphany.Persona`, runtime `epiphany-starfire` (F6), renamed to `eureka.*` and `eureka-yggdrasil` in the mouth rebinding (`mouth-crossing-on-yggdrasil`). Inbound: Bifrost persona-feedback deliveries for target `epiphany` (F6). Persona-session tools reach both (`body-includes-discord`). | Request, permit, post, receipt (Bifrost journal). Inbound deliveries: **unprobed** who records that a delivery was answered; `run-is-the-grant` forbids her organ from holding that fact. Blocks the Persona cuts only, not the cuts admitted today. | Bifrost owns posting and delivery records; her permit issuer grants while the brake is released. | | Units and instance templates | `mind-wake@.timer/.service`, `mind-self@: