From bf3f610b66e4ac038eeb202501d82a9535942f32 Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 11:58:12 +0000 Subject: [PATCH] mind-stream: the thought's model comes from the Projector; connector key reader and membrane pin tested; swarm stop line derived Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/bin/mind-stream.rs | 14 +++---- crates/mind-body/src/cli.rs | 2 +- crates/mind-body/src/launch.rs | 2 +- crates/mind-body/src/stream.rs | 16 +++++++- crates/mind-body/src/stream/tests.rs | 50 +++++++++++++++++++++-- crates/mind-body/tests/cli.rs | 4 +- crates/mind-body/tests/public_surface.txt | 5 ++- crates/mind-body/tests/stream_bin.rs | 44 ++++++++++++++++++++ 8 files changed, 118 insertions(+), 19 deletions(-) create mode 100644 crates/mind-body/tests/stream_bin.rs diff --git a/crates/mind-body/src/bin/mind-stream.rs b/crates/mind-body/src/bin/mind-stream.rs index 1d86334..2604b40 100644 --- a/crates/mind-body/src/bin/mind-stream.rs +++ b/crates/mind-body/src/bin/mind-stream.rs @@ -24,7 +24,7 @@ use eureka_pipeline::Slug; use eureka_state::HuginnClient; use mind_body::control::read_effective; use mind_body::queue::HuginnMind; -use mind_body::stream::{HuginnPersonaSource, KitProjector, Mode, OllamaContinuer, Organs, Outcome, STREAM_MIN_GAP_S, STREAM_POLL_S, tick}; +use mind_body::stream::{HuginnPersonaSource, KitProjector, Mode, OllamaContinuer, Organs, Outcome, STREAM_MIN_GAP_S, STREAM_POLL_S, connector_key, tick}; const TIMEOUT: Duration = Duration::from_secs(30); /// How long the kit may take to answer one Projector turn. @@ -35,12 +35,10 @@ fn setting(name: &str) -> Option { std::env::var(name).ok().filter(|value| !value.is_empty()) } -/// The connector key: the credential file, read once and trimmed. An absence names the variable, never a value. +/// The credentials directory systemd hands the unit; its absence names the variable. #[allow(clippy::disallowed_methods, reason = "the binary reads its own environment, and hands it to the library")] -fn connector_key() -> Result { - let directory = std::env::var_os("CREDENTIALS_DIRECTORY").map(PathBuf::from).ok_or_else(|| anyhow!("CREDENTIALS_DIRECTORY is not set"))?; - let key = std::fs::read_to_string(directory.join("connector-key")).map_err(|_| anyhow!("the connector-key credential could not be read"))?; - Ok(key.trim().to_owned()) +fn credentials_directory() -> Result { + std::env::var_os("CREDENTIALS_DIRECTORY").map(PathBuf::from).ok_or_else(|| anyhow!("CREDENTIALS_DIRECTORY is not set")) } fn required(name: &str) -> Result { @@ -78,11 +76,11 @@ fn run() -> Result<()> { let projector_model = required("MIND_STREAM_PROJECTOR_MODEL")?; let base_model = required("MIND_STREAM_BASE_MODEL")?; let connector: SocketAddr = required("MIND_STREAM_CONNECTOR_ENDPOINT")?.parse().map_err(|_| anyhow!("MIND_STREAM_CONNECTOR_ENDPOINT is not :"))?; - let projector = KitProjector::open(connector, connector_key()?, &instance.0, &projector_model, PROJECTOR_TIMEOUT)?; + let projector = KitProjector::open(connector, connector_key(&credentials_directory()?)?, &instance.0, &projector_model, PROJECTOR_TIMEOUT)?; let continuer = OllamaContinuer::open(&required("MIND_STREAM_ENDPOINT")?, &base_model)?; let mind = HuginnMind::new(HuginnClient::new(address, instance.clone(), TIMEOUT)); let persona = HuginnPersonaSource::new(HuginnClient::new(address, instance.clone(), TIMEOUT)); - let organs = Organs { mind: &mind, persona: &persona, projector: &projector, continuer: &continuer, projector_model: &projector_model, base_model: &base_model }; + let organs = Organs { mind: &mind, persona: &persona, projector: &projector, continuer: &continuer, base_model: &base_model }; let mode = if args.dry_run { Mode::DryRun } else { Mode::Live }; let once = args.once || args.dry_run; loop { diff --git a/crates/mind-body/src/cli.rs b/crates/mind-body/src/cli.rs index 9087478..6227784 100644 --- a/crates/mind-body/src/cli.rs +++ b/crates/mind-body/src/cli.rs @@ -151,7 +151,7 @@ fn swarm( let written = SwarmWriter::open(&path).and_then(|mut writer| writer.set_brake(false, now, set_by)); let stopped = stop_units(StopScope::Swarm); if written.is_ok() && stopped.is_ok() { - writeln!(out, "units: stopped every mind-self@, mind-persona@ and mind-stream@ unit")?; + writeln!(out, "units: stopped every {}", stop_args(StopScope::Swarm)?[2..].join(" "))?; } held((written, stopped)) } diff --git a/crates/mind-body/src/launch.rs b/crates/mind-body/src/launch.rs index 281815e..5b13288 100644 --- a/crates/mind-body/src/launch.rs +++ b/crates/mind-body/src/launch.rs @@ -153,7 +153,7 @@ pub enum StopScope<'a> { /// `eureka-two`). The swarm's: each template's glob over the loaded units, /// which is exactly every running Mind and exits 0 when none is loaded. No /// `--no-block`: `systemctl stop` returns when the units are down. The -/// instance is validated by `unit()`. +/// instance is validated by `named()`. pub(crate) fn stop_args(scope: StopScope<'_>) -> Result> { let mut args = vec!["stop".to_string(), "--".to_string()]; args.extend(held_templates().into_iter().map(|template| match scope { diff --git a/crates/mind-body/src/stream.rs b/crates/mind-body/src/stream.rs index 21d5f83..8e15ed1 100644 --- a/crates/mind-body/src/stream.rs +++ b/crates/mind-body/src/stream.rs @@ -100,6 +100,8 @@ impl Sampling { /// The Projector: state to narrative. Its one owner is `KitProjector`, her own caller on the connector. A fault is an error: /// the tick ends, the process does not. pub trait Projector { + /// The model id every projection of this Projector names; the thought records it, so it is the model that ran. + fn model(&self) -> &str; /// One turn: `prompt` is the turn's one user text and `request_id` is its identity on the connector. fn project(&self, request_id: &str, prompt: &str) -> Result; } @@ -200,6 +202,13 @@ fn request_id(instance: &Slug, label: &str, mode: Mode, now: DateTime) -> S } } +/// The connector key: the credential file in systemd's credentials directory, read once and trimmed. Its error names the +/// credential and never a byte of the file. +pub fn connector_key(credentials: &std::path::Path) -> Result { + let key = std::fs::read_to_string(credentials.join("connector-key")).map_err(|_| anyhow!("the connector-key credential could not be read"))?; + Ok(key.trim().to_owned()) +} + /// The Projector as one tool-less connector turn by `caller` (her instance) on `model`. Loopback only, as the kit is. /// Each projection is its own conversation: nothing is carried from one tick to the next. pub struct KitProjector { @@ -220,6 +229,10 @@ impl KitProjector { } impl Projector for KitProjector { + fn model(&self) -> &str { + &self.model + } + fn project(&self, request_id: &str, prompt: &str) -> Result { let mut request = ProviderRequest::new(request_id, request_id, &self.model, PROJECTOR_INSTRUCTIONS, PROJECTOR_MAX_OUTPUT_TOKENS); request.input = vec![InputItem::UserText { text: prompt.to_owned() }]; @@ -594,7 +607,6 @@ pub struct Organs<'a> { pub persona: &'a dyn PersonaSource, pub projector: &'a dyn Projector, pub continuer: &'a dyn Continuer, - pub projector_model: &'a str, pub base_model: &'a str, } @@ -685,7 +697,7 @@ pub fn tick(organs: &Organs, brake: &dyn Fn() -> Effective, mode: Mode, now: Dat identity: Line::from(seed.identity.as_str()), context: Para::from(seed.context.as_str()), stimulus: Para::from(seed.stimulus.as_str()), - projector_model: Short(organs.projector_model.to_owned()), + projector_model: Short(organs.projector.model().to_owned()), projection: Para::from(projection.as_str()), base_model: Short(organs.base_model.to_owned()), temperature_permille: Sampling::STREAM.temperature_permille, diff --git a/crates/mind-body/src/stream/tests.rs b/crates/mind-body/src/stream/tests.rs index 958dcef..c83bb2b 100644 --- a/crates/mind-body/src/stream/tests.rs +++ b/crates/mind-body/src/stream/tests.rs @@ -73,6 +73,10 @@ impl Fake { } impl Projector for Fake { + fn model(&self) -> &str { + "projector:test" + } + fn project(&self, request_id: &str, prompt: &str) -> Result { self.projected.borrow_mut().push((request_id.into(), prompt.into())); if self.fault { Err(anyhow!("the connector is down")) } else { Ok(self.projection.clone()) } @@ -202,7 +206,7 @@ fn mind_with_thought(thought_at: i64) -> TestMind { } fn organs<'a>(mind: &'a dyn MindPort, persona: &'a Persona, model: &'a Fake) -> Organs<'a> { - Organs { mind, persona, projector: model, continuer: model, projector_model: "projector:test", base_model: "base:test" } + Organs { mind, persona, projector: model, continuer: model, base_model: "base:test" } } fn thoughts(mind: &TestMind) -> Vec { @@ -553,7 +557,7 @@ fn a_kit_refusal_is_a_tick_fault() { let continuer = Fake::saying("unused", &["unused"]); let persona = persona(); let wrap = Wrap::over(&mind, &[]); - let organs = Organs { mind: &wrap, persona: &persona, projector: &projector, continuer: &continuer, projector_model: KIT_MODEL, base_model: "base:test" }; + let organs = Organs { mind: &wrap, persona: &persona, projector: &projector, continuer: &continuer, base_model: "base:test" }; let fault = format!("{:#}", tick(&organs, &|| released(), Mode::Live, t(1000 + STREAM_IDLE_GAP_S)).unwrap_err()); assert!(fault.contains("ModelNotAllowed"), "{fault}"); assert!(backend.seen.lock().unwrap().is_empty(), "the provider was never reached"); @@ -561,6 +565,46 @@ fn a_kit_refusal_is_a_tick_fault() { assert!(wrap.admitted.borrow().is_empty()); } +#[test] +fn a_thought_names_the_model_the_kit_ran() { + let backend = Scripted::ending(StopReason::EndTurn, &["I turn it over."]); + let (projector, _root) = kit(Arc::clone(&backend), KIT_MODEL); + let mind = mind_with_thought(1000); + let continuer = Fake::saying("unused", &["A drift"]); + let persona = persona(); + let wrap = Wrap::over(&mind, &[0.1]); + let organs = Organs { mind: &wrap, persona: &persona, projector: &projector, continuer: &continuer, base_model: "base:test" }; + let Outcome::Thought { thought, .. } = tick(&organs, &|| released(), Mode::Live, t(1000 + STREAM_IDLE_GAP_S)).unwrap() else { panic!("a thought") }; + let ran = backend.seen.lock().unwrap()[0].1.model.clone(); + assert_eq!(ran, KIT_MODEL); + assert_eq!(thought.projector_model.0, ran, "the recorded model is the one the kit was asked for"); +} + +#[test] +fn the_instructions_restate_no_membrane_text() { + let membrane = build_projector_prompt(&ProjectorPrompt { identity: "", typed_context: "", visible_stimulus: "", domain_guidance: STREAM_GUIDANCE, word_budget: PROJECTOR_WORDS }).to_lowercase(); + let words: Vec<&str> = PROJECTOR_INSTRUCTIONS.split_whitespace().collect(); + assert!(!PROJECTOR_INSTRUCTIONS.contains('\n') && words.len() <= 20, "one short line"); + for run in words.windows(4) { + let run = run.join(" ").to_lowercase(); + assert!(!membrane.contains(&run), "Ghostlight's membrane text reached the instructions: {run}"); + } +} + +#[test] +fn the_connector_key_is_read_trimmed_and_its_absence_leaks_nothing() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("connector-key"), "canary-key-7f3a91\n").unwrap(); + assert_eq!(connector_key(dir.path()).unwrap(), "canary-key-7f3a91"); + let empty = tempfile::tempdir().unwrap(); + let fault = format!("{:#}", connector_key(empty.path()).unwrap_err()); + assert!(fault.contains("connector-key") && !fault.contains(empty.path().to_str().unwrap()), "{fault}"); + let bad = tempfile::tempdir().unwrap(); + std::fs::write(bad.path().join("connector-key"), b"canary-key-7f3a91\xff\xfe").unwrap(); + let fault = format!("{:#}", connector_key(bad.path()).unwrap_err()); + assert!(!fault.contains("canary"), "no byte of the key reaches an error: {fault}"); +} + #[test] fn a_projection_cut_short_is_a_fault() { for stop in [StopReason::MaxOutputTokens, StopReason::Refusal, StopReason::ToolUse, StopReason::ContextWindow] { @@ -837,7 +881,7 @@ fn old_repo_activity_does_not_make_a_tick_due() { let model = Fake::saying("I turn it over.", &["A new drift"]); let persona = CountedPersona(persona(), std::cell::Cell::new(0)); let wrap = Wrap::over(&mind, &[0.1]).with_forge(vec![push(1, 500)]); - let organs = Organs { mind: &wrap, persona: &persona, projector: &model, continuer: &model, projector_model: "projector:test", base_model: "base:test" }; + let organs = Organs { mind: &wrap, persona: &persona, projector: &model, continuer: &model, base_model: "base:test" }; assert_eq!(tick(&organs, &|| released(), Mode::Live, t(1901)).unwrap(), Outcome::NotDue); assert_eq!((model.calls(), persona.1.get(), thoughts(&mind).len()), (0, 0, 1), "no model call, no seed, no thought"); } diff --git a/crates/mind-body/tests/cli.rs b/crates/mind-body/tests/cli.rs index f23495f..1376db7 100644 --- a/crates/mind-body/tests/cli.rs +++ b/crates/mind-body/tests/cli.rs @@ -275,7 +275,7 @@ fn brake_hold_stops_once_with_the_instance_and_the_store_already_held() { result.unwrap(); assert_eq!(calls, [Some("eureka".to_string())]); assert!(seen_at_stop.contains("BrakeHeld"), "the write comes before the stop: {seen_at_stop}"); - assert!(out.contains("units: stopped mind-self@eureka.service mind-persona@eureka.service mind-stream@eureka.service"), "{out}"); + assert!(out.lines().any(|line| line == "units: stopped mind-self@eureka.service mind-persona@eureka.service mind-stream@eureka.service"), "{out}"); } #[test] @@ -475,7 +475,7 @@ fn swarm_brake_hold_stops_once_with_the_swarm_scope_and_the_swarm_store_already_ result.unwrap(); assert_eq!(calls, [None], "one stop, for the swarm"); assert!(seen_at_stop.contains("BrakeHeld"), "the write comes before the stop: {seen_at_stop}"); - assert!(out.contains("units: stopped every mind-self@, mind-persona@ and mind-stream@ unit"), "{out}"); + assert!(out.lines().any(|line| line == "units: stopped every mind-self@*.service mind-persona@*.service mind-stream@*.service"), "the line names exactly what stop_args stopped: {out}"); } #[test] diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 77a3a3a..10d7fb9 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -327,6 +327,7 @@ mind_body::stream: const AGENT : & str mind_body::stream: const STREAM_IDLE_GAP_S : i64 mind_body::stream: const STREAM_MIN_GAP_S : i64 mind_body::stream: const STREAM_POLL_S : i64 +mind_body::stream: fn connector_key (credentials : & std :: path :: Path) -> Result < String > mind_body::stream: fn tick (organs : & Organs , brake : & dyn Fn () -> Effective , mode : Mode , now : DateTime < Utc >) -> Result < Outcome > mind_body::stream: impl Continuer for OllamaContinuer mind_body::stream: impl HuginnPersonaSource @@ -341,11 +342,11 @@ mind_body::stream: impl Sampling mind_body::stream: impl Seed mind_body::stream: pub trait Continuer { fn continue_raw (& self , prompt : & str , sampling : & Sampling , n : usize) -> Result < Vec < String > > ; } mind_body::stream: pub trait PersonaSource { fn persona (& self) -> Result < PersonaRead > ; } -mind_body::stream: pub trait Projector { fn project (& self , request_id : & str , prompt : & str) -> Result < String > ; } +mind_body::stream: pub trait Projector { fn model (& self) -> & str ; fn project (& self , request_id : & str , prompt : & str) -> Result < String > ; } mind_body::stream: struct HuginnPersonaSource pub-fields[] mind_body::stream: struct KitProjector pub-fields[] mind_body::stream: struct OllamaContinuer pub-fields[] -mind_body::stream: struct Organs< 'a > pub-fields[mind: & 'a dyn MindPort, persona: & 'a dyn PersonaSource, projector: & 'a dyn Projector, continuer: & 'a dyn Continuer, projector_model: & 'a str, base_model: & 'a str] +mind_body::stream: struct Organs< 'a > pub-fields[mind: & 'a dyn MindPort, persona: & 'a dyn PersonaSource, projector: & 'a dyn Projector, continuer: & 'a dyn Continuer, base_model: & 'a str] mind_body::stream: struct PersonaRead pub-fields[state: Value, receipt_id: String] mind_body::stream: struct Sampling # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub-fields[temperature_permille: u16, min_p_permille: u16, max_tokens: u16] mind_body::tempo: const DEFAULT_BEAT_S : u32 diff --git a/crates/mind-body/tests/stream_bin.rs b/crates/mind-body/tests/stream_bin.rs new file mode 100644 index 0000000..012f833 --- /dev/null +++ b/crates/mind-body/tests/stream_bin.rs @@ -0,0 +1,44 @@ +//! The `mind-stream` binary's startup reads the connector key from systemd's credentials directory: +//! an absent directory names the variable, and the key's bytes never reach stderr. + +use std::process::{Command, Output}; + +const CANARY: &str = "canary-key-7f3a91"; + +fn run(credentials: Option<&std::path::Path>, connector: &str) -> Output { + let mut command = Command::new(env!("CARGO_BIN_EXE_mind-stream")); + command + .args(["--instance", "eureka", "--once"]) + .env_clear() + .env("HUGINN_ENDPOINT", "rudp://127.0.0.1:9") + .env("MIND_STREAM_CONNECTOR_ENDPOINT", connector) + .env("MIND_STREAM_PROJECTOR_MODEL", "claude-haiku-5-5") + .env("MIND_STREAM_BASE_MODEL", "base:test") + .env("MIND_STREAM_ENDPOINT", "http://127.0.0.1:1"); + if let Some(dir) = credentials { + command.env("CREDENTIALS_DIRECTORY", dir); + } + command.output().unwrap() +} + +#[test] +fn a_missing_credentials_directory_names_the_variable_and_a_missing_key_names_the_credential() { + let output = run(None, "127.0.0.1:4103"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert_eq!(output.status.code(), Some(1)); + assert!(stderr.contains("CREDENTIALS_DIRECTORY"), "{stderr}"); + let empty = tempfile::tempdir().unwrap(); + let stderr = String::from_utf8_lossy(&run(Some(empty.path()), "127.0.0.1:4103").stderr).to_string(); + assert!(stderr.contains("connector-key"), "{stderr}"); +} + +#[test] +fn the_key_is_never_printed() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("connector-key"), format!("{CANARY}\n")).unwrap(); + // A non-loopback connector ends startup after the key was read and handed over. + let output = run(Some(dir.path()), "10.77.0.1:4103"); + let shown = format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)); + assert!(shown.contains("not a loopback address"), "the key was read and startup went on to the endpoint check: {shown}"); + assert!(!shown.contains(CANARY), "{shown}"); +}