GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

wip persona-session (local, does not build tests)

Browse source
This commit is contained in:
Meta Vens 2026-11-01 11:54:25 +01:00
commit ee0bf1201d
11 changed files with 5330 additions and 5092 deletions
Download patch file Download diff file Expand all files Collapse all files

157
crates/mind-body/Cargo.toml
View file

@ -1,80 +1,77 @@
[package]
name = "mind-body"
version = "0.1.0"
edition.workspace = true
license.workspace = true
publish.workspace = true
# One crate for her organs (rationale "One crate for her organs"). It holds the
# operator's brakes: a root-owned per-instance CultCache
# store (`control`), the operator's CLI over it (`cli`, the `mind-control`
# binary) and the one fail-closed reader every acting organ calls
# (`control::read_effective`). It also holds the launch primitive (`launch`,
# `queue`): the one way her organs open a run and start its unit. The heartbeat
# (`heartbeat`, the `mind-heartbeat` binary) runs the shared heartbeat over them. The actuator
# (`self_run`, the `mind-self-run` binary) carries her one open Self run.
#
# The Persona organ (`persona`, the `mind-persona` binary) drains her pending chat
# through ports. The connector transport is ModelConnector's kit client, pinned to
# the kit's revision (`connector`); the chat transport is cultnet-rs's RUDP client
# to Bifrost's mouth (`chat_rudp`). `rmp-serde` is the chat documents' positional
# MessagePack codec (`chat_client`); `sha2` derives the mouth's connection id from
# the chat credential. The kit pins its own CultLib revision, so the graph holds
# two until the workspace pin and the kit's meet.
# `rmp-serde` is also the lease wire (`lease_client`): the connector's MessagePack frames.
#
# `serde_json` reads Claude Code's stream-json, a foreign owner's wire (`self_run`);
# `uuid`'s v5 makes a Self run's session id derivable from the run's id.
[dependencies]
anyhow = "1"
chrono = { version = "0.4.44", features = ["serde"] }
cultcache-rs.workspace = true
cultnet-rs.workspace = true
eureka-pipeline = { path = "../eureka-pipeline" }
eureka-state = { path = "../eureka-state", default-features = false }
huginn-mind = { path = "../huginn-mind" }
model-connector = { git = "https://eureka.gamecult.org/forge/GameCult/ModelConnector.git", rev = "c1802d85bee3ae04b0c5737563c8343c1fef515d", package = "model-connector" }
rmp-serde = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha2 = "0.10"
uuid = { version = "1", default-features = false, features = ["v5"] }
[target.'cfg(unix)'.dependencies]
libc = "0.2"
# The lease socket: SCM_RIGHTS receive with MSG_CMSG_CLOEXEC and the descriptor
# duplicates, as safe calls (the library forbids unsafe code). Already in the
# graph through tempfile; this adds its `net` feature only.
rustix = { version = "1", default-features = false, features = ["std", "net"] }
# `syn`, `quote` and `proc-macro2` read the library's own source for
# tests/public_surface.rs, which pins its public surface. All three are already
# in the workspace's graph (serde's derive uses them); they are test-only.
[dev-dependencies]
# The kit's serve loop and service, for the test that runs the connector client
# against the real server with a fake provider behind it.
model-connector = { git = "https://eureka.gamecult.org/forge/GameCult/ModelConnector.git", rev = "c1802d85bee3ae04b0c5737563c8343c1fef515d", package = "model-connector", features = ["daemon"] }
proc-macro2 = "1"
quote = "1"
syn = { version = "2", features = ["full"] }
tempfile = "3"
[lib]
name = "mind_body"
path = "src/lib.rs"
[[bin]]
name = "mind-control"
path = "src/main.rs"
[[bin]]
name = "mind-self-run"
path = "src/bin/mind-self-run.rs"
[[bin]]
name = "mind-persona"
path = "src/bin/mind-persona.rs"
[[bin]]
name = "mind-heartbeat"
path = "src/bin/mind-heartbeat.rs"
[package]
name = "mind-body"
version = "0.1.0"
edition.workspace = true
license.workspace = true
publish.workspace = true
# One crate for her organs (rationale "One crate for her organs"). It holds the
# operator's brakes: a root-owned per-instance CultCache
# store (`control`), the operator's CLI over it (`cli`, the `mind-control`
# binary) and the one fail-closed reader every acting organ calls
# (`control::read_effective`). It also holds the launch primitive (`launch`,
# `queue`): the one way her organs open a run and start its unit. The heartbeat
# (`heartbeat`, the `mind-heartbeat` binary) runs the shared heartbeat over them. The actuator
# (`self_run`, the `mind-self-run` binary) carries her one open Self run.
#
# The Persona organ (`persona`, the `mind-persona` binary) drains her pending chat
# through ports. The connector transport is ModelConnector's kit client, pinned to
# the kit's revision (`connector`); the chat transport is cultnet-rs's RUDP client
# to Bifrost's mouth (`chat_rudp`). `rmp-serde` is the chat documents' positional
# MessagePack codec (`chat_client`); `sha2` derives the mouth's connection id from
# the chat credential. The kit pins its own CultLib revision, so the graph holds
# two until the workspace pin and the kit's meet.
# `rmp-serde` is also the lease wire (`lease_client`): the connector's MessagePack frames.
#
# `serde_json` reads Claude Code's stream-json, a foreign owner's wire (`self_run`);
# `uuid`'s v5 makes a Self run's session id derivable from the run's id.
[dependencies]
anyhow = "1"
chrono = { version = "0.4.44", features = ["serde"] }
cultcache-rs.workspace = true
cultnet-rs.workspace = true
eureka-pipeline = { path = "../eureka-pipeline" }
eureka-state = { path = "../eureka-state", default-features = false }
huginn-mind = { path = "../huginn-mind" }
model-connector = { git = "https://eureka.gamecult.org/forge/GameCult/ModelConnector.git", rev = "c1802d85bee3ae04b0c5737563c8343c1fef515d", package = "model-connector" }
rmp-serde = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha2 = "0.10"
uuid = { version = "1", default-features = false, features = ["v5"] }
[target.'cfg(unix)'.dependencies]
libc = "0.2"
# The lease socket: SCM_RIGHTS receive with MSG_CMSG_CLOEXEC and the descriptor
# duplicates, as safe calls (the library forbids unsafe code). Already in the
# graph through tempfile; this adds its `net` feature only.
rustix = { version = "1", default-features = false, features = ["std", "net"] }
# `syn`, `quote` and `proc-macro2` read the library's own source for
# tests/public_surface.rs, which pins its public surface. All three are already
# in the workspace's graph (serde's derive uses them); they are test-only.
[dev-dependencies]
proc-macro2 = "1"
quote = "1"
syn = { version = "2", features = ["full"] }
tempfile = "3"
[lib]
name = "mind_body"
path = "src/lib.rs"
[[bin]]
name = "mind-control"
path = "src/main.rs"
[[bin]]
name = "mind-self-run"
path = "src/bin/mind-self-run.rs"
[[bin]]
name = "mind-persona"
path = "src/bin/mind-persona.rs"
[[bin]]
name = "mind-heartbeat"
path = "src/bin/mind-heartbeat.rs"

202
crates/mind-body/src/bin/mind-persona.rs
View file

