From cebac9d426e87c3d0d78a21a553eb0b10e2eaf5a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 8 Nov 2026 11:01:39 +0100 Subject: [PATCH] mind-body: a Mind sleeps once a night, in her Persona slot, disposing of her day by typed operations The spawn prelude leaves one_process for a shared start_session. sleep.rs is the one owner of a night: the night line (her day whole, open items, the tide), SleepReply and its schema, the decoder's ledger (every day memory exactly once, ids her memories in force, one conversation per consolidation, told back, third refusal ends it), admission of whole operations in batches that never split a consolidation, and the closing of the run with her line. The heartbeat opens one Sleep run per listed idle Mind per window (HEARTBEAT_SLEEP_UTC, HEARTBEAT_SLEEP_MINDS); a missed night is not made up and a failed one not retried. Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/bin/mind-heartbeat.rs | 10 +- crates/mind-body/src/heartbeat.rs | 86 ++- crates/mind-body/src/heartbeat/tests.rs | 93 +++- crates/mind-body/src/lib.rs | 1 + crates/mind-body/src/persona.rs | 130 +++-- crates/mind-body/src/persona/tests.rs | 213 ++++++++ crates/mind-body/src/sleep.rs | 578 +++++++++++++++++++++ crates/mind-body/src/sleep/tests.rs | 316 +++++++++++ crates/mind-body/tests/public_surface.txt | 9 +- 9 files changed, 1376 insertions(+), 60 deletions(-) create mode 100644 crates/mind-body/src/sleep.rs create mode 100644 crates/mind-body/src/sleep/tests.rs diff --git a/crates/mind-body/src/bin/mind-heartbeat.rs b/crates/mind-body/src/bin/mind-heartbeat.rs index 0a2b466..96e303f 100644 --- a/crates/mind-body/src/bin/mind-heartbeat.rs +++ b/crates/mind-body/src/bin/mind-heartbeat.rs @@ -22,7 +22,7 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; use eureka_state::HuginnClient; use mind_body::chat_rudp::{RudpChat, connection_id}; -use mind_body::heartbeat::{Config, FileControl, FileStore, Organs, PersonaId, Seat, HeartbeatView, poll}; +use mind_body::heartbeat::{Config, FileControl, FileStore, HeartbeatView, Night, Organs, PersonaId, Seat, poll}; use mind_body::launch::{SystemClock, SystemdLauncher}; use mind_body::queue::HuginnMind; @@ -61,7 +61,13 @@ fn config() -> Result { .map(PersonaId::parse) .collect::>>()?; let operators = required("MIND_PERSONA_OPERATOR")?.split(',').map(|id| id.trim().to_string()).filter(|id| !id.is_empty()).collect(); - Config::new(personas, operators) + // The nightly sleep is both settings or neither. + let night = match (setting("HEARTBEAT_SLEEP_UTC"), setting("HEARTBEAT_SLEEP_MINDS")) { + (Some(window), Some(minds)) => Some(Night::parse(&window, &minds)?), + (None, None) => None, + _ => bail!("HEARTBEAT_SLEEP_UTC and HEARTBEAT_SLEEP_MINDS are set together or not at all"), + }; + Config::new(personas, operators, night) } fn store() -> Result { diff --git a/crates/mind-body/src/heartbeat.rs b/crates/mind-body/src/heartbeat.rs index aa28d40..b6af0a5 100644 --- a/crates/mind-body/src/heartbeat.rs +++ b/crates/mind-body/src/heartbeat.rs @@ -33,7 +33,7 @@ use std::net::SocketAddr; use std::path::PathBuf; use anyhow::{Context, Result, anyhow, bail}; -use chrono::{DateTime, SecondsFormat, Utc}; +use chrono::{DateTime, SecondsFormat, Timelike, Utc}; use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; use cultnet_rs::Selection; use eureka_pipeline::{PipelineKind, PipelineRef, ResolutionOutcome, RunOperator, RunTurn, Slug}; @@ -104,11 +104,58 @@ pub struct Config { pub personas: Vec, /// Sender ids that are the operator. pub operators: Vec, + /// The nightly window and the Minds that sleep in it; none, no one sleeps. + pub night: Option, +} + +/// The heartbeat's night: a daily window in UTC minutes (it may wrap midnight) and the Minds that sleep in it. +/// The heartbeat alone decides who sleeps and when: one Sleep run per Mind per window, never made up if the window +/// passes without one. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Night { + pub from_min: u32, + pub to_min: u32, + pub minds: Vec, +} + +impl Night { + /// `window` is `HH:MM-HH:MM` in UTC, `minds` the slugs comma-separated. + pub fn parse(window: &str, minds: &str) -> Result { + let minute = |text: &str| -> Result { + let (hour, minute) = text.trim().split_once(':').ok_or_else(|| anyhow!("the night window is HH:MM-HH:MM"))?; + let (hour, minute) = (hour.parse::()?, minute.parse::()?); + if hour > 23 || minute > 59 { + bail!("the night window is HH:MM-HH:MM"); + } + Ok(hour * 60 + minute) + }; + let (from, to) = window.split_once('-').ok_or_else(|| anyhow!("the night window is HH:MM-HH:MM"))?; + let (from_min, to_min) = (minute(from)?, minute(to)?); + if from_min == to_min { + bail!("the night window is empty"); + } + let minds: Vec = minds.split(',').map(|slug| Slug(slug.trim().to_string())).filter(|slug| !slug.0.is_empty()).collect(); + if minds.is_empty() { + bail!("no Mind is listed to sleep"); + } + Ok(Self { from_min, to_min, minds }) + } + + /// When the window `now` is inside opened, or none outside it. + fn opened(&self, now: DateTime) -> Option> { + let minute = now.hour() * 60 + now.minute(); + let inside = if self.from_min < self.to_min { (self.from_min..self.to_min).contains(&minute) } else { minute >= self.from_min || minute < self.to_min }; + if !inside { + return None; + } + let day = if minute >= self.from_min { now.date_naive() } else { now.date_naive().pred_opt()? }; + Some(day.and_hms_opt(self.from_min / 60, self.from_min % 60, 0)?.and_utc()) + } } impl Config { /// Refuses no Persona, or a Persona twice. - pub fn new(personas: Vec, operators: Vec) -> Result { + pub fn new(personas: Vec, operators: Vec, night: Option) -> Result { if personas.is_empty() { bail!("no persona is configured"); } @@ -124,6 +171,7 @@ impl Config { Ok(Self { personas, operators, + night, }) } } @@ -430,6 +478,13 @@ fn last_turn(mind: &dyn MindPort, instance: &Slug) -> Result Result>> { + let operator = format!("{:?}", RunOperator::Mind); + let runs = any_of(any_of(any_of(of_kind(PipelineKind::Run), "root", &[&instance.0]), "turn", &["Sleep"]), "operated_by", &[&operator]); + newest(mind, runs)?.map(|header| parse_time(&header.admission.admitted_at)).transpose() +} + /// The newest document a selection holds, by admission ordinal: one query. fn newest(mind: &dyn MindPort, selection: Selection) -> Result> { let asking = Selection { descending: true, limit: Some(1), ..selection }; @@ -527,6 +582,8 @@ struct MindRead { /// The urgency she declared and no turn has consumed. urgency: u16, self_end: Option>, + /// When her newest Sleep run was admitted. + last_sleep: Option>, } fn read_mind(organs: &Organs, mind: &dyn MindPort, slug: &Slug) -> Result { @@ -540,6 +597,7 @@ fn read_mind(organs: &Organs, mind: &dyn MindPort, slug: &Slug) -> Result Result { } } + // The night: inside the window, one Sleep run for a listed Mind who is idle and released and has not slept + // since the window opened (a failed night counts). The operator's mention went first above: she is busy then. + // A window that passes without a night is not made up. + if let Some(night) = &config.night + && let Some(opened) = night.opened(now) + { + for row in &mut rows { + if !row.idle() || !night.minds.contains(row.slug()) || row.read.as_ref().is_some_and(|read| read.last_sleep.is_some_and(|slept| slept >= opened)) { + continue; + } + match launch(organs, row, RunTurn::Sleep, vec![])? { + Launched::Run(label) => { + row.note = format!("asleep {label}"); + row.busy = true; + launched.push(row.slug().clone()); + launches.push(label); + break; + } + Launched::Busy => row.busy = true, + Launched::Refused(why) => row.note = why, + } + } + } + // On the beat: when the beat has passed, one launch for the winner of the pick. let due = (now - state.last_beat_at).num_seconds() >= i64::from(beat_s); let standings: Vec = rows diff --git a/crates/mind-body/src/heartbeat/tests.rs b/crates/mind-body/src/heartbeat/tests.rs index a00947e..1e3bc87 100644 --- a/crates/mind-body/src/heartbeat/tests.rs +++ b/crates/mind-body/src/heartbeat/tests.rs @@ -8,7 +8,7 @@ use std::cell::{Cell, RefCell}; -use chrono::Duration; +use chrono::{Duration, TimeZone}; use eureka_pipeline::PipelineKind as K; use huginn_mind::PipelineSelectionPage; use tempfile::TempDir; @@ -192,7 +192,7 @@ impl Rig { launcher: Recording::default(), clock: Fixed(Cell::new(at(1_000_000))), store: MemStore::default(), - config: Config::new(vec![persona("ann", "Ann"), persona("bob", "Bob")], vec!["@op:chat".into()]).unwrap(), + config: Config::new(vec![persona("ann", "Ann"), persona("bob", "Bob")], vec!["@op:chat".into()], None).unwrap(), labels: Cell::new(0), }) } @@ -1424,3 +1424,92 @@ fn the_file_store_reads_back_the_addressing_it_wrote_to_the_view() { assert_eq!(store.addressing().unwrap(), Some(table.clone())); assert_eq!(store.load_addressing().unwrap(), Some(table)); } + +/// A night over the rig's clock, 13:31 UTC: the window holds it, `ann` sleeps. +fn nights(window: &str, minds: &str) -> Box { + let mut rig = Rig::new(); + rig.config.night = Some(Night::parse(window, minds).unwrap()); + rig.turn(ANN, 10); + rig.turn(BOB, 10); + rig.store.save(&HeartbeatState { last_beat_at: rig.now(), handled_mentions: vec![] }).unwrap(); + rig +} + +/// Ends the run the heartbeat launched for `who` (the newest by admission, not by key). +fn end_launched(rig: &Rig, who: usize) { + let launched = rig.minds[who].ids(K::Run).into_iter().find(|id| id.contains(":run:mind-")).expect("a launched run"); + rig.minds[who].committed(vec![close(reference(K::Run, &launched), recorded())]); +} + +fn started_turns(rig: &Rig) -> Vec<(String, RunTurn)> { + rig.launcher.started.borrow().iter().map(|(slug, turn)| (slug.0.clone(), *turn)).collect() +} + +#[test] +fn the_window_wraps_midnight_and_names_when_it_opened() { + let night = Night::parse("22:00-04:00", "ann").unwrap(); + let on = |hour, minute| Utc.with_ymd_and_hms(2026, 10, 9, hour, minute, 0).unwrap(); + assert_eq!(night.opened(on(23, 0)), Some(on(22, 0))); + assert_eq!(night.opened(on(2, 0)), Some(on(22, 0) - Duration::days(1)), "after midnight it opened yesterday"); + assert_eq!(night.opened(on(4, 0)), None, "the end is outside"); + assert_eq!(night.opened(on(13, 0)), None); + assert_eq!(Night::parse("13:00-14:00", "ann").unwrap().opened(on(13, 0)), Some(on(13, 0)), "the start is inside"); + assert!(Night::parse("13:00-13:00", "ann").is_err() && Night::parse("25:00-03:00", "ann").is_err() && Night::parse("13:00-14:00", " , ").is_err()); +} + +#[test] +fn a_listed_idle_mind_sleeps_once_per_window_and_a_missed_night_is_not_made_up() { + let rig = nights("13:00-14:00", "ann"); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("ann".into())] }); + assert_eq!(started_turns(&rig), vec![("ann".to_string(), RunTurn::Sleep)], "bob is not listed, and the beat has just fallen"); + assert!(rig.minds[ANN].view(K::Run, rig.minds[ANN].ids(K::Run).last().unwrap()).is_some()); + // Busy while the night lasts: nothing more. Then her night ends (failed or slept, it counts) and the window is spent for her. + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![] }); + end_launched(&rig, ANN); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![] }, "one night per window, whatever became of it"); + assert_eq!(started_turns(&rig).len(), 1); + // The next window opens a day later and she sleeps again. + rig.advance(24 * 3600); + rig.store.save(&HeartbeatState { last_beat_at: rig.now(), handled_mentions: vec![] }).unwrap(); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("ann".into())] }); +} + +#[test] +fn no_night_outside_the_window_for_an_unlisted_mind_or_after_the_window_passed_busy() { + let outside = nights("22:00-04:00", "ann,bob"); + assert_eq!(outside.poll(), Outcome::Polled { launched: vec![] }); + assert!(started_turns(&outside).is_empty()); + // She is awake (a live Persona turn) when the window opens and until it closes: the night is not made up after. + let rig = nights("13:00-14:00", "ann"); + rig.minds[ANN].committed(vec![run_in("ann", "awake", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); + rig.launcher.revive_as(&Slug("ann".into()), RunTurn::PersonaTurn); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![] }); + rig.advance(3600); + rig.launcher.kill_as(&Slug("ann".into()), RunTurn::PersonaTurn); + rig.end_run(ANN); + rig.store.save(&HeartbeatState { last_beat_at: rig.now(), handled_mentions: vec![] }).unwrap(); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![] }); + assert!(started_turns(&rig).is_empty()); +} + +#[test] +fn the_operator_goes_first_and_the_night_waits_for_her_to_be_idle() { + let rig = nights("13:00-14:00", "ann"); + rig.say(ANN, delivery(1, "@op:chat", "Ann, a word", &[])); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("ann".into())] }); + assert_eq!(started_turns(&rig), vec![("ann".to_string(), RunTurn::PersonaTurn)], "the mention is answered awake, no Sleep that poll"); + end_launched(&rig, ANN); + rig.chats[ANN].deliveries.borrow_mut().clear(); + rig.advance(60); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("ann".into())] }); + assert_eq!(started_turns(&rig).last(), Some(&("ann".to_string(), RunTurn::Sleep))); +} + +#[test] +fn a_sleeping_mind_is_not_eligible_for_the_beat() { + let rig = nights("13:00-14:00", "ann"); + rig.store.save(&HeartbeatState { last_beat_at: DateTime::::UNIX_EPOCH, handled_mentions: vec![] }).unwrap(); + rig.pulse(ANN, 5, None, Some(1000)); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("ann".into()), Slug("bob".into())] }); + assert_eq!(started_turns(&rig), vec![("ann".to_string(), RunTurn::Sleep), ("bob".to_string(), RunTurn::PersonaTurn)], "the beat went to the Mind who was awake, though ann's urgency was the higher"); +} diff --git a/crates/mind-body/src/lib.rs b/crates/mind-body/src/lib.rs index 01dd16a..57bcf25 100644 --- a/crates/mind-body/src/lib.rs +++ b/crates/mind-body/src/lib.rs @@ -48,6 +48,7 @@ pub mod persona; pub mod queue; mod recall; pub mod self_run; +mod sleep; pub mod stream; pub mod tempo; diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index 3ca9c16..e13a968 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -88,12 +88,12 @@ use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest, ContextLine, Refus use crate::connector::render_plan; use crate::control::{ControlSource, Effective, HeldReason, derive}; use crate::heartbeat::{HeartbeatView, PollTrace}; -use crate::lease_client::{LeaseRequest, Lender, Lent}; +use crate::lease_client::{Lease, LeaseRequest, Lender, Lent}; use crate::launch::{Clock, LaunchOutcome, LaunchRequest, Launcher, Ports, holders, open_and_launch}; use crate::queue::{MindPort, any_of, citing, headers, in_force, live_claims, of_kind}; use crate::recall; use crate::self_run::session::{Polled, SessionInbox, drive}; -use crate::self_run::{MAX_TURNS, ROTATE_MARGIN_S, SESSION_NAMESPACE, Spawner, Why}; +use crate::self_run::{Child, MAX_TURNS, ROTATE_MARGIN_S, SESSION_NAMESPACE, Spawner, Why}; /// The admitting agent's name on everything this organ admits. pub const AGENT: &str = "mind-persona"; @@ -162,18 +162,18 @@ const FOR_ITEM_FAULT: &str = "start_self_run.for_item: not an open item shown, o const REMEMBER_MAX: usize = 8; const SETTLE_MAX: usize = 8; /// A memory's text in bytes, the people it is about, and a settle's reason in characters. -const REMEMBER_TEXT_MAX_BYTES: usize = 1000; -const ABOUT_MAX: usize = 8; -const WHY_MAX_CHARS: usize = 300; +pub(crate) const REMEMBER_TEXT_MAX_BYTES: usize = 1000; +pub(crate) const ABOUT_MAX: usize = 8; +pub(crate) const WHY_MAX_CHARS: usize = 300; /// The longest id or alias the kind's `Short` holds, in bytes. -const SHORT_MAX_BYTES: usize = 200; +pub(crate) const SHORT_MAX_BYTES: usize = 200; /// The most bytes each open-item list of the responsibilities section carries, outside `GOALS_MAX_BYTES`. const OPEN_COMMITTED_MAX_BYTES: usize = 4 * 1024; const OPEN_ASKED_MAX_BYTES: usize = 4 * 1024; /// The most open items one list fetches; the rest are counted. const OPEN_FETCH: u32 = 64; /// The most event ids a Persona memory cites as its sources. -const SOURCES_MAX: usize = 8; +pub(crate) const SOURCES_MAX: usize = 8; /// The longest raw reply a trace keeps. const TRACE_RAW_MAX: usize = 2000; @@ -407,22 +407,27 @@ const SETTLE_DESCRIPTION: &str = "Close an open item shown under 'You committed const START_SELF_RUN_DESCRIPTION: &str = "A brief for your Self: it reaches your live Self within a minute, or starts one. It is also how you answer a `reply` line; your Self's replies to your briefs come back as `reply` lines."; const FOR_ITEM_DESCRIPTION: &str = "The open commitment or ask this Self run answers: null, an open item id shown in this session, or new: for remember[i] (sort Asked or Committed) written in this same reply. An item taken on leaves the open lists."; +/// One thing she remembers, as a schema: the shape her waking reply and her night share. +pub(crate) fn remember_item_schema() -> Value { + json!({ + "type": "object", + "additionalProperties": false, + "required": ["sort", "text", "about", "salience_permille"], + "properties": { + "sort": { "type": "string", "enum": MemorySort::NAMES }, + "text": { "type": "string", "minLength": 1, "description": format!("At most {REMEMBER_TEXT_MAX_BYTES} bytes.") }, + "about": { "type": "array", "maxItems": ABOUT_MAX, "uniqueItems": true, "items": { "type": "string" } }, + "salience_permille": { "type": "integer", "minimum": 0, "maximum": SALIENCE_MAX_PERMILLE, "description": "How much it matters, in thousandths; with how recent it is, it decides which memories you see when not all fit." } + } + }) +} + pub fn reply_schema(rooms: &[String], campaigns: &[String]) -> String { let remember = json!({ "description": REMEMBER_DESCRIPTION, "type": "array", "maxItems": REMEMBER_MAX, - "items": { - "type": "object", - "additionalProperties": false, - "required": ["sort", "text", "about", "salience_permille"], - "properties": { - "sort": { "type": "string", "enum": MemorySort::NAMES }, - "text": { "type": "string", "minLength": 1, "description": format!("At most {REMEMBER_TEXT_MAX_BYTES} bytes.") }, - "about": { "type": "array", "maxItems": ABOUT_MAX, "uniqueItems": true, "items": { "type": "string" } }, - "salience_permille": { "type": "integer", "minimum": 0, "maximum": SALIENCE_MAX_PERMILLE, "description": "How much it matters, in thousandths; with how recent it is, it decides which memories you see when not all fit." } - } - } + "items": remember_item_schema() }); let settle = json!({ "description": SETTLE_DESCRIPTION, @@ -774,19 +779,19 @@ struct Refused { } /// The instance brake and the swarm store read once: what they allow. -fn read(control: &impl ControlSource, swarm: &impl ControlSource, instance: &Slug) -> Effective { +pub(crate) fn read(control: &impl ControlSource, swarm: &impl ControlSource, instance: &Slug) -> Effective { derive(control, swarm, instance) } -fn label_of(run: &PipelineRef) -> &str { +pub(crate) fn label_of(run: &PipelineRef) -> &str { run.id.0.rsplit(':').next().unwrap_or_default() } -fn date(now: DateTime) -> Date { +pub(crate) fn date(now: DateTime) -> Date { Date(now.format("%Y-%m-%d").to_string()) } -fn unix_ms(at: DateTime) -> u64 { +pub(crate) fn unix_ms(at: DateTime) -> u64 { u64::try_from(at.timestamp_millis()).unwrap_or(0) } @@ -816,10 +821,14 @@ pub fn carry(organs: &Organs, control: &impl ControlSource, swarm: &impl Control Effective::Held { reason } => return Ok(Outcome::Dead(reason)), Effective::Released(_) => {} } - // The run is the one the heartbeat opened for this unit; none held, none run. + // The run is the one the heartbeat opened for this unit, a waking session's or her night's; none held, none run. let Some(holder) = holders(organs.mind, &instance, RunTurn::PersonaTurn)?.into_iter().next() else { return Ok(Outcome::NoRun) }; + let night = matches!(holder.facts, PipelineFacts::Run { turn: RunTurn::Sleep, .. }); let run = holder.id; make_sandbox(organs.sandbox)?; + if night { + return crate::sleep::carry_night(organs, control, swarm, config, &run); + } let mut progress = Progress::default(); let started = organs.clock.now(); // The replies a session opened alone for, this start: a reply that cannot be resolved never loops processes. @@ -880,38 +889,16 @@ fn one_process( remaining_s: u64, progress: &mut Progress, ) -> Result<()> { - let Ok(persona) = organs.persona.instructions() else { - progress.failure = Some(Failure::Persona); - return Ok(()); - }; - let doctrine = match read_doctrine(&config.doctrine) { - Ok(doctrine) => doctrine, + let session_id = Uuid::new_v5(&SESSION_NAMESPACE, format!("{}:p{process}", run.id.0).as_bytes()); + let rooms = reply_rooms(config, &session.conversation); + let schema_of = |campaigns: &[(String, String)]| reply_schema(&rooms, &campaigns.iter().map(|(slug, _)| slug.clone()).collect::>()); + let Started { child, lease, started_ms } = match start_session(organs, config, session_id, remaining_s, schema_of)? { + Ok(started) => started, Err(why) => { progress.failure = Some(why); return Ok(()); } }; - let campaigns = campaigns_in_force(organs.mind)?; - let rooms = reply_rooms(config, &session.conversation); - let schema = reply_schema(&rooms, &campaigns.iter().map(|(slug, _)| slug.clone()).collect::>()); - let prompt = organs.scratch.join(INSTRUCTIONS_FILE); - if std::fs::write(&prompt, format!("{doctrine}{persona}\n\n{}", render_campaigns(&campaigns))).is_err() { - progress.failure = Some(Failure::Session); - return Ok(()); - } - let started_ms = unix_ms(organs.clock.now()); - let lease = match organs.lender.lend(&LeaseRequest { expires_at_unix_ms: started_ms.saturating_add(remaining_s.saturating_mul(1000)) }) { - Ok(Lent::Granted(lease)) => lease, - Ok(Lent::Refused(_)) | Err(_) => { - progress.failure = Some(Failure::Lease); - return Ok(()); - } - }; - let session_id = Uuid::new_v5(&SESSION_NAMESPACE, format!("{}:p{process}", run.id.0).as_bytes()); - let Ok(child) = organs.spawner.spawn(&persona_args(session_id, &config.model, &schema, &prompt), lease.token()) else { - progress.failure = Some(Failure::Session); - return Ok(()); - }; let mut inbox = PersonaInbox::new(organs, control, swarm, config, run, progress, session, started_ms, render_plan(lease.plan_quota(), started_ms)); let driven = drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout()); inbox.settle_replies(); @@ -919,6 +906,45 @@ fn one_process( Ok(()) } +/// A session started: her process, the lease its token came from (held until the session ends) and when it began. +pub(crate) struct Started { + pub(crate) child: Box, + pub(crate) lease: Box, + pub(crate) started_ms: u64, +} + +/// The one prelude of a session, her waking process's and her night's alike: her persona and doctrine and the +/// campaigns in force written as the instructions file, a lease for the time left, and her `claude` under +/// `session_id` with the schema `schema_of` makes of the campaigns. The fault is the `Failure` the start earned. +pub(crate) fn start_session( + organs: &Organs, + config: &Config, + session_id: Uuid, + remaining_s: u64, + schema_of: impl FnOnce(&[(String, String)]) -> String, +) -> Result> { + let Ok(persona) = organs.persona.instructions() else { return Ok(Err(Failure::Persona)) }; + let doctrine = match read_doctrine(&config.doctrine) { + Ok(doctrine) => doctrine, + Err(why) => return Ok(Err(why)), + }; + let campaigns = campaigns_in_force(organs.mind)?; + let schema = schema_of(&campaigns); + let prompt = organs.scratch.join(INSTRUCTIONS_FILE); + if std::fs::write(&prompt, format!("{doctrine}{persona} + +{}", render_campaigns(&campaigns))).is_err() { + return Ok(Err(Failure::Session)); + } + let started_ms = unix_ms(organs.clock.now()); + let lease = match organs.lender.lend(&LeaseRequest { expires_at_unix_ms: started_ms.saturating_add(remaining_s.saturating_mul(1000)) }) { + Ok(Lent::Granted(lease)) => lease, + Ok(Lent::Refused(_)) | Err(_) => return Ok(Err(Failure::Lease)), + }; + let Ok(child) = organs.spawner.spawn(&persona_args(session_id, &config.model, &schema, &prompt), lease.token()) else { return Ok(Err(Failure::Session)) }; + Ok(Ok(Started { child, lease, started_ms })) +} + /// What her session is started with: the Self's argument list is the model for it, with her /// model, her schema and her instructions file, and no resume, continue or fork. fn persona_args(session: Uuid, model: &str, schema: &str, system_prompt_file: &Path) -> Vec { @@ -1717,7 +1743,7 @@ fn refused_note(why: Failure) -> String { } /// The structured output of a result line, as the text `decode_reply` reads; none when it has none. -fn structured_output(line: &str) -> Option { +pub(crate) fn structured_output(line: &str) -> Option { let event: Value = serde_json::from_str(line).ok()?; event.get("structured_output").filter(|output| !output.is_null()).map(Value::to_string) } @@ -2115,7 +2141,7 @@ fn close_run(organs: &Organs, run: &PipelineRef, answered: u32, last_intent: Opt } /// The next resolution sequence of a subject: the writer's, one past the latest record of it. -fn next_sequence(mind: &dyn MindPort, subject: &PipelineRef) -> Result { +pub(crate) fn next_sequence(mind: &dyn MindPort, subject: &PipelineRef) -> Result { let closures = Selection { cites: Some(citing(subject, Some("subject"))), ..of_kind(PipelineKind::Resolution) }; let latest = headers(mind, &closures)? .iter() diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 90d06a2..ecb4620 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -7035,3 +7035,216 @@ fn a_flush_after_an_image_line_carries_no_image() { assert!(flush.images.is_empty(), "the flush does not repeat the image"); assert_eq!(requests.iter().filter(|told| !told.images.is_empty()).count(), 1); } + +// ---- the night ---- + +use crate::sleep::{SLEEP_GUIDANCE, context, night_line, sleep_schema}; + +const HOUR_MS: i64 = 3_600_000; +const NIGHT_AT_MS: i64 = 48 * HOUR_MS; + +fn memory_id(label: &str) -> String { + format!("{INSTANCE}:memory:{label}") +} + +fn dm(room: &str, with: &[&str]) -> MemoryOrigin { + MemoryOrigin::Direct { room: room.into(), with: with.iter().map(|who| (*who).into()).collect() } +} + +impl Rig { + /// One of her memories, admitted `hours_ago` hours before the night opens, with sources. + #[allow(clippy::too_many_arguments, reason = "a fixture states every field of a memory")] + fn remembered(&self, label: &str, sort: MemorySort, text: &str, about: &[&str], origin: MemoryOrigin, hours_ago: i64, sources: &[&str]) { + self.mind.admit_at(at(NIGHT_AT_MS - hours_ago * HOUR_MS)); + let PipelineDocument::Memory(mut memory) = memory_from(label, sort, text, about, origin, 500) else { unreachable!() }; + memory.sources = sources.iter().map(|source| (*source).into()).collect(); + self.mind.committed(vec![PipelineDocument::Memory(memory)]); + } + + /// The heartbeat's part for a night: the Sleep run opens at `NIGHT_AT_MS`, and the clock stands ten minutes on. + fn open_the_night(&self) -> PipelineRef { + self.mind.admit_at(at(NIGHT_AT_MS)); + self.clock.now.set(NIGHT_AT_MS + 600_000); + let effective = read(&FileSource::new(self.store()), &FileSource::new(self.swarm_store()), self.mind.instance()); + let Effective::Released(grant) = effective else { panic!("released") }; + let ports = Ports { mind: &self.mind, launcher: &Recording::default(), clock: &self.clock, host: "yggdrasil-host" }; + let request = LaunchRequest { turn: RunTurn::Sleep, claims: vec![], agent: "mind-heartbeat".into(), grant }; + let LaunchOutcome::Launched { run } = open_and_launch(&ports, request).unwrap() else { panic!("not launched") }; + run + } + + fn memory_in_force(&self, label: &str) -> bool { + let id = memory_id(label); + self.mind.view(K::Memory, &id).is_some_and(|view| matches!(view.status, huginn_mind::PipelineStatus::InForce)) + } + + fn the_memory(&self, label: &str) -> PipelineMemory { + let Some(view) = self.mind.view(K::Memory, &memory_id(label)) else { panic!("{label} is not in her mind") }; + let PipelineDocument::Memory(memory) = view.document else { unreachable!() }; + memory + } + + /// How the Sleep run was closed. + fn night_closing(&self, run: &PipelineRef) -> String { + let label = label_of(run); + let view = self.mind.view(K::Resolution, &format!("{INSTANCE}:resolution:run.{label}.n1")).expect("the night is closed"); + let PipelineDocument::Resolution(resolution) = view.document else { unreachable!() }; + let ResolutionOutcome::Recorded { reason } = resolution.outcome else { panic!("a night is closed Recorded") }; + reason.0 + } + + fn the_night(&self) -> Outcome { + self.start_with(&config()) + } +} + +fn night_reply(night: &str, keep: &[&str], consolidate: serde_json::Value, retire: serde_json::Value, remember: serde_json::Value) -> String { + let keep: Vec = keep.iter().map(|label| memory_id(label)).collect(); + json!({"night": night, "keep": keep, "consolidate": consolidate, "retire": retire, "remember": remember}).to_string() +} + +fn consolidating(consumes: &[&str], text: &str) -> serde_json::Value { + let consumes: Vec = consumes.iter().map(|label| memory_id(label)).collect(); + json!({"consumes": consumes, "sort": "Learned", "text": text, "about": [], "salience_permille": 700}) +} + +#[test] +fn the_night_line_holds_the_day_whole_and_prescribes_nothing() { + let rig = Rig::new(vec![], &[]); + let room = || MemoryOrigin::Room { rooms: "swarm".into(), heard_from: vec!["@op:chat".into()] }; + // Before the day: an open ask, and forty older memories (the backlog the tide moves through). + rig.remembered("ask", MemorySort::Asked, "Find the release owner.", &["@op:chat"], room(), 60, &[]); + for n in 0..40 { + rig.remembered(&format!("old{n:02}"), MemorySort::Learned, &format!("Older fact {n}."), &[], room(), 50, &[]); + } + // An earlier night of hers, opened 30 hours before this one, closed. + rig.mind.admit_at(at(NIGHT_AT_MS - 30 * HOUR_MS)); + rig.mind.committed(vec![crate::testkit::run("slept-before", RunTurn::Sleep, RunOperator::Mind, &[])]); + rig.mind.committed(vec![crate::testkit::close(crate::testkit::run_ref("slept-before"), crate::testkit::recorded())]); + // Her day: after that night opened. A Sleep-origin memory of it is not part of it. + rig.remembered("d1", MemorySort::Decided, "Ship on Friday.", &["@op:chat"], room(), 20, &["$e1"]); + rig.remembered("d2", MemorySort::Learned, "A private thing.", &["@kiko:chat"], dm("!dm:chat", &["@kiko:chat"]), 10, &[]); + rig.remembered("formed", MemorySort::Idea, "Formed in sleep.", &[], MemoryOrigin::Sleep { run: "slept-before".into() }, 5, &[]); + let run = rig.open_the_night(); + let now = at(NIGHT_AT_MS + HOUR_MS); + let night = context(&rig.mind, &run, now).unwrap(); + let ids = |held: &[crate::sleep::Held]| held.iter().map(|held| held.id.clone()).collect::>(); + assert_eq!(ids(&night.day), vec![memory_id("d1"), memory_id("d2")], "oldest first, the Sleep-origin memory left out"); + assert_eq!(ids(&night.open), vec![memory_id("ask")]); + let tide = ids(&night.tide); + assert_eq!(tide.len(), 32); + assert_eq!((tide[0].as_str(), tide[8].as_str()), (memory_id("old32").as_str(), memory_id("old00").as_str()), "one earlier night moves the tide 32 on and it wraps the backlog"); + let line = night_line(&night); + let expected_day = format!( + "Your day, whole, oldest first (2 memories):\n- {} [Decided, 21h ago, salience 500] in #swarm, with \"@op:chat\"; about @op:chat; sources $e1\n Ship on Friday.\n- {} [Learned, 11h ago, salience 500] privately, with \"@kiko:chat\"; about @kiko:chat\n A private thing.\n\n", + memory_id("d1"), + memory_id("d2") + ); + assert!(line.starts_with("It is night, 2026-10-11 14:15 UTC.\n\nYour memories in force: Asked 1, Decided 1, Idea 1, Learned 41.\n\n"), "{line}"); + assert!(line.contains(&expected_day), "{line}"); + assert!(line.contains("Your open asks and commitments from before your day, oldest first (1):\n"), "{line}"); + assert!(line.ends_with(SLEEP_GUIDANCE), "the guidance closes the line"); + assert!(!line.contains("Formed in sleep.")); + assert!(!SLEEP_GUIDANCE.chars().any(|c| c.is_ascii_digit()), "the night names no count"); + assert!(!SLEEP_GUIDANCE.contains("theme") && !SLEEP_GUIDANCE.contains("topic"), "nor a theme"); +} + +#[test] +fn an_uncovered_night_is_told_back_by_id_and_the_third_refusal_ends_it_with_nothing_admitted() { + let uncovered = night_reply("A night.", &["d1"], json!([]), json!([]), json!([])); + let rig = Rig::new(vec![operator_says(1, "do not read me")], &[uncovered.clone(), uncovered.clone(), uncovered]); + rig.remembered("d1", MemorySort::Learned, "One.", &[], MemoryOrigin::Wake, 5, &[]); + rig.remembered("d2", MemorySort::Learned, "Two.", &[], MemoryOrigin::Wake, 4, &[]); + let run = rig.open_the_night(); + let before = rig.mind.ids(K::Memory).len(); + assert!(matches!(rig.the_night(), Outcome::Served { failure: Some(_), .. })); + let told = rig.model.requests.borrow(); + assert_eq!(told.len(), 3, "the start line and two retells"); + assert!(told[1].input.contains(&format!("in keep, consumes or retire: {}", memory_id("d2"))), "{}", told[1].input); + drop(told); + let reason = rig.night_closing(&run); + assert!(reason.starts_with("failed after 3 refusals: ") && reason.contains(&memory_id("d2")), "{reason}"); + assert_eq!(rig.mind.ids(K::Memory).len(), before, "nothing was admitted"); + assert!(rig.memory_in_force("d1") && rig.memory_in_force("d2")); + assert_eq!((rig.chat.pending_calls.get(), rig.chat.acked.borrow().len()), (0, 0), "a night neither reads nor acks chat"); +} + +#[test] +fn a_consolidation_keeps_the_conversation() { + let mixed = night_reply("Tried to mix.", &[], json!([consolidating(&["a", "b"], "Mixed.")]), json!([]), json!([])); + let kept = night_reply("Kept apart.", &["b"], json!([consolidating(&["a", "c"], "What kiko and the other member settled.")]), json!([]), json!([])); + let rig = Rig::new(vec![], &[mixed, kept]); + rig.model.linger.set(true); + rig.remembered("a", MemorySort::Learned, "Private one.", &[], dm("!dm:chat", &["@kiko:chat"]), 6, &["$a1", "$a2"]); + rig.remembered("b", MemorySort::Learned, "Public.", &[], MemoryOrigin::Wake, 5, &[]); + rig.remembered("c", MemorySort::Learned, "Private two.", &[], dm("!dm:chat", &["@libby:chat", "@kiko:chat"]), 4, &["$a2", "$c1"]); + let run = rig.open_the_night(); + let served = rig.the_night(); + assert!(matches!(served, Outcome::Served { failure: None, .. }), "{}", rig.night_closing(&run)); + let told = rig.model.requests.borrow(); + assert!(told[1].input.contains("one conversation only") && told[1].input.contains(&memory_id("a")), "{}", told[1].input); + drop(told); + let formed = rig.the_memory(&format!("{}-c0", label_of(&run))); + assert_eq!(formed.origin, dm("!dm:chat", &["@kiko:chat", "@libby:chat"]), "the DM stays its own, the union of who spoke in it"); + assert_eq!(formed.sources.iter().map(|source| source.0.as_str()).collect::>(), vec!["$c1", "$a2", "$a1"], "the consumed memories' sources, newest first, once each"); + assert_eq!(formed.consolidates.iter().map(|consumed| consumed.id.0.clone()).collect::>(), vec![memory_id("a"), memory_id("c")]); + assert!(!rig.memory_in_force("a") && !rig.memory_in_force("c") && rig.memory_in_force("b")); +} + +#[test] +fn a_decoded_night_admits_whole_operations_and_the_run_closes_with_her_line() { + let reply = night_reply( + "I looked at the release again and let the rest go.", + &["keep"], + json!([consolidating(&["m1", "m2"], "The release owner is unknown; ask the operator Friday.")]), + json!([ + {"item": memory_id("done"), "outcome": "done", "why": "It shipped."}, + {"item": memory_id("faded"), "outcome": "faded", "why": "Nobody asked again."}, + {"item": memory_id("declined"), "outcome": "declined", "why": "Not mine."}, + {"item": memory_id("wrong"), "outcome": "wrong", "why": "It was mistaken."} + ]), + json!([{"sort": "Idea", "text": "Try the other index.", "about": [], "salience_permille": 400}]), + ); + let rig = Rig::new(vec![operator_says(1, "pending, never read")], &[reply]); + for (n, label) in ["m1", "m2", "keep", "done", "faded", "declined", "wrong"].iter().enumerate() { + let sort = if ["done", "faded", "declined", "wrong"].contains(label) { MemorySort::Committed } else { MemorySort::Learned }; + rig.remembered(label, sort, &format!("About {label}."), &[], MemoryOrigin::Wake, 7 - n as i64, &[]); + } + let run = rig.open_the_night(); + assert!(matches!(rig.the_night(), Outcome::Served { failure: None, .. })); + let tag = label_of(&run).to_string(); + assert!(rig.memory_in_force("keep") && rig.memory_in_force(&format!("{tag}-c0")) && rig.memory_in_force(&format!("{tag}-r0"))); + for label in ["m1", "m2", "done", "faded", "declined", "wrong"] { + assert!(!rig.memory_in_force(label), "{label} left force"); + } + let outcome = |label: &str| { + let view = rig.mind.view(K::Resolution, &format!("{INSTANCE}:resolution:memory.{label}.n1")).unwrap(); + let PipelineDocument::Resolution(resolution) = view.document else { unreachable!() }; + resolution.outcome + }; + assert!(matches!(outcome("m1"), ResolutionOutcome::Superseded { ref by } if by[0].id.0 == memory_id(&format!("{tag}-c0")))); + assert!(matches!(outcome("done"), ResolutionOutcome::Recorded { .. }) && matches!(outcome("faded"), ResolutionOutcome::Recorded { .. })); + assert!(matches!(outcome("declined"), ResolutionOutcome::Withdrawn { .. }) && matches!(outcome("wrong"), ResolutionOutcome::Withdrawn { .. })); + assert_eq!(rig.the_memory(&format!("{tag}-c0")).origin, MemoryOrigin::Sleep { run: tag.as_str().into() }, "a public consolidation is formed in sleep"); + assert_eq!(rig.the_memory(&format!("{tag}-r0")).origin, MemoryOrigin::Sleep { run: tag.as_str().into() }); + assert_eq!(rig.night_closing(&run), "slept: I looked at the release again and let the rest go."); + // She reads it on waking, as the line of her last runs. + let lines = recent_runs(&rig.mind, &Slug(INSTANCE.into())).unwrap(); + assert!(lines[0].starts_with(&format!("{tag} Sleep: slept: I looked")), "{lines:?}"); + assert_eq!((rig.chat.pending_calls.get(), rig.chat.acked.borrow().len(), rig.sent_texts().len()), (0, 0, 0), "pending chat was neither read into the night nor acked"); +} + +#[test] +fn a_sleep_holder_opens_one_process_with_the_night_schema_and_her_tools_never_a_conversation() { + let reply = night_reply("Nothing to do.", &[], json!([]), json!([]), json!([])); + let rig = Rig::new(vec![operator_says(1, "hello, wake up")], &[reply]); + rig.open_the_night(); + rig.the_night(); + let spawns = rig.model.spawns.borrow(); + assert_eq!(spawns.len(), 1); + let after = |flag: &str| spawns[0].iter().position(|arg| arg == flag).map(|at| spawns[0][at + 1].clone()).unwrap(); + assert_eq!(after("--json-schema"), sleep_schema()); + assert_eq!(after("--tools"), "Bash,WebFetch,WebSearch"); + assert_eq!(rig.chat.pending_calls.get(), 0); + assert!(rig.model.requests.borrow()[0].input.contains("This is your night.")); +} diff --git a/crates/mind-body/src/sleep.rs b/crates/mind-body/src/sleep.rs new file mode 100644 index 0000000..6a5e78a --- /dev/null +++ b/crates/mind-body/src/sleep.rs @@ -0,0 +1,578 @@ +//! A Mind's night. Once in the heartbeat's window a Sleep run opens in her Persona slot (the heartbeat decides +//! who and when); this module is the one organ that carries it: one session over the same prelude her waking +//! session starts with, told her day whole, answering with a [`SleepReply`] that this module turns into +//! admissions (ruling `eureka-body:ruling:nightly-sleep-turn-self-directed`). +//! +//! What she contemplates is hers: the night line prescribes no reading, theme, text or count. What the code owns +//! is the ledger of her day. Every memory of it is disposed of exactly once (kept, consumed by a consolidation, +//! or retired); an id is one of her memories in force; a consolidation merges one conversation's memories only +//! and supersedes each source in the batch that admits it (admission refuses the unpaired). Code never chooses +//! what to merge and never writes a memory's text. A night is one process: a missed night is not made up and a +//! failed night is not retried. + +use std::collections::{BTreeMap, BTreeSet}; + +use anyhow::{Result, anyhow, bail}; +use chrono::{DateTime, Duration, Utc}; +use cultnet_rs::Selection; +use eureka_pipeline::{Label, Line, MemoryOrigin, MemorySort, PipelineDocument, PipelineKind, PipelineMemory, PipelineRef, PipelineResolution, ResolutionOutcome, SALIENCE_MAX_PERMILLE, Short}; +use huginn_mind::{BATCH_MAX, PipelineAdmissionOutcome, PipelinePageItems}; +use serde::Deserialize; +use serde_json::json; + +use crate::control::ControlSource; +use crate::persona::{ + ABOUT_MAX, AGENT, Config, Conversation, Failure, Organs, Outcome, PERSONA_POLL_S, REMEMBER_TEXT_MAX_BYTES, Remember, SOURCES_MAX, Started, WHY_MAX_CHARS, date, label_of, next_sequence, remember_item_schema, start_session, + structured_output, +}; +use crate::queue::{MindPort, any_of, headers, in_force, of_kind}; +use crate::recall; +use crate::self_run::session::{Polled, SessionInbox, drive}; +use crate::self_run::{SESSION_NAMESPACE, Why}; + +/// The longest a night runs, whatever the unit's cap. +pub(crate) const SLEEP_CAP_S: u64 = 3600; +/// The most memories of her day one night line shows; the rest are older by the next night. +pub(crate) const DAY_MAX: usize = 96; +/// The most open Asked and Committed items older than her day the line shows. +const OPEN_MAX: usize = 48; +/// The older memories of other sorts one night shows, and the step the tide moves each night. +const TIDE: usize = 32; +const NIGHT_MAX_BYTES: usize = 900; +const CONSOLIDATE_MAX: usize = 32; +const CONSUMES_MAX: usize = 16; +const RETIRE_MAX: usize = 128; +const REMEMBER_MAX: usize = 16; +const REFUSALS_PER_NIGHT: u32 = 3; + +/// What the night tells her, in her second person. It names no text, theme, reading or count. +pub(crate) const SLEEP_GUIDANCE: &str = "\ +This is your night. What you think about, read or look into is yours: your own mind with `huginn`, your shell, the web, or nothing at all. No reading is assigned and nothing is expected of it. + +Your day is shown whole above. Every memory of it gets exactly one disposition: keep it (name it in `keep`), fold it into a consolidation (name it in that consolidation's `consumes`), or retire it with an outcome and a reason (`retire`: done, declined, faded or wrong). A consolidation is one new memory in your words that takes the place of the memories it consumes, so write it to carry what those memories were for: the concrete target, the claim or the question, the live tension, why it changes what you do next, and the anchors (ids, names, places) it rests on. It reduces bulk, not meaning. Consolidate only memories of one conversation; a private conversation stays its own. Anything else you want to remember goes in `remember`. + +Older items shown, and any other memory of yours in force, are yours to touch or leave. Name every memory by its id exactly as shown, once in the whole reply. + +End with one line of what the night was (`night`)."; + +/// The night's reply, as the one type decoded. Her day's ledger is `keep`, `consolidate` and `retire`. +#[derive(Clone, Debug, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct SleepReply { + pub(crate) night: String, + pub(crate) keep: Vec, + pub(crate) consolidate: Vec, + pub(crate) retire: Vec, + pub(crate) remember: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Consolidate { + pub(crate) consumes: Vec, + pub(crate) sort: MemorySort, + pub(crate) text: String, + pub(crate) about: Vec, + pub(crate) salience_permille: u16, +} + +#[derive(Clone, Debug, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Retire { + pub(crate) item: String, + pub(crate) outcome: RetireOutcome, + pub(crate) why: String, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub(crate) enum RetireOutcome { + Done, + Declined, + Faded, + Wrong, +} + +/// The JSON Schema of [`SleepReply`], with the bounds the decoder holds. +pub(crate) fn sleep_schema() -> String { + let id = json!({ "type": "string" }); + let remember = |maximum: usize| json!({ "type": "array", "maxItems": maximum, "items": remember_item_schema() }); + let mut consolidate = remember_item_schema(); + consolidate["required"] = json!(["consumes", "sort", "text", "about", "salience_permille"]); + consolidate["properties"]["consumes"] = json!({ "type": "array", "minItems": 1, "maxItems": CONSUMES_MAX, "uniqueItems": true, "items": id }); + json!({ + "type": "object", + "additionalProperties": false, + "required": ["night", "keep", "consolidate", "retire", "remember"], + "properties": { + "night": { "type": "string", "minLength": 1, "description": format!("One line of what the night was. At most {NIGHT_MAX_BYTES} bytes.") }, + "keep": { "type": "array", "uniqueItems": true, "items": id }, + "consolidate": { "type": "array", "maxItems": CONSOLIDATE_MAX, "items": consolidate }, + "retire": { + "type": "array", + "maxItems": RETIRE_MAX, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["item", "outcome", "why"], + "properties": { + "item": id, + "outcome": { "type": "string", "enum": ["done", "declined", "faded", "wrong"] }, + "why": { "type": "string", "minLength": 1, "maxLength": WHY_MAX_CHARS } + } + } + }, + "remember": remember(REMEMBER_MAX) + } + }) + .to_string() +} + +/// One of her memories in force, with when it was admitted. +#[derive(Clone, Debug)] +pub(crate) struct Held { + pub(crate) id: String, + pub(crate) memory: PipelineMemory, + pub(crate) at: DateTime, + pub(crate) ordinal: u64, +} + +/// What a night is told and what its reply may name: derived from her mind and nothing stored. +pub(crate) struct Context { + pub(crate) now: DateTime, + pub(crate) label: String, + pub(crate) counts: BTreeMap, + pub(crate) day: Vec, + pub(crate) open: Vec, + pub(crate) tide: Vec, + /// Every memory of hers in force, by id. + pub(crate) all: BTreeMap, +} + +fn parse_time(at: &str) -> Result> { + Ok(DateTime::parse_from_rfc3339(at).map_err(|_| anyhow!("an admission time could not be read"))?.with_timezone(&Utc)) +} + +/// Her memories in force, oldest admitted first: the one read of what she holds. +fn memories_in_force(mind: &dyn MindPort) -> Result> { + let mut found = Vec::new(); + let mut asking = Selection { projection: "document".into(), ..in_force(of_kind(PipelineKind::Memory)) }; + loop { + let page = mind.query(&asking)?; + let PipelinePageItems::Documents(views) = page.items else { bail!("the mind answered headers to a document selection") }; + for view in views { + if let PipelineDocument::Memory(memory) = view.document { + found.push(Held { id: view.id.id.0, memory, at: parse_time(&view.admission.admitted_at)?, ordinal: view.admission.ordinal }); + } + } + match page.next { + Some(cursor) => asking.cursor = Some(cursor), + None => break, + } + } + found.sort_by_key(|held| held.ordinal); + Ok(found) +} + +/// Her Sleep runs, any status, oldest first, with when each was admitted. +fn sleeps(mind: &dyn MindPort) -> Result)>> { + let runs = any_of(any_of(any_of(of_kind(PipelineKind::Run), "root", &[&mind.instance().0]), "turn", &["Sleep"]), "operated_by", &["Mind"]); + let mut found = headers(mind, &runs)?.into_iter().map(|run| Ok((run.id, run.admission.ordinal, parse_time(&run.admission.admitted_at)?))).collect::>>()?; + found.sort_by_key(|(_, ordinal, _)| *ordinal); + Ok(found.into_iter().map(|(id, _, at)| (id, at)).collect()) +} + +fn is_open_sort(sort: MemorySort) -> bool { + matches!(sort, MemorySort::Asked | MemorySort::Committed) +} + +fn is_sleep_made(held: &Held) -> bool { + matches!(held.memory.origin, MemoryOrigin::Sleep { .. }) +} + +/// The night, derived. Her day: her memories in force admitted after her previous Sleep run opened (else in the +/// 24 hours before this run opened), formed otherwise than in sleep, oldest first, at most `DAY_MAX`. Her open +/// Asked and Committed items older than the day. A tide of `TIDE` older memories of other sorts, from an offset +/// that moves by `TIDE` with each earlier Sleep run of hers. +pub(crate) fn context(mind: &dyn MindPort, run: &PipelineRef, now: DateTime) -> Result { + let nights = sleeps(mind)?; + let run_at = nights.iter().find(|(other, _)| other == run).map(|(_, at)| *at).ok_or_else(|| anyhow!("the night's run is not in her mind"))?; + let earlier: Vec<&(PipelineRef, DateTime)> = nights.iter().filter(|(other, _)| other != run).collect(); + let day_start = earlier.last().map_or(run_at - Duration::hours(24), |(_, at)| *at); + let memories = memories_in_force(mind)?; + let mut counts = BTreeMap::new(); + for held in &memories { + *counts.entry(format!("{:?}", held.memory.sort)).or_insert(0) += 1; + } + let day: Vec = memories.iter().filter(|held| held.at > day_start && !is_sleep_made(held)).take(DAY_MAX).cloned().collect(); + let open: Vec = memories.iter().filter(|held| held.at <= day_start && is_open_sort(held.memory.sort)).take(OPEN_MAX).cloned().collect(); + let backlog: Vec<&Held> = memories.iter().filter(|held| held.at <= day_start && !is_open_sort(held.memory.sort) && !is_sleep_made(held)).collect(); + let tide: Vec = if backlog.is_empty() { + Vec::new() + } else { + let offset = earlier.len() * TIDE % backlog.len(); + (0..TIDE.min(backlog.len())).map(|step| backlog[(offset + step) % backlog.len()].clone()).collect() + }; + let all = memories.into_iter().map(|held| (held.id.clone(), held)).collect(); + Ok(Context { now, label: label_of(run).to_string(), counts, day, open, tide, all }) +} + +fn age(now: DateTime, at: DateTime) -> String { + let minutes = (now - at).num_minutes().max(0); + match minutes { + 0..=59 => format!("{minutes}m"), + 60..=2879 => format!("{}h", minutes / 60), + _ => format!("{}d", minutes / 1440), + } +} + +fn entry(now: DateTime, held: &Held) -> String { + let memory = &held.memory; + let about = if memory.about.is_empty() { String::new() } else { format!("; about {}", memory.about.iter().map(|who| who.0.as_str()).collect::>().join(", ")) }; + let sources = if memory.sources.is_empty() { String::new() } else { format!("; sources {}", memory.sources.iter().map(|source| source.0.as_str()).collect::>().join(", ")) }; + format!( + "- {} [{:?}, {} ago, salience {}] {}{about}{sources}\n {}\n", + held.id, + memory.sort, + age(now, held.at), + memory.salience_permille, + recall::origin_phrase(&memory.origin, &BTreeMap::new()), + memory.text.0 + ) +} + +/// The night line: the date, her counts in force, her day whole, her open items, the tide, and the guidance. +pub(crate) fn night_line(context: &Context) -> String { + let now = context.now; + let counts = if context.counts.is_empty() { "none".to_string() } else { context.counts.iter().map(|(sort, count)| format!("{sort} {count}")).collect::>().join(", ") }; + let mut line = format!("It is night, {} UTC.\n\nYour memories in force: {counts}.\n\n", now.format("%Y-%m-%d %H:%M")); + if context.day.is_empty() { + line.push_str("Your day: nothing new has been admitted since your last night.\n\n"); + } else { + line.push_str(&format!("Your day, whole, oldest first ({} memories):\n", context.day.len())); + context.day.iter().for_each(|held| line.push_str(&entry(now, held))); + line.push('\n'); + } + if !context.open.is_empty() { + line.push_str(&format!("Your open asks and commitments from before your day, oldest first ({}):\n", context.open.len())); + context.open.iter().for_each(|held| line.push_str(&entry(now, held))); + line.push('\n'); + } + if !context.tide.is_empty() { + line.push_str(&format!("Older memories of yours, as they come round ({}):\n", context.tide.len())); + context.tide.iter().for_each(|held| line.push_str(&entry(now, held))); + line.push('\n'); + } + line.push_str(SLEEP_GUIDANCE); + line +} + +fn bytes_fault(field: &str, text: &str, most: usize) -> Option { + (text.trim().is_empty() || text.len() > most).then(|| format!("{field}: empty, or over {most} bytes")) +} + +/// The reply decoded and held to the ledger. The fault is told back to her whole. +pub(crate) fn decode(text: &str, context: &Context) -> std::result::Result { + let reply: SleepReply = serde_json::from_str(text).map_err(|error| format!("the reply is not a SleepReply: {error}"))?; + let mut faults: Vec = Vec::new(); + faults.extend(bytes_fault("night", &reply.night, NIGHT_MAX_BYTES)); + if reply.consolidate.len() > CONSOLIDATE_MAX || reply.retire.len() > RETIRE_MAX || reply.remember.len() > REMEMBER_MAX { + faults.push(format!("at most {CONSOLIDATE_MAX} consolidations, {RETIRE_MAX} retirements and {REMEMBER_MAX} remembered")); + } + let shown: BTreeSet = context.day.iter().chain(&context.open).chain(&context.tide).flat_map(|held| recall::people_of(&held.memory)).collect(); + let about_fault = |about: &[String]| about.len() > ABOUT_MAX || about.iter().any(|who| !shown.contains(who)); + for consolidation in &reply.consolidate { + if consolidation.consumes.is_empty() || consolidation.consumes.len() > CONSUMES_MAX { + faults.push(format!("consumes: between 1 and {CONSUMES_MAX} ids")); + } + faults.extend(bytes_fault("consolidate.text", &consolidation.text, REMEMBER_TEXT_MAX_BYTES)); + if consolidation.salience_permille > SALIENCE_MAX_PERMILLE || about_fault(&consolidation.about) { + faults.push("consolidate.about or salience_permille: out of bounds (about is an id of a person a memory shown to you names)".into()); + } + let conversations: Vec = consolidation.consumes.iter().filter_map(|id| context.all.get(id)).map(|held| Conversation::of_origin(Some(&held.memory.origin))).collect(); + if conversations.windows(2).any(|pair| pair[0] != pair[1]) { + faults.push(format!("a consolidation merges the memories of one conversation only: {} are of two", consolidation.consumes.join(", "))); + } + } + for retirement in &reply.retire { + if retirement.why.trim().is_empty() || retirement.why.chars().count() > WHY_MAX_CHARS { + faults.push(format!("retire {}: why is empty or over {WHY_MAX_CHARS} characters", retirement.item)); + } + } + for remembered in &reply.remember { + faults.extend(bytes_fault("remember.text", &remembered.text, REMEMBER_TEXT_MAX_BYTES)); + if remembered.salience_permille > SALIENCE_MAX_PERMILLE || about_fault(&remembered.about) { + faults.push("remember.about or salience_permille: out of bounds (about is an id of a person a memory shown to you names)".into()); + } + } + let named: Vec<&String> = reply.keep.iter().chain(reply.consolidate.iter().flat_map(|consolidation| &consolidation.consumes)).chain(reply.retire.iter().map(|retirement| &retirement.item)).collect(); + let mut seen = BTreeSet::new(); + for id in &named { + if !seen.insert(id.as_str()) { + faults.push(format!("{id} is named more than once; every memory is named once in the whole reply")); + } + } + for id in &named { + if !context.all.contains_key(id.as_str()) { + faults.push(format!("{id} is not a memory of yours in force")); + } + } + let missing: Vec<&str> = context.day.iter().map(|held| held.id.as_str()).filter(|id| !seen.contains(id)).collect(); + if !missing.is_empty() { + faults.push(format!("your day is not covered; dispose of each of these in keep, consumes or retire: {}", missing.join(", "))); + } + if faults.is_empty() { Ok(reply) } else { Err(faults.join("\n")) } +} + +fn reference(id: &str) -> PipelineRef { + PipelineRef { kind: PipelineKind::Memory, id: Short(id.to_string()) } +} + +/// The reply as whole operations, each a unit of documents that stands or falls together: a consolidation (its +/// memory and a Superseded resolution of each memory it consumes), a retirement, a remembered memory. +pub(crate) fn operations(mind: &dyn MindPort, context: &Context, reply: &SleepReply) -> Result>> { + let instance = mind.instance().clone(); + let on = date(context.now); + let run = Short(context.label.clone()); + let mut units = Vec::new(); + for (index, consolidation) in reply.consolidate.iter().enumerate() { + let mut consumed: Vec<&Held> = consolidation.consumes.iter().filter_map(|id| context.all.get(id)).collect(); + consumed.sort_by_key(|held| held.ordinal); + let origin = match Conversation::of_origin(consumed.first().map(|held| &held.memory.origin)) { + Conversation::Direct(room) => { + let mut with: Vec = Vec::new(); + for who in consumed.iter().flat_map(|held| match &held.memory.origin { + MemoryOrigin::Direct { with, .. } => with.clone(), + _ => Vec::new(), + }) { + if !with.contains(&who) && with.len() < ABOUT_MAX { + with.push(who); + } + } + MemoryOrigin::Direct { room: Short(room), with } + } + Conversation::Public => MemoryOrigin::Sleep { run: run.clone() }, + }; + let mut sources: Vec = Vec::new(); + for source in consumed.iter().rev().flat_map(|held| held.memory.sources.iter().rev()) { + if !sources.contains(source) && sources.len() < SOURCES_MAX { + sources.push(source.clone()); + } + } + let label = Label(format!("{}-c{index}", context.label)); + let memory = PipelineMemory { + instance: instance.clone(), + label: label.clone(), + sort: consolidation.sort, + text: Line(consolidation.text.clone()), + about: consolidation.about.iter().cloned().map(Short).collect(), + origin, + sources, + salience_permille: consolidation.salience_permille, + consolidates: consolidation.consumes.iter().map(|id| reference(id)).collect(), + }; + let by = reference(&format!("{}:memory:{}", instance.0, label.0)); + let mut unit = vec![PipelineDocument::Memory(memory)]; + for id in &consolidation.consumes { + let subject = reference(id); + unit.push(PipelineDocument::Resolution(PipelineResolution { + sequence: next_sequence(mind, &subject)?, + subject, + outcome: ResolutionOutcome::Superseded { by: vec![by.clone()] }, + rationale: "Consolidated in sleep.".into(), + resolved_on: on.clone(), + })); + } + units.push(unit); + } + for retirement in &reply.retire { + let subject = reference(&retirement.item); + let reason = Line(retirement.why.clone()); + units.push(vec![PipelineDocument::Resolution(PipelineResolution { + sequence: next_sequence(mind, &subject)?, + subject, + outcome: match retirement.outcome { + RetireOutcome::Done | RetireOutcome::Faded => ResolutionOutcome::Recorded { reason }, + RetireOutcome::Declined | RetireOutcome::Wrong => ResolutionOutcome::Withdrawn { reason }, + }, + rationale: retirement.why.as_str().into(), + resolved_on: on.clone(), + })]); + } + for (index, remembered) in reply.remember.iter().enumerate() { + units.push(vec![PipelineDocument::Memory(PipelineMemory { + instance: instance.clone(), + label: Label(format!("{}-r{index}", context.label)), + sort: remembered.sort, + text: Line(remembered.text.clone()), + about: remembered.about.iter().cloned().map(Short).collect(), + origin: MemoryOrigin::Sleep { run: run.clone() }, + sources: Vec::new(), + salience_permille: remembered.salience_permille, + consolidates: Vec::new(), + })]); + } + Ok(units) +} + +/// Operations packed in order into batches of at most `BATCH_MAX`, none split. +pub(crate) fn pack(units: Vec>) -> Vec> { + let mut batches: Vec> = Vec::new(); + for unit in units { + match batches.last_mut() { + Some(last) if last.len() + unit.len() <= BATCH_MAX => last.extend(unit), + _ => batches.push(unit), + } + } + batches +} + +/// How a night ended. +enum Finish { + Slept { reply: SleepReply }, + Failed { refusals: u32, why: String }, +} + +struct SleepInbox<'a, C, S> { + organs: &'a Organs<'a>, + control: &'a C, + swarm: &'a S, + context: &'a Context, + line: String, + told: Option, + refusals: u32, + finish: Option, +} + +impl SleepInbox<'_, C, S> { + /// Her reply admitted whole operations at a time, in batches; a refused batch ends the night, the earlier ones standing. + fn admit(&self, reply: &SleepReply) -> Result<()> { + let batches = pack(operations(self.organs.mind, self.context, reply)?); + for batch in batches { + match self.organs.mind.admit(AGENT, &self.context.label, batch)? { + PipelineAdmissionOutcome::Committed { .. } | PipelineAdmissionOutcome::AlreadyAdmitted { .. } => {} + refused => bail!("her mind refused a batch of the night: {refused:?}"), + } + } + Ok(()) + } + + fn refuse(&mut self, fault: String) -> (bool, bool) { + self.refusals += 1; + if self.refusals >= REFUSALS_PER_NIGHT { + self.finish = Some(Finish::Failed { refusals: self.refusals, why: fault }); + return (false, true); + } + self.told = Some(format!("Your reply was not taken:\n{fault}\n\nReply again, whole.")); + (true, false) + } +} + +impl SessionInbox for SleepInbox<'_, C, S> { + fn start_line(&mut self) -> Result { + Ok(self.line.clone()) + } + + fn poll_every_s(&self) -> i64 { + PERSONA_POLL_S + } + + /// No chat is read into the night or acked by it; the brakes still stop it. + fn poll(&mut self) -> Result { + if let crate::control::Effective::Held { reason } = crate::persona::read(self.control, self.swarm, self.organs.mind.instance()) { + self.finish.get_or_insert(Finish::Failed { refusals: self.refusals, why: format!("held: {reason:?}") }); + } + Ok(Polled { pending: self.told.is_some(), other: false, other_first: false, yield_now: self.finish.is_some() }) + } + + fn tell(&mut self, _switch: bool) -> Result { + Ok(self.told.take().unwrap_or_default()) + } + + fn flush_text(&mut self, _why: Why) -> String { + "Your time is nearly spent. Reply now with your whole SleepReply: every memory of your day disposed of, and the night in one line.".into() + } + + fn result(&mut self, line: &str, ok: bool, blocked: bool) -> Result<(bool, bool)> { + if self.finish.is_some() { + return Ok((false, true)); + } + if blocked || !ok { + self.finish = Some(Finish::Failed { refusals: self.refusals, why: "the session failed".into() }); + return Ok((false, true)); + } + let Some(text) = structured_output(line) else { return Ok(self.refuse("the result carried no reply".into())) }; + match decode(&text, self.context) { + Ok(reply) => match self.admit(&reply) { + Ok(()) => { + self.finish = Some(Finish::Slept { reply }); + Ok((true, false)) + } + Err(error) => { + self.finish = Some(Finish::Failed { refusals: self.refusals, why: error.to_string() }); + Ok((false, true)) + } + }, + Err(fault) => Ok(self.refuse(fault)), + } + } +} + +/// Closes the night's run Recorded with her line and the counts, or the failure. +fn close(organs: &Organs, run: &PipelineRef, finish: &Finish) -> Result<()> { + let (reason, rationale) = match finish { + Finish::Slept { reply } => ( + format!("slept: {}", reply.night), + format!( + "kept {}, consolidated {} from {}, retired {}, formed {}", + reply.keep.len(), + reply.consolidate.len(), + reply.consolidate.iter().map(|consolidation| consolidation.consumes.len()).sum::(), + reply.retire.len(), + reply.remember.len() + ), + ), + Finish::Failed { refusals, why } => (format!("failed after {refusals} refusals: {why}"), "The night ended without a reply the ledger took.".to_string()), + }; + let resolution = PipelineResolution { + subject: run.clone(), + sequence: next_sequence(organs.mind, run)?, + outcome: ResolutionOutcome::Recorded { reason: reason.as_str().into() }, + rationale: rationale.as_str().into(), + resolved_on: date(organs.clock.now()), + }; + match organs.mind.admit(AGENT, label_of(run), vec![PipelineDocument::Resolution(resolution)])? { + PipelineAdmissionOutcome::Committed { .. } => Ok(()), + _ => bail!("the Sleep run could not be closed"), + } +} + +/// One night, in the Persona unit, under the Sleep run the heartbeat opened: one process, no retry. +pub(crate) fn carry_night(organs: &Organs, control: &impl ControlSource, swarm: &impl ControlSource, config: &Config, run: &PipelineRef) -> Result { + let remaining_s = organs.run_cap_s.min(SLEEP_CAP_S); + let finish = match context(organs.mind, run, organs.clock.now()) { + Err(error) => Finish::Failed { refusals: 0, why: format!("her mind could not be read for the night: {error}") }, + Ok(context) => { + let session_id = uuid::Uuid::new_v5(&SESSION_NAMESPACE, format!("{}:sleep", run.id.0).as_bytes()); + match start_session(organs, config, session_id, remaining_s, |_| sleep_schema())? { + Err(why) => Finish::Failed { refusals: 0, why: format!("the night could not start: {why}") }, + Ok(Started { child, lease, .. }) => { + let mut inbox = SleepInbox { organs, control, swarm, context: &context, line: night_line(&context), told: None, refusals: 0, finish: None }; + let driven = drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout()); + drop(lease); + match (inbox.finish.take(), driven) { + (Some(finish), _) => finish, + (None, Err(error)) => Finish::Failed { refusals: inbox.refusals, why: format!("the session broke: {error}") }, + (None, Ok(_)) => Finish::Failed { refusals: inbox.refusals, why: "the night ended without an answer".into() }, + } + } + } + } + }; + close(organs, run, &finish)?; + let failure = matches!(finish, Finish::Failed { .. }).then_some(Failure::Session); + Ok(Outcome::Served { run: run.clone(), answered: u32::from(failure.is_none()), failure, self_run: None }) +} + +#[cfg(test)] +mod tests; diff --git a/crates/mind-body/src/sleep/tests.rs b/crates/mind-body/src/sleep/tests.rs new file mode 100644 index 0000000..bd28a8e --- /dev/null +++ b/crates/mind-body/src/sleep/tests.rs @@ -0,0 +1,316 @@ +//! The night's ledger rules, one behaviour each: what the decoder refuses and tells back, what an operation +//! admits, how operations pack into batches, and which memories a night is shown. The whole night through a +//! scripted child is in `persona::tests`. + +use eureka_pipeline::{MemoryOrigin, MemorySort, PipelineDocument, ResolutionOutcome, RunOperator, RunTurn}; +use serde_json::json; + +use super::*; +use crate::testkit::{INSTANCE, TestMind, at, memory_from, run, run_ref}; + +const T0: i64 = 1_000_000_000; + +fn dm(room: &str, with: &[&str]) -> MemoryOrigin { + MemoryOrigin::Direct { room: room.into(), with: with.iter().map(|who| (*who).into()).collect() } +} + +fn wake() -> MemoryOrigin { + MemoryOrigin::Wake +} + +fn id(label: &str) -> String { + format!("{INSTANCE}:memory:{label}") +} + +fn held(label: &str, sort: MemorySort, about: &[&str], origin: MemoryOrigin, ordinal: u64) -> Held { + let PipelineDocument::Memory(memory) = memory_from(label, sort, "Text.", about, origin, 500) else { unreachable!() }; + Held { id: id(label), memory, at: at(T0 + i64::try_from(ordinal).unwrap()), ordinal } +} + +/// A night over four memories: two in her day, one open ask, three elsewhere (public, and two DMs). +fn ledger() -> Context { + let day = vec![held("d1", MemorySort::Learned, &["@kiko:chat"], wake(), 4), held("d2", MemorySort::Decided, &[], wake(), 5)]; + let open = vec![held("ask", MemorySort::Asked, &["@op:chat"], wake(), 1)]; + let others = vec![held("x1", MemorySort::Learned, &[], wake(), 2), held("dmA", MemorySort::Learned, &[], dm("!a:chat", &["@a:chat"]), 3), held("dmB", MemorySort::Learned, &[], dm("!b:chat", &["@b:chat"]), 6)]; + let all = day.iter().chain(&open).chain(&others).map(|held| (held.id.clone(), held.clone())).collect(); + Context { now: at(T0 + 100), label: "mind-night".into(), counts: BTreeMap::new(), day, open, tide: Vec::new(), all } +} + +fn reply(keep: &[&str]) -> serde_json::Value { + json!({"night": "A night.", "keep": keep.iter().map(|label| id(label)).collect::>(), "consolidate": [], "retire": [], "remember": []}) +} + +fn fault(reply: &serde_json::Value) -> String { + decode(&reply.to_string(), &ledger()).expect_err("the reply should be refused") +} + +#[test] +fn every_memory_of_her_day_is_disposed_of_exactly_once() { + assert!(decode(&reply(&["d1", "d2"]).to_string(), &ledger()).is_ok()); + let missing = fault(&reply(&["d1"])); + assert!(missing.contains("not covered") && missing.contains(&id("d2")) && !missing.contains(&format!("{}, ", id("d1"))), "{missing}"); + // Disposed of in any of the three ways counts. + let mut by_retire = reply(&["d1"]); + by_retire["retire"] = json!([{"item": id("d2"), "outcome": "faded", "why": "It faded."}]); + assert!(decode(&by_retire.to_string(), &ledger()).is_ok()); + let mut by_consolidation = reply(&[]); + by_consolidation["consolidate"] = json!([{"consumes": [id("d1"), id("d2")], "sort": "Learned", "text": "Both.", "about": [], "salience_permille": 500}]); + assert!(decode(&by_consolidation.to_string(), &ledger()).is_ok()); + // Twice, within a list or across them, is refused and the id named. + let mut twice = reply(&["d1", "d2"]); + twice["retire"] = json!([{"item": id("d2"), "outcome": "done", "why": "Done."}]); + assert!(fault(&twice).contains(&format!("{} is named more than once", id("d2")))); + assert!(fault(&reply(&["d1", "d1", "d2"])).contains("named more than once")); + let mut across = reply(&["d2"]); + across["consolidate"] = json!([ + {"consumes": [id("d1")], "sort": "Learned", "text": "One.", "about": [], "salience_permille": 500}, + {"consumes": [id("d1")], "sort": "Learned", "text": "Again.", "about": [], "salience_permille": 500} + ]); + assert!(fault(&across).contains("named more than once")); +} + +#[test] +fn an_id_must_be_a_memory_of_hers_in_force_whether_or_not_it_is_of_the_day() { + assert!(decode(&reply(&["d1", "d2", "ask", "x1"]).to_string(), &ledger()).is_ok(), "older items are hers to touch"); + let ghost = fault(&reply(&["d1", "d2", "ghost"])); + assert!(ghost.contains("ghost is not a memory of yours in force"), "{ghost}"); + let mut retiring = reply(&["d1", "d2"]); + retiring["retire"] = json!([{"item": "yggdrasil:run:not-a-memory", "outcome": "done", "why": "No."}]); + assert!(fault(&retiring).contains("not a memory of yours")); +} + +#[test] +fn a_consolidation_merges_one_conversation_only() { + let consolidate = |consumes: &[&str]| { + let mut r = reply(&["d1", "d2"].iter().filter(|label| !consumes.contains(label)).copied().collect::>()); + r["consolidate"] = json!([{"consumes": consumes.iter().map(|label| id(label)).collect::>(), "sort": "Learned", "text": "Merged.", "about": [], "salience_permille": 500}]); + r + }; + assert!(fault(&consolidate(&["d1", "dmA"])).contains("one conversation only"), "public with a DM"); + assert!(fault(&consolidate(&["dmA", "dmB", "d1", "d2"])).contains("one conversation only"), "two DMs"); + for same in [&["d1", "d2", "x1"][..], &["dmA"][..]] { + let mut r = consolidate(same); + r["keep"] = json!(["d1", "d2"].iter().filter(|label| !same.contains(label)).map(|label| id(label)).collect::>()); + assert!(decode(&r.to_string(), &ledger()).is_ok(), "{same:?}"); + } +} + +#[test] +fn a_reply_holds_the_bounds_its_schema_states() { + let with = |edit: &dyn Fn(&mut serde_json::Value)| { + let mut r = reply(&["d1", "d2"]); + edit(&mut r); + r + }; + let long = "x".repeat(901); + assert!(fault(&with(&|r| r["night"] = json!(long.clone()))).contains("night")); + assert!(fault(&with(&|r| r["night"] = json!(" "))).contains("night")); + assert!(decode(&with(&|r| r["night"] = json!("x".repeat(900))).to_string(), &ledger()).is_ok()); + let item = |text: &str| json!({"sort": "Idea", "text": text, "about": [], "salience_permille": 100}); + assert!(fault(&with(&|r| r["remember"] = json!((0..17).map(|n| item(&format!("n{n}"))).collect::>()))).contains("at most")); + assert!(decode(&with(&|r| r["remember"] = json!((0..16).map(|n| item(&format!("n{n}"))).collect::>())).to_string(), &ledger()).is_ok()); + assert!(fault(&with(&|r| r["remember"] = json!([item(&"y".repeat(1001))]))).contains("remember.text")); + assert!(fault(&with(&|r| r["remember"] = json!([{"sort": "Idea", "text": "t", "about": [], "salience_permille": 1001}]))).contains("salience")); + assert!(fault(&with(&|r| r["retire"] = json!([{"item": id("x1"), "outcome": "done", "why": ""}]))).contains("why")); + assert!(fault(&with(&|r| r["retire"] = json!([{"item": id("x1"), "outcome": "done", "why": "w".repeat(301)}]))).contains("why")); + let many: Vec = (0..129).map(|n| json!({"item": format!("missing{n}"), "outcome": "done", "why": "w"})).collect(); + assert!(fault(&with(&|r| r["retire"] = json!(many.clone()))).contains("at most")); + let empty = json!({"consumes": [], "sort": "Learned", "text": "t", "about": [], "salience_permille": 1}); + assert!(fault(&with(&|r| r["consolidate"] = json!([empty.clone()]))).contains("consumes")); + let wide = json!({"consumes": (0..17).map(|n| format!("c{n}")).collect::>(), "sort": "Learned", "text": "t", "about": [], "salience_permille": 1}); + assert!(fault(&with(&|r| r["consolidate"] = json!([wide.clone()]))).contains("consumes")); + assert!(fault(&with(&|r| r["consolidate"] = json!(vec![empty.clone(); 33]))).contains("at most")); + assert!(fault(&with(&|r| r["extra"] = json!(1))).contains("not a SleepReply"), "no unknown field"); + assert!(fault(&json!({"night": "n"})).contains("not a SleepReply")); +} + +#[test] +fn an_about_is_a_person_a_shown_memory_names() { + let mut named = reply(&["d1", "d2"]); + named["remember"] = json!([{"sort": "Rapport", "text": "She likes it plain.", "about": ["@kiko:chat", "@op:chat"], "salience_permille": 300}]); + assert!(decode(&named.to_string(), &ledger()).is_ok(), "d1 is about kiko and the ask is about the operator"); + named["remember"][0]["about"] = json!(["@stranger:chat"]); + assert!(fault(&named).contains("remember.about")); + let mut consolidated = reply(&["d2"]); + consolidated["consolidate"] = json!([{"consumes": [id("d1")], "sort": "Learned", "text": "t", "about": ["@stranger:chat"], "salience_permille": 100}]); + assert!(fault(&consolidated).contains("consolidate.about")); + named["remember"][0]["about"] = json!(vec!["@kiko:chat"; 9]); + assert!(fault(&named).contains("remember.about"), "more than eight"); +} + +#[test] +fn faults_are_told_together_so_one_retell_can_mend_them() { + let both = fault(&json!({"night": "", "keep": [id("d1"), "ghost"], "consolidate": [], "retire": [], "remember": []})); + assert!(both.contains("night") && both.contains("ghost") && both.contains(&id("d2")), "{both}"); +} + +fn op(consumes: usize) -> Vec { + (0..=consumes).map(|n| memory_from(&format!("u{n}"), MemorySort::Idea, "x", &[], wake(), 100)).collect() +} + +#[test] +fn operations_pack_into_batches_of_at_most_the_batch_max_and_never_split_one() { + let units: Vec> = vec![op(16), op(16), op(16), op(16), op(1)]; + let sizes: Vec = units.iter().map(Vec::len).collect(); + assert_eq!(sizes, vec![17, 17, 17, 17, 2], "seventy documents"); + let batches = pack(units); + assert_eq!(batches.iter().map(Vec::len).collect::>(), vec![51, 19], "three whole consolidations fit, the fourth starts the next batch"); + assert!(batches.iter().all(|batch| batch.len() <= huginn_mind::BATCH_MAX)); + assert_eq!(pack(vec![op(62), op(0)]).iter().map(Vec::len).collect::>(), vec![63 + 1], "a batch fills to the max exactly"); + assert_eq!(pack(vec![op(62), op(1)]).iter().map(Vec::len).collect::>(), vec![63, 2]); + assert!(pack(Vec::new()).is_empty()); +} + +fn consolidated(context: &Context, consumes: &[&str]) -> PipelineMemory { + let reply = SleepReply { + night: "n".into(), + keep: vec![], + consolidate: vec![Consolidate { consumes: consumes.iter().map(|label| id(label)).collect(), sort: MemorySort::Learned, text: "One.".into(), about: vec![], salience_permille: 400 }], + retire: vec![], + remember: vec![], + }; + let units = operations(&TestMind::seeded(), context, &reply).unwrap(); + assert_eq!(units.len(), 1); + assert_eq!(units[0].len(), 1 + consumes.len(), "the memory and a resolution of each consumed memory"); + let PipelineDocument::Memory(memory) = units[0][0].clone() else { panic!("the memory comes first") }; + for (resolution, consumed) in units[0][1..].iter().zip(consumes) { + let PipelineDocument::Resolution(resolution) = resolution else { panic!("a resolution") }; + assert_eq!(resolution.subject.id.0, id(consumed)); + assert!(matches!(&resolution.outcome, ResolutionOutcome::Superseded { by } if by.len() == 1 && by[0].id.0 == id(&memory.label.0)), "{:?}", resolution.outcome); + } + memory +} + +#[test] +fn a_consolidation_carries_at_most_eight_sources_newest_first_and_eight_voices() { + let mut context = ledger(); + let mut old = held("s0", MemorySort::Learned, &[], dm("!a:chat", &[]), 10); + old.memory.sources = (0..6).map(|n| format!("$old{n}").as_str().into()).collect(); + old.memory.origin = dm("!a:chat", &["@v0:chat", "@v1:chat", "@v2:chat", "@v3:chat", "@v4:chat"]); + let mut new = held("s1", MemorySort::Learned, &[], dm("!a:chat", &[]), 11); + new.memory.sources = ["$new0", "$old5", "$new1", "$new2", "$new3"].map(Into::into).to_vec(); + new.memory.origin = dm("!a:chat", &["@v3:chat", "@v5:chat", "@v6:chat", "@v7:chat", "@v8:chat"]); + for held in [&old, &new] { + context.all.insert(held.id.clone(), held.clone()); + } + let memory = consolidated(&context, &["s0", "s1"]); + let sources: Vec<&str> = memory.sources.iter().map(|source| source.0.as_str()).collect(); + assert_eq!(sources, vec!["$new3", "$new2", "$new1", "$old5", "$new0", "$old4", "$old3", "$old2"], "newest memory's last source first, once each, eight"); + let MemoryOrigin::Direct { room, with } = &memory.origin else { panic!("a DM stays its own") }; + assert_eq!(room.0, "!a:chat"); + assert_eq!(with.iter().map(|who| who.0.as_str()).collect::>(), vec!["@v0:chat", "@v1:chat", "@v2:chat", "@v3:chat", "@v4:chat", "@v5:chat", "@v6:chat", "@v7:chat"], "the union, in the order they spoke, eight"); + assert_eq!(memory.consolidates.iter().map(|consumed| consumed.id.0.clone()).collect::>(), vec![id("s0"), id("s1")]); + assert_eq!(memory.label.0, "mind-night-c0"); + let public = consolidated(&context, &["d1", "d2"]); + assert_eq!(public.origin, MemoryOrigin::Sleep { run: "mind-night".into() }); + assert_eq!((public.sort, public.text.0.as_str(), public.salience_permille), (MemorySort::Learned, "One.", 400)); +} + +#[test] +fn retirements_and_remembered_things_become_resolutions_and_sleep_memories() { + let context = ledger(); + let reply = SleepReply { + night: "n".into(), + keep: vec![], + consolidate: vec![], + retire: [RetireOutcome::Done, RetireOutcome::Declined, RetireOutcome::Faded, RetireOutcome::Wrong] + .into_iter() + .zip(["d1", "d2", "ask", "x1"]) + .map(|(outcome, label)| Retire { item: id(label), outcome, why: format!("why {label}") }) + .collect(), + remember: vec![Remember { sort: MemorySort::Idea, text: "Later.".into(), about: vec![], salience_permille: 200 }], + }; + let units = operations(&TestMind::seeded(), &context, &reply).unwrap(); + assert_eq!(units.len(), 5); + let kinds: Vec<(&str, bool)> = units[..4] + .iter() + .map(|unit| { + let PipelineDocument::Resolution(resolution) = &unit[0] else { panic!("a resolution") }; + assert_eq!(resolution.rationale.0, format!("why {}", resolution.subject.id.0.rsplit(':').next().unwrap()), "her reason"); + match &resolution.outcome { + ResolutionOutcome::Recorded { reason } => (reason.0.as_str().rsplit(' ').next().unwrap(), true), + ResolutionOutcome::Withdrawn { reason } => (reason.0.as_str().rsplit(' ').next().unwrap(), false), + other => panic!("{other:?}"), + } + }) + .collect(); + assert_eq!(kinds, vec![("d1", true), ("d2", false), ("ask", true), ("x1", false)], "done and faded are Recorded, declined and wrong Withdrawn"); + let PipelineDocument::Memory(remembered) = &units[4][0] else { panic!("a memory") }; + assert_eq!((remembered.label.0.as_str(), &remembered.origin), ("mind-night-r0", &MemoryOrigin::Sleep { run: "mind-night".into() })); +} + +/// Her mind with a Sleep run opened at `T0 + 1000` (and `earlier` Sleep runs before it), and `memories` +/// admitted at the given offsets. +fn night_over(earlier: usize, memories: &[(String, MemorySort, MemoryOrigin, i64)]) -> (TestMind, PipelineRef) { + let mind = TestMind::seeded(); + for n in 0..earlier { + mind.admit_at(at(T0 - 100 * 3_600_000 + n as i64)); + mind.committed(vec![run(&format!("before{n}"), RunTurn::Sleep, RunOperator::Mind, &[])]); + mind.committed(vec![crate::testkit::close(run_ref(&format!("before{n}")), crate::testkit::recorded())]); + } + for (label, sort, origin, offset) in memories { + mind.admit_at(at(T0 + offset)); + mind.committed(vec![memory_from(label, *sort, "Text.", &[], origin.clone(), 500)]); + } + mind.admit_at(at(T0 + 1000)); + mind.committed(vec![run("tonight", RunTurn::Sleep, RunOperator::Mind, &[])]); + (mind, run_ref("tonight")) +} + +#[test] +fn the_day_is_capped_oldest_first_the_open_items_too_and_the_rest_wait() { + // No earlier night: her day is the 24 hours before the run opened; the asks are older than that. + let asks = (0..60).map(|n| (format!("ask{n:03}"), MemorySort::Asked, wake(), -(25 * 3_600_000) + n as i64)); + let days = (0..100).map(|n| (format!("day{n:03}"), MemorySort::Learned, wake(), 10 + n as i64)); + let (mind2, run2) = night_over(0, &asks.chain(days).collect::>()); + let night = context(&mind2, &run2, at(T0 + 5000)).unwrap(); + assert_eq!(night.day.len(), DAY_MAX); + assert_eq!((night.day[0].id.as_str(), night.day[95].id.as_str()), (id("day000").as_str(), id("day095").as_str()), "oldest first; day096 and on wait"); + assert_eq!(night.open.len(), OPEN_MAX); + assert_eq!(night.open[0].id, id("ask000")); + assert_eq!(night.counts.get("Learned"), Some(&100)); + assert_eq!(night.counts.get("Asked"), Some(&60)); +} + +#[test] +fn the_tide_moves_thirty_two_memories_a_night_over_the_older_backlog_and_is_whole_when_small() { + let old = |n: usize| (format!("old{n:02}"), MemorySort::Learned, wake(), -(150 * 3_600_000) + n as i64); + let backlog: Vec<_> = (0..40).map(old).collect(); + let (mind, run) = night_over(2, &backlog); + let night = context(&mind, &run, at(T0 + 5000)).unwrap(); + let ids: Vec<&str> = night.tide.iter().map(|held| held.id.as_str()).collect(); + assert_eq!(ids.len(), 32); + let picked: Vec = [0, 7, 8, 31].map(|at| ids[at].to_string()).to_vec(); + assert_eq!(picked, vec![id("old24"), id("old31"), id("old32"), id("old15")], "two earlier nights: offset 64 mod 40 is 24, and it wraps"); + let (mind, run) = night_over(1, &backlog[..10]); + assert_eq!(context(&mind, &run, at(T0 + 5000)).unwrap().tide.len(), 10, "a small backlog is shown whole, once each"); + let (mind, run) = night_over(0, &[]); + assert!(context(&mind, &run, at(T0 + 5000)).unwrap().tide.is_empty()); +} + +#[test] +fn the_day_begins_when_her_previous_night_opened_and_sleep_made_memories_are_not_in_it() { + let hour = 3_600_000; + let memories = vec![ + ("before".to_string(), MemorySort::Learned, wake(), -(120 * hour)), + ("formed".to_string(), MemorySort::Idea, MemoryOrigin::Sleep { run: "x".into() }, -hour), + ("after".to_string(), MemorySort::Learned, wake(), -hour), + ]; + // The earlier night opened 100 hours before tonight's anchor: `before` is older than it, `after` newer. + let (mind, run) = night_over(1, &memories); + let night = context(&mind, &run, at(T0 + 5000)).unwrap(); + assert_eq!(night.day.iter().map(|held| held.id.clone()).collect::>(), vec![id("after")]); + assert!(night.tide.iter().all(|held| held.id != id("formed")), "no Sleep-origin memory comes round"); +} + +#[test] +fn the_night_schema_states_the_reply_the_decoder_takes() { + let schema: serde_json::Value = serde_json::from_str(&sleep_schema()).unwrap(); + assert_eq!(schema["required"], json!(["night", "keep", "consolidate", "retire", "remember"])); + assert_eq!(schema["additionalProperties"], json!(false)); + assert_eq!(schema["properties"]["consolidate"]["maxItems"], json!(32)); + assert_eq!(schema["properties"]["consolidate"]["items"]["properties"]["consumes"]["maxItems"], json!(16)); + assert_eq!(schema["properties"]["retire"]["maxItems"], json!(128)); + assert_eq!(schema["properties"]["remember"]["maxItems"], json!(16)); + assert_eq!(schema["properties"]["retire"]["items"]["properties"]["outcome"]["enum"], json!(["done", "declined", "faded", "wrong"])); +} diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 2b4966c..669cbc0 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -90,12 +90,14 @@ mind_body::heartbeat: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Outco mind_body::heartbeat: const AGENT : & str mind_body::heartbeat: fn poll (organs : & Organs , config : & Config) -> Result < Outcome > mind_body::heartbeat: impl Config -mind_body::heartbeat: impl Config: fn new (personas : Vec < PersonaId > , operators : Vec < String >) -> Result < Self > +mind_body::heartbeat: impl Config: fn new (personas : Vec < PersonaId > , operators : Vec < String > , night : Option < Night >) -> Result < Self > mind_body::heartbeat: impl Control for FileControl mind_body::heartbeat: impl FileStore mind_body::heartbeat: impl FileStore: fn check (& self) -> Result < () > mind_body::heartbeat: impl FileStore: fn new (path : impl Into < PathBuf >) -> Self mind_body::heartbeat: impl HeartbeatView for FileStore +mind_body::heartbeat: impl Night +mind_body::heartbeat: impl Night: fn parse (window : & str , minds : & str) -> Result < Self > mind_body::heartbeat: impl PersonaId mind_body::heartbeat: impl PersonaId: fn parse (spec : & str) -> Result < Self > mind_body::heartbeat: impl Row < '_ > @@ -106,11 +108,12 @@ mind_body::heartbeat: pub trait Control { fn envelope (& self) -> SwarmEffective mind_body::heartbeat: pub trait HeartbeatView { fn latest (& self) -> Result < Option < PollTrace > > ; fn addressing (& self) -> Result < Option < Addressing > > ; } mind_body::heartbeat: pub trait Store { fn load (& self) -> Result < Option < HeartbeatState > > ; fn save (& self , state : & HeartbeatState) -> Result < () > ; fn record (& self , trace : & PollTrace) -> Result < () > ; fn load_addressing (& self) -> Result < Option < Addressing > > ; fn save_addressing (& self , addressing : & Addressing) -> Result < () > ; } mind_body::heartbeat: struct Addressing # [derive (Clone , Debug , Default , PartialEq , Eq , DatabaseEntry)] pub-fields[minds: Vec < Addressee >] -mind_body::heartbeat: struct Config # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[personas: Vec < PersonaId >, operators: Vec < String >] +mind_body::heartbeat: struct Config # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[personas: Vec < PersonaId >, operators: Vec < String >, night: Option < Night >] mind_body::heartbeat: struct FileControl pub-fields[] mind_body::heartbeat: struct FileStore pub-fields[] mind_body::heartbeat: struct HeartbeatState # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[last_beat_at: DateTime < Utc >, handled_mentions: Vec < String >] mind_body::heartbeat: struct InstanceTrace # [derive (Clone , Debug , PartialEq , Eq , serde :: Serialize , serde :: Deserialize)] pub-fields[instance: String, vote_beat_s: Option < u32 >, urgency_permille: u16, addressed: bool, busy: bool, released: bool, idle_s: i64, score: String, note: String] +mind_body::heartbeat: struct Night # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[from_min: u32, to_min: u32, minds: Vec < Slug >] mind_body::heartbeat: struct Organs< 'a > pub-fields[seats: & 'a [Seat < 'a >], control: & 'a dyn Control, launcher: & 'a dyn Launcher, clock: & 'a dyn Clock, host: & 'a str, store: & 'a dyn Store] mind_body::heartbeat: struct PersonaId # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[slug: Slug, public_name: String, handle: String, matrix_id: String, huginn: SocketAddr] mind_body::heartbeat: struct PollTrace # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[at: DateTime < Utc >, tempo_s: u32, instances: Vec < InstanceTrace >, launches: Vec < String >] @@ -286,6 +289,8 @@ mind_body::self_run::inbox: fn derive_counting_held (mind : & dyn MindPort , hol mind_body::self_run::inbox: fn render (briefs : & [Brief] , plan : & str) -> String mind_body::self_run::inbox: fn resolving_sentence (id : & str) -> String mind_body::self_run::inbox: struct Brief # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[id: String, campaign: String, repo: String, item: String, ordinal: u64] +mind_body::sleep: impl< C : ControlSource , S : ControlSource > SessionInbox for SleepInbox < '_ , C , S > +mind_body::sleep: impl< C : ControlSource , S : ControlSource > SleepInbox < '_ , C , S > mind_body::stream: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Mode { Live , DryRun , } mind_body::stream: # [derive (Clone , Debug , PartialEq)] pub enum Outcome { Held (crate :: control :: HeldReason) , NotDue , Blank , Thought { thought : Box < PipelineThought > , admitted : bool } , } mind_body::stream: const AGENT : & str