mind-stream: the Projector is her own connector caller; Ollama keeps only the raw continuer
KitProjector sends one tool-less kit turn per tick as caller <instance>; OllamaStream's /api/chat Projector and StreamModel are deleted, replaced by the Projector and Continuer ports. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
995a5b88f9
commit
ee72561716
7 changed files with 290 additions and 64 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,25 @@ 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, 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 connector key: the credential file, read once and trimmed. An absence names the variable, never a value.
|
||||
#[allow(clippy::disallowed_methods, reason = "the binary reads its own environment, and hands it to the library")]
|
||||
fn connector_key() -> Result<String> {
|
||||
let directory = std::env::var_os("CREDENTIALS_DIRECTORY").map(PathBuf::from).ok_or_else(|| anyhow!("CREDENTIALS_DIRECTORY is not set"))?;
|
||||
let key = std::fs::read_to_string(directory.join("connector-key")).map_err(|_| anyhow!("the connector-key credential could not be read"))?;
|
||||
Ok(key.trim().to_owned())
|
||||
}
|
||||
|
||||
fn required(name: &str) -> Result<String> {
|
||||
setting(name).ok_or_else(|| anyhow!("{name} is not set"))
|
||||
}
|
||||
|
|
@ -63,10 +77,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()?, &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, projector_model: &projector_model, base_model: &base_model };
|
||||
let mode = if args.dry_run { Mode::DryRun } else { Mode::Live };
|
||||
let once = args.once || args.dry_run;
|
||||
loop {
|
||||
|
|
|
|||
|
|
@ -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,29 @@ 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 {
|
||||
/// 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 +143,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 +158,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 +179,71 @@ 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 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 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,7 +592,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: &'a dyn Projector,
|
||||
pub continuer: &'a dyn Continuer,
|
||||
pub projector_model: &'a str,
|
||||
pub base_model: &'a str,
|
||||
}
|
||||
|
|
@ -586,8 +653,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 +663,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,7 +678,7 @@ 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()),
|
||||
|
|
@ -633,7 +701,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,14 @@ 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 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 +202,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, projector_model: "projector:test", base_model: "base:test" }
|
||||
}
|
||||
|
||||
fn thoughts(mind: &TestMind) -> Vec<String> {
|
||||
|
|
@ -414,8 +416,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 +467,147 @@ 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, projector_model: KIT_MODEL, 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_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();
|
||||
let id = model.projected.borrow()[0].0.clone();
|
||||
id
|
||||
};
|
||||
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 +645,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 +660,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 +670,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 +692,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 +838,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, projector_model: "projector:test", 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 +909,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");
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -328,19 +328,24 @@ 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 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 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, projector_model: & 'a str, 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
|
||||
|
|
|
|||
|
|
@ -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