GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 2 Projects Releases Packages Wiki Activity Actions

stream: LF again; the stale boundary docs name the label as the gap's origin

Browse source 
Restores LF on stream.rs, stream/tests.rs and bin/mind-stream.rs (5defa18 converted them). Renames the capped-repo test for the rule built; build_seed's since is the newest thought's label tick time; STREAM_MIN_GAP_S counts from the label, so a model run of M seconds leaves 900 - M s of quiet after admission. No behaviour change.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-07 04:44:48 +01:00
commit e2e0488abb
3 changed files with 1616 additions and 1614 deletions
Download patch file Download diff file Expand all files Collapse all files

218
crates/mind-body/src/bin/mind-stream.rs
View file

@ -1,109 +1,109 @@
//! `mind-stream --instance <slug> [--once] [--dry-run]`: her stream of thought. //! `mind-stream --instance <slug> [--once] [--dry-run]`: her stream of thought.
//! //!
//! Without `--once` it runs forever: one `mind_body::stream::tick` per `STREAM_POLL_S`, or per //! Without `--once` it runs forever: one `mind_body::stream::tick` per `STREAM_POLL_S`, or per
//! `STREAM_MIN_GAP_S` after a blank or a fault. `--dry-run` runs one tick that admits nothing and //! `STREAM_MIN_GAP_S` after a blank or a fault. `--dry-run` runs one tick that admits nothing and
//! prints the thought it would admit as JSON. A model or mind fault ends the tick, never the //! prints the thought it would admit as JSON. A model or mind fault ends the tick, never the
//! process: it is one line on stderr. This reads the process's arguments and environment, which //! process: it is one line on stderr. This reads the process's arguments and environment, which
//! the library never does. //! the library never does.
//! //!
//! Environment: `HUGINN_ENDPOINT` (`rudp://<ip>:<port>`, her own daemon), `EUREKA_INSTANCE` //! Environment: `HUGINN_ENDPOINT` (`rudp://<ip>:<port>`, her own daemon), `EUREKA_INSTANCE`
//! (when `--instance` is not given), `MIND_STREAM_ENDPOINT` (`http://<loopback>:<port>`, the //! (when `--instance` is not given), `MIND_STREAM_ENDPOINT` (`http://<loopback>:<port>`, the
//! model host), `MIND_STREAM_PROJECTOR_MODEL` and `MIND_STREAM_BASE_MODEL`. //! model host), `MIND_STREAM_PROJECTOR_MODEL` and `MIND_STREAM_BASE_MODEL`.
use std::net::SocketAddr; use std::net::SocketAddr;
use std::process::ExitCode; use std::process::ExitCode;
use std::time::Duration; use std::time::Duration;
use anyhow::{Result, anyhow, bail}; use anyhow::{Result, anyhow, bail};
use chrono::Utc; use chrono::Utc;
use eureka_pipeline::Slug; use eureka_pipeline::Slug;
use eureka_state::HuginnClient; use eureka_state::HuginnClient;
use mind_body::control::read_effective; use mind_body::control::read_effective;
use mind_body::queue::HuginnMind; 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, Mode, OllamaStream, Organs, Outcome, STREAM_MIN_GAP_S, STREAM_POLL_S, tick};
const TIMEOUT: Duration = Duration::from_secs(30); const TIMEOUT: Duration = Duration::from_secs(30);
#[allow(clippy::disallowed_methods, reason = "the binary reads its own environment, and hands it to the library")] #[allow(clippy::disallowed_methods, reason = "the binary reads its own environment, and hands it to the library")]
fn setting(name: &str) -> Option<String> { fn setting(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|value| !value.is_empty()) std::env::var(name).ok().filter(|value| !value.is_empty())
} }
fn required(name: &str) -> Result<String> { fn required(name: &str) -> Result<String> {
setting(name).ok_or_else(|| anyhow!("{name} is not set")) setting(name).ok_or_else(|| anyhow!("{name} is not set"))
} }
struct Args { struct Args {
instance: Option<String>, instance: Option<String>,
once: bool, once: bool,
dry_run: bool, dry_run: bool,
} }
fn arguments() -> Result<Args> { fn arguments() -> Result<Args> {
let mut args = Args { instance: None, once: false, dry_run: false }; let mut args = Args { instance: None, once: false, dry_run: false };
let mut given = std::env::args().skip(1); let mut given = std::env::args().skip(1);
while let Some(argument) = given.next() { while let Some(argument) = given.next() {
match argument.as_str() { match argument.as_str() {
"--instance" => args.instance = Some(given.next().ok_or_else(|| anyhow!("--instance needs a slug"))?), "--instance" => args.instance = Some(given.next().ok_or_else(|| anyhow!("--instance needs a slug"))?),
"--once" => args.once = true, "--once" => args.once = true,
"--dry-run" => args.dry_run = true, "--dry-run" => args.dry_run = true,
other => bail!("unknown argument {other}; usage: mind-stream --instance <slug> [--once] [--dry-run]"), other => bail!("unknown argument {other}; usage: mind-stream --instance <slug> [--once] [--dry-run]"),
} }
} }
Ok(args) Ok(args)
} }
fn run() -> Result<()> { fn run() -> Result<()> {
let args = arguments()?; let args = arguments()?;
let instance = Slug(args.instance.map_or_else(|| required("EUREKA_INSTANCE"), Ok)?); let instance = Slug(args.instance.map_or_else(|| required("EUREKA_INSTANCE"), Ok)?);
let address: SocketAddr = required("HUGINN_ENDPOINT")? let address: SocketAddr = required("HUGINN_ENDPOINT")?
.strip_prefix("rudp://") .strip_prefix("rudp://")
.and_then(|address| address.parse().ok()) .and_then(|address| address.parse().ok())
.ok_or_else(|| anyhow!("HUGINN_ENDPOINT is not rudp://<ip>:<port>"))?; .ok_or_else(|| anyhow!("HUGINN_ENDPOINT is not rudp://<ip>:<port>"))?;
let projector_model = required("MIND_STREAM_PROJECTOR_MODEL")?; let projector_model = required("MIND_STREAM_PROJECTOR_MODEL")?;
let base_model = required("MIND_STREAM_BASE_MODEL")?; let base_model = required("MIND_STREAM_BASE_MODEL")?;
let model = OllamaStream::open(&required("MIND_STREAM_ENDPOINT")?, &projector_model, &base_model)?; let model = OllamaStream::open(&required("MIND_STREAM_ENDPOINT")?, &projector_model, &base_model)?;
let mind = HuginnMind::new(HuginnClient::new(address, instance.clone(), TIMEOUT)); let mind = HuginnMind::new(HuginnClient::new(address, instance.clone(), TIMEOUT));
let persona = HuginnPersonaSource::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, model: &model, projector_model: &projector_model, base_model: &base_model };
let mode = if args.dry_run { Mode::DryRun } else { Mode::Live }; let mode = if args.dry_run { Mode::DryRun } else { Mode::Live };
let once = args.once || args.dry_run; let once = args.once || args.dry_run;
loop { loop {
let mut faulted = false; let mut faulted = false;
let waits = match tick(&organs, &|| read_effective(&instance), mode, Utc::now()) { let waits = match tick(&organs, &|| read_effective(&instance), mode, Utc::now()) {
Ok(Outcome::Thought { thought, admitted }) => { Ok(Outcome::Thought { thought, admitted }) => {
if admitted { if admitted {
println!("admitted {}:thought:{}", instance.0, thought.label.0); println!("admitted {}:thought:{}", instance.0, thought.label.0);
} else { } else {
println!("{}", serde_json::to_string_pretty(&*thought)?); println!("{}", serde_json::to_string_pretty(&*thought)?);
} }
false false
} }
Ok(Outcome::Blank) => true, Ok(Outcome::Blank) => true,
Ok(Outcome::Held(reason)) => { Ok(Outcome::Held(reason)) => {
eprintln!("mind-stream: held ({reason:?}); nothing ticked"); eprintln!("mind-stream: held ({reason:?}); nothing ticked");
false false
} }
Ok(Outcome::NotDue) => false, Ok(Outcome::NotDue) => false,
Err(fault) => { Err(fault) => {
eprintln!("mind-stream: {}", format!("{fault:#}").replace('\n', " ")); eprintln!("mind-stream: {}", format!("{fault:#}").replace('\n', " "));
faulted = true; faulted = true;
true true
} }
}; };
if once { if once {
return if faulted { Err(anyhow!("the tick faulted")) } else { Ok(()) }; return if faulted { Err(anyhow!("the tick faulted")) } else { Ok(()) };
} }
std::thread::sleep(Duration::from_secs(if waits { STREAM_MIN_GAP_S as u64 } else { STREAM_POLL_S as u64 })); std::thread::sleep(Duration::from_secs(if waits { STREAM_MIN_GAP_S as u64 } else { STREAM_POLL_S as u64 }));
} }
} }
fn main() -> ExitCode { fn main() -> ExitCode {
match run() { match run() {
Ok(()) => ExitCode::SUCCESS, Ok(()) => ExitCode::SUCCESS,
Err(fault) => { Err(fault) => {
eprintln!("mind-stream: {}", format!("{fault:#}").replace('\n', " ")); eprintln!("mind-stream: {}", format!("{fault:#}").replace('\n', " "));
ExitCode::FAILURE ExitCode::FAILURE
} }
} }
} }

1286
crates/mind-body/src/stream.rs
View file

File diff suppressed because it is too large Load diff

1726
crates/mind-body/src/stream/tests.rs
View file

File diff suppressed because it is too large Load diff