diff --git a/Cargo.lock b/Cargo.lock index e563a9a..e792fdd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1178,6 +1178,7 @@ source = "git+https://eureka.gamecult.org/forge/GameCult/ModelConnector.git?rev= dependencies = [ "aes-gcm", "base64", + "cultcache-rs 0.3.1", "cultnet-rs 0.2.0", "rand_core 0.6.4", "rmp-serde", diff --git a/crates/mind-body/Cargo.toml b/crates/mind-body/Cargo.toml index a8fdd9e..ffacef1 100644 --- a/crates/mind-body/Cargo.toml +++ b/crates/mind-body/Cargo.toml @@ -25,7 +25,8 @@ publish.workspace = true # # The stream organ (`stream`, the `mind-stream` binary) builds its Projector prompt with Ghostlight's # `ghostlight-persona-projection`, pinned to a revision, so the membrane's text stays Ghostlight's; it -# speaks to a model host on loopback over `ureq` 3 with default features off (plain HTTP, no TLS), as the +# asks its Projector through the connector kit's client (`model-connector`, one turn per tick) and speaks to the base +# model's host on loopback over `ureq` 3 with default features off (plain HTTP, no TLS), as the # daemon's embedder does, and stays below 3.4 for the same reason. # # `serde_json` reads Claude Code's stream-json, a foreign owner's wire (`self_run`); @@ -63,6 +64,9 @@ proc-macro2 = "1" quote = "1" syn = { version = "2", features = ["full"] } tempfile = "3" +# A real kit server (`serve` over a scripted `ProviderBackend`) proves the stream's Projector wire, not a fake of it. +# Same revision as the dependency; the `daemon` feature is test-only. +model-connector = { git = "https://eureka.gamecult.org/forge/GameCult/ModelConnector.git", rev = "c1802d85bee3ae04b0c5737563c8343c1fef515d", package = "model-connector", features = ["daemon"] } [lib] name = "mind_body" diff --git a/crates/mind-body/src/bin/mind-stream.rs b/crates/mind-body/src/bin/mind-stream.rs index aeeec1f..2604b40 100644 --- a/crates/mind-body/src/bin/mind-stream.rs +++ b/crates/mind-body/src/bin/mind-stream.rs @@ -7,10 +7,14 @@ //! the library never does. //! //! Environment: `HUGINN_ENDPOINT` (`rudp://:`, her own daemon), `EUREKA_INSTANCE` -//! (when `--instance` is not given), `MIND_STREAM_ENDPOINT` (`http://:`, the -//! model host), `MIND_STREAM_PROJECTOR_MODEL` and `MIND_STREAM_BASE_MODEL`. +//! (when `--instance` is not given), `MIND_STREAM_CONNECTOR_ENDPOINT` (`:`, the +//! connector kit the Projector is asked through), `MIND_STREAM_PROJECTOR_MODEL` (the model id the +//! kit's caller row allows), `MIND_STREAM_ENDPOINT` (`http://:`, the model host) and +//! `MIND_STREAM_BASE_MODEL`. The connector key is the file `connector-key` in `CREDENTIALS_DIRECTORY` +//! (systemd's `LoadCredential`), read once; the caller id is the instance. use std::net::SocketAddr; +use std::path::PathBuf; use std::process::ExitCode; use std::time::Duration; @@ -20,15 +24,23 @@ use eureka_pipeline::Slug; use eureka_state::HuginnClient; use mind_body::control::read_effective; use mind_body::queue::HuginnMind; -use mind_body::stream::{HuginnPersonaSource, Mode, OllamaStream, 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. +const PROJECTOR_TIMEOUT: Duration = Duration::from_secs(300); #[allow(clippy::disallowed_methods, reason = "the binary reads its own environment, and hands it to the library")] fn setting(name: &str) -> Option { std::env::var(name).ok().filter(|value| !value.is_empty()) } +/// 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 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 { setting(name).ok_or_else(|| anyhow!("{name} is not set")) } @@ -63,10 +75,12 @@ fn run() -> Result<()> { .ok_or_else(|| anyhow!("HUGINN_ENDPOINT is not rudp://:"))?; let projector_model = required("MIND_STREAM_PROJECTOR_MODEL")?; let base_model = required("MIND_STREAM_BASE_MODEL")?; - let model = OllamaStream::open(&required("MIND_STREAM_ENDPOINT")?, &projector_model, &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(&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, model: &model, 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 35b048b..6227784 100644 --- a/crates/mind-body/src/cli.rs +++ b/crates/mind-body/src/cli.rs @@ -28,12 +28,12 @@ use std::path::Path; use anyhow::{Result, anyhow, bail}; use chrono::{DateTime, Utc}; -use eureka_pipeline::{RunTurn, Slug}; +use eureka_pipeline::Slug; use crate::control::{ ControlWriter, Effective, FileSource, SwarmWriter, control_path, derive, derive_swarm, load_state, load_swarm_state, swarm_path, }; -use crate::launch::{StopScope, unit}; +use crate::launch::{StopScope, stop_args}; const USAGE: &str = "usage: mind-control [--root ] --instance show | check | brake hold | brake release | mind-control [--root ] swarm show | brake hold | brake release"; @@ -110,7 +110,7 @@ pub fn run( let stopped = stop_units(StopScope::Mind(&instance)); match (written, stopped) { (Ok(()), Ok(())) => { - writeln!(out, "units: stopped {} {}", unit(&instance, RunTurn::SelfRun)?, unit(&instance, RunTurn::PersonaTurn)?)?; + writeln!(out, "units: stopped {}", stop_args(StopScope::Mind(&instance))?[2..].join(" "))?; Ok(()) } outcome => held(outcome), @@ -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@ and mind-persona@ 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 b954134..5b13288 100644 --- a/crates/mind-body/src/launch.rs +++ b/crates/mind-body/src/launch.rs @@ -124,15 +124,23 @@ fn template(turn: RunTurn) -> &'static str { } } -/// The unit of one turn: the template by turn, then the instance, held to the -/// leaf's grammar first so it cannot escape the name. The error names nothing -/// of the value. Start and alive both name the unit here. -pub(crate) fn unit(instance: &Slug, turn: RunTurn) -> Result { +/// The stream's unit template: not a run slot, so only a brake hold names it. +const STREAM_TEMPLATE: &str = "mind-stream"; + +/// A template's unit for one instance, the instance held to the leaf's grammar +/// first so it cannot escape the name. The error names nothing of the value. +fn named(template: &str, instance: &Slug) -> Result { instance.validate_slug().map_err(|_| anyhow!("the instance is not a unit name"))?; - Ok(format!("{}@{}.service", template(turn), instance.0)) + Ok(format!("{template}@{}.service", instance.0)) } -/// What one stop reaches: the two units of one Mind, or every Mind's. +/// The unit of one turn: the template by turn, then the instance. Start and +/// alive both name the unit here. +pub(crate) fn unit(instance: &Slug, turn: RunTurn) -> Result { + named(template(turn), instance) +} + +/// What one stop reaches: the three units of one Mind, or every Mind's. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum StopScope<'a> { Mind(&'a Slug), @@ -141,22 +149,25 @@ pub enum StopScope<'a> { /// The arguments of the one synchronous stop. A Mind's: the exact unit `unit()` /// names for each turn of this instance, the same names start and alive use, -/// with no glob (a prefix glob also matches another instance, `eureka*` matches +/// and its stream's, with no glob (a prefix glob also matches another instance, `eureka*` matches /// `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()]; - for turn in [RunTurn::SelfRun, RunTurn::PersonaTurn] { - args.push(match scope { - StopScope::Mind(instance) => unit(instance, turn)?, - StopScope::Swarm => format!("{}@*.service", template(turn)), - }); - } + args.extend(held_templates().into_iter().map(|template| match scope { + StopScope::Mind(instance) => named(template, instance), + StopScope::Swarm => Ok(format!("{template}@*.service")), + }).collect::>>()?); Ok(args) } +/// The templates a hold stops: both run slots and the stream. +fn held_templates() -> [&'static str; 3] { + [template(RunTurn::SelfRun), template(RunTurn::PersonaTurn), STREAM_TEMPLATE] +} + impl SystemdLauncher { /// Stops the units of one Mind, or of every Mind, and waits for them. An /// inherent method, not on `Launcher`: only a brake hold calls it, through @@ -173,10 +184,10 @@ impl SystemdLauncher { } let states = match scope { StopScope::Mind(instance) => { - let states = [RunTurn::SelfRun, RunTurn::PersonaTurn] + let states = held_templates() .into_iter() - .map(|turn| { - let unit = unit(instance, turn)?; + .map(|template| { + let unit = named(template, instance)?; let state = self.unit_state(&unit); Ok(format!("{unit} is {state}")) }) @@ -1486,10 +1497,12 @@ mod tests { launcher.start(&instance, turn).unwrap(); started.push(spy.calls.borrow().last().unwrap().last().unwrap().clone()); } - assert_eq!(args[2..], started[..], "stop names exactly the units start names, for both turns"); + assert_eq!(args[2..4], started[..], "stop names exactly the units start names, for both turns"); + assert_eq!(args[4], format!("mind-stream@{name}.service"), "and the stream of that instance"); + assert_eq!(args.len(), 5); assert!(args[2..].iter().all(|unit| !unit.contains(['*', '?', '['])), "a glob can match another instance or none: {args:?}"); } - assert_eq!(stop_args(StopScope::Mind(&eureka())).unwrap(), ["stop", "--", "mind-self@eureka.service", "mind-persona@eureka.service"]); + assert_eq!(stop_args(StopScope::Mind(&eureka())).unwrap(), ["stop", "--", "mind-self@eureka.service", "mind-persona@eureka.service", "mind-stream@eureka.service"]); } #[test] @@ -1503,7 +1516,7 @@ mod tests { fn the_swarm_stop_is_one_stop_of_both_templates_globs() { let (launcher, spy) = spied(); launcher.stop(StopScope::Swarm).unwrap(); - assert_eq!(*spy.calls.borrow(), vec![vec!["stop", "--", "mind-self@*.service", "mind-persona@*.service"]]); + assert_eq!(*spy.calls.borrow(), vec![vec!["stop", "--", "mind-self@*.service", "mind-persona@*.service", "mind-stream@*.service"]]); } #[test] @@ -1535,12 +1548,12 @@ mod tests { let (launcher, spy) = spied(); spy.success.set(false); spy.code.set(Some(5)); - spy.states.borrow_mut().extend([("mind-self@eureka.service", "inactive\n"), ("mind-persona@eureka.service", "active\n")]); + spy.states.borrow_mut().extend([("mind-self@eureka.service", "inactive\n"), ("mind-persona@eureka.service", "active\n"), ("mind-stream@eureka.service", "failed\n")]); let shown = format!("{:#}", launcher.stop(StopScope::Mind(&eureka())).unwrap_err()); assert!(shown.contains("exit status 5"), "{shown}"); - assert!(shown.contains("mind-self@eureka.service is inactive") && shown.contains("mind-persona@eureka.service is active"), "{shown}"); + assert!(shown.contains("mind-self@eureka.service is inactive") && shown.contains("mind-persona@eureka.service is active") && shown.contains("mind-stream@eureka.service is failed"), "{shown}"); let asked: Vec> = spy.calls.borrow().iter().skip(1).cloned().collect(); - assert_eq!(asked, [["is-active", "mind-self@eureka.service"], ["is-active", "mind-persona@eureka.service"]], "one query per unit, after the one stop"); + assert_eq!(asked, [["is-active", "mind-self@eureka.service"], ["is-active", "mind-persona@eureka.service"], ["is-active", "mind-stream@eureka.service"]], "one query per unit, after the one stop"); // A state word outside the three, and anything systemctl prints with it, is not carried. spy.states.borrow_mut().clear(); *spy.stdout.borrow_mut() = "canary-output-7f3a91\n".to_string(); diff --git a/crates/mind-body/src/stream.rs b/crates/mind-body/src/stream.rs index ddef05f..8e15ed1 100644 --- a/crates/mind-body/src/stream.rs +++ b/crates/mind-body/src/stream.rs @@ -13,7 +13,7 @@ //! and never starts a run. Delivery to a Persona is another organ's. use std::collections::BTreeMap; -use std::net::IpAddr; +use std::net::{IpAddr, SocketAddr}; use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; @@ -25,7 +25,9 @@ use eureka_pipeline::{ use eureka_state::HuginnClient; use ghostlight_persona_projection::{ProjectorPrompt, build_projector_prompt}; use huginn_mind::{HuginnMindResponse, PipelineAdmissionOutcome, PipelineDocumentView, PipelinePageItems}; +use model_connector::{ConnectorClient, InputItem, ProviderRequest, ReasoningEffort, StopReason, TransportDisposition, TransportEventPayload, TransportInvocation, TransportOutcome}; use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; use crate::control::Effective; use crate::persona::{Responsibilities, outcome_words, quoted, recent_runs, render_responsibilities, repo_notice_line}; @@ -95,25 +97,31 @@ impl Sampling { const STREAM: Self = Self { temperature_permille: TEMPERATURE_PERMILLE, min_p_permille: MIN_P_PERMILLE, max_tokens: MAX_TOKENS }; } -/// The two models of a tick. Faults are errors: the tick ends, the process does not. -pub trait StreamModel { - /// The Projector: one chat prompt in, one projection out. - fn project(&self, prompt: &str) -> Result; - /// The continuer: `n` raw continuations of `prompt`, no chat template. +/// 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; +} + +/// The continuer: raw continuation of a frame. Its one owner is `OllamaContinuer`, the base model on this host. +pub trait Continuer { + /// `n` raw continuations of `prompt`, no chat template. fn continue_raw(&self, prompt: &str, sampling: &Sampling, n: usize) -> Result>; } -/// Ollama on this host. Loopback only, as Ghostlight's local binding is. -pub struct OllamaStream { +/// The base model on Ollama on this host. Loopback only, as Ghostlight's local binding is. +pub struct OllamaContinuer { base: String, - projector_model: String, base_model: String, agent: ureq::Agent, } -impl OllamaStream { +impl OllamaContinuer { /// `endpoint` is `http://:`; any other address is refused here. - pub fn open(endpoint: &str, projector_model: &str, base_model: &str) -> Result { + pub fn open(endpoint: &str, base_model: &str) -> Result { // Parsed as the HTTP client parses it, so the host checked is the host connected to. let uri: ureq::http::Uri = endpoint.parse().map_err(|_| anyhow!("the stream endpoint is not a URL"))?; let authority = uri.authority().filter(|_| uri.scheme_str() == Some("http")).ok_or_else(|| anyhow!("the stream endpoint must be http://:"))?; @@ -137,7 +145,7 @@ impl OllamaStream { .max_redirects(0) .build() .into(); - Ok(Self { base: format!("http://{authority}"), projector_model: projector_model.to_owned(), base_model: base_model.to_owned(), agent }) + Ok(Self { base: format!("http://{authority}"), base_model: base_model.to_owned(), agent }) } fn post(&self, path: &str, body: &Value) -> Result { @@ -152,12 +160,7 @@ impl OllamaStream { } } -impl StreamModel for OllamaStream { - fn project(&self, prompt: &str) -> Result { - let answer = self.post("/api/chat", &json!({ "model": self.projector_model, "stream": false, "messages": [{ "role": "user", "content": prompt }] }))?; - answer["message"]["content"].as_str().map(str::to_owned).ok_or_else(|| anyhow!("the Projector's answer has no message content")) - } - +impl Continuer for OllamaContinuer { fn continue_raw(&self, prompt: &str, sampling: &Sampling, n: usize) -> Result> { let body = json!({ "model": self.base_model, @@ -178,6 +181,82 @@ impl StreamModel for OllamaStream { } } +// --------------------------------------------------------------------------------------------- +// The Projector: her own caller on the connector. + +/// What the connector's request asks of the model beside the user text. It names no membrane text: that is Ghostlight's. +const PROJECTOR_INSTRUCTIONS: &str = "Write only the text the request asks for."; +/// Room for the model's adaptive thinking; the projection itself is at most `PROJECTOR_WORDS` words. +const PROJECTOR_MAX_OUTPUT_TOKENS: u32 = 1024; +/// The sealed frame the client will carry: the prompt out and the projection back fit with wide room. +const CONNECTOR_FRAME_BYTES: usize = 256 * 1024; +/// How long an invocation stays admissible at the connector. +const INVOCATION_LIFE_MS: u64 = 60_000; + +/// The identity of one projection on the connector. A live tick's is derived from the instance and the tick label alone; a +/// dry run's also carries the moment, because the kit's replay store answers a repeated id byte-identically. +fn request_id(instance: &Slug, label: &str, mode: Mode, now: DateTime) -> String { + match mode { + Mode::Live => format!("{AGENT}:{}:{label}", instance.0), + Mode::DryRun => format!("{AGENT}:{}:{label}:dry{}", instance.0, now.timestamp_millis()), + } +} + +/// 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 { + client: ConnectorClient, + caller: String, + model: String, +} + +impl KitProjector { + /// `endpoint` is the kit's loopback address; `connection_key` is the caller's key, read by the binary and never printed. + pub fn open(endpoint: SocketAddr, connection_key: String, caller: &str, model: &str, timeout: Duration) -> Result { + if !endpoint.ip().is_loopback() { + bail!("the connector endpoint is not a loopback address"); + } + let client = ConnectorClient::new(endpoint, connection_key, CONNECTOR_FRAME_BYTES, Some(timeout)).map_err(|_| anyhow!("the connector client could not be configured"))?; + Ok(Self { client, caller: caller.to_owned(), model: model.to_owned() }) + } +} + +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() }]; + request.reasoning_effort = Some(ReasoningEffort::Low); + let expires = u64::try_from(Utc::now().timestamp_millis()).unwrap_or_default() + INVOCATION_LIFE_MS; + let invocation = TransportInvocation::new(&self.caller, expires, Sha256::digest(prompt.as_bytes()).into(), request).map_err(|_| anyhow!("the Projector's request is not a valid connector request"))?; + let result = self.client.execute(&invocation).map_err(|fault| anyhow!("the connector did not answer the Projector: {fault}"))?; + match result.disposition { + TransportDisposition::Refused(reason) => bail!("the connector refused the Projector: {reason:?}"), + TransportDisposition::Transported { events, receipt } => match receipt.outcome { + TransportOutcome::Completed { stop_reason: StopReason::EndTurn, .. } => Ok(events + .into_iter() + .filter_map(|event| match event.payload { + TransportEventPayload::TextDelta { text } => Some(text), + _ => None, + }) + .collect()), + TransportOutcome::Completed { stop_reason, .. } => bail!("the Projector's turn ended early: {stop_reason:?}"), + TransportOutcome::Failed { kind, .. } => bail!("the provider failed the Projector: {kind:?}"), + }, + } + } +} + // --------------------------------------------------------------------------------------------- // Her persona state, read and never put. @@ -526,8 +605,8 @@ pub enum Mode { pub struct Organs<'a> { pub mind: &'a dyn MindPort, pub persona: &'a dyn PersonaSource, - pub model: &'a dyn StreamModel, - pub projector_model: &'a str, + pub projector: &'a dyn Projector, + pub continuer: &'a dyn Continuer, pub base_model: &'a str, } @@ -586,8 +665,9 @@ pub fn tick(organs: &Organs, brake: &dyn Fn() -> Effective, mode: Mode, now: Dat let persona = organs.persona.persona()?; let seed = build_seed(mind, &persona, since, now, &responsibilities, fresh)?; + let label = format!("t{}", now.timestamp()); let began = std::time::Instant::now(); - let projection = bounded(&organs.model.project(&seed.projector_prompt())?, PARA_MAX_BYTES); + let projection = bounded(&organs.projector.project(&request_id(mind.instance(), &label, mode, now), &seed.projector_prompt())?, PARA_MAX_BYTES); let projector_ms = u32::try_from(began.elapsed().as_millis()).unwrap_or(u32::MAX); if projection.is_empty() { return Ok(Outcome::Blank); @@ -595,7 +675,7 @@ pub fn tick(organs: &Organs, brake: &dyn Fn() -> Effective, mode: Mode, now: Dat let frame = continuer_frame(text_of(&persona.state, &["publicName"]), &now.format(FRAME_DATE_FORMAT).to_string(), &projection); let began = std::time::Instant::now(); let candidates: Vec = organs - .model + .continuer .continue_raw(&frame, &Sampling::STREAM, CONTINUATIONS)? .iter() .map(|text| bounded(text, LINE_MAX_BYTES)) @@ -610,14 +690,14 @@ pub fn tick(organs: &Organs, brake: &dyn Fn() -> Effective, mode: Mode, now: Dat let chosen = least_near(&nearest_permille); let thought = PipelineThought { instance: mind.instance().clone(), - label: Label::from(format!("t{}", now.timestamp()).as_str()), + label: Label::from(label.as_str()), text: Para::from(candidates[chosen].as_str()), reads: seed.reads.clone(), persona_receipt: Short(persona.receipt_id.clone()), 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, @@ -633,7 +713,6 @@ pub fn tick(organs: &Organs, brake: &dyn Fn() -> Effective, mode: Mode, now: Dat if mode == Mode::DryRun { return Ok(Outcome::Thought { thought: Box::new(thought), admitted: false }); } - let label = thought.label.0.clone(); match mind.admit(AGENT, &format!("stream-{label}"), vec![PipelineDocument::Thought(thought.clone())])? { PipelineAdmissionOutcome::Committed { .. } => Ok(Outcome::Thought { thought: Box::new(thought), admitted: true }), other => Err(anyhow!("the mind did not admit the thought: {other:?}")), diff --git a/crates/mind-body/src/stream/tests.rs b/crates/mind-body/src/stream/tests.rs index 36797cf..da8d620 100644 --- a/crates/mind-body/src/stream/tests.rs +++ b/crates/mind-body/src/stream/tests.rs @@ -58,7 +58,7 @@ struct Fake { projection: String, candidates: Vec, fault: bool, - projected: RefCell>, + projected: RefCell>, continued: RefCell>, } @@ -72,12 +72,18 @@ impl Fake { } } -impl StreamModel for Fake { - fn project(&self, prompt: &str) -> Result { - self.projected.borrow_mut().push(prompt.into()); - if self.fault { Err(anyhow!("the model host is down")) } else { Ok(self.projection.clone()) } +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()) } + } +} + +impl Continuer for Fake { fn continue_raw(&self, prompt: &str, sampling: &Sampling, n: usize) -> Result> { self.continued.borrow_mut().push((prompt.into(), *sampling, n)); Ok(self.candidates.iter().take(n).cloned().collect()) @@ -200,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, 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 { @@ -414,8 +420,8 @@ fn frame_holds_projection_verbatim() { assert!(!prompt.contains(foreign), "{foreign} reached the continuer"); } let projected = model.projected.borrow(); - assert!(projected[0].contains("private Projector") && projected[0].contains(STREAM_GUIDANCE), "the Projector prompt is Ghostlight's builder over her seed"); - assert!(projected[0].contains("Libby holds the sounds")); + assert!(projected[0].1.contains("private Projector") && projected[0].1.contains(STREAM_GUIDANCE), "the Projector prompt is Ghostlight's builder over her seed"); + assert!(projected[0].1.contains("Libby holds the sounds")); } #[test] @@ -465,7 +471,202 @@ fn stream_admits_only_thoughts() { } // --------------------------------------------------------------------------------------------- -// The model port against a fake HTTP server. +// The Projector against a real kit server. + +use model_connector::{CallerAdmission, ProviderBackend, ProviderCall, ServeLimits, StopReason, TransportEvent, TransportResult, Usage, serve as serve_kit}; + +const KIT_KEY: &str = "stream-projector-test-key"; +const KIT_MODEL: &str = "claude-haiku-5-5"; + +/// Answers every call as scripted and records the caller and the request the provider would have been shown. +struct Scripted { + outcome: TransportOutcome, + words: Vec<&'static str>, + seen: Mutex>, +} + +impl Scripted { + fn ending(stop_reason: StopReason, words: &[&'static str]) -> Arc { + let usage = Usage { input_tokens: 1, cache_read_tokens: 0, cache_write_tokens: 0, output_tokens: 1, reasoning_tokens: None }; + Arc::new(Self { outcome: TransportOutcome::Completed { provider_response_id: Some("msg-1".into()), stop_reason, usage }, words: words.to_vec(), seen: Mutex::default() }) + } +} + +impl ProviderBackend for Scripted { + fn ready(&self) -> Result<(), &'static str> { + Ok(()) + } + + fn provider(&self) -> &str { + "anthropic-messages" + } + + fn execute(&self, call: &ProviderCall) -> TransportResult { + self.seen.lock().unwrap().push((call.caller_runtime_id().to_owned(), call.request().clone())); + let events = self.words.iter().enumerate().map(|(at, text)| TransportEvent { sequence: at as u64, payload: TransportEventPayload::TextDelta { text: (*text).into() } }).collect(); + call.transported(events, self.outcome.clone()) + } +} + +/// A running kit whose caller row for `instance` allows `allowed`, and a Projector opened on it. +fn kit(backend: Arc, allowed: &str) -> (KitProjector, tempfile::TempDir) { + let root = tempfile::tempdir().unwrap(); + let store = root.path().join("replay.cc"); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = listener.local_addr().unwrap(); + let admission = CallerAdmission::new(INSTANCE, KIT_KEY, 1, [allowed.to_owned()], 2, 64 * 1024, 4096).unwrap(); + let limits = ServeLimits { max_frame_bytes: 512 * 1024, max_connections: 8, socket_timeout_ms: 5000, max_expiry_skew_ms: 300_000 }; + std::thread::spawn(move || serve_kit(listener, limits, vec![admission], &store, backend)); + (KitProjector::open(endpoint, KIT_KEY.into(), INSTANCE, KIT_MODEL, Duration::from_secs(5)).unwrap(), root) +} + +#[test] +fn projector_is_one_kit_turn() { + let backend = Scripted::ending(StopReason::EndTurn, &["I turn ", "it over."]); + let (projector, _root) = kit(Arc::clone(&backend), KIT_MODEL); + let id = "mind-stream:eureka:t100"; + assert_eq!(projector.project(id, "The projector prompt.").unwrap(), "I turn it over."); + let seen = backend.seen.lock().unwrap(); + let (caller, request) = &seen[0]; + assert_eq!(caller, INSTANCE, "her instance is the caller"); + assert_eq!((request.request_id.as_str(), request.conversation_id.as_str(), request.model.as_str()), (id, id, KIT_MODEL)); + assert_eq!(request.input, [InputItem::UserText { text: "The projector prompt.".into() }]); + assert_eq!(request.instructions, PROJECTOR_INSTRUCTIONS); + assert_eq!(request.reasoning_effort, Some(ReasoningEffort::Low)); + assert!(request.tools.is_empty() && request.output_schema_json.is_none()); + assert_eq!(request.max_output_tokens, PROJECTOR_MAX_OUTPUT_TOKENS); +} + +#[test] +fn each_projection_is_its_own_conversation() { + let backend = Scripted::ending(StopReason::EndTurn, &["One."]); + let (projector, _root) = kit(Arc::clone(&backend), KIT_MODEL); + projector.project("mind-stream:eureka:t100", "first").unwrap(); + projector.project("mind-stream:eureka:t200", "second").unwrap(); + let seen = backend.seen.lock().unwrap(); + assert_ne!(seen[0].1.request_id, seen[1].1.request_id); + assert_ne!(seen[0].1.conversation_id, seen[1].1.conversation_id); + assert_eq!(seen[1].1.input, [InputItem::UserText { text: "second".into() }], "no transcript is carried"); +} + +#[test] +fn a_kit_refusal_is_a_tick_fault() { + let backend = Scripted::ending(StopReason::EndTurn, &["Never said."]); + let (projector, _root) = kit(Arc::clone(&backend), "some-other-model"); + let mind = mind_with_thought(1000); + 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, 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"); + assert_eq!(continuer.calls(), 0, "the continuer is not called"); + 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"); + for (written, read) in [("canary-key-7f3a91\r\n", "canary-key-7f3a91"), ("canary-key-7f3a91\n\n\n", "canary-key-7f3a91"), (" \t\ncanary-key-7f3a91 \t\r\n\n", "canary-key-7f3a91"), ("canary-key-7f3a91", "canary-key-7f3a91")] { + std::fs::write(dir.path().join("connector-key"), written).unwrap(); + assert_eq!(connector_key(dir.path()).unwrap(), read, "{written:?}"); + } + 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 no_open_error_carries_any_part_of_the_key() { + let key = "zq7Kx-9wVtm3Pd"; + // Every way open() can fail with the key in hand: a non-loopback endpoint and a configuration the client refuses. + for endpoint in ["10.77.0.1:4103", "127.0.0.1:0", "0.0.0.0:4103"] { + let fault = format!("{:#}", KitProjector::open(endpoint.parse().unwrap(), key.into(), INSTANCE, KIT_MODEL, Duration::from_secs(1)).err().unwrap()); + for window in key.as_bytes().windows(4) { + assert!(!fault.contains(std::str::from_utf8(window).unwrap()), "{endpoint}: part of the key reached the error: {fault}"); + } + } +} + +#[test] +fn a_projection_cut_short_is_a_fault() { + for stop in [StopReason::MaxOutputTokens, StopReason::Refusal, StopReason::ToolUse, StopReason::ContextWindow] { + let (projector, _root) = kit(Scripted::ending(stop, &["Half a thou"]), KIT_MODEL); + let fault = format!("{:#}", projector.project("mind-stream:eureka:t1", "p").unwrap_err()); + assert!(fault.contains("ended early") && !fault.contains("Half a thou"), "{stop:?}: {fault}"); + } + let failed = Arc::new(Scripted { outcome: TransportOutcome::Failed { kind: model_connector::FailureKind::RateLimited, message: "slow down".into(), usage: None }, words: vec![], seen: Mutex::default() }); + let (projector, _root) = kit(failed, KIT_MODEL); + assert!(format!("{:#}", projector.project("mind-stream:eureka:t1", "p").unwrap_err()).contains("RateLimited")); +} + +#[test] +fn non_loopback_connector_refused() { + for refused in ["10.77.0.1:4103", "192.0.2.1:4103", "[2001:db8::1]:4103"] { + assert!(KitProjector::open(refused.parse().unwrap(), KIT_KEY.into(), INSTANCE, KIT_MODEL, Duration::from_secs(1)).is_err(), "{refused} must be refused"); + } + for opened in ["127.0.0.1:4103", "[::1]:4103"] { + assert!(KitProjector::open(opened.parse().unwrap(), KIT_KEY.into(), INSTANCE, KIT_MODEL, Duration::from_secs(1)).is_ok(), "{opened} is loopback"); + } +} + +#[test] +fn request_ids_name_the_tick_and_a_dry_run_never_takes_the_live_one() { + let mind = mind_with_thought(1000); + let persona = persona(); + let at = t(1000 + STREAM_IDLE_GAP_S); + let run = |mode, now: DateTime| { + let model = Fake::saying("I turn it over.", &["A drift"]); + let wrap = Wrap::over(&mind, &[0.1]); + tick(&organs(&wrap, &persona, &model), &|| released(), mode, now).unwrap(); + model.projected.into_inner().remove(0).0 + }; + let live = run(Mode::Live, at); + assert_eq!(live, format!("mind-stream:{INSTANCE}:t{}", at.timestamp())); + let dry = run(Mode::DryRun, at); + let later_dry = run(Mode::DryRun, at + Span::milliseconds(5)); + assert_ne!(dry, live, "the kit's replay store would answer the live id byte-identically"); + assert!(dry.starts_with(&format!("{live}:dry")), "{dry}"); + assert_ne!(dry, later_dry, "two dry runs of one label do not collide"); + let next = run(Mode::Live, at + Span::seconds(STREAM_IDLE_GAP_S)); + assert_ne!(next, live, "each tick is its own conversation"); +} + +// --------------------------------------------------------------------------------------------- +// The continuer against a fake HTTP server. type Seen = Arc>>; @@ -503,7 +704,7 @@ fn serve(answers: Vec) -> (String, Seen) { #[test] fn raw_generate_is_raw() { let (endpoint, seen) = serve(vec![json!({ "response": "one" }).to_string(), json!({ "response": "two" }).to_string()]); - let model = OllamaStream::open(&endpoint, "projector:x", "base:y").unwrap(); + let model = OllamaContinuer::open(&endpoint, "base:y").unwrap(); let out = model.continue_raw("The frame\n\n", &Sampling::STREAM, 2).unwrap(); assert_eq!(out, ["one", "two"]); let seen = seen.lock().unwrap(); @@ -518,16 +719,6 @@ fn raw_generate_is_raw() { } } -#[test] -fn the_projector_is_asked_in_chat() { - let (endpoint, seen) = serve(vec![json!({ "message": { "role": "assistant", "content": "I hold it." } }).to_string()]); - let model = OllamaStream::open(&endpoint, "projector:x", "base:y").unwrap(); - assert_eq!(model.project("Be the Projector.").unwrap(), "I hold it."); - let seen = seen.lock().unwrap(); - assert_eq!(seen[0].0, "/api/chat"); - assert_eq!(seen[0].1, json!({ "model": "projector:x", "stream": false, "messages": [{ "role": "user", "content": "Be the Projector." }] })); -} - #[test] fn a_model_host_that_answers_badly_is_a_fault() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -538,8 +729,8 @@ fn a_model_host_that_answers_badly_is_a_fault() { let _ = socket.read(&mut buffer); let _ = socket.write_all(b"HTTP/1.1 500 Internal Server Error\r\ncontent-length: 4\r\nconnection: close\r\n\r\nboom"); }); - let model = OllamaStream::open(&endpoint, "p", "b").unwrap(); - let fault = model.project("x").unwrap_err().to_string(); + let model = OllamaContinuer::open(&endpoint, "b").unwrap(); + let fault = model.continue_raw("x", &Sampling::STREAM, 1).unwrap_err().to_string(); assert!(fault.contains("500"), "{fault}"); } @@ -560,13 +751,13 @@ fn non_loopback_endpoint_refused() { "http://", "", ] { - assert!(OllamaStream::open(refused, "p", "b").is_err(), "{refused} must be refused"); + assert!(OllamaContinuer::open(refused, "b").is_err(), "{refused} must be refused"); } for refused in ["http://10.77.0.1:11434", "http://example.com:11434", "http://0.0.0.0:11434", "https://127.0.0.1:11434", "127.0.0.1:11434", "http://127.0.0.1:11434/api", "http://[2001:db8::1]:11434"] { - assert!(OllamaStream::open(refused, "p", "b").is_err(), "{refused} must be refused"); + assert!(OllamaContinuer::open(refused, "b").is_err(), "{refused} must be refused"); } for opened in ["http://127.0.0.1:11434", "http://localhost:11434", "http://[::1]:11434", "http://127.0.0.2:1/", "http://LOCALHOST:11434/"] { - assert!(OllamaStream::open(opened, "p", "b").is_ok(), "{opened} is loopback"); + assert!(OllamaContinuer::open(opened, "b").is_ok(), "{opened} is loopback"); } } @@ -706,7 +897,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, model: &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"); } @@ -777,7 +968,7 @@ fn every_thought_records_the_fixed_parts_of_its_prompts() { assert!(version.len() <= LINE_MAX_BYTES, "{}", version.len()); assert!(include_str!("../../Cargo.toml").contains(GHOSTLIGHT_REV), "the recorded revision is the manifest's pin"); let rebuilt = build_projector_prompt(&ProjectorPrompt { identity: &thought.identity.0, typed_context: &thought.context.0, visible_stimulus: &thought.stimulus.0, domain_guidance: STREAM_GUIDANCE, word_budget: PROJECTOR_WORDS }); - assert_eq!(model.projected.borrow()[0], rebuilt, "the trace rebuilds the Projector's exact prompt"); + assert_eq!(model.projected.borrow()[0].1, rebuilt, "the trace rebuilds the Projector's exact prompt"); } diff --git a/crates/mind-body/tests/cli.rs b/crates/mind-body/tests/cli.rs index fa838a4..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"), "{out}"); + assert!(out.lines().any(|line| line == "units: stopped mind-self@eureka.service mind-persona@eureka.service mind-stream@eureka.service"), "{out}"); } #[test] @@ -362,8 +362,8 @@ fn hold_through_the_binary(exit: i32) -> (std::process::Output, Vec, tem fn the_binary_hold_runs_systemctl_stop_on_exactly_the_units_it_starts() { let (output, recorded, _root) = hold_through_the_binary(0); assert_eq!(output.status.code(), Some(0), "{}", String::from_utf8_lossy(&output.stderr)); - assert_eq!(recorded, ["stop", "--", "mind-self@eureka.service", "mind-persona@eureka.service"], "a stop that succeeds asks no state"); - assert!(String::from_utf8_lossy(&output.stdout).contains("units: stopped mind-self@eureka.service mind-persona@eureka.service")); + assert_eq!(recorded, ["stop", "--", "mind-self@eureka.service", "mind-persona@eureka.service", "mind-stream@eureka.service"], "a stop that succeeds asks no state"); + assert!(String::from_utf8_lossy(&output.stdout).contains("units: stopped mind-self@eureka.service mind-persona@eureka.service mind-stream@eureka.service")); } #[cfg(unix)] @@ -373,11 +373,11 @@ fn a_partial_stop_names_each_units_state_and_keeps_the_brake_held_with_exit_2() assert_eq!(output.status.code(), Some(2)); assert_eq!( recorded, - ["stop", "--", "mind-self@eureka.service", "mind-persona@eureka.service", "is-active", "mind-self@eureka.service", "is-active", "mind-persona@eureka.service"] + ["stop", "--", "mind-self@eureka.service", "mind-persona@eureka.service", "mind-stream@eureka.service", "is-active", "mind-self@eureka.service", "is-active", "mind-persona@eureka.service", "is-active", "mind-stream@eureka.service"] ); let stderr = String::from_utf8_lossy(&output.stderr); assert!(stderr.contains("exit status 5") && stderr.contains("the brake is held"), "{stderr}"); - assert!(stderr.contains("mind-self@eureka.service is inactive") && stderr.contains("mind-persona@eureka.service is active"), "{stderr}"); + assert!(stderr.contains("mind-self@eureka.service is inactive") && stderr.contains("mind-persona@eureka.service is active") && stderr.contains("mind-stream@eureka.service is active"), "{stderr}"); assert!(!stderr.contains("were not stopped"), "{stderr}"); assert!(!String::from_utf8_lossy(&output.stdout).contains("stopped")); assert!(shown(root.path()).contains("BrakeHeld")); @@ -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@ and mind-persona@ 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 2382c83..693c9a8 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -328,20 +328,26 @@ 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 mind_body::stream: impl HuginnPersonaSource: fn new (client : HuginnClient) -> Self -mind_body::stream: impl OllamaStream -mind_body::stream: impl OllamaStream: fn open (endpoint : & str , projector_model : & str , base_model : & str) -> Result < Self > +mind_body::stream: impl KitProjector +mind_body::stream: impl KitProjector: fn open (endpoint : SocketAddr , connection_key : String , caller : & str , model : & str , timeout : Duration) -> Result < Self > +mind_body::stream: impl OllamaContinuer +mind_body::stream: impl OllamaContinuer: fn open (endpoint : & str , base_model : & str) -> Result < Self > mind_body::stream: impl PersonaSource for HuginnPersonaSource +mind_body::stream: impl Projector for KitProjector mind_body::stream: impl Sampling mind_body::stream: impl Seed -mind_body::stream: impl StreamModel for OllamaStream +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 StreamModel { fn project (& self , prompt : & str) -> Result < String > ; fn continue_raw (& self , prompt : & str , sampling : & Sampling , n : usize) -> Result < Vec < 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 OllamaStream pub-fields[] -mind_body::stream: struct Organs< 'a > pub-fields[mind: & 'a dyn MindPort, persona: & 'a dyn PersonaSource, model: & 'a dyn StreamModel, projector_model: & 'a str, base_model: & 'a str] +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, 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..ab41af9 --- /dev/null +++ b/crates/mind-body/tests/stream_bin.rs @@ -0,0 +1,51 @@ +//! 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 = "zq7Kx-9wVtm3Pd"; + +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}"); +} + +/// Every four-byte run of the key: a partial key is as much a leak as the whole. +fn leaks(shown: &str) -> Option<&str> { + CANARY.as_bytes().windows(4).map(|window| std::str::from_utf8(window).unwrap()).find(|run| shown.contains(run)) +} + +#[test] +fn no_byte_run_of_the_key_is_printed_on_any_startup_path() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("connector-key"), format!("{CANARY}\n")).unwrap(); + // A non-loopback connector and a configuration the client refuses both end startup after the key was read and handed over. + for (connector, ending) in [("10.77.0.1:4103", "not a loopback address"), ("127.0.0.1:0", "could not be configured")] { + let output = run(Some(dir.path()), connector); + let shown = format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)); + assert!(shown.contains(ending), "{connector}: the key was read and startup went on: {shown}"); + assert_eq!(leaks(&shown), None, "{connector}: {shown}"); + } +} diff --git a/crates/mind-body/tests/stream_proxy.rs b/crates/mind-body/tests/stream_proxy.rs index 088c9c7..2d44a4d 100644 --- a/crates/mind-body/tests/stream_proxy.rs +++ b/crates/mind-body/tests/stream_proxy.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; -use mind_body::stream::{OllamaStream, StreamModel}; +use mind_body::stream::{Continuer, OllamaContinuer, Sampling}; /// Counts the connections a loopback listener accepts, answering each with `reply` when given. fn listen(reply: Option) -> (String, Arc) { @@ -67,7 +67,8 @@ fn the_stream_agent_ignores_the_proxy_environment() { let plain: ureq::Agent = ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(2))).build().into(); let _ = plain.get(&endpoint).call(); } else { - let _ = OllamaStream::open(&endpoint, "p", "b").unwrap().project("hello"); + let sampling = Sampling { temperature_permille: 1000, min_p_permille: 50, max_tokens: 16 }; + let _ = OllamaContinuer::open(&endpoint, "b").unwrap().continue_raw("hello", &sampling, 1); } return; }