mind-self-via-lease: a Self run borrows the one Claude token from the connector over the lease socket; the child gets a cleared environment and a fresh per-spawn config; the unit's RuntimeMaxSec is the one run cap; settle is claude's last modelUsage plus the tail after it; an Expired lease withdraws its run (squash of eureka-body/mind-self-via-lease fcaf0a1..a91fc1c; Soul s1-s3)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Cargo.lock

@@ -1191,6 +1191,7 @@ dependencies = [
  "quote",
  "rmp-serde",
  "rust_decimal",
+ "rustix",
  "serde",
  "serde_json",
  "sha2",

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

crates/mind-body/src/bin/mind-self-run.rs

@@ -3,18 +3,25 @@
 //! `mind-self@<instance>`. 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 <slug> [--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<i32> {
     let args: Vec<String> = 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())
 }

crates/mind-body/src/launch.rs

@@ -275,9 +275,18 @@ fn past_grace(admitted_at: &str, now: DateTime<Utc>) -> 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<Utc>) -> Result<PipelineDocument> {
+    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<Utc>, outcome: ResolutionOutcome, rationale: &str) -> Result<PipelineDocument> {
     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<Utc>) -> 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),
     }))
 }

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<u64>,
+}
+
+/// 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<ModelUsage>,
+}
+
+#[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<ModelUsage>,
+}
+
+#[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<Vec<u8>> {
+    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<LeaseMessage> {
+    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<Inheritable> {
+        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<Inheritable> {
+        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<Self>, settlement: Settlement) -> Result<u64>;
+}
+
+/// The connector's answer to a request.
+pub enum Lent {
+    Granted(Box<dyn Lease>),
+    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<Lent>;
+}
+
+/// The connector's lease socket.
+pub struct SocketLender {
+    path: PathBuf,
+}
+
+impl SocketLender {
+    pub fn new(path: impl Into<PathBuf>) -> Self {
+        Self { path: path.into() }
+    }
+}
+
+#[cfg(not(unix))]
+impl Lender for SocketLender {
+    fn lend(&self, _request: &LeaseRequest) -> Result<Lent> {
+        let _ = &self.path;
+        Err(anyhow!("the credential lease needs a unix host"))
+    }
+}
+
+#[cfg(unix)]
+impl Lender for SocketLender {
+    fn lend(&self, request: &LeaseRequest) -> Result<Lent> {
+        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<Lent> {
+        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<OwnedFd>)> {
+        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<Self>, settlement: Settlement) -> Result<u64> {
+            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<LeaseMessage> {
+    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(())
+    }
+}

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<T: Send + 'static>(script: impl FnOnce(UnixStream) -> T + Send + 'static) -> (TempDir, SocketLender, JoinHandle<T>) {
+        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<u8> {
+        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<u8> {
+        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<u8>| {
+            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();
+    }
+}

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@<instance>`: 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@<instance>`: 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;

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<String> {
     [
@@ -100,6 +135,8 @@ pub fn claude_args(session: Uuid, budget_usd: &str) -> Vec<String> {
         &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<Box<dyn Child>>;
+    /// Starts the child with `token` as its credential descriptor.
+    fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>>;
 }
 
 /// 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<i32>, 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<Option<(PipelineRef, String)>> {
     }
 }
 
-/// 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<SelfRunOutcome> {
+/// 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<String, (String, TokenUsage)>,
+    reported: Option<BTreeMap<String, TokenUsage>>,
+    /// The message ids heard when `reported` was stated: its total already holds them.
+    counted: BTreeSet<String>,
+}
+
+/// 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::<Value>(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<String, Value>) {
+        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<String, TokenUsage> = 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<String, TokenUsage> = 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<Utc>) -> 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<SelfRunOutcome> {
+    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<SelfRunOutcome> {
     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<dyn Child>, run: &PipelineRef, spent: &mut Spent, out: &mut dyn Write) -> Result<SelfRunOutcome> {
     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<PathBuf>,
+}
 
-impl Spawner for ClaudeSpawner {
-    fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>> {
-        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<Item = (OsString, OsString)>) -> Result<Self> {
+        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<Box<dyn Child>> {
+impl Spawner for ClaudeSpawner {
+    fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
+        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<Self> {
+        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<Box<dyn Child>> {
+    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<Box<dyn Child>> {
     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<Read>,
     /// 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 {

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<Box<dyn Child>> {
+    fn spawn(&self, args: &[String], _token: &TokenFd) -> Result<Box<dyn Child>> {
         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<RefCell<Vec<Event>>>,
+    asked: RefCell<Vec<LeaseRequest>>,
+    answer: Answer,
+}
+
+impl FakeLender {
+    fn new(answer: Answer) -> Self {
+        Self::logging(Rc::new(RefCell::new(Vec::new())), answer)
+    }
+
+    fn logging(log: Rc<RefCell<Vec<Event>>>, answer: Answer) -> Self {
+        Self { log, asked: RefCell::new(Vec::new()), answer }
+    }
+}
+
+impl Lender for FakeLender {
+    fn lend(&self, request: &LeaseRequest) -> Result<Lent> {
+        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<RefCell<Vec<Event>>>,
+    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<Self>, settlement: Settlement) -> Result<u64> {
+        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<SelfRunOutcome>,
     out: String,
     events: Vec<Event>,
+    /// Every event, the lease's included.
+    all: Vec<Event>,
     spawns: Vec<Vec<String>>,
     admits: usize,
+    asked: Vec<LeaseRequest>,
 }
 
 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<Step>, code: Option<i32>, 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<chrono::Utc> {
@@ -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<Step> = 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<dyn Child> {
+        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<dyn Child> {
         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<Box<dyn Child>> {
-            Ok(Box::new(Prompt(start(&self.0, args))))
+        fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
+            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<Box<dyn Child>> {
-            Ok(Box::new(Bounded(start(&self.0, args), std::time::Instant::now())))
+        fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
+            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;

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::<String>::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<usize> {
+            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);
+}

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<String> = 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<String> {
+    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);
+}

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

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<Mutex<Vec<Vec<u8>>>>,
+    }
+
+    fn read_frame(stream: &mut std::os::unix::net::UnixStream) -> std::io::Result<Vec<u8>> {
+        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");

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@<instance>.timer/.service`, `mind-self@<instance>:<label>.service`, `mind-persona@<instance>:<label>.service` (one-shot per Persona turn; `RuntimeMaxSec` set at install bounds a hung unit), `mind-permit@<instance>`; users `mind-<instance>` (organs) and `mind-<instance>-self` (Self runs); state `/var/lib/gamecult/minds/<instance>/`; control `/etc/gamecult/minds/<instance>/`. Instance `eureka`. | Hand-installed by gamecult-ops now; Idunn targets declared before the first hosted member, which deletes the hand install (`mind-units-templated-interim`). | gamecult-ops installs; Idunn later. Starting: `open_and_launch` through `SystemdLauncher::start`, as her users, who are granted start only. The instance in a unit name is always the mind's (`MindPort::instance`), passed to every `Launcher` call; `SystemdLauncher` holds none. The instance and label are validated by the leaf's grammar before any `systemctl` argv is built. Stopping: only the operator's `brake hold`, as root (cut brake-stop). The unit-level brake gate and the start-only grant are follow-up `mind-units-brake-gate`. None of this exists on Yggdrasil yet (F19, F22). Each piece's identity, lifecycle and authority: the section "The mind units" below (cut mind-units-install; unit names pending question `mind-unit-names`). |
-| Self-run credential | `CLAUDE_CODE_OAUTH_TOKEN` in a root-0600 EnvironmentFile per instance. | Created by `claude setup-token`; rotated by the operator. | **Open: question `self-run-token-reach`.** The Self run's `claude` process can read its own environment, so `token-out-of-her-reach` holds only if the operator names this unit (F4 of the prior-art file, follow-up `token-reach-two-units`). |
+| Self-run credential | None in her unit. The connector holds the one subscription token; a Self run borrows it as a lease over a unix socket (`CLAUDE_LEASE_SOCKET`), and the token reaches `claude` only as a descriptor it reads once (`CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR`). The child's environment is cleared and rebuilt from an allow-list (`PATH`, `LANG`, `LC_*`, `TERM`, and its own `HOME` and `CLAUDE_CONFIG_DIR`), so no credential variable in the unit can outrank the lease. | Rotated by the operator at the connector, never in her unit: no `EnvironmentFile` carries a token for `mind-self@` (rulings `self-runs-through-connector`, `self-run-credential-lease`; `self-run-token-reach` is superseded). | The connector's: one lease per unit invocation, bound to the peer; the unit holds no credential variable. Her `claude` runs her tools, so anything in the unit's environment is readable from `/proc/<claude>/environ` by her Bash (ruling `claude-code-runs-her-tools-2`). |
 | Wake frame | Fixed prompt frame (provenance, jurisdiction, brake obedience) in `/etc/gamecult/minds/<instance>/wake-frame.md`, root-owned. Composed by the Persona organ from host config; the waker passes the unit only the instance and the run label. | Changed only by an install. | Host config (ruling `self-waker` rider 2); never a file her runs merge into. |
 | Epiphany residue | F5, F8, F11, as they stood before the teardown. | Teardown landed (gamecult-ops `9c12adb`); what it did not name is follow-up `eureka-body:follow_up:epiphany-body-leftovers`; the repo is archived last. | gamecult-ops for host and repo tooling; Idunn for its legacy target list (follow-up). |
 
@@ -533,7 +533,7 @@ hers per turn (`one-live-self-run`) and only then can `User=` name a per-instanc
 | --- | --- | --- | --- |
 | Waker template | `/etc/systemd/system/mind-wake@.service` and `mind-wake@.timer`, sources `systemd/` in gamecult-ops. Instance unit `mind-wake@<i>.service`, `User=mind-%i`. | Installed and replaced by `scripts/install-mind-units-yggdrasil.sh`. The timer is a fixed poll (1 min after the last cycle ends); it decides nothing: the cadence is the dial's, read by the waker from the newest Persona run (mind-wake r3). Schedules and rest (follow-up `burn-schedule-and-rest`) belong in the dial, so they need no unit change. Each tick runs one cycle, `Type=oneshot`, `TimeoutStartSec=5min`, `Restart=no`. Not gated at the unit: the waker reads the brake itself and returns `Held` before any consequence, and a second reading would protect nothing. | Install, change: root via the script. Enable or disable `mind-wake@<i>.timer`: root only, by hand in the runbook. The script never enables it, so installing starts nothing. Start the service: the timer (systemd as root). Her users have no grant on `mind-wake@`. |
 | Persona template | `/etc/systemd/system/mind-persona@.service`. Instance unit `mind-persona@<i>.service`, `User=mind-%i`, `Type=exec`, no `[Install]`, so it cannot be enabled. | Inactive until started for one Persona turn. Ends by its own exit, at `RuntimeMaxSec=30min`, or by the brake hold's stop (`TimeoutStopSec=30s`, then SIGKILL to the whole cgroup). `Restart=no`, and nothing else restarts it. Every start first runs the gate (below); while the brake is not Released the unit is skipped and ends inactive. `ExecStart` names the Persona organ's binary in the Huginn release; that cut ships the binary and owns the argv. Until then a released start fails at exec (203/EXEC), and the closing launch counts it as a dead run. | Start: `mind-<i>` only, through `open_and_launch` (the waker), by polkit's verb `start` on this one unit name. Stop, kill, restart: root only, which in practice means `mind-control brake hold` (cut brake-stop). Change: root via the script. Enable: nobody. |
-| Self template | `/etc/systemd/system/mind-self@.service`. Instance unit `mind-self@<i>.service`, `User=mind-%i-self`, `Type=exec`, no `[Install]`. | As the Persona unit, with `RuntimeMaxSec=4h`. `ExecStart` names the Self-run actuator's binary in the Huginn release; that cut adds the token `EnvironmentFile` (ruling `self-run-token-reach`), the state directory and the argv. | Start: `mind-<i>` only (the Persona's start-Self-run tool through `open_and_launch`), by polkit, verb `start`. Stop, kill, restart: root only. Change: root via the script. Enable: nobody. |
+| Self template | `/etc/systemd/system/mind-self@.service`. Instance unit `mind-self@<i>.service`, `User=mind-%i-self`, `Type=exec`, no `[Install]`. | As the Persona unit, with `RuntimeMaxSec=4h`. `ExecStart` names the Self-run actuator's binary in the Huginn release; that cut adds the state directory, `RuntimeDirectory=` (each child's fresh configuration), `MIND_SELF_RUN_CAP_S` equal to `RuntimeMaxSec` in seconds, `CLAUDE_LEASE_SOCKET` and the argv. It adds no token and no `EnvironmentFile` with a credential. | Start: `mind-<i>` only (the Persona's start-Self-run tool through `open_and_launch`), by polkit, verb `start`. Stop, kill, restart: root only. Change: root via the script. Enable: nobody. |
 | Organ user | `mind-<i>`: system user and group, shell `/usr/sbin/nologin`, home `/nonexistent`. | Created by the install script if absent and never modified by it after that. Deleted by root at the mind's retirement. Runs the waker and Persona units. | Root creates and deletes it. It may start `mind-persona@<i>` and `mind-self@<i>`, and nothing else. It cannot write the control store (root 0644, dir 0755) or stop a unit. It can signal its own processes with `kill(2)`, as any Unix user can. That only ends spend, and the ended run is closed Recorded by the next launch, so it adds no second stopper of her units. |
 | Self user | `mind-<i>-self`: system user and group, nologin, home `/nonexistent`. The install script refuses an instance slug ending in `-self`, so this name never collides with another mind's organ user. | As the organ user. | Root creates and deletes it. It starts no unit. Polkit refuses it every verb, and the rule refuses any subject whose name ends in `-self`. |
 | `mind-control` | `/opt/gamecult/huginn/releases/<sha>/mind-control`, root:root 0700, reached by the `current` symlink; `/usr/local/sbin/mind-control` links to `/opt/gamecult/huginn/current/mind-control`. | Built and installed with each Huginn release by the Release step of `runbooks/huginn-yggdrasil.md`. Replaced when `current` flips. The previous release is kept until the new one verifies. | Install, replace: root via the Release step. Run: root only (mode 0700), which covers the operator's verbs and the gate, which systemd runs as root through the `+` prefix. No user of hers can execute it. |
@@ -864,13 +864,12 @@ operator's token, then A6 and the bring-up; A3, A4 and A1 run beside it):
    mind, work the run's claims, flush (admit what the mind lacks, close her own run), end.
    Waits on nothing but A2's message contract (the two prompt constants in
    `cut-mind-self-run.r1`).
-5. `self-run-environment` (gamecult-ops): the `mind-self@` lines A2 leaves to ops (token
-   `EnvironmentFile`, `StateDirectory`, `WorkingDirectory`, `CLAUDE_CONFIG_DIR`,
-   `HUGINN_ENDPOINT`), Claude Code and the `huginn` CLI on Yggdrasil (follow-up
+5. `self-run-environment` (gamecult-ops): the `mind-self@` lines A2 leaves to ops (`StateDirectory`, `WorkingDirectory`, `RuntimeDirectory`,
+   `MIND_SELF_RUN_CAP_S` (= `RuntimeMaxSec`), optional `MIND_SELF_CONFIG_SEED`, `CLAUDE_LEASE_SOCKET`, `HUGINN_ENDPOINT`), Claude Code and the `huginn` CLI on Yggdrasil (follow-up
    `eureka-substrate:follow_up:huginn-cli-on-yggdrasil`), the skill installed from A4, git and
    forge reach for her work, the root-owned managed settings. Waits on A1, A2, question
-   `question:self-run-permissions`, and an operator action: her subscription token
-   (`claude setup-token`, ruling `self-run-token-reach`).
+   `question:self-run-permissions`, and the connector's `--lease-max-ms` at least the cap plus
+   slack (rulings `self-runs-through-connector`, `self-run-credential-lease`: no token in her unit).
 6. `mind-launch-self` (Huginn): the one binary that opens a Self run through
    `open_and_launch`; the Persona's start-Self tool later, the operator's bring-up now.
    Waits on question `question:first-self-run-starter`.