From 7972d971e4120e31165197c5f99bcd27bc69a46b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 05:35:34 +0100 Subject: [PATCH] integration tests follow the usage-only surface; connector_kit test removed (WIP) Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/self_run/tests.rs | 1 + crates/mind-body/src/self_run/tests/lease.rs | 42 ++++- crates/mind-body/tests/connector_kit.rs | 180 ------------------- crates/mind-body/tests/persona_bin.rs | 9 +- crates/mind-body/tests/self_run.rs | 36 ++-- 5 files changed, 62 insertions(+), 206 deletions(-) delete mode 100644 crates/mind-body/tests/connector_kit.rs diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 44a2987..4cabed5 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -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::*; diff --git a/crates/mind-body/src/self_run/tests/lease.rs b/crates/mind-body/src/self_run/tests/lease.rs index 37d931a..3841e46 100644 --- a/crates/mind-body/src/self_run/tests/lease.rs +++ b/crates/mind-body/src/self_run/tests/lease.rs @@ -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"); } diff --git a/crates/mind-body/tests/connector_kit.rs b/crates/mind-body/tests/connector_kit.rs deleted file mode 100644 index e991015..0000000 --- a/crates/mind-body/tests/connector_kit.rs +++ /dev/null @@ -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>, -} - -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, - _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"); -} diff --git a/crates/mind-body/tests/persona_bin.rs b/crates/mind-body/tests/persona_bin.rs index 64f4961..e78c2e3 100644 --- a/crates/mind-body/tests/persona_bin.rs +++ b/crates/mind-body/tests/persona_bin.rs @@ -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)); } diff --git a/crates/mind-body/tests/self_run.rs b/crates/mind-body/tests/self_run.rs index b8a0e52..aebb3d8 100644 --- a/crates/mind-body/tests/self_run.rs +++ b/crates/mind-body/tests/self_run.rs @@ -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 { + 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>>>, } fn read_frame(stream: &mut std::os::unix::net::UnixStream) -> std::io::Result> { @@ -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.