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>
This commit is contained in:
Meta Vens 2026-10-30 00:15:14 +01:00
commit d54ba01103
19 changed files with 1712 additions and 85 deletions

1
Cargo.lock generated
View file

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

View file

@ -25,6 +25,7 @@ publish.workspace = true
# MessagePack codec (`chat_client`); `sha2` derives the mouth's connection id from # 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 # the chat credential. The kit pins its own CultLib revision, so the graph holds
# two until the workspace pin and the kit's meet. # 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`); # `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. # `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] [target.'cfg(unix)'.dependencies]
libc = "0.2" 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 # `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 # tests/public_surface.rs, which pins its public surface. All three are already

View file

@ -3,18 +3,25 @@
//! `mind-self@<instance>`. This reads the process's arguments and environment, //! `mind-self@<instance>`. This reads the process's arguments and environment,
//! the things the library never reads, and prints the request lines and the //! 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 //! 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 //! a signal, 3 when she has no single open Self run (no child started), 4 when
//! when the mind cannot be reached or the arguments are wrong. Without //! the connector refused the run's credential lease (the run is closed, no child
//! `--endpoint` the daemon is `HUGINN_ENDPOINT`. //! 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 std::time::Duration;
use anyhow::Result; use anyhow::{Result, anyhow};
use eureka_state::HuginnClient; use eureka_state::HuginnClient;
use mind_body::cli::instance_and_endpoint; use mind_body::cli::instance_and_endpoint;
use mind_body::launch::SystemClock; use mind_body::launch::SystemClock;
use mind_body::lease_client::{LEASE_SOCKET_VARIABLE, SocketLender};
use mind_body::queue::HuginnMind; 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. /// How long one call to the mind may take.
const TIMEOUT: Duration = Duration::from_secs(15); 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 // 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). // 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> { fn run() -> Result<i32> {
let args: Vec<String> = std::env::args().skip(1).collect(); 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 (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 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}"); println!("{outcome}");
Ok(outcome.exit_status()) Ok(outcome.exit_status())
} }

View file

@ -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 /// 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 /// a unit that keeps dying counts as an empty run and trips the breaker.
/// sequence is the run's next, counting a closure that was itself withdrawn.
fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime<Utc>) -> Result<PipelineDocument> { 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 closures = Selection { cites: Some(citing(run, Some("subject"))), ..of_kind(PipelineKind::Resolution) };
let latest = headers(mind, &closures)? let latest = headers(mind, &closures)?
.iter() .iter()
@ -290,8 +299,8 @@ fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime<Utc>) -> Res
Ok(PipelineDocument::Resolution(PipelineResolution { Ok(PipelineDocument::Resolution(PipelineResolution {
subject: run.clone(), subject: run.clone(),
sequence: latest + 1, sequence: latest + 1,
outcome: ResolutionOutcome::Recorded { reason: "unit not active".into() }, outcome,
rationale: "The run's unit was not active when its turn's slot was next wanted.".into(), rationale: rationale.into(),
resolved_on: date(now), resolved_on: date(now),
})) }))
} }

View file

@ -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(())
}
}

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View file

@ -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();
}
}

View file

@ -16,7 +16,8 @@
//! the chat wire and port, `chat_rudp` its RUDP transport to Bifrost's mouth, and //! the chat wire and port, `chat_rudp` its RUDP transport to Bifrost's mouth, and
//! `connector` the model transport over ModelConnector's kit client). //! `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 //! `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. //! 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 connector;
pub mod control; pub mod control;
pub mod launch; pub mod launch;
pub mod lease_client;
pub mod persona; pub mod persona;
pub mod queue; pub mod queue;
pub mod self_run; pub mod self_run;

View file

@ -17,27 +17,43 @@
//! ruling `eureka-body:ruling:self-run-decider`). `carry` is the loop that turns //! 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 operating system's facts into events and executes the actions.
//! //!
//! The actuator admits nothing, reads neither the brake nor the environment and //! The child's credential is a lease, not a variable (rulings
//! stops no unit: the gate and the brake hold own those. It never echoes an //! `eureka-body:ruling:self-runs-through-connector` and
//! argument, an endpoint or a stream line. Stream-json is Claude Code's wire, a //! `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 //! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing
//! parsed is stored. //! parsed is stored.
use std::collections::{BTreeMap, BTreeSet};
use std::ffi::{OsStr, OsString};
use std::fmt; use std::fmt;
use std::io::{BufRead, BufReader, Write}; use std::io::{BufRead, BufReader, Write};
use std::path::Path; use std::path::{Path, PathBuf};
use std::process::{ChildStdin, Command, Stdio}; use std::process::{ChildStdin, Command, Stdio};
use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; use std::sync::mpsc::{Receiver, RecvTimeoutError, channel};
use std::time::Duration as StdDuration; use std::time::Duration as StdDuration;
use anyhow::{Context, Result, anyhow, bail}; use anyhow::{Context, Result, anyhow, bail};
use chrono::{DateTime, Utc};
use cultnet_rs::Selection; use cultnet_rs::Selection;
use eureka_pipeline::{PipelineDocument, PipelineRef, RunTurn, Slug}; use eureka_pipeline::{PipelineDocument, PipelineRef, ResolutionOutcome, RunTurn, Slug};
use huginn_mind::PipelinePageItems; use huginn_mind::{PipelineAdmissionOutcome, PipelinePageItems};
use serde_json::{Value, json}; use serde_json::{Value, json};
use uuid::Uuid; 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 crate::queue::MindPort;
use decide::{Action, Event, State, decide, next_deadline}; 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; pub const EXIT_DRAIN_S: i64 = 2;
/// A backstop `claude` enforces; the ceiling is reached first. /// A backstop `claude` enforces; the ceiling is reached first.
pub const MAX_TURNS: u32 = 500; 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. /// The name the start message tells her to admit as.
pub const AGENT: &str = "mind-self-run"; pub const AGENT: &str = "mind-self-run";
/// The namespace of the session id: the run id is the name in it. /// 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 /// 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. /// `budget_usd` is the run's, as stored.
pub fn claude_args(session: Uuid, budget_usd: &str) -> Vec<String> { 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(), &MAX_TURNS.to_string(),
"--permission-prompts", "--permission-prompts",
"none", "none",
"--setting-sources",
"user",
] ]
.map(String::from) .map(String::from)
.to_vec() .to_vec()
@ -178,7 +215,8 @@ impl Stop {
} }
pub trait Spawner { 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. /// Why a run ended.
@ -199,15 +237,18 @@ pub enum Cause {
pub enum SelfRunOutcome { pub enum SelfRunOutcome {
/// She has not exactly one open Self run: nothing was started. /// She has not exactly one open Self run: nothing was started.
NoRun, 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 }, Ended { cause: Cause, code: Option<i32>, requests: u64, prefix: u64 },
} }
impl SelfRunOutcome { impl SelfRunOutcome {
/// 0 when the child exited 0, 3 when no child was started, else 1 (a /// 0 when the child exited 0, 3 when no child was started, 4 when the
/// child killed by a signal has no code). /// connector refused the lease, else 1 (a child killed by a signal has no code).
pub fn exit_status(&self) -> i32 { pub fn exit_status(&self) -> i32 {
match self { match self {
Self::NoRun => 3, Self::NoRun => 3,
Self::LeaseRefused { .. } => 4,
Self::Ended { code: Some(0), .. } => 0, Self::Ended { code: Some(0), .. } => 0,
Self::Ended { .. } => 1, Self::Ended { .. } => 1,
} }
@ -218,6 +259,7 @@ impl fmt::Display for SelfRunOutcome {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self { match self {
Self::NoRun => write!(f, "no run: she has no single open Self run"), 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 } => { Self::Ended { cause, code, requests, prefix } => {
let cause = match cause { let cause = match cause {
Cause::Own => "own", 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 /// What a run used, from its stream. `claude`'s own account comes first: a
/// its end. The request lines go to `out`. A mind that cannot be reached or a /// result record's `modelUsage` is its cumulative per-model total as of that
/// child that cannot be started is an error; every other end is an outcome. /// result, so the last result that has one is the run's usage up to then, plus
/// This is only the loop: facts in, one clock read per fact, actions out. /// the tokens of every assistant message after it (the flush turn, or a turn the
pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, spawner: &dyn Spawner, out: &mut dyn Write) -> Result<SelfRunOutcome> { /// 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 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(); 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 state = State::default();
let mut now = clock.now(); let mut now = clock.now();
loop { loop {
let wait = next_deadline(&state).map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL)); let wait = next_deadline(&state).map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL));
let fact = match child.next(wait) { let fact = match child.next(wait) {
Read::Line(line) => match read_usage(&line) { Read::Line(line) => {
Usage::Request { id, prefix } => Event::Request { id, prefix }, spent.hear(&line);
Usage::Result => Event::Result, match read_usage(&line) {
Usage::Other | Usage::Unreadable => Event::Tick, Usage::Request { id, prefix } => Event::Request { id, prefix },
}, Usage::Result => Event::Result,
Usage::Other | Usage::Unreadable => Event::Tick,
}
}
Read::Timeout => Event::Tick, Read::Timeout => Event::Tick,
Read::Eof => Event::StdoutEof, Read::Eof => Event::StdoutEof,
Read::Exited(code) => Event::ChildExited(code), 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 /// The real spawner: the `claude` on the unit's `PATH`, its stderr in the
/// environment (the library names no variable), its stderr in the journal. /// journal. Its credential is the lease's descriptor, named in the one variable
pub struct ClaudeSpawner; /// 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 { impl ClaudeSpawner {
fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>> { /// From the unit's environment: it must name [`SCRATCH_VARIABLE`], and may name [`SEED_VARIABLE`].
spawn_program(Path::new("claude"), args) 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); let mut command = Command::new(program);
command.args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit()); 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. // Its own process group, so that a signal reaches what the child started too.
#[cfg(unix)] #[cfg(unix)]
std::os::unix::process::CommandExt::process_group(&mut command, 0); std::os::unix::process::CommandExt::process_group(&mut command, 0);
let mut process = command.spawn().context("claude could not be started")?; let mut process = command.spawn().context("claude could not be started")?;
drop(inheritable);
let group = process.id(); let group = process.id();
let stdin = process.stdin.take(); let stdin = process.stdin.take();
let stdout = process.stdout.take().ok_or_else(|| anyhow!("claude has no stdout"))?; 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 || { std::thread::spawn(move || {
let _ = sender.send(Read::Exited(process.wait().ok().and_then(|status| status.code()))); 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 { struct ClaudeProcess {
@ -336,6 +627,8 @@ struct ClaudeProcess {
facts: Receiver<Read>, facts: Receiver<Read>,
/// The child's pid, which is its process group's id. /// The child's pid, which is its process group's id.
group: u32, group: u32,
/// Its configuration, removed when the child's handle goes.
_config: FreshConfig,
} }
impl Child for ClaudeProcess { impl Child for ClaudeProcess {

View file

@ -13,8 +13,12 @@ use eureka_pipeline::{PipelineDocument, PipelineKind as K, PipelineRef, RunOpera
use huginn_mind::{PipelineAdmissionOutcome, PipelineSelectionPage}; use huginn_mind::{PipelineAdmissionOutcome, PipelineSelectionPage};
use super::*; use super::*;
use crate::lease_client::{Lease, LeaseRefused, token_holding};
use crate::testkit::*; 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 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"); 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. /// A signal was attempted, delivered or not.
Stopped(Stop), Stopped(Stop),
Asked(StdDuration), 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)] #[derive(Clone)]
@ -171,12 +183,84 @@ struct Fake {
} }
impl Spawner for 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()); self.spawns.borrow_mut().push(args.to_vec());
if self.refuse { if self.refuse {
return Err(anyhow!("refused")); 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>, outcome: Result<SelfRunOutcome>,
out: String, out: String,
events: Vec<Event>, events: Vec<Event>,
/// Every event, the lease's included.
all: Vec<Event>,
spawns: Vec<Vec<String>>, spawns: Vec<Vec<String>>,
admits: usize, admits: usize,
asked: Vec<LeaseRequest>,
} }
impl Ran { impl Ran {
@ -248,6 +335,10 @@ impl Ran {
.collect() .collect()
} }
fn count_all(&self, wanted: &Event) -> usize {
self.all.iter().filter(|event| *event == wanted).count()
}
fn count(&self, wanted: &Event) -> usize { fn count(&self, wanted: &Event) -> usize {
self.events.iter().filter(|event| *event == wanted).count() 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()), spawns: RefCell::new(Vec::new()),
refuse: false, refuse: false,
}; };
let lender = FakeLender::logging(log.clone(), Answer::Grant);
let mut out = Vec::new(); let mut out = Vec::new();
let outcome = carry(seen, &*clock, &fake, &mut out); let outcome = carry(seen, &*clock, &lender, &fake, TEST_CAP_S, &mut out);
let events = log.borrow().clone(); 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(); 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> { 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 mind = with_run();
let clock = Rc::new(Fixed(Cell::new(at_start()))); let clock = Rc::new(Fixed(Cell::new(at_start())));
let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true }; 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"); assert_eq!(refusing.spawns.borrow().len(), 1, "it tried exactly once");
struct Down; 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 }; 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()); 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", "500",
"--permission-prompts", "--permission-prompts",
"none", "none",
"--setting-sources",
"user",
] ]
); );
// Those belong to the run measure or to the unit's managed settings. // 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); let ran = go(steps);
assert_eq!( assert_eq!(
ran.out, 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(); 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"); 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 // 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. // 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.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!(jump.ended().0, Cause::Signalled(Stop::Interrupt));
assert_eq!(flushes(&jump), 1); 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 ran = go(steps);
let (_, _, requests, prefix) = ran.ended(); let (_, _, requests, prefix) = ran.ended();
assert_eq!((requests, prefix), (12, 1_000)); 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] #[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 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 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 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 events = log.borrow().clone();
let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32; 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. // One read for the lease's expiry, one before the first wait, then one per fact heard:
assert_eq!(clock.1.get(), 1 + reads, "{events:?}"); // 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 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. // 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(); 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(at("m1", 10)),
Out(result_line()), 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(); let outcome = ran.outcome.as_ref().unwrap().to_string();
assert!(!ran.out.contains(canary) && !outcome.contains(canary)); assert!(!ran.out.contains(canary) && !outcome.contains(canary));
assert!(!ran.written().iter().any(|text| text.contains(canary)), "nothing from the stream is written back"); 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)] #[cfg(unix)]
mod process { mod process {
mod lease_child;
use std::os::unix::fs::PermissionsExt; use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf; use std::path::PathBuf;
@ -865,8 +964,25 @@ mod process {
/// Spawns it, retrying the one error a freshly written executable can give /// Spawns it, retrying the one error a freshly written executable can give
/// while another test's fork still holds its write handle. /// while another test's fork still holds its write handle.
fn start(program: &Path, args: &[String]) -> Box<dyn Child> { 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 { 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, Ok(child) => return child,
Err(error) if format!("{error:#}").contains("busy") => std::thread::sleep(StdDuration::from_millis(100)), Err(error) if format!("{error:#}").contains("busy") => std::thread::sleep(StdDuration::from_millis(100)),
Err(error) => panic!("{error:#}"), Err(error) => panic!("{error:#}"),
@ -934,7 +1050,9 @@ mod process {
#[test] #[test]
fn a_program_that_cannot_start_is_an_error_that_names_no_argument() { fn a_program_that_cannot_start_is_an_error_that_names_no_argument() {
let canary = "canary-argument-6b0d"; 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:#}"); assert!(!format!("{error:#}").contains(canary) && !format!("{error:#}").contains("nonexistent"), "{error:#}");
} }
@ -961,8 +1079,8 @@ mod process {
struct Spawn(PathBuf); struct Spawn(PathBuf);
impl Spawner for Spawn { impl Spawner for Spawn {
fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>> { fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
Ok(Box::new(Prompt(start(&self.0, args)))) Ok(Box::new(Prompt(start_with(&self.0, args, token))))
} }
} }
@ -987,8 +1105,8 @@ mod process {
struct Patient(PathBuf); struct Patient(PathBuf);
impl Spawner for Patient { impl Spawner for Patient {
fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>> { fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
Ok(Box::new(Bounded(start(&self.0, args), std::time::Instant::now()))) 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 program = script(&dir, &body);
let mind = with_run(); let mind = with_run();
let mut out = Vec::new(); 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!(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] #[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 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 started = std::time::Instant::now();
let mut out = Vec::new(); 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, SelfRunOutcome::Ended { cause: Cause::Own, code: Some(1), requests: 0, prefix: 0 });
assert_eq!(outcome.exit_status(), 1); 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"); 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 program = script(&dir, &format!("{before}cat <<'EOF'\n{result}\nEOF\nsleep 30 &\necho $! > {}\nexit 0", pids.display()));
let started = std::time::Instant::now(); let started = std::time::Instant::now();
let mut out = Vec::new(); 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 holder = std::fs::read_to_string(&pids).unwrap().trim().to_string();
let cause = if flushed { Cause::FlushRequested } else { Cause::Own }; let cause = if flushed { Cause::FlushRequested } else { Cause::Own };
let (requests, prefix) = if flushed { (1, CEILING_TOKENS) } else { (0, 0) }; let (requests, prefix) = if flushed { (1, CEILING_TOKENS) } else { (0, 0) };
@ -1085,3 +1203,5 @@ mod process {
} }
} }
} }
mod lease;

View file

@ -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);
}

View file

@ -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);
}

View file

@ -4,6 +4,7 @@ mind_body: mod cli
mind_body: mod connector mind_body: mod connector
mind_body: mod control mind_body: mod control
mind_body: mod launch mind_body: mod launch
mind_body: mod lease_client
mind_body: mod persona mind_body: mod persona
mind_body: mod queue mind_body: mod queue
mind_body: mod self_run 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 SystemClock pub-fields[]
mind_body::launch: struct SystemctlReply pub-fields[success: bool, code: Option < i32 >, stdout: String] mind_body::launch: struct SystemctlReply pub-fields[success: bool, code: Option < i32 >, stdout: String]
mind_body::launch: struct SystemdLauncher pub-fields[] 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 (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: # [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 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 , 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 , 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 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: # [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 AGENT : & str
mind_body::self_run: const CEILING_TOKENS : u64 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_HEADROOM_TOKENS : u64
mind_body::self_run: const FLUSH_MESSAGE : & str mind_body::self_run: const FLUSH_MESSAGE : & str
mind_body::self_run: const MAX_TURNS : u32 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: 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 claude_args (session : Uuid , budget_usd : & str) -> Vec < String >
mind_body::self_run: fn read_usage (line : & str) -> Usage 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 session_id (run : & PipelineRef) -> Uuid
mind_body::self_run: fn start_message (instance : & Slug , run : & PipelineRef) -> String 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 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
mind_body::self_run: impl SelfRunOutcome: fn exit_status (& self) -> i32 mind_body::self_run: impl SelfRunOutcome: fn exit_status (& self) -> i32
mind_body::self_run: impl Spawner for ClaudeSpawner 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 Stop
mind_body::self_run: impl fmt :: Display for SelfRunOutcome 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 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::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: # [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 mind_body::wake: const AGENT : & str

View file

@ -25,6 +25,27 @@ fn all_output(output: &Output) -> String {
format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)) 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] #[test]
fn without_an_instance_or_with_a_stray_argument_it_exits_2_with_the_usage_and_echoes_nothing() { fn without_an_instance_or_with_a_stray_argument_it_exits_2_with_the_usage_and_echoes_nothing() {
let canary = "canary-argument-3f9a"; 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)] #[cfg(unix)]
mod lane { mod lane {
use std::io::{IoSlice, Read, Write};
use std::mem::MaybeUninit;
use std::net::UdpSocket; use std::net::UdpSocket;
use std::os::unix::net::UnixListener;
use std::sync::{Arc, Mutex};
use std::path::Path; use std::path::Path;
use std::process::{Child, Command, Output, Stdio}; use std::process::{Child, Command, Output, Stdio};
use std::time::Duration; 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 /// 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. /// 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 exe = Path::new(env!("CARGO_BIN_EXE_mind-self-run"));
let path = format!("{}:/usr/bin:/bin", bin.display()); let path = format!("{}:/usr/bin:/bin", bin.display());
for _ in 0..40 { for _ in 0..40 {
@ -196,6 +221,9 @@ mod lane {
.args(["--instance", INSTANCE, "--endpoint", endpoint]) .args(["--instance", INSTANCE, "--endpoint", endpoint])
.env("PATH", &path) .env("PATH", &path)
.env("SELF_RUN_ENV_CANARY", ENV_CANARY) .env("SELF_RUN_ENV_CANARY", ENV_CANARY)
.env("CLAUDE_LEASE_SOCKET", socket)
.env("MIND_SELF_RUN_CAP_S", "14400")
.env("RUNTIME_DIRECTORY", bin)
.output() .output()
.unwrap(); .unwrap();
if attempt.status.code() != Some(2) { if attempt.status.code() != Some(2) {
@ -206,6 +234,61 @@ mod lane {
panic!("the daemon never answered"); 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"; const ENV_CANARY: &str = "canary-unit-environment-b94e";
/// Writes an executable `claude` into `bin` and returns the path of the record it keeps. /// Writes an executable `claude` into `bin` and returns the path of the record it keeps.
@ -237,20 +320,22 @@ mod lane {
) )
}; };
let body = format!( let body = format!(
"echo journal-line >&2 "read -r -u \"$CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR\" cred
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", 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), first = request("msg_1", 100),
second = request("msg_2", 299_999), second = request("msg_2", 299_999),
record = bin.path().join("record").display(), record = bin.path().join("record").display(),
); );
let record = fake_claude(bin.path(), &body); 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); drop(daemon);
let stdout = String::from_utf8_lossy(&output.stdout).to_string(); 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!(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_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's environment is the child's, not the bin's output"); 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"); 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(); 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"); assert!(lines.next().unwrap().contains("yggdrasil:run:selfrun-lane"), "the start message names her run");
let flush = lines.next().unwrap(); let flush = lines.next().unwrap();
assert!(flush.contains(FLUSH_MESSAGE) && flush.contains("\"type\":\"user\""), "the flush went to the child's stdin: {flush}"); 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. /// 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 (daemon, _state, endpoint) = serve(documents);
let bin = TempDir::new().unwrap(); let bin = TempDir::new().unwrap();
let record = fake_claude(bin.path(), "exit 0"); 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); drop(daemon);
assert_eq!(output.status.code(), Some(3), "{}", all_output(&output)); 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"); assert_eq!(String::from_utf8_lossy(&output.stdout), "no run: she has no single open Self run\n");

View file

@ -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. | | 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. | | 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`). | | 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. | | 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). | | 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@`. | | 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. | | 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. | | 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`. | | 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. | | `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. 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 Waits on nothing but A2's message contract (the two prompt constants in
`cut-mind-self-run.r1`). `cut-mind-self-run.r1`).
5. `self-run-environment` (gamecult-ops): the `mind-self@` lines A2 leaves to ops (token 5. `self-run-environment` (gamecult-ops): the `mind-self@` lines A2 leaves to ops (`StateDirectory`, `WorkingDirectory`, `RuntimeDirectory`,
`EnvironmentFile`, `StateDirectory`, `WorkingDirectory`, `CLAUDE_CONFIG_DIR`, `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
`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 `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 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 `question:self-run-permissions`, and the connector's `--lease-max-ms` at least the cap plus
(`claude setup-token`, ruling `self-run-token-reach`). 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 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. `open_and_launch`; the Persona's start-Self tool later, the operator's bring-up now.
Waits on question `question:first-self-run-starter`. Waits on question `question:first-self-run-starter`.