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 <noreply@anthropic.com>
This commit is contained in:
parent
4e9d1dbd23
commit
bf3f610b66
8 changed files with 118 additions and 19 deletions
|
|
@ -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<String> {
|
|||
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<String> {
|
||||
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<PathBuf> {
|
||||
std::env::var_os("CREDENTIALS_DIRECTORY").map(PathBuf::from).ok_or_else(|| anyhow!("CREDENTIALS_DIRECTORY is not set"))
|
||||
}
|
||||
|
||||
fn required(name: &str) -> Result<String> {
|
||||
|
|
@ -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 <ip>:<port>"))?;
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Vec<String>> {
|
||||
let mut args = vec!["stop".to_string(), "--".to_string()];
|
||||
args.extend(held_templates().into_iter().map(|template| match scope {
|
||||
|
|
|
|||
|
|
@ -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<String>;
|
||||
}
|
||||
|
|
@ -200,6 +202,13 @@ fn request_id(instance: &Slug, label: &str, mode: Mode, now: DateTime<Utc>) -> 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<String> {
|
||||
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<String> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<String> {
|
||||
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<String> {
|
||||
|
|
@ -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");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
44
crates/mind-body/tests/stream_bin.rs
Normal file
44
crates/mind-body/tests/stream_bin.rs
Normal file
|
|
@ -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}");
|
||||
}
|
||||
Loading…
Reference in a new issue