@ -1,97 +1,105 @@
//! `mind-persona --instance <slug> [--endpoint rudp://<ip>:<port>] [--max-turns n]`:
//! one start of the Persona organ (`mind_body::persona::start`) inside
//! `mind-persona@<instance>`. This reads the process's arguments and
//! environment, the things the library never reads, and prints the outcome. It
//! exits 0 for every outcome, including a unit that is dead because the brake is
//! held or the dial out of bounds, and 2 when the mind cannot be reached, the
//! arguments or the environment are wrong or a transport is not configured.
//!
//! The unit's `persona.env` and credentials name everything else:
//! `MIND_PERSONA_ROOMS` (a,b), `MIND_PERSONA_OPERATOR`
//! (id[,id]), `MIND_PERSONA_MINDS` (id[,id]: every Mind of the swarm, required) and `MIND_PERSONA_SELF_SENDER`, each also a flag that overrides it
//! (`--rooms`, `--operator`, `--minds`, `--self-sender`), and `MIND_HEARTBEAT_STATE` (optional: the path of the
//! heartbeat's `heartbeat.cc`, which the unit reads and never writes, for the tempo in her input); `MIND_PERSONA_DOCTRINE`
//! (optional, absolute paths separated by `:`) names the doctrine files read at each
//! turn start and put ahead of her persona;
//! `MIND_CONNECTOR_ENDPOINT` (`<ip>:<port>`, the connector's loopback listener),
//! `MIND_CONNECTOR_KEY_FILE` and `MIND_CONNECTOR_MODEL`; `BIFROST_CHAT_ENDPOINT`
//! (`rudp://<ip>:<port>`, the mouth) and `BIFROST_CHAT_KEY_FILE`. Without
//! `--endpoint` the daemon is `HUGINN_ENDPOINT`; the trace is `persona-trace.cc`
//! under `STATE_DIRECTORY`. Key files are read here and their text goes to the
//! transports, which hold it; nothing is printed.
use std::net::SocketAddr;
use std::path::Path;
use std::time::Duration;
use anyhow::{Context, Result, anyhow, bail};
use eureka_state::HuginnClient;
use mind_body::chat_rudp::{RudpChat, connection_id};
use mind_body::cli::instance_and_endpoint;
use mind_body::connector::{self, KitConnector};
use mind_body::control::{CONTROL_ROOT, FileSource, control_path, swarm_path};
use mind_body::launch::{SystemClock, SystemdLauncher};
use mind_body::heartbeat::FileStore;
use mind_body::persona::{Config, FileTrace, HuginnPersona, Organs, start};
use mind_body::queue::HuginnMind;
/// How long one call to the mind may take.
const TIMEOUT: Duration = Duration::from_secs(15);
/// How long the mouth may take to answer one chat request (a send is a Matrix PUT).
const CHAT_TIMEOUT: Duration = Duration::from_secs(20);
const USAGE: &str = "usage: mind-persona --instance <slug> [--endpoint rudp://<ip>:<port>] [--max-turns n] [--rooms a,b] [--operator id[,id]] [--minds id[,id]] [--self-sender id]";
fn host() -> String {
std::fs::read_to_string("/proc/sys/kernel/hostname").map(|name| name.trim().to_string()).ok().filter(|name| !name.is_empty()).unwrap_or_else(|| "unknown".into())
}
// The binary is the one place that reads the process: its arguments and
// the environment. The library reads neither (clippy.toml, `#![forbid]` in lib.rs).
#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")]
fn setting(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|value| !value.is_empty())
}
fn required(name: &str) -> Result<String> {
setting(name).ok_or_else(|| anyhow!("{name} is not set"))
}
/// The text of the credential file named by `name`; never echoed.
fn credential(name: &str) -> Result<String> {
std::fs::read_to_string(required(name)?).with_context(|| format!("the file named by {name} could not be read"))
}
#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")]
fn run() -> Result<String> {
let args: Vec<String> = std::env::args().skip(1).collect();
let (config, rest) = Config::from_args_and_env(&args, setting, USAGE)?;
let (instance, address) = instance_and_endpoint(&rest, setting("HUGINN_ENDPOINT"), USAGE)?;
let Some(state) = setting("STATE_DIRECTORY") else { bail!("no STATE_DIRECTORY") };
let chat_endpoint: SocketAddr = required("BIFROST_CHAT_ENDPOINT")?
.strip_prefix("rudp://")
.and_then(|address| address.parse().ok())
.ok_or_else(|| anyhow!("BIFROST_CHAT_ENDPOINT is not rudp://<ip>:<port>"))?;
let chat = RudpChat::new(chat_endpoint, connection_id(&credential("BIFROST_CHAT_KEY_FILE")?), instance.0.clone(), CHAT_TIMEOUT);
let connector = KitConnector::new(connector::endpoint(&required("MIND_CONNECTOR_ENDPOINT")?).context("MIND_CONNECTOR_ENDPOINT must be <ip>:<port> or tcp://<ip>:<port>")?, &credential("MIND_CONNECTOR_KEY_FILE")?, required("MIND_CONNECTOR_MODEL")?, instance.0.clone())?;
let source = FileSource::new(control_path(Path::new(CONTROL_ROOT), &instance)?);
let mind = HuginnMind::new(HuginnClient::new(address, instance.clone(), TIMEOUT));
let persona = HuginnPersona::new(HuginnClient::new(address, instance, TIMEOUT));
let trace = FileTrace::new(Path::new(&state).join("persona-trace.cc"));
let launcher = SystemdLauncher::new();
let host = host();
let tempo = setting("MIND_HEARTBEAT_STATE").map(FileStore::new);
let organs = Organs { mind: &mind, launcher: &launcher, clock: &SystemClock, host: &host, chat: &chat, connector: &connector, persona: &persona, trace: &trace, tempo: tempo.as_ref().map(|store| store as &dyn mind_body::heartbeat::TempoView) };
let swarm = FileSource::new(swarm_path(Path::new(CONTROL_ROOT)));
Ok(start(&organs, &source, &swarm, &config)?.to_string())
}
fn main() {
match run() {
Ok(outcome) => println!("{outcome}"),
Err(error) => {
eprintln!("mind-persona: {error:#}");
std::process::exit(2);
}
}
}
//! `mind-persona --instance <slug> [--endpoint rudp://<ip>:<port>]`:
//! one start of the Persona organ (`mind_body::persona::carry`) inside
//! `mind-persona@<instance>`. This reads the process's arguments and
//! environment, the things the library never reads, and prints the outcome. It
//! exits 0 for every outcome, including a unit that is dead because the brake is
//! held or the dial out of bounds, and 2 when the mind cannot be reached, the
//! arguments or the environment are wrong or a transport is not configured.
//!
//! The unit's `persona.env` and credentials name everything else:
//! `MIND_PERSONA_ROOMS` (a,b), `MIND_PERSONA_OPERATOR`
//! (id[,id]), `MIND_PERSONA_MINDS` (id[,id]: every Mind of the swarm, required) and `MIND_PERSONA_SELF_SENDER`, each also a flag that overrides it
//! (`--rooms`, `--operator`, `--minds`, `--self-sender`), and `MIND_HEARTBEAT_STATE` (optional: the path of the
//! heartbeat's `heartbeat.cc`, which the unit reads and never writes, for the tempo in her input); `MIND_PERSONA_DOCTRINE`
//! (optional, absolute paths separated by `:`) names the doctrine files read at each
//! process start and put ahead of her persona; `MIND_PERSONA_MODEL` is the model her session runs;
//! `MIND_PERSONA_RUN_CAP_S` is the unit's `RuntimeMaxSec` in seconds, the one length of a start;
//! `MIND_PERSONA_CONFIG_SEED` optionally names a root-owned directory copied into each
//! process's fresh configuration; `CLAUDE_LEASE_SOCKET` is the connector's lease socket (the token itself
//! is never in this process's environment, only the descriptor a lease grants); `RUNTIME_DIRECTORY` (the
//! unit's `RuntimeDirectory=`) holds each process's configuration and instructions; `BIFROST_CHAT_ENDPOINT`
//! (`rudp://<ip>:<port>`, the mouth) and `BIFROST_CHAT_KEY_FILE`. Without
//! `--endpoint` the daemon is `HUGINN_ENDPOINT`; the trace is `persona-trace.cc`
//! under `STATE_DIRECTORY`, and her sandbox is `STATE_DIRECTORY/sandbox`. The key file is read here and its
//! text goes to the transport, which holds it; nothing is printed.
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, anyhow, bail};
use eureka_state::HuginnClient;
use mind_body::chat_rudp::{RudpChat, connection_id};
use mind_body::cli::instance_and_endpoint;
use mind_body::control::{CONTROL_ROOT, FileSource, control_path, swarm_path};
use mind_body::launch::{SystemClock, SystemdLauncher};
use mind_body::heartbeat::FileStore;
use mind_body::lease_client::{LEASE_SOCKET_VARIABLE, SocketLender};
use mind_body::persona::{Config, ENV_CONFIG_SEED, ENV_RUN_CAP, FileTrace, HuginnPersona, Organs, SANDBOX_DIR, carry};
use mind_body::queue::HuginnMind;
use mind_body::self_run::{ClaudeSpawner, SCRATCH_VARIABLE};
/// How long one call to the mind may take.
const TIMEOUT: Duration = Duration::from_secs(15);
/// How long the mouth may take to answer one chat request (a send is a Matrix PUT).
const CHAT_TIMEOUT: Duration = Duration::from_secs(20);
const USAGE: &str = "usage: mind-persona --instance <slug> [--endpoint rudp://<ip>:<port>] [--rooms a,b] [--operator id[,id]] [--minds id[,id]] [--self-sender id]";
fn host() -> String {
std::fs::read_to_string("/proc/sys/kernel/hostname").map(|name| name.trim().to_string()).ok().filter(|name| !name.is_empty()).unwrap_or_else(|| "unknown".into())
}
// The binary is the one place that reads the process: its arguments and
// the environment. The library reads neither (clippy.toml, `#![forbid]` in lib.rs).
#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")]
fn setting(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|value| !value.is_empty())
}
fn required(name: &str) -> Result<String> {
setting(name).ok_or_else(|| anyhow!("{name} is not set"))
}
/// The text of the credential file named by `name`; never echoed.
fn credential(name: &str) -> Result<String> {
std::fs::read_to_string(required(name)?).with_context(|| format!("the file named by {name} could not be read"))
}
#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")]
fn run() -> Result<String> {
let args: Vec<String> = std::env::args().skip(1).collect();
let (config, rest) = Config::from_args_and_env(&args, setting, USAGE)?;
let (instance, address) = instance_and_endpoint(&rest, setting("HUGINN_ENDPOINT"), USAGE)?;
let Some(state) = setting("STATE_DIRECTORY") else { bail!("no STATE_DIRECTORY") };
let chat_endpoint: SocketAddr = required("BIFROST_CHAT_ENDPOINT")?
.strip_prefix("rudp://")
.and_then(|address| address.parse().ok())
.ok_or_else(|| anyhow!("BIFROST_CHAT_ENDPOINT is not rudp://<ip>:<port>"))?;
let chat = RudpChat::new(chat_endpoint, connection_id(&credential("BIFROST_CHAT_KEY_FILE")?), instance.0.clone(), CHAT_TIMEOUT);
let run_cap_s: u64 = setting(ENV_RUN_CAP).and_then(|seconds| seconds.parse().ok()).filter(|seconds| *seconds > 0).ok_or_else(|| anyhow!("{ENV_RUN_CAP} is not set to the unit's RuntimeMaxSec in seconds"))?;
let scratch = PathBuf::from(required(SCRATCH_VARIABLE)?);
let sandbox = Path::new(&state).join(SANDBOX_DIR);
let lender = SocketLender::new(required(LEASE_SOCKET_VARIABLE)?);
let spawner = ClaudeSpawner::in_sandbox(std::env::vars_os(), ENV_CONFIG_SEED, Some(sandbox.clone()))?;
let source = FileSource::new(control_path(Path::new(CONTROL_ROOT), &instance)?);
let mind = HuginnMind::new(HuginnClient::new(address, instance.clone(), TIMEOUT));
let persona = HuginnPersona::new(HuginnClient::new(address, instance, TIMEOUT));
let trace = FileTrace::new(Path::new(&state).join("persona-trace.cc"));
let launcher = SystemdLauncher::new();
let host = host();
let tempo = setting("MIND_HEARTBEAT_STATE").map(FileStore::new);
let organs = Organs { mind: &mind, launcher: &launcher, clock: &SystemClock, host: &host, chat: &chat, persona: &persona, trace: &trace, tempo: tempo.as_ref().map(|store| store as &dyn mind_body::heartbeat::TempoView), lender: &lender, spawner: &spawner, run_cap_s, sandbox: &sandbox, scratch: &scratch };
let swarm = FileSource::new(swarm_path(Path::new(CONTROL_ROOT)));
Ok(carry(&organs, &source, &swarm, &config)?.to_string())
}
fn main() {
match run() {
Ok(outcome) => println!("{outcome}"),
Err(error) => {
eprintln!("mind-persona: {error:#}");
std::process::exit(2);
}
}
}

160
crates/mind-body/src/connector.rs
View file

@ -1,129 +1,31 @@
//! The Persona organ's model transport: ModelConnector's kit client.
//!
//! `KitConnector` is the `Connector` port over `model_connector::ConnectorClient`,
//! the wire-v4 client of the connector daemon (CultNet operation frames,
//! AES-GCM sealed, one request per connection). A turn is one tool-less
//! `ProviderRequest` for the instance as caller, its output constrained to the
//! turn's schema, and its answer the assistant text of a `Completed` result; a
//! refusal, a failure or a stop other than the end of the turn is an error. The
//! readout is the kit's usage readout (`gamecult.model.usage_readout.v1`): the
//! subscription plan's own windows as the connector last saw them, asked as this
//! client's own caller. The connector meters nothing in dollars.
//!
//! The connector reads its connection key as the file's text without the
//! trailing line break; the same rule applies here (`key`). The endpoint is the
//! connector's loopback TCP listener, `host:port` or `tcp://host:port`.
use std::net::SocketAddr;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use anyhow::{Result, anyhow, bail};
use model_connector::{
ConnectorClient, InputItem, PlanQuota, ProviderRequest, StopReason, TransportDisposition, TransportEventPayload, TransportInvocation, TransportOutcome,
provider_request_sha256,
};
use crate::persona::{Connector, TurnRequest, UsageReadout};
/// The largest frame either side of a connection may carry.
const MAX_FRAME_BYTES: usize = 4 * 1024 * 1024;
/// How long a connection waits for the connector: a turn runs a model.
const RESPONSE_TIMEOUT: Duration = Duration::from_secs(600);
/// How long an invocation may wait to be admitted.
const EXPIRES_IN_MS: u64 = 60_000;
/// The longest answer one turn may ask for.
pub const MAX_OUTPUT_TOKENS: u32 = 4096;
/// The connection key as the connector reads it: the file's text without its
/// trailing line break.
pub fn key(key_file_text: &str) -> String {
key_file_text.trim_end_matches(['\r', '\n']).to_string()
}
/// The connector's address from its configured endpoint: `host:port` or
/// `tcp://host:port`, an IP literal. A `rudp://` endpoint is refused: the
/// connector speaks framed TCP.
pub fn endpoint(text: &str) -> Result<SocketAddr> {
text.strip_prefix("tcp://").unwrap_or(text).parse::<SocketAddr>().map_err(|_| anyhow!("the connector endpoint is not <ip>:<port> or tcp://<ip>:<port>"))
}
/// `Connector` over the kit's client.
pub struct KitConnector {
client: ConnectorClient,
model: String,
caller: String,
}
impl KitConnector {
/// `key_text` is the connection key file's text (`key` is applied here);
/// `caller` is the caller id the key belongs to: the asker of every call.
pub fn new(endpoint: SocketAddr, key_text: &str, model: impl Into<String>, caller: impl Into<String>) -> Result<Self> {
let client = ConnectorClient::new(endpoint, key(key_text), MAX_FRAME_BYTES, Some(RESPONSE_TIMEOUT)).map_err(|_| anyhow!("the connector client could not be configured"))?;
Ok(Self { client, model: model.into(), caller: caller.into() })
}
}
fn now_ms() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |since| u64::try_from(since.as_millis()).unwrap_or(u64::MAX))
}
impl Connector for KitConnector {
fn turn(&self, request: &TurnRequest) -> Result<String> {
let mut provider = ProviderRequest::new(request.request_id.as_str(), request.caller.as_str(), self.model.as_str(), request.instructions.as_str(), MAX_OUTPUT_TOKENS);
provider.input = vec![InputItem::UserText { text: request.input.clone() }];
provider.output_schema_json = Some(request.output_schema_json.clone());
// The caller has no provider-native form of its request: the native digest is the request's own.
let digest = provider_request_sha256(&provider).map_err(|_| anyhow!("the connector request does not digest"))?;
let invocation = TransportInvocation::new(request.caller.as_str(), now_ms() + EXPIRES_IN_MS, digest, provider)
.map_err(|_| anyhow!("the connector invocation is not valid"))?;
let result = self.client.execute(&invocation).map_err(|error| anyhow!("the connector call failed: {error}"))?;
match result.disposition {
TransportDisposition::Refused(reason) => bail!("the connector refused the call: {reason:?}"),
TransportDisposition::Transported { events, receipt } => {
match &receipt.outcome {
TransportOutcome::Completed { stop_reason: StopReason::EndTurn, .. } => {}
TransportOutcome::Completed { stop_reason, .. } => bail!("the connector stopped the turn early: {stop_reason:?}"),
TransportOutcome::Failed { kind, .. } => bail!("the connector's provider failed: {kind:?}"),
}
let mut text = String::new();
for event in events {
if let TransportEventPayload::TextDelta { text: piece } = event.payload {
text.push_str(&piece);
}
}
Ok(text)
}
}
}
fn usage_readout(&self) -> Result<UsageReadout> {
self.client.usage_readout(&self.caller).map_err(|error| anyhow!("the usage readout failed: {error}"))
}
}
/// The plan windows as one line for a Mind: each window's percent used, its
/// status, when it resets and how old the reading is. Rendered from the readout
/// and nothing else, so no number in it is ever a paraphrase; the guidance that
/// follows it states the swarm's aim and that nothing enforces it.
pub fn render_plan(windows: &[PlanQuota], now_ms: u64) -> String {
let guidance = "The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.";
if windows.is_empty() {
return format!("Plan usage: unobserved. {guidance}");
}
let rendered: Vec<String> = windows
.iter()
.map(|plan| match plan.utilization_permille {
Some(permille) => format!(
"{} {}.{}% {}, resets at unix ms {}, read {}s ago",
plan.window,
permille / 10,
permille % 10,
plan.status,
plan.resets_at_unix_ms,
now_ms.saturating_sub(plan.observed_at_unix_ms) / 1000
),
None => format!("{} expired, resets at unix ms {}", plan.window, plan.resets_at_unix_ms),
})
.collect();
format!("Plan usage: {}. {guidance}", rendered.join("; "))
}
//! The plan windows as one line for a Mind: the one rendering both organs put in front of their
//! sessions, from the windows a credential lease granted (`lease_client`).
use model_connector::PlanQuota;
/// The plan windows as one line for a Mind: each window's percent used, its
/// status, when it resets and how old the reading is. Rendered from the readout
/// and nothing else, so no number in it is ever a paraphrase; the guidance that
/// follows it states the swarm's aim and that nothing enforces it.
pub fn render_plan(windows: &[PlanQuota], now_ms: u64) -> String {
let guidance = "The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.";
if windows.is_empty() {
return format!("Plan usage: unobserved. {guidance}");
}
let rendered: Vec<String> = windows
.iter()
.map(|plan| match plan.utilization_permille {
Some(permille) => format!(
"{} {}.{}% {}, resets at unix ms {}, read {}s ago",
plan.window,
permille / 10,
permille % 10,
plan.status,
plan.resets_at_unix_ms,
now_ms.saturating_sub(plan.observed_at_unix_ms) / 1000
),
None => format!("{} expired, resets at unix ms {}", plan.window, plan.resets_at_unix_ms),
})
.collect();
format!("Plan usage: {}. {guidance}", rendered.join("; "))
}

96
crates/mind-body/src/lib.rs
View file

