Merge pull request 'mind-stream: the Projector is her own connector caller on Haiku; a brake hold stops mind-stream@ too' (#47) from hands/eureka-body-stream-projector-kit into main
This commit is contained in:
commit
f4951db649
11 changed files with 459 additions and 99 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -7,10 +7,14 @@
|
|||
//! the library never does.
|
||||
//!
|
||||
//! Environment: `HUGINN_ENDPOINT` (`rudp://<ip>:<port>`, her own daemon), `EUREKA_INSTANCE`
|
||||
//! (when `--instance` is not given), `MIND_STREAM_ENDPOINT` (`http://<loopback>:<port>`, the
|
||||
//! model host), `MIND_STREAM_PROJECTOR_MODEL` and `MIND_STREAM_BASE_MODEL`.
|
||||
//! (when `--instance` is not given), `MIND_STREAM_CONNECTOR_ENDPOINT` (`<loopback ip>:<port>`, 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://<loopback>:<port>`, 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<String> {
|
||||
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<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> {
|
||||
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://<ip>:<port>"))?;
|
||||
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 <ip>:<port>"))?;
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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 <dir>] --instance <slug> show | check | brake hold | brake release | mind-control [--root <dir>] 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))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String> {
|
||||
/// 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<String> {
|
||||
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<String> {
|
||||
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<Vec<String>> {
|
||||
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::<Result<Vec<_>>>()?);
|
||||
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<Vec<String>> = 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();
|
||||
|
|
|
|||
|
|
@ -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<String>;
|
||||
/// 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<String>;
|
||||
}
|
||||
|
||||
/// 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<Vec<String>>;
|
||||
}
|
||||
|
||||
/// 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://<loopback host>:<port>`; any other address is refused here.
|
||||
pub fn open(endpoint: &str, projector_model: &str, base_model: &str) -> Result<Self> {
|
||||
pub fn open(endpoint: &str, base_model: &str) -> Result<Self> {
|
||||
// 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://<loopback>:<port>"))?;
|
||||
|
|
@ -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<Value> {
|
||||
|
|
@ -152,12 +160,7 @@ impl OllamaStream {
|
|||
}
|
||||
}
|
||||
|
||||
impl StreamModel for OllamaStream {
|
||||
fn project(&self, prompt: &str) -> Result<String> {
|
||||
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<Vec<String>> {
|
||||
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<Utc>) -> 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<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 {
|
||||
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<Self> {
|
||||
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<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() }];
|
||||
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<String> = 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:?}")),
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ struct Fake {
|
|||
projection: String,
|
||||
candidates: Vec<String>,
|
||||
fault: bool,
|
||||
projected: RefCell<Vec<String>>,
|
||||
projected: RefCell<Vec<(String, String)>>,
|
||||
continued: RefCell<Vec<(String, Sampling, usize)>>,
|
||||
}
|
||||
|
||||
|
|
@ -72,12 +72,18 @@ impl Fake {
|
|||
}
|
||||
}
|
||||
|
||||
impl StreamModel for Fake {
|
||||
fn project(&self, prompt: &str) -> Result<String> {
|
||||
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<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()) }
|
||||
}
|
||||
}
|
||||
|
||||
impl Continuer for Fake {
|
||||
fn continue_raw(&self, prompt: &str, sampling: &Sampling, n: usize) -> Result<Vec<String>> {
|
||||
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<String> {
|
||||
|
|
@ -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<Vec<(String, ProviderRequest)>>,
|
||||
}
|
||||
|
||||
impl Scripted {
|
||||
fn ending(stop_reason: StopReason, words: &[&'static str]) -> Arc<Self> {
|
||||
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<Scripted>, 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<Utc>| {
|
||||
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<Mutex<Vec<(String, Value)>>>;
|
||||
|
||||
|
|
@ -503,7 +704,7 @@ fn serve(answers: Vec<String>) -> (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");
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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<String>, 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]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
51
crates/mind-body/tests/stream_bin.rs
Normal file
51
crates/mind-body/tests/stream_bin.rs
Normal file
|
|
@ -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}");
|
||||
}
|
||||
}
|
||||
|
|
@ -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>) -> (String, Arc<AtomicUsize>) {
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue