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

integration tests follow the usage-only surface; connector_kit test removed (WIP)

Browse source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 05:35:34 +01:00
commit 7972d971e4
5 changed files with 62 additions and 206 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -14,6 +14,7 @@ use huginn_mind::{PipelineAdmissionOutcome, PipelineSelectionPage};
use super::*;
use cultnet_rs::Selection;
use huginn_mind::PipelinePageItems;
use crate::lease_client::{Lease, LeaseRefused, token_holding};
use model_connector::PlanQuota;
use crate::testkit::*;

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

@ -74,6 +74,44 @@ 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})
}
fn read_total(settlement: &Settlement) -> u64 {
settlement.usage.iter().map(|used| used.usage.cache_read_tokens).sum()
#[test]
fn a_lease_is_asked_for_before_claude_starts_for_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 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 refused_lease_spawns_nothing() {
for kind in [LeaseRefusalKind::UnknownCaller, LeaseRefusalKind::Busy, LeaseRefusalKind::ProviderRejected] {
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 the_lease_is_held_until_the_child_ends_and_then_dropped_with_nothing_settled() {
let ran = go(vec![Out(result_line())]);
assert_eq!(ran.all.last(), Some(&Event::Abandoned), "the lease ends by being dropped");
assert!(!ran.out.contains("settle"), "no settle report exists");
}

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

@ -1,180 +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, the ledger and
//! the spend readout are the kit's.
use std::net::TcpListener;
use std::sync::{Arc, Mutex};
use mind_body::heartbeat::Models;
use rust_decimal::Decimal;
use mind_body::connector::{KitConnector, endpoint, key};
use mind_body::persona::{Connector, TurnRequest};
use model_connector::{
CallerAdmission, ModelPrice, PlanQuota, PlanQuotaSnapshot, PriceTable, ProviderBackend, ProviderCall, ProviderRequest, Rates, ServeLimits, SpendCeiling, SpendWindowMs,
StopReason, TransportEvent, TransportEventPayload, TransportOutcome, TransportResult, Usage, serve,
};
const CALLER: &str = "eureka";
const MODEL: &str = "claude-test-model";
const KEY: &str = "connector-key-for-test";
const HEARTBEAT: &str = "heartbeat";
const HEARTBEAT_KEY: &str = "the-heartbeats-own-key";
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 rate = Rates { input: 1_000_000, cache_write_short: 1_000_000, cache_write_long: 1_000_000, cache_read: 1_000_000, output: 1_000_000 };
let prices = PriceTable::from([(MODEL.to_string(), ModelPrice { rates: rate, long_context: None, overhead_tokens: 0 })]);
let admission =
CallerAdmission::new(CALLER, key(KEY), 1, [MODEL.to_string()], 2, 64 * 1024, 8192, SpendCeiling(1 << 40), SpendWindowMs(3_600_000)).unwrap();
let observer =
CallerAdmission::new(HEARTBEAT, key(HEARTBEAT_KEY), 1, [MODEL.to_string()], 2, 64 * 1024, 8192, SpendCeiling(1 << 40), SpendWindowMs(3_600_000)).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, observer], prices, &replay, backend);
});
Daemon { endpoint, provider, _store: store }
}
fn turn_request() -> TurnRequest {
TurnRequest {
caller: CALLER.into(),
request_id: "run1-t0".into(),
max_cost_micro_usd: 5_000_000,
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_spend_readout_is_the_kits_for_the_caller_after_a_billed_turn() {
let daemon = daemon(Behaviour::Reply("{}"));
let connector = connector(&daemon, MODEL);
connector.turn(&turn_request()).unwrap();
let readout = connector.spend_readout().unwrap();
let row = readout.callers.iter().find(|row| row.caller_id == CALLER).expect("the caller's row");
assert_eq!(row.spent_micro_usd, 13, "10 input and 3 output tokens at 1 USD per million each");
assert_eq!(row.ceiling_micro_usd, 1 << 40);
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")]);
}
#[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.spend_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");
}
#[test]
fn the_heartbeat_reads_a_personas_spend_as_itself_and_picks_her_row() {
let daemon = daemon(Behaviour::Reply("{}"));
connector(&daemon, MODEL).turn(&turn_request()).unwrap();
let heartbeat = KitConnector::new(endpoint(&daemon.endpoint).unwrap(), HEARTBEAT_KEY, MODEL, HEARTBEAT).unwrap();
let spent = Models::spent(&heartbeat, CALLER).expect("a persona's row, asked under the heartbeat's own key");
assert_eq!(spent.usd, Decimal::new(13, 6));
assert_eq!(Models::spent(&heartbeat, HEARTBEAT).unwrap().usd, Decimal::ZERO, "its own row is there too");
assert!(Models::spent(&heartbeat, "nobody").is_err(), "a caller with no row is an error, not zero");
}

9
crates/mind-body/tests/persona_bin.rs
View file

@ -14,8 +14,7 @@ fn run(dir: &Path, drop: &[&str], extra: &[(&str, &str)], args: &[&str]) -> Outp
let mut env = vec![
("HUGINN_ENDPOINT", "rudp://127.0.0.1:9".to_string()),
("STATE_DIRECTORY", dir.to_string_lossy().into_owned()),
("MIND_PERSONA_ROOMS", "#swarm:chat,#burn:chat".to_string()),
("MIND_PERSONA_BURN_ROOM", "#burn:chat".to_string()),
("MIND_PERSONA_ROOMS", "#swarm:chat,#ops:chat".to_string()),
("MIND_PERSONA_OPERATOR", "@op:chat".to_string()),
("MIND_PERSONA_SELF_SENDER", "@eureka:chat".to_string()),
("MIND_CONNECTOR_ENDPOINT", "127.0.0.1:4110".to_string()),
@ -47,10 +46,8 @@ fn a_unit_configured_from_its_environment_alone_reaches_the_gate_and_ends_dead()
#[test]
fn a_flag_overrides_the_environment() {
let dir = tempfile::tempdir().unwrap();
// The burn room from the environment is not among the rooms the flag names.
let output = run(dir.path(), &[], &[], &["--instance", "eureka", "--rooms", "#swarm:chat"]);
assert_eq!(output.status.code(), Some(2));
let flagged = run(dir.path(), &[], &[], &["--instance", "eureka", "--rooms", "#swarm:chat,#burn:chat"]);
// The rooms flag replaces the rooms the environment names.
let flagged = run(dir.path(), &[], &[], &["--instance", "eureka", "--rooms", "#swarm:chat"]);
assert_eq!(flagged.status.code(), Some(0), "{}", text(&flagged.stderr));
}

36
crates/mind-body/tests/self_run.rs
View file

@ -164,7 +164,7 @@ mod lane {
turn: RunTurn::SelfRun,
host: "test".into(),
started_on: date(),
budget_usd: "7.25".into(),
budget_usd: "0".into(),
claims: vec![],
campaigns: vec![Slug(CAMPAIGN.into())],
}),
@ -234,17 +234,22 @@ mod lane {
panic!("the daemon never answered");
}
const GRANT: &[u8] = include_bytes!("../src/lease_client/lease_v1/grant.bin");
const SETTLED: &[u8] = include_bytes!("../src/lease_client/lease_v1/settled.bin");
/// The connector grant frame (lease v2): a length, then the named-map MessagePack of the grant with one plan window.
fn grant_frame() -> Vec<u8> {
let body = rmp_serde::to_vec_named(&serde_json::json!({"kind": "claude_lease_grant", "value": {"lease_id": "lease:0123456789abcdef0123456789abcdef", "plan_quota": [{
"window": "five_hour", "utilization_permille": 80, "resets_at_unix_ms": 1_793_307_000_000_u64, "status": "allowed", "observed_at_unix_ms": 1_793_306_000_000_u64
}]}})).unwrap();
let mut frame = u32::try_from(body.len()).unwrap().to_be_bytes().to_vec();
frame.extend_from_slice(&body);
frame
}
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.
/// The connector's lease socket: every connection gets the grant with a pipe holding
/// `LANE_TOKEN` attached as SCM_RIGHTS, and is then held open until the run lets go.
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>> {
@ -278,15 +283,13 @@ mod lane {
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);
}
rustix::net::sendmsg(&stream, &[IoSlice::new(&grant_frame())], &mut control, rustix::net::SendFlags::empty()).unwrap();
// The lease lasts while this connection is open: the run drops it to end it.
let _ = read_frame(&mut stream);
});
}
});
FakeConnector { _dir: dir, path, settles }
FakeConnector { _dir: dir, path }
}
const ENV_CANARY: &str = "canary-unit-environment-b94e";
@ -334,14 +337,14 @@ echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start
let stdout = String::from_utf8_lossy(&output.stdout).to_string();
assert_eq!(output.status.code(), Some(0), "{stdout}\n{}", String::from_utf8_lossy(&output.stderr));
assert_eq!(stdout, "request 1 prefix 101\nrequest 2 prefix 300000\nsettled: billed 123456 micro-USD\nended: flush-requested code 0 requests 2 prefix 300000\n");
assert_eq!(stdout, "request 1 prefix 101\nrequest 2 prefix 300000\nended: flush-requested code 0 requests 2 prefix 300000\n");
assert!(!all_output(&output).contains(ENV_CANARY), "the unit environment is not echoed");
assert!(String::from_utf8_lossy(&output.stderr).contains("journal-line"), "the child's stderr is the unit's journal");
let recorded = std::fs::read_to_string(&record).unwrap();
let mut lines = recorded.lines();
let argv = lines.next().unwrap();
assert!(argv.contains(&format!("--session-id {LANE_SESSION}")) && argv.contains("--max-budget-usd 7.25"), "{argv}");
assert!(argv.contains(&format!("--session-id {LANE_SESSION}")) && !argv.contains("--max-budget-usd"), "{argv}");
assert!(argv.starts_with("-p --input-format stream-json"), "{argv}");
assert!(lines.next().unwrap().contains("yggdrasil:run:selfrun-lane"), "the start message names her run");
let flush = lines.next().unwrap();
@ -349,9 +352,6 @@ echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start
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.