@ -1,48 +1,48 @@
//! `mind-body`: the organs of one Mind's body that live in Huginn's workspace.
//!
//! Today that is `control`, the operator's brakes, and `cli`, the operator's
//! hand on them. The brakes are the operator's consent, not her body: the store
//! is root-owned and no organ of hers writes it.
//!
//! `launch` and `queue` are the launch primitive: the work queue, the
//! repetition breaker, the one-live-Self-run rule, and the single path that
//! opens a run in the mind and only then starts its unit.
//!
//! `heartbeat` runs the swarm's shared heartbeat over them: the one process,
//! the `mind-heartbeat` binary, that decides who wakes and when, by the rule in
//! `tempo` (the Minds' votes make the beat; the pick chooses who takes it).
//!
//! `persona` is the Persona organ: one start of `mind-persona@<instance>` drains her
//! pending chat through the connector, one typed reply per turn (`chat_client` is
//! the chat wire and port, `chat_rudp` its RUDP transport to Bifrost's mouth, and
//! `connector` the model transport over ModelConnector's kit client).
//!
//! `lease_client` asks the Claude connector for a Self run's credential and the
//! plan's usage windows. `self_run` carries one Self run inside `mind-self@<instance>`: the
//! `mind-self-run` binary runs `claude` over stream-json for her one open Self
//! run, asks for the flush at the context ceiling and stops a child that outlives it.
// The control root is a constant, so the library reads no part of the process
// environment (clippy.toml lists the readers); `forbid` keeps an `allow` from
// putting one back. `forbid(unsafe_code)` closes the readers the list cannot
// name: a foreign `getenv` or any other FFI call needs `unsafe`. The rest of the
// Grant rule is ruling `eureka-body:ruling:mind-launch-forge-scanner-bound`.
#![forbid(clippy::disallowed_methods)]
#![forbid(unsafe_code)]
pub mod chat_client;
pub mod chat_rudp;
pub mod cli;
pub mod connector;
pub mod control;
pub mod heartbeat;
pub mod launch;
pub mod lease_client;
pub mod persona;
pub mod queue;
mod recall;
pub mod self_run;
pub mod tempo;
#[cfg(test)]
mod testkit;
//! `mind-body`: the organs of one Mind's body that live in Huginn's workspace.
//!
//! Today that is `control`, the operator's brakes, and `cli`, the operator's
//! hand on them. The brakes are the operator's consent, not her body: the store
//! is root-owned and no organ of hers writes it.
//!
//! `launch` and `queue` are the launch primitive: the work queue, the
//! repetition breaker, the one-live-Self-run rule, and the single path that
//! opens a run in the mind and only then starts its unit.
//!
//! `heartbeat` runs the swarm's shared heartbeat over them: the one process,
//! the `mind-heartbeat` binary, that decides who wakes and when, by the rule in
//! `tempo` (the Minds' votes make the beat; the pick chooses who takes it).
//!
//! `persona` is the Persona organ: one start of `mind-persona@<instance>` carries her live
//! `claude` session under bubblewrap in her sandbox, driven by the Self's own driver, with her
//! pending chat as the inbox and one typed reply per result (`chat_client` is the chat wire and
//! port and `chat_rudp` its RUDP transport to Bifrost's mouth).
//!
//! `lease_client` asks the Claude connector for a session's credential and the
//! plan's usage windows. `self_run` carries one Self run inside `mind-self@<instance>`: the
//! `mind-self-run` binary runs `claude` over stream-json for her one open Self
//! run, asks for the flush at the context ceiling and stops a child that outlives it.
// The control root is a constant, so the library reads no part of the process
// environment (clippy.toml lists the readers); `forbid` keeps an `allow` from
// putting one back. `forbid(unsafe_code)` closes the readers the list cannot
// name: a foreign `getenv` or any other FFI call needs `unsafe`. The rest of the
// Grant rule is ruling `eureka-body:ruling:mind-launch-forge-scanner-bound`.
#![forbid(clippy::disallowed_methods)]
#![forbid(unsafe_code)]
pub mod chat_client;
pub mod chat_rudp;
pub mod cli;
pub mod connector;
pub mod control;
pub mod heartbeat;
pub mod launch;
pub mod lease_client;
pub mod persona;
pub mod queue;
mod recall;
pub mod self_run;
pub mod tempo;
#[cfg(test)]
mod testkit;

3166
crates/mind-body/src/persona.rs
View file

File diff suppressed because it is too large Load diff

1865
crates/mind-body/src/self_run.rs
View file

File diff suppressed because it is too large Load diff

692
crates/mind-body/src/self_run/decide.rs
View file

@ -1,341 +1,351 @@
//! The Self-run actuator's lifecycle decision, and nothing else.
//!
//! `decide(state, event, now) -> (state, action)` is the one owner of when the
//! flush is asked for, when the stdin closes, when a signal goes out and when
//! the run is over (ruling `eureka-body:ruling:self-run-decider`). It performs
//! no I/O and reads no clock: the loop in `carry` turns operating-system facts
//! into events, reads the clock once per event, and executes the action. A
//! timing rule is pinned here, by tables and by an enumeration of event orders.
//!
//! Delivery is counted, not guessed, and at most one write is outstanding (ruling
//! `eureka-body:ruling:self-session-one-outstanding-write`): the CLI joins the
//! lines queued behind a request into one echo, so a second line is never
//! written before the first one's echo. Every stdin line the actuator writes (the
//! start message, a write of tells and the flush) increments `written`; every
//! replay echo increments `consumed`. The echo is the point the CLI takes a line
//! into a request, folded into the running turn or queued for the next. What
//! waits (the briefs, the flush ask) is coalesced into the next single write.
//! Idle is derived: no turn open and no write outstanding.
//!
//! The rules, in this order, while stdin is open and no signal is out:
//! (1) a result that is blocked, or an error before any answer, closes stdin: no
//! cache is worth keeping and a flush would fail the same way; (2) the first
//! result after the flush's echo closes stdin, folded or queued alike; (3) the
//! cap less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (4)
//! idle with no brief waiting, at the last result's cache lifetime less a margin:
//! ask the flush (idle); (5) with nothing outstanding, write what waits: the
//! briefs and the flush ask if there is one, as one line. The ceiling asks the
//! same flush from a request. Compaction is off in the CLI (`DISABLE_COMPACT=1`),
//! so the flush is the only way a session ends short of its cache.
//!
//! The headroom counts the requests after the flush's echo, from the prefix the
//! flush was asked at. The grace interrupt is at the earlier of the grace
//! after the flush write and the cap less `LADDER_S`, so the signal ladder
//! finishes before the unit's cap (ruling
//! `eureka-body:ruling:flush-always-gets-a-turn`). At most one action comes out
//! of an event; a deadline that is already due yields its action at the next
//! event, and `next_deadline` says it is now.
use std::collections::HashSet;
use chrono::{DateTime, Duration, Utc};
use super::{
CEILING_TOKENS, Cause, ESCALATE_AFTER_S, EXIT_DRAIN_S, FLUSH_GRACE_S, FLUSH_HEADROOM_TOKENS, LADDER_S, ROTATE_MARGIN_S, SelfRunOutcome, Stop, TTL_1H_S, TTL_5M_S, TTL_MARGIN_S,
};
/// Why the flush was asked: the session's own end, three ways.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Why {
/// The context reached the ceiling.
Ceiling,
/// The unit's cap is near.
Rotation,
/// The cache would be evicted and nothing waits.
Idle,
}
/// What the loop saw, already turned into a fact the decision can use.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Event {
/// A request of the main thread: its message id and the context it read.
Request { id: String, prefix: u64 },
/// The CLI took one of our stdin lines into a request (its replay echo).
Echo,
/// A turn ended. `ok`: not an error. `blocked`: the CLI refused the context. `ttl_1h`: the last request wrote its cache for an hour, else five minutes.
Result { ok: bool, blocked: bool, ttl_1h: bool },
/// A poll of the mind: whether a brief waits to be told.
Inbox { pending: bool },
/// `waitpid` returned: the child's exit code, none when a signal ended it.
ChildExited(Option<i32>),
StdoutEof,
/// Time passed, or a line that no rule reads.
Tick,
/// What the operating system said of the signal the last action sent.
Signalled { how: Stop, delivered: bool },
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Action {
None,
/// The one write: the waiting briefs and the flush ask, coalesced into one user line.
Write { tells: bool, flush: Option<Why> },
CloseStdin,
/// To the child's whole process group.
Signal(Stop),
Finish(SelfRunOutcome),
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct Flush {
/// The flush's line number once written (its echo is the `seq`-th); none while it waits for the outstanding write's echo.
seq: Option<u32>,
/// The prefix at which the flush was asked: the headroom's base.
prefix: u64,
asked: DateTime<Utc>,
/// Its echo has come: the CLI took it into a request.
consumed: bool,
/// A request after its echo reached the headroom.
exhausted: bool,
why: Why,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct State {
seen: HashSet<String>,
pub(super) requests: u64,
pub(super) prefix: u64,
/// Lines written to stdin, and lines the CLI echoed back as taken.
pub(super) written: u32,
pub(super) consumed: u32,
turn_open: bool,
/// An ok result has come: the session holds an answer worth its cache.
answered_once: bool,
last_result_at: Option<DateTime<Utc>>,
/// The cache lifetime the last result left, in seconds.
pub(super) ttl_s: i64,
pub(super) inbox: bool,
/// When the process started and the unit's cap in seconds; none where no cap is known.
cap: Option<(DateTime<Utc>, i64)>,
stdin_closed: bool,
flush: Option<Flush>,
/// The last signal attempted and when.
stopping: Option<(Stop, DateTime<Utc>)>,
/// The strongest signal delivered.
delivered: Option<Stop>,
exited: Option<(Option<i32>, DateTime<Utc>)>,
eof: bool,
finished: bool,
}
impl State {
/// A session whose start message was just written.
pub(super) fn started(at: DateTime<Utc>, run_cap_s: u64) -> Self {
Self { written: 1, cap: Some((at, i64::try_from(run_cap_s).unwrap_or(i64::MAX))), ..Self::default() }
}
/// A line written whose echo has not come.
fn outstanding(&self) -> bool {
self.written > self.consumed
}
fn idle(&self) -> bool {
!self.turn_open && !self.outstanding()
}
/// Seconds idle since the last result, none while a line or a turn is outstanding.
pub(super) fn idle_s(&self, now: DateTime<Utc>) -> Option<i64> {
self.idle().then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds()))
}
}
fn outcome(state: &State, cause: Cause, code: Option<i32>) -> SelfRunOutcome {
SelfRunOutcome::Ended { cause, code, requests: state.requests, prefix: state.prefix }
}
/// How the run ended, by what was done to the child: the strongest signal delivered, else the flush, else nothing.
fn cause(state: &State) -> Cause {
match (state.delivered, &state.flush) {
(Some(how), _) => Cause::Signalled(how),
(None, Some(_)) => Cause::FlushRequested,
(None, None) => Cause::Own,
}
}
/// When the flush's grace runs out: the grace after the write, never later than
/// the point the signal ladder needs to finish before the unit's cap.
fn grace_deadline(state: &State, flush: &Flush) -> DateTime<Utc> {
let grace = flush.asked + Duration::seconds(FLUSH_GRACE_S);
match state.cap {
Some((started, cap)) => grace.min(started + Duration::seconds(cap.saturating_sub(LADDER_S))),
None => grace,
}
}
/// When the flush is next asked on its own, and why: the cap's margin, or the
/// cache's end while idle with no brief waiting.
fn due(state: &State) -> Option<(DateTime<Utc>, Why)> {
if state.stdin_closed || state.exited.is_some() || state.flush.is_some() {
return None;
}
let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Why::Rotation));
let idle = (state.idle() && !state.inbox).then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Why::Idle));
[rotation, idle].into_iter().flatten().min_by_key(|(at, _)| *at)
}
/// The moment the loop must next be heard at, if any. A moment already past
/// means an action is due now.
pub(super) fn next_deadline(state: &State) -> Option<DateTime<Utc>> {
if state.finished {
return None;
}
if let Some((_, exited)) = state.exited {
return (!state.eof).then(|| exited + Duration::seconds(EXIT_DRAIN_S));
}
let signals = match (&state.stopping, &state.flush) {
(Some((_, sent)), _) => return Some(*sent + Duration::seconds(ESCALATE_AFTER_S)),
(None, Some(flush)) => Some(grace_deadline(state, flush)),
(None, None) => None,
};
signals.into_iter().chain(due(state).map(|(at, _)| at)).min()
}
/// The flush is asked once, by the ceiling, the cap's margin or the cache's end;
/// never to a child whose input is closed or that has exited. Asking writes
/// nothing: `send` writes it when no write is outstanding.
fn ask_flush(state: &mut State, now: DateTime<Utc>, why: Why) -> bool {
if state.flush.is_some() || state.stdin_closed || state.exited.is_some() {
return false;
}
state.flush = Some(Flush { seq: None, prefix: state.prefix, asked: now, consumed: false, exhausted: false, why });
true
}
/// The lifecycle rules 3 to 5: ask the flush where it is due, then write what waits.
fn lifecycle(state: &mut State, now: DateTime<Utc>) -> Action {
if let Some((at, why)) = due(state)
&& at <= now
{
ask_flush(state, now, why);
}
send(state)
}
/// With nothing outstanding, the one write: the waiting briefs (until the flush
/// is written) and the flush, if it has not been written, as one line.
fn send(state: &mut State) -> Action {
if state.stdin_closed || state.exited.is_some() || state.outstanding() || state.flush.as_ref().is_some_and(|flush| flush.seq.is_some()) {
return Action::None;
}
let flush = state.flush.as_ref().map(|flush| flush.why);
if !state.inbox && flush.is_none() {
return Action::None;
}
state.written += 1;
let seq = state.written;
if let Some(flush) = state.flush.as_mut() {
flush.seq = Some(seq);
}
Action::Write { tells: std::mem::take(&mut state.inbox), flush }
}
pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) {
if state.finished {
return (state, Action::None);
}
let mut action = Action::None;
match event {
Event::Request { id, prefix } => {
if state.seen.insert(id) {
state.requests += 1;
state.prefix = prefix;
if let Some(flush) = state.flush.as_mut()
&& flush.consumed
&& flush.prefix.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit)
{
flush.exhausted = true;
}
if prefix >= CEILING_TOKENS {
ask_flush(&mut state, now, Why::Ceiling);
}
}
}
Event::Echo => {
state.consumed += 1;
state.turn_open = true;
let consumed = state.consumed;
if let Some(flush) = state.flush.as_mut()
&& flush.seq.is_some_and(|seq| consumed >= seq)
{
flush.consumed = true;
}
}
Event::Result { ok, blocked, ttl_1h } => {
state.turn_open = false;
state.last_result_at = Some(now);
state.ttl_s = if ttl_1h { TTL_1H_S } else { TTL_5M_S };
let no_answer = !ok && !state.answered_once;
state.answered_once |= ok;
let flush_answered = state.flush.as_ref().is_some_and(|flush| flush.consumed);
if (blocked || no_answer || flush_answered) && !state.stdin_closed && state.exited.is_none() {
state.stdin_closed = true;
action = Action::CloseStdin;
}
}
Event::Inbox { pending } => state.inbox = pending,
Event::ChildExited(code) => {
state.exited.get_or_insert((code, now));
}
Event::StdoutEof => state.eof = true,
Event::Tick => {}
Event::Signalled { how, delivered } => {
if delivered {
state.delivered = Some(how);
}
}
}
// A child that has exited is never written to or signalled: the run is over
// once its stream has ended, or once the stream has had its drain.
if let Some((code, exited)) = state.exited {
if state.eof || now >= exited + Duration::seconds(EXIT_DRAIN_S) {
state.finished = true;
let outcome = outcome(&state, cause(&state), code);
return (state, Action::Finish(outcome));
}
return (state, Action::None);
}
if action != Action::None {
return (state, action);
}
if let Some((Stop::Kill, sent)) = state.stopping {
// SIGKILL went out and the child still has not been reaped: stop waiting for it.
if now >= sent + Duration::seconds(ESCALATE_AFTER_S) {
state.finished = true;
let cause = if state.delivered == Some(Stop::Kill) { Cause::Signalled(Stop::Kill) } else { Cause::Unstoppable };
let outcome = outcome(&state, cause, None);
return (state, Action::Finish(outcome));
}
return (state, Action::None);
}
if state.stopping.is_none() {
let next = lifecycle(&mut state, now);
if next != Action::None {
return (state, next);
}
}
let how = match (&state.stopping, &state.flush) {
(Some((last, sent)), _) => (now >= *sent + Duration::seconds(ESCALATE_AFTER_S)).then(|| last.next()),
(None, Some(flush)) => (flush.exhausted || now >= grace_deadline(&state, flush)).then_some(Stop::Interrupt),
(None, None) => None,
};
match how {
Some(how) => {
state.stopping = Some((how, now));
(state, Action::Signal(how))
}
None => (state, Action::None),
}
}
#[cfg(test)]
mod tests;
//! The Self-run actuator's lifecycle decision, and nothing else.
//!
//! `decide(state, event, now) -> (state, action)` is the one owner of when the
//! flush is asked for, when the stdin closes, when a signal goes out and when
//! the run is over (ruling `eureka-body:ruling:self-run-decider`). It performs
//! no I/O and reads no clock: the loop in `carry` turns operating-system facts
//! into events, reads the clock once per event, and executes the action. A
//! timing rule is pinned here, by tables and by an enumeration of event orders.
//!
//! Delivery is counted, not guessed, and at most one write is outstanding (ruling
//! `eureka-body:ruling:self-session-one-outstanding-write`): the CLI joins the
//! lines queued behind a request into one echo, so a second line is never
//! written before the first one's echo. Every stdin line the actuator writes (the
//! start message, a write of tells and the flush) increments `written`; every
//! replay echo increments `consumed`. The echo is the point the CLI takes a line
//! into a request, folded into the running turn or queued for the next. What
//! waits (the briefs, the flush ask) is coalesced into the next single write.
//! Idle is derived: no turn open and no write outstanding.
//!
//! The rules, in this order, while stdin is open and no signal is out:
//! (1) a result that is blocked, or an error before any answer, closes stdin: no
//! cache is worth keeping and a flush would fail the same way; (2) the first
//! result after the flush's echo closes stdin, folded or queued alike; (3) the
//! cap less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (4)
//! idle with no brief waiting, at the last result's cache lifetime less a margin:
//! ask the flush (idle); (4b) a session told to yield closes stdin at idle, with nothing outstanding and no flush written; (5) with nothing outstanding, write what waits: the
//! briefs and the flush ask if there is one, as one line. The ceiling asks the
//! same flush from a request. Compaction is off in the CLI (`DISABLE_COMPACT=1`),
//! so the flush is the only way a session ends short of its cache.
//!
//! The headroom counts the requests after the flush's echo, from the prefix the
//! flush was asked at. The grace interrupt is at the earlier of the grace
//! after the flush write and the cap less `LADDER_S`, so the signal ladder
//! finishes before the unit's cap (ruling
//! `eureka-body:ruling:flush-always-gets-a-turn`). At most one action comes out
//! of an event; a deadline that is already due yields its action at the next
//! event, and `next_deadline` says it is now.
use std::collections::HashSet;
use chrono::{DateTime, Duration, Utc};
use super::{
CEILING_TOKENS, Cause, ESCALATE_AFTER_S, EXIT_DRAIN_S, FLUSH_GRACE_S, FLUSH_HEADROOM_TOKENS, LADDER_S, ROTATE_MARGIN_S, SelfRunOutcome, Stop, TTL_1H_S, TTL_5M_S, TTL_MARGIN_S,
};
/// Why the flush was asked: the session's own end, three ways.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Why {
/// The context reached the ceiling.
Ceiling,
/// The unit's cap is near.
Rotation,
/// The cache would be evicted and nothing waits.
Idle,
}
/// What the loop saw, already turned into a fact the decision can use.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Event {
/// A request of the main thread: its message id and the context it read.
Request { id: String, prefix: u64 },
/// The CLI took one of our stdin lines into a request (its replay echo).
Echo,
/// A turn ended. `ok`: not an error. `blocked`: the CLI refused the context. `ttl_1h`: the last request wrote its cache for an hour, else five minutes.
Result { ok: bool, blocked: bool, ttl_1h: bool },
/// A poll of the mind: whether a brief waits to be told.
Inbox { pending: bool },
/// `waitpid` returned: the child's exit code, none when a signal ended it.
ChildExited(Option<i32>),
StdoutEof,
/// Time passed, or a line that no rule reads.
Tick,
/// What the operating system said of the signal the last action sent.
Signalled { how: Stop, delivered: bool },
/// The inbox's owner wants the session over at its next idle: what is oldest and unanswered is another conversation's. Only the Persona sends it.
Yield,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Action {
None,
/// The one write: the waiting briefs and the flush ask, coalesced into one user line.
Write { tells: bool, flush: Option<Why> },
CloseStdin,
/// To the child's whole process group.
Signal(Stop),
Finish(SelfRunOutcome),
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct Flush {
/// The flush's line number once written (its echo is the `seq`-th); none while it waits for the outstanding write's echo.
seq: Option<u32>,
/// The prefix at which the flush was asked: the headroom's base.
prefix: u64,
asked: DateTime<Utc>,
/// Its echo has come: the CLI took it into a request.
consumed: bool,
/// A request after its echo reached the headroom.
exhausted: bool,
why: Why,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct State {
seen: HashSet<String>,
pub(super) requests: u64,
pub(super) prefix: u64,
/// Lines written to stdin, and lines the CLI echoed back as taken.
pub(super) written: u32,
pub(super) consumed: u32,
turn_open: bool,
/// An ok result has come: the session holds an answer worth its cache.
answered_once: bool,
last_result_at: Option<DateTime<Utc>>,
/// The cache lifetime the last result left, in seconds.
pub(super) ttl_s: i64,
pub(super) inbox: bool,
/// The session is to end at its next idle (`Event::Yield`); once asked it stays asked.
yielding: bool,
/// When the process started and the unit's cap in seconds; none where no cap is known.
cap: Option<(DateTime<Utc>, i64)>,
stdin_closed: bool,
flush: Option<Flush>,
/// The last signal attempted and when.
stopping: Option<(Stop, DateTime<Utc>)>,
/// The strongest signal delivered.
delivered: Option<Stop>,
exited: Option<(Option<i32>, DateTime<Utc>)>,
eof: bool,
finished: bool,
}
impl State {
/// A session whose start message was just written.
pub(super) fn started(at: DateTime<Utc>, run_cap_s: u64) -> Self {
Self { written: 1, cap: Some((at, i64::try_from(run_cap_s).unwrap_or(i64::MAX))), ..Self::default() }
}
/// A line written whose echo has not come.
fn outstanding(&self) -> bool {
self.written > self.consumed
}
fn idle(&self) -> bool {
!self.turn_open && !self.outstanding()
}
/// Seconds idle since the last result, none while a line or a turn is outstanding.
pub(super) fn idle_s(&self, now: DateTime<Utc>) -> Option<i64> {
self.idle().then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds()))
}
}
fn outcome(state: &State, cause: Cause, code: Option<i32>) -> SelfRunOutcome {
SelfRunOutcome::Ended { cause, code, requests: state.requests, prefix: state.prefix }
}
/// How the run ended, by what was done to the child: the strongest signal delivered, else the flush, else nothing.
fn cause(state: &State) -> Cause {
match (state.delivered, &state.flush) {
(Some(how), _) => Cause::Signalled(how),
(None, Some(_)) => Cause::FlushRequested,
(None, None) => Cause::Own,
}
}
/// When the flush's grace runs out: the grace after the write, never later than
/// the point the signal ladder needs to finish before the unit's cap.
fn grace_deadline(state: &State, flush: &Flush) -> DateTime<Utc> {
let grace = flush.asked + Duration::seconds(FLUSH_GRACE_S);
match state.cap {
Some((started, cap)) => grace.min(started + Duration::seconds(cap.saturating_sub(LADDER_S))),
None => grace,
}
}
/// When the flush is next asked on its own, and why: the cap's margin, or the
/// cache's end while idle with no brief waiting.
fn due(state: &State) -> Option<(DateTime<Utc>, Why)> {
if state.stdin_closed || state.exited.is_some() || state.flush.is_some() {
return None;
}
let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Why::Rotation));
let idle = (state.idle() && !state.inbox).then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Why::Idle));
[rotation, idle].into_iter().flatten().min_by_key(|(at, _)| *at)
}
/// The moment the loop must next be heard at, if any. A moment already past
/// means an action is due now.
pub(super) fn next_deadline(state: &State) -> Option<DateTime<Utc>> {
if state.finished {
return None;
}
if let Some((_, exited)) = state.exited {
return (!state.eof).then(|| exited + Duration::seconds(EXIT_DRAIN_S));
}
let signals = match (&state.stopping, &state.flush) {
(Some((_, sent)), _) => return Some(*sent + Duration::seconds(ESCALATE_AFTER_S)),
(None, Some(flush)) => Some(grace_deadline(state, flush)),
(None, None) => None,
};
signals.into_iter().chain(due(state).map(|(at, _)| at)).min()
}
/// The flush is asked once, by the ceiling, the cap's margin or the cache's end;
/// never to a child whose input is closed or that has exited. Asking writes
/// nothing: `send` writes it when no write is outstanding.
fn ask_flush(state: &mut State, now: DateTime<Utc>, why: Why) -> bool {
if state.flush.is_some() || state.stdin_closed || state.exited.is_some() {
return false;
}
state.flush = Some(Flush { seq: None, prefix: state.prefix, asked: now, consumed: false, exhausted: false, why });
true
}
/// The lifecycle rules 3 to 5: ask the flush where it is due, then write what waits.
fn lifecycle(state: &mut State, now: DateTime<Utc>) -> Action {
if let Some((at, why)) = due(state)
&& at <= now
{
ask_flush(state, now, why);
}
// A yielding session ends at idle: its own text waits for the next one. A flush already written keeps its own end.
if state.yielding && state.idle() && !state.stdin_closed && state.exited.is_none() && state.flush.as_ref().is_none_or(|flush| flush.seq.is_none()) {
state.stdin_closed = true;
return Action::CloseStdin;
}
send(state)
}
/// With nothing outstanding, the one write: the waiting briefs (until the flush
/// is written) and the flush, if it has not been written, as one line.
fn send(state: &mut State) -> Action {
if state.stdin_closed || state.exited.is_some() || state.outstanding() || state.flush.as_ref().is_some_and(|flush| flush.seq.is_some()) {
return Action::None;
}
let flush = state.flush.as_ref().map(|flush| flush.why);
if !state.inbox && flush.is_none() {
return Action::None;
}
state.written += 1;
let seq = state.written;
if let Some(flush) = state.flush.as_mut() {
flush.seq = Some(seq);
}
Action::Write { tells: std::mem::take(&mut state.inbox), flush }
}
pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) {
if state.finished {
return (state, Action::None);
}
let mut action = Action::None;
match event {
Event::Request { id, prefix } => {
if state.seen.insert(id) {
state.requests += 1;
state.prefix = prefix;
if let Some(flush) = state.flush.as_mut()
&& flush.consumed
&& flush.prefix.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit)
{
flush.exhausted = true;
}
if prefix >= CEILING_TOKENS {
ask_flush(&mut state, now, Why::Ceiling);
}
}
}
Event::Echo => {
state.consumed += 1;
state.turn_open = true;
let consumed = state.consumed;
if let Some(flush) = state.flush.as_mut()
&& flush.seq.is_some_and(|seq| consumed >= seq)
{
flush.consumed = true;
}
}
Event::Result { ok, blocked, ttl_1h } => {
state.turn_open = false;
state.last_result_at = Some(now);
state.ttl_s = if ttl_1h { TTL_1H_S } else { TTL_5M_S };
let no_answer = !ok && !state.answered_once;
state.answered_once |= ok;
let flush_answered = state.flush.as_ref().is_some_and(|flush| flush.consumed);
if (blocked || no_answer || flush_answered) && !state.stdin_closed && state.exited.is_none() {
state.stdin_closed = true;
action = Action::CloseStdin;
}
}
Event::Inbox { pending } => state.inbox = pending,
Event::Yield => state.yielding = true,
Event::ChildExited(code) => {
state.exited.get_or_insert((code, now));
}
Event::StdoutEof => state.eof = true,
Event::Tick => {}
Event::Signalled { how, delivered } => {
if delivered {
state.delivered = Some(how);
}
}
}
// A child that has exited is never written to or signalled: the run is over
// once its stream has ended, or once the stream has had its drain.
if let Some((code, exited)) = state.exited {
if state.eof || now >= exited + Duration::seconds(EXIT_DRAIN_S) {
state.finished = true;
let outcome = outcome(&state, cause(&state), code);
return (state, Action::Finish(outcome));
}
return (state, Action::None);
}
if action != Action::None {
return (state, action);
}
if let Some((Stop::Kill, sent)) = state.stopping {
// SIGKILL went out and the child still has not been reaped: stop waiting for it.
if now >= sent + Duration::seconds(ESCALATE_AFTER_S) {
state.finished = true;
let cause = if state.delivered == Some(Stop::Kill) { Cause::Signalled(Stop::Kill) } else { Cause::Unstoppable };
let outcome = outcome(&state, cause, None);
return (state, Action::Finish(outcome));
}
return (state, Action::None);
}
if state.stopping.is_none() {
let next = lifecycle(&mut state, now);
if next != Action::None {
return (state, next);
}
}
let how = match (&state.stopping, &state.flush) {
(Some((last, sent)), _) => (now >= *sent + Duration::seconds(ESCALATE_AFTER_S)).then(|| last.next()),
(None, Some(flush)) => (flush.exhausted || now >= grace_deadline(&state, flush)).then_some(Stop::Interrupt),
(None, None) => None,
};
match how {
Some(how) => {
state.stopping = Some((how, now));
(state, Action::Signal(how))
}
None => (state, Action::None),
}
}
#[cfg(test)]
mod tests;

138
crates/mind-body/src/self_run/session.rs Normal file
View file

@ -0,0 +1,138 @@
//! The one session driver, for both organs: it writes a live `claude` child's stdin,
//! reads its stream and executes what `decide` says. The Self's briefs and the
//! Persona's chat are the two inboxes it serves (the [`SessionInbox`] each
//! organ implements); the loop, the one outstanding write and every timing
//! decision are the same code (rulings `eureka-body:ruling:self-session-one-outstanding-write`
//! and `eureka-body:ruling:persona-session-full-sandbox`).
use std::collections::VecDeque;
use std::io::Write;
use std::time::Duration as StdDuration;
use anyhow::Result;
use chrono::Duration as ChronoDuration;
use serde_json::json;
use super::decide::{Action, Event, State, Why, decide, next_deadline};
use super::{Child, Read, SelfRunOutcome, Usage, read_usage, write_result, writes_hour_cache};
use crate::launch::Clock;
/// What one poll of an inbox found.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct Polled {
/// Something waits to be told.
pub(crate) pending: bool,
/// The session should end at its next idle: what is oldest and unanswered is not its business.
pub(crate) yield_now: bool,
}
/// What the session is told, and what its results mean: the organ's side of the loop.
pub(crate) trait SessionInbox {
/// The first line written, whole.
fn start_line(&mut self) -> Result<String>;
/// Seconds between polls.
fn poll_every_s(&self) -> i64;
fn poll(&mut self) -> Result<Polled>;
/// The text of what waits, which is thereby told.
fn tell(&mut self) -> Result<String>;
fn flush_text(&self, why: Why) -> String;
/// Every line of the child's stream, before it is read.
fn line(&mut self, _line: &str) {}
/// The CLI took one of our lines into a request.
fn echoed(&mut self) {}
/// A turn ended with `line`; the `(ok, blocked)` the decision sees. An error is the organ's world failing, not the session's.
fn result(&mut self, _line: &str, ok: bool, blocked: bool) -> Result<(bool, bool)> {
Ok((ok, blocked))
}
}
/// The loop: facts in, one clock read per fact, actions out. The one writer into the session.
pub(crate) fn drive(clock: &dyn Clock, mut child: Box<dyn Child>, inbox: &mut dyn SessionInbox, run_cap_s: u64, out: &mut dyn Write) -> Result<SelfRunOutcome> {
let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string();
child.write_line(&user_line(&inbox.start_line()?));
let poll_every = ChronoDuration::seconds(inbox.poll_every_s());
let longest = StdDuration::from_secs(u64::try_from(inbox.poll_every_s()).unwrap_or(1));
let mut now = clock.now();
let mut state = State::started(now, run_cap_s);
let mut polled = now;
let mut ttl_1h = false;
let mut closed = false;
loop {
let poll_due = polled + poll_every;
let wake = next_deadline(&state).into_iter().chain((!closed).then_some(poll_due)).min();
let wait = wake.map_or(longest, |at| (at - now).to_std().unwrap_or_default().min(longest));
let fact = match child.next(wait) {
Read::Line(line) => {
inbox.line(&line);
write_result(&line, out)?;
if let Some(hour) = writes_hour_cache(&line) {
ttl_1h = hour;
}
match read_usage(&line) {
Usage::Request { id, prefix } => Event::Request { id, prefix },
Usage::Echo => {
inbox.echoed();
Event::Echo
}
Usage::Result { ok, blocked } => {
let (ok, blocked) = inbox.result(&line, ok, blocked)?;
Event::Result { ok, blocked, ttl_1h }
}
Usage::Other | Usage::Unreadable => Event::Tick,
}
}
Read::Timeout => Event::Tick,
Read::Eof => Event::StdoutEof,
Read::Exited(code) => Event::ChildExited(code),
};
now = clock.now();
let mut events = VecDeque::from([fact]);
if !closed && now >= poll_due {
polled = now;
match inbox.poll() {
Ok(found) => {
events.push_back(Event::Inbox { pending: found.pending });
if found.yield_now {
events.push_back(Event::Yield);
}
}
Err(_) => writeln!(out, "inbox poll failed")?,
}
}
while let Some(heard_event) = events.pop_front() {
let counted = state.requests;
// The decision's inputs, as they stood when the event came.
let (ttl_s, inbox_pending, idle_s) = (state.ttl_s, state.inbox || matches!(&heard_event, Event::Inbox { pending: true }), state.idle_s(now));
let (next, action) = decide(state, heard_event, now);
state = next;
if state.requests != counted {
writeln!(out, "request {} prefix {}", state.requests, state.prefix)?;
}
match &action {
Action::None => {}
Action::Write { tells, flush } => {
let mut parts = Vec::new();
if *tells {
parts.push(inbox.tell()?);
}
parts.extend(flush.map(|why| inbox.flush_text(why)));
child.write_line(&user_line(&parts.join("\n\n")));
}
Action::CloseStdin => {
child.close_input();
closed = true;
}
Action::Signal(how) => events.push_front(Event::Signalled { how: *how, delivered: child.stop(*how) }),
Action::Finish(_) => {}
}
if !matches!(action, Action::None | Action::Finish(_)) {
let idle = idle_s.map_or_else(|| "working".to_string(), |seconds| seconds.to_string());
writeln!(out, "decision {action:?} prefix {} ttl_s {ttl_s} idle_s {idle} inbox {inbox_pending} written {} consumed {}", state.prefix, state.written, state.consumed)?;
}
if let Action::Finish(outcome) = action {
return Ok(outcome);
}
}
}
}

3579
crates/mind-body/src/self_run/tests.rs
View file

File diff suppressed because it is too large Load diff

212
crates/mind-body/src/testkit.rs
View file

@ -3,8 +3,13 @@
//! so a field the leaf adds to a kind is one edit.
use std::cell::{Cell, RefCell};
use std::collections::VecDeque;
use std::rc::Rc;
use std::time::Duration as StdDuration;
use anyhow::{Result, anyhow};
use chrono::Duration as ChronoDuration;
use model_connector::PlanQuota;
use chrono::{DateTime, Duration, TimeZone, Utc};
use cultnet_rs::Selection;
use eureka_pipeline::{
@ -19,7 +24,9 @@ use huginn_mind::{
use tempfile::TempDir;
use crate::launch::{Clock, Launcher};
use crate::lease_client::{Lease, LeaseRefusalKind, LeaseRefused, LeaseRequest, Lender, Lent, TokenFd, token_holding};
use crate::queue::MindPort;
use crate::self_run::{Child, Read, Spawner, Stop};
pub(crate) const INSTANCE: &str = "yggdrasil";
pub(crate) const CAMPAIGN: &str = "eureka-body";
@ -472,3 +479,208 @@ pub(crate) fn follow_up(label: &str, source: &PipelineRef) -> PipelineDocument {
owner: "Hands".into(),
})
}
// The scripted child, the fake spawner and the fake lease: the stream and the lease the Self and the Persona organs are tested against.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum Event {
Wrote(String),
InputClosed,
/// A signal was attempted, delivered or not.
Stopped(Stop),
Asked(StdDuration),
/// The connector was asked for a lease and answered one.
Lent,
/// A child was started.
Spawned,
/// The lease was dropped, which ends it: nothing is settled.
Abandoned,
}
#[derive(Clone)]
pub(crate) enum Step {
Line(String),
/// Time passes with no output: the clock moves and the read times out.
Idle(i64),
/// The clock moves, then this line arrives.
After(i64, String),
/// The child is reaped with this code; its stream stays open.
Exit(Option<i32>),
/// Its stdout ends; the child is not reaped yet.
Eof,
/// The clock moves, this runs (the mind changes, a tool call's hook fires), and the read times out.
Do(i64, Rc<dyn Fn()>),
}
pub(crate) struct Scripted {
pub(crate) steps: VecDeque<Step>,
pub(crate) log: Rc<RefCell<Vec<Event>>>,
pub(crate) clock: Rc<Fixed>,
pub(crate) code: Option<i32>,
/// The weakest signal that ends it; weaker ones are ignored.
pub(crate) dies_on: Option<Stop>,
/// Signals that cannot be delivered (no `kill` program, say).
pub(crate) refuses: Vec<Stop>,
/// When the steps run out it neither exits nor closes its stream: time just passes.
pub(crate) hangs: bool,
pub(crate) exited: bool,
pub(crate) eof: bool,
/// How many times it was asked: a loop that never ends is a failure, not a hang.
pub(crate) asked: u32,
}
impl Child for Scripted {
fn write_line(&mut self, line: &str) {
self.log.borrow_mut().push(Event::Wrote(line.to_string()));
}
fn close_input(&mut self) {
self.log.borrow_mut().push(Event::InputClosed);
}
fn next(&mut self, timeout: StdDuration) -> Read {
self.log.borrow_mut().push(Event::Asked(timeout));
self.asked += 1;
assert!(self.asked < 500, "the loop never ended");
match self.steps.pop_front() {
Some(Step::Line(line)) => Read::Line(line),
Some(Step::Idle(seconds)) => {
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds));
Read::Timeout
}
Some(Step::After(seconds, line)) => {
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds));
Read::Line(line)
}
Some(Step::Exit(code)) => {
self.exited = true;
Read::Exited(code)
}
Some(Step::Do(seconds, act)) => {
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds));
act();
Read::Timeout
}
Some(Step::Eof) => {
self.eof = true;
Read::Eof
}
// The stream is over: the child is reaped, then its stdout ends.
None if self.hangs => {
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(timeout.as_secs() as i64));
Read::Timeout
}
None if !self.exited => {
self.exited = true;
Read::Exited(self.code)
}
None if !self.eof => {
self.eof = true;
Read::Eof
}
None => panic!("the loop kept waiting on a child that ended"),
}
}
fn stop(&mut self, how: Stop) -> bool {
self.log.borrow_mut().push(Event::Stopped(how));
if self.refuses.contains(&how) {
return false;
}
if self.dies_on.is_some_and(|weakest| how >= weakest) {
self.steps.clear();
self.hangs = false;
}
true
}
}
pub(crate) struct Fake {
pub(crate) child: RefCell<Option<Scripted>>,
pub(crate) spawns: RefCell<Vec<Vec<String>>>,
pub(crate) refuse: bool,
}
impl Fake {
pub(crate) fn with(child: Option<Scripted>, refuse: bool) -> Self {
Self { child: RefCell::new(child), spawns: RefCell::new(Vec::new()), refuse }
}
}
impl Spawner for Fake {
fn spawn(&self, args: &[String], _token: &TokenFd) -> Result<Box<dyn Child>> {
self.spawns.borrow_mut().push(args.to_vec());
if self.refuse {
return Err(anyhow!("refused"));
}
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)]
pub(crate) enum Answer {
Grant,
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.
pub(crate) struct FakeLender {
pub(crate) log: Rc<RefCell<Vec<Event>>>,
pub(crate) asked: RefCell<Vec<LeaseRequest>>,
pub(crate) answer: Answer,
}
impl FakeLender {
pub(crate) fn new(answer: Answer) -> Self {
Self::logging(Rc::new(RefCell::new(Vec::new())), answer)
}
pub(crate) 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 => {
self.log.borrow_mut().push(Event::Lent);
Ok(Lent::Granted(Box::new(FakeLease { log: self.log.clone(), token: token_holding("tok-canary"), plan: vec![plan_window()] })))
}
Answer::Refuse(kind) => Ok(Lent::Refused(LeaseRefused { kind, detail: "detail-canary".into() })),
Answer::Unreachable => Err(anyhow!("down")),
}
}
}
pub(crate) fn plan_window() -> PlanQuota {
PlanQuota { window: "five_hour".into(), utilization_permille: Some(423), resets_at_unix_ms: 1_793_307_000_000, status: "allowed".into(), observed_at_unix_ms: 1_793_306_000_000 }
}
pub(crate) struct FakeLease {
pub(crate) log: Rc<RefCell<Vec<Event>>>,
pub(crate) token: TokenFd,
pub(crate) plan: Vec<PlanQuota>,
}
impl Lease for FakeLease {
fn token(&self) -> &TokenFd {
&self.token
}
fn plan_quota(&self) -> &[PlanQuota] {
&self.plan
}
}
impl Drop for FakeLease {
fn drop(&mut self) {
self.log.borrow_mut().push(Event::Abandoned);
}
}

155
crates/mind-body/tests/connector_kit.rs
View file

@ -1,155 +0,0 @@
//! The connector transport against the kit's real server: `serve` over a real
//! replay store and caller admission, with a fake provider behind it. What is
//! faked is the model; the contract, the sealing, the admission and the usage
//! readout are the kit's.
use std::net::TcpListener;
use std::sync::{Arc, Mutex};
use mind_body::connector::{KitConnector, endpoint, key};
use mind_body::persona::{Connector, TurnRequest};
use model_connector::{
CallerAdmission, PlanQuota, PlanQuotaSnapshot, ProviderBackend, ProviderCall, ProviderRequest, ServeLimits, StopReason, TransportEvent, TransportEventPayload, TransportOutcome, TransportResult, Usage,
serve,
};
const CALLER: &str = "eureka";
const MODEL: &str = "claude-test-model";
const KEY: &str = "connector-key-for-test";
enum Behaviour {
Reply(&'static str),
Stop(StopReason),
Fail,
}
struct FakeProvider {
behaviour: Behaviour,
seen: Mutex<Vec<ProviderRequest>>,
}
impl ProviderBackend for FakeProvider {
fn ready(&self) -> Result<(), &'static str> {
Ok(())
}
fn provider(&self) -> &str {
"fake-provider"
}
fn plan_quota(&self) -> PlanQuotaSnapshot {
PlanQuotaSnapshot {
windows: vec![
PlanQuota { window: "five_hour".into(), utilization_permille: Some(423), resets_at_unix_ms: 9, status: "allowed".into(), observed_at_unix_ms: 5 },
PlanQuota { window: "seven_day".into(), utilization_permille: None, resets_at_unix_ms: 1, status: "expired".into(), observed_at_unix_ms: 2 },
],
}
}
fn execute(&self, call: &ProviderCall) -> TransportResult {
self.seen.lock().unwrap().push(call.request().clone());
let usage = Usage { input_tokens: 10, cache_read_tokens: 0, cache_write_tokens: 0, output_tokens: 3, reasoning_tokens: None };
match &self.behaviour {
Behaviour::Reply(text) => call.transported(
vec![TransportEvent { sequence: 0, payload: TransportEventPayload::TextDelta { text: (*text).to_string() } }],
TransportOutcome::Completed { provider_response_id: None, stop_reason: StopReason::EndTurn, usage },
),
Behaviour::Stop(stop_reason) => call.transported(
vec![TransportEvent { sequence: 0, payload: TransportEventPayload::TextDelta { text: "cut".into() } }],
TransportOutcome::Completed { provider_response_id: None, stop_reason: *stop_reason, usage },
),
Behaviour::Fail => call.transported(
vec![],
TransportOutcome::Failed { kind: model_connector::FailureKind::Unavailable, message: "down".into(), usage: Some(usage) },
),
}
}
}
struct Daemon {
endpoint: String,
provider: Arc<FakeProvider>,
_store: tempfile::TempDir,
}
fn daemon(behaviour: Behaviour) -> Daemon {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let endpoint = listener.local_addr().unwrap().to_string();
let store = tempfile::tempdir().unwrap();
let replay = store.path().join("replay.cc");
let admission = CallerAdmission::new(CALLER, key(KEY), 1, [MODEL.to_string()], 2, 64 * 1024, 8192).unwrap();
let limits = ServeLimits { max_frame_bytes: 256 * 1024, max_connections: 8, socket_timeout_ms: 5_000, max_expiry_skew_ms: 300_000 };
let provider = Arc::new(FakeProvider { behaviour, seen: Mutex::default() });
let backend = provider.clone();
std::thread::spawn(move || {
let _ = serve(listener, limits, vec![admission], &replay, backend);
});
Daemon { endpoint, provider, _store: store }
}
fn turn_request() -> TurnRequest {
TurnRequest {
caller: CALLER.into(),
request_id: "run1-t0".into(),
instructions: "You are Eureka.".into(),
input: "Rooms you may speak in: operator.".into(),
output_schema_json: r#"{"type":"object"}"#.into(),
}
}
fn connector(daemon: &Daemon, model: &str) -> KitConnector {
KitConnector::new(endpoint(&daemon.endpoint).unwrap(), &format!("{KEY}\n"), model, CALLER).unwrap()
}
#[test]
fn a_turn_is_one_tool_less_schema_constrained_call_and_its_answer_is_the_assistant_text() {
let daemon = daemon(Behaviour::Reply(r#"{"say":[],"start_self_run":null}"#));
let text = connector(&daemon, MODEL).turn(&turn_request()).unwrap();
assert_eq!(text, r#"{"say":[],"start_self_run":null}"#);
let seen = daemon.provider.seen.lock().unwrap();
assert_eq!(seen.len(), 1);
let request = &seen[0];
assert_eq!((request.request_id.as_str(), request.conversation_id.as_str(), request.model.as_str()), ("run1-t0", CALLER, MODEL));
assert_eq!(request.instructions, "You are Eureka.");
assert_eq!(request.input, vec![model_connector::InputItem::UserText { text: "Rooms you may speak in: operator.".into() }]);
assert_eq!(request.output_schema_json.as_deref(), Some(r#"{"type":"object"}"#));
assert!(request.tools.is_empty());
}
#[test]
fn the_usage_readout_is_the_plan_windows_the_connector_last_saw_and_is_asked_as_the_callers_own() {
let daemon = daemon(Behaviour::Reply("{}"));
let readout = connector(&daemon, MODEL).usage_readout().unwrap();
let windows: Vec<_> = readout.plan_quota.iter().map(|plan| (plan.window.as_str(), plan.utilization_permille, plan.status.as_str())).collect();
assert_eq!(windows, [("five_hour", Some(423), "allowed"), ("seven_day", None, "expired")]);
assert!(daemon.provider.seen.lock().unwrap().is_empty(), "a readout reaches no provider");
}
#[test]
fn a_turn_the_connector_refuses_or_the_provider_fails_or_stops_early_is_an_error() {
let refused = daemon(Behaviour::Reply("{}"));
assert!(connector(&refused, "a-model-she-is-not-allowed").turn(&turn_request()).is_err());
assert!(refused.provider.seen.lock().unwrap().is_empty(), "a refused call reaches no provider");
assert!(connector(&daemon(Behaviour::Fail), MODEL).turn(&turn_request()).is_err());
assert!(connector(&daemon(Behaviour::Stop(StopReason::MaxOutputTokens)), MODEL).turn(&turn_request()).is_err(), "a cut-off answer is not an answer");
}
#[test]
fn a_wrong_connection_key_gets_nothing() {
let daemon = daemon(Behaviour::Reply("{}"));
let stranger = KitConnector::new(endpoint(&daemon.endpoint).unwrap(), "not-the-key", MODEL, CALLER).unwrap();
assert!(stranger.turn(&turn_request()).is_err());
assert!(stranger.usage_readout().is_err());
assert!(daemon.provider.seen.lock().unwrap().is_empty());
}
#[test]
fn endpoints_are_the_connectors_loopback_tcp_address_and_a_rudp_url_is_refused() {
assert_eq!(endpoint("127.0.0.1:4110").unwrap().port(), 4110);
assert_eq!(endpoint("tcp://127.0.0.1:4110").unwrap().port(), 4110);
for bad in ["rudp://127.0.0.1:4110", "localhost:4110", "127.0.0.1", ""] {
assert!(endpoint(bad).is_err(), "{bad}");
}
assert_eq!(key("secret\r\n"), "secret");
assert_eq!(key("a b\n\n"), "a b");
}