stream: soul pass s2 fixes (thinking window, capped per repo, no proxy env, identical lines, forge read at gap marks)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/mind-body/src/bin/mind-stream.rs

@@ -1,6 +1,6 @@
 //! `mind-stream --instance <slug> [--once] [--dry-run]`: her stream of thought.
 //!
-//! Without `--once` it runs forever: one `mind_body::stream::tick` per `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
 //! 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
@@ -20,10 +20,9 @@ use eureka_pipeline::Slug;
 use eureka_state::HuginnClient;
 use mind_body::control::read_effective;
 use mind_body::queue::HuginnMind;
-use mind_body::stream::{HuginnPersonaSource, Mode, OllamaStream, Organs, Outcome, STREAM_MIN_GAP_S, 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 POLL_S: u64 = 60;
 
 #[allow(clippy::disallowed_methods, reason = "the binary reads its own environment, and hands it to the library")]
 fn setting(name: &str) -> Option<String> {
@@ -95,7 +94,7 @@ fn run() -> Result<()> {
         if once {
             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 { POLL_S }));
+        std::thread::sleep(Duration::from_secs(if waits { STREAM_MIN_GAP_S as u64 } else { STREAM_POLL_S as u64 }));
     }
 }
 

crates/mind-body/src/stream.rs

@@ -36,6 +36,8 @@ pub const AGENT: &str = "mind-stream";
 
 /// No tick sooner than this after her newest thought, whatever has happened.
 pub const STREAM_MIN_GAP_S: i64 = 900;
+/// How often the stream process polls: the unit in which a poll "crosses" a `STREAM_MIN_GAP_S` mark.
+pub const STREAM_POLL_S: i64 = 60;
 /// A tick is due after this long even when nothing has happened.
 pub const STREAM_IDLE_GAP_S: i64 = 3600;
 /// The most bytes of context and of stimulus the Projector is shown.
@@ -128,6 +130,9 @@ impl OllamaStream {
             .timeout_global(Some(Duration::from_secs(900)))
             .timeout_connect(Some(Duration::from_secs(3)))
             .http_status_as_error(false)
+            // The host checked above is the host connected to: no proxy from the environment, no redirect to elsewhere.
+            .proxy(None)
+            .max_redirects(0)
             .build()
             .into();
         Ok(Self { base: format!("http://{authority}"), projector_model: projector_model.to_owned(), base_model: base_model.to_owned(), agent })
@@ -331,7 +336,7 @@ fn replies(mind: &dyn MindPort, instance: &Slug) -> Result<Vec<(PipelineRef, Str
 }
 
 /// What her mind admitted since `since`, newest first: one line and one citation each.
-fn news(mind: &dyn MindPort, instance: &Slug, since: Option<DateTime<Utc>>, responsibilities: &Responsibilities, now: DateTime<Utc>) -> Result<Vec<(String, Option<PipelineRef>)>> {
+fn mind_news(mind: &dyn MindPort, instance: &Slug, since: Option<DateTime<Utc>>) -> Result<Vec<(String, Option<PipelineRef>)>> {
     let after = |view: &PipelineDocumentView| since.is_none_or(|since| admitted_at(view).is_some_and(|at| at > since));
     let mut found: Vec<(u64, String, PipelineRef)> = Vec::new();
     for view in documents(mind, in_force(mine(PipelineKind::Memory, instance)), RECENT_READ)?.into_iter().filter(&after) {
@@ -353,24 +358,26 @@ fn news(mind: &dyn MindPort, instance: &Slug, since: Option<DateTime<Utc>>, resp
         }
     }
     found.sort_by_key(|(ordinal, ..)| std::cmp::Reverse(*ordinal));
-    let mut out: Vec<(String, Option<PipelineRef>)> = found.into_iter().map(|(_, line, id)| (line, Some(id))).collect();
+    Ok(found.into_iter().map(|(_, line, id)| (line, Some(id))).collect())
-    // Her stewarded repos' activity by others is the forge's word, not a document; a daemon with no forge adds nothing.
+}
-    // Only what happened after her newest thought is news, and the notice is stamped with when it happened.
+
-    if !responsibilities.repos.is_empty()
+/// Her stewarded repos' activity by others is the forge's word, not a document; a daemon with no forge adds nothing.
-        && let Ok(mut activity) = mind.stewarded_activity(&responsibilities.repos, None)
+/// Only what happened after `since` is news, and the notice is stamped with when it happened.
-    {
+fn repo_news(mind: &dyn MindPort, since: Option<DateTime<Utc>>, responsibilities: &Responsibilities, now: DateTime<Utc>) -> Option<(String, Option<PipelineRef>)> {
-        let since_ms = since.map_or(0, |since| u64::try_from(since.timestamp_millis()).unwrap_or(0));
+    if responsibilities.repos.is_empty() {
-        activity.events.retain(|event| since.is_none() || event.at > since_ms);
+        return None;
-        // A capped repo has no time of its own: it is news only beside an event that is.
-        if activity.events.is_empty() && since.is_some() {
-            activity.capped.clear();
-        }
-        let happened = activity.events.iter().map(|event| event.at).max().unwrap_or_else(|| u64::try_from(now.timestamp_millis()).unwrap_or(0));
-        if let Some(line) = repo_notice_line(&activity, happened) {
-            out.push((format!("- {line}"), None));
-        }
     }
-    Ok(out)
+    let mut activity = mind.stewarded_activity(&responsibilities.repos, None).ok()?;
+    if let Some(since) = since {
+        let since_ms = u64::try_from(since.timestamp_millis()).unwrap_or(0);
+        // A capped repo's unread entries are older than the oldest it showed. Once the read shows an event of the repo at or before
+        // `since` they are older than `since` too; until then the cap is disclosed, whatever other repos showed.
+        let decided: Vec<String> = activity.events.iter().filter(|event| event.at <= since_ms).map(|event| event.repo.identity()).collect();
+        activity.capped.retain(|repo| !decided.contains(&repo.identity()));
+        activity.events.retain(|event| event.at > since_ms);
+    }
+    let happened = activity.events.iter().map(|event| event.at).max().unwrap_or_else(|| u64::try_from(now.timestamp_millis()).unwrap_or(0));
+    repo_notice_line(&activity, happened).map(|line| (format!("- {line}"), None))
 }
 
 /// One in-force memory of hers older than `UNBIDDEN_AFTER_DAYS`, picked uniformly by `pick`.
@@ -422,8 +429,13 @@ fn build_seed(mind: &dyn MindPort, persona: &PersonaRead, since: Option<DateTime
     ];
     let kept: Vec<String> = sections.iter().zip(BUDGETS).map(|(text, (_, budget))| cap(text, budget)).collect();
     // The reads are what the Projector was shown after the bounds: the recall lines and replies the capped sections kept.
-    for entry in recalled.entries().filter(|entry| kept[5].contains(entry.line.trim_end())) {
+    // Each shown line is one citation: a memory reading like another takes its own line, not a share of the first's.
-        cite(PipelineRef { kind: PipelineKind::Memory, id: Short(entry.id.clone()) });
+    let mut shown: Vec<&str> = kept[5].lines().collect();
+    for entry in recalled.entries() {
+        if let Some(at) = shown.iter().position(|line| *line == entry.line.trim_end()) {
+            shown.remove(at);
+            cite(PipelineRef { kind: PipelineKind::Memory, id: Short(entry.id.clone()) });
+        }
     }
     let replies_shown = kept[6].lines().count().saturating_sub(1);
     heard.iter().take(replies_shown).for_each(|(id, _)| cite(id.clone()));
@@ -528,10 +540,19 @@ fn due(since: Option<DateTime<Utc>>, now: DateTime<Utc>, news: bool) -> bool {
     }
 }
 
+/// When her newest thought was ticked: its label's instant, never later than its admission. The models take minutes between the
+/// two, and what the mind or the forge admitted meanwhile is news to the next thought, which the admission time would hide.
 fn newest_thought(mind: &dyn MindPort) -> Result<Option<DateTime<Utc>>> {
     let newest = cultnet_rs::Selection { descending: true, limit: Some(1), ..mine(PipelineKind::Thought, mind.instance()) };
     let PipelinePageItems::Headers(found) = mind.query(&newest)?.items else { bail!("the mind answered documents to a header selection") };
-    Ok(found.iter().filter_map(|header| DateTime::parse_from_rfc3339(&header.admission.admitted_at).ok()).map(|at| at.with_timezone(&Utc)).max())
+    Ok(found
+        .iter()
+        .filter_map(|header| {
+            let admitted = DateTime::parse_from_rfc3339(&header.admission.admitted_at).ok()?.with_timezone(&Utc);
+            let label = header.id.id.0.rsplit(':').next().and_then(|label| label.strip_prefix('t')).and_then(|label| label.parse::<i64>().ok()).and_then(|label| DateTime::from_timestamp(label, 0));
+            Some(label.map_or(admitted, |at| at.min(admitted)))
+        })
+        .max())
 }
 
 fn bounded(text: &str, max: usize) -> String {
@@ -550,8 +571,14 @@ pub fn tick(organs: &Organs, brake: &dyn Fn() -> Effective, mode: Mode, now: Dat
     }
     // Due-ness is decided on the cheap reads; the seed (recall, runs, replies) is built only for a tick that may run.
     let responsibilities = render_responsibilities(mind, mind.instance(), &BTreeMap::new())?;
-    let fresh = news(mind, mind.instance(), since, &responsibilities, now)?;
+    let mut fresh = mind_news(mind, mind.instance(), since)?;
-    if mode == Mode::Live && !due(since, now, !fresh.is_empty()) {
+    // The forge is read when a tick will run anyway, and on a quiet stretch only by the poll that crosses each minimum-gap mark.
+    let quiet = mode == Mode::Live && !due(since, now, !fresh.is_empty());
+    if quiet && since.is_some_and(|at| (now - at).num_seconds() % STREAM_MIN_GAP_S >= STREAM_POLL_S) {
+        return Ok(Outcome::NotDue);
+    }
+    fresh.extend(repo_news(mind, since, &responsibilities, now));
+    if quiet && fresh.is_empty() {
         return Ok(Outcome::NotDue);
     }
     let persona = organs.persona.persona()?;

crates/mind-body/src/stream/tests.rs

@@ -19,6 +19,11 @@ fn t(secs: i64) -> DateTime<Utc> {
     at(secs * 1000)
 }
 
+/// The label of a thought ticked at test-clock `secs`: the Unix seconds of that instant.
+fn ticked(secs: i64) -> String {
+    format!("t{}", t(secs).timestamp())
+}
+
 fn released() -> Effective {
     Effective::Released(Grant::for_test(Slug(INSTANCE.into())))
 }
@@ -86,11 +91,12 @@ struct Wrap<'a> {
     selections: RefCell<Vec<Selection>>,
     admitted: RefCell<Vec<Vec<PipelineDocument>>>,
     forge: Option<RepoActivity>,
+    forge_reads: std::cell::Cell<usize>,
 }
 
 impl<'a> Wrap<'a> {
     fn over(inner: &'a TestMind, scores: &[f32]) -> Self {
-        Self { inner, scores: RefCell::new(scores.iter().copied().collect()), selections: RefCell::default(), admitted: RefCell::default(), forge: None }
+        Self { inner, scores: RefCell::new(scores.iter().copied().collect()), selections: RefCell::default(), admitted: RefCell::default(), forge: None, forge_reads: std::cell::Cell::new(0) }
     }
 
     /// The same mind with a forge that answers `events`, whatever the asker's cursor.
@@ -122,6 +128,7 @@ impl MindPort for Wrap<'_> {
     }
 
     fn stewarded_activity(&self, repos: &[huginn_mind::eureka_pipeline::OrgRepo], after: Option<u64>) -> Result<RepoActivity> {
+        self.forge_reads.set(self.forge_reads.get() + 1);
         match &self.forge {
             Some(activity) => Ok(activity.clone()),
             None => self.inner.stewarded_activity(repos, after),
@@ -137,7 +144,8 @@ impl MindPort for Wrap<'_> {
 /// The seed as the tick builds it: the cheap reads first, then the rest.
 fn seed_of(mind: &dyn MindPort, read: &PersonaRead, since: Option<DateTime<Utc>>, now: DateTime<Utc>) -> Result<Seed> {
     let responsibilities = render_responsibilities(mind, mind.instance(), &BTreeMap::new())?;
-    let fresh = news(mind, mind.instance(), since, &responsibilities, now)?;
+    let mut fresh = mind_news(mind, mind.instance(), since)?;
+    fresh.extend(repo_news(mind, since, &responsibilities, now));
     build_seed(mind, read, since, now, &responsibilities, fresh)
 }
 
@@ -187,7 +195,7 @@ fn mind_with_thought(thought_at: i64) -> TestMind {
     mind.admit_at(t(10));
     mind.committed(vec![memory("m1", MemorySort::Learned, "Libby wants the sounds before the shaders", &["@libby:chat"])]);
     mind.admit_at(t(thought_at));
-    mind.committed(vec![thought_doc("t1", "An earlier drift")]);
+    mind.committed(vec![thought_doc(&ticked(thought_at), "An earlier drift")]);
     mind
 }
 
@@ -271,7 +279,7 @@ fn stimulus_is_what_is_new() {
     mind.admit_at(t(100));
     mind.committed(vec![memory("m1", MemorySort::Learned, "Before the thought, the kettle", &[])]);
     mind.admit_at(t(200));
-    mind.committed(vec![thought_doc("t1", "A drift")]);
+    mind.committed(vec![thought_doc(&ticked(200), "A drift")]);
     mind.admit_at(t(300));
     mind.committed(vec![memory("m2", MemorySort::Decided, "After the thought, the window", &[]), run("w9", RunTurn::SelfRun, RunOperator::Mind, &[])]);
     mind.admit_at(t(350));
@@ -639,9 +647,9 @@ fn a_flood_of_news_fills_the_stimulus_with_whole_lines_newest_first() {
 fn the_newest_thought_sets_the_cadence_not_the_oldest() {
     let mind = TestMind::seeded();
     mind.admit_at(t(100));
-    mind.committed(vec![thought_doc("t1", "An old drift")]);
+    mind.committed(vec![thought_doc(&ticked(100), "An old drift")]);
     mind.admit_at(t(10_000));
-    mind.committed(vec![thought_doc("t2", "A recent drift")]);
+    mind.committed(vec![thought_doc(&ticked(10_000), "A recent drift")]);
     let model = Fake::saying("I turn it over.", &["A new drift"]);
     let persona = persona();
     let wrap = Wrap::over(&mind, &[0.1]);
@@ -710,11 +718,11 @@ fn new_repo_activity_makes_a_tick_due_and_the_hourly_memory_comes_for_an_active_
     mind.admit_at(t(100));
     mind.committed(vec![memory("m1", MemorySort::Learned, "An old thing she holds", &[])]);
     mind.admit_at(t(week));
-    mind.committed(vec![thought_doc("t1", "An earlier drift")]);
+    mind.committed(vec![thought_doc(&ticked(week), "An earlier drift")]);
     let model = Fake::saying("I turn it over.", &["A new drift", "Another"]);
     let persona = persona();
     let fresh = Wrap::over(&mind, &[0.1, 0.2]).with_forge(vec![push(2, week + 1000)]);
-    let Outcome::Thought { thought, admitted: true } = tick(&organs(&fresh, &persona, &model), &|| released(), Mode::Live, t(week + 1000 + STREAM_MIN_GAP_S)).unwrap() else { panic!("new activity past the minimum gap") };
+    let Outcome::Thought { thought, admitted: true } = tick(&organs(&fresh, &persona, &model), &|| released(), Mode::Live, t(week + 2 * STREAM_MIN_GAP_S)).unwrap() else { panic!("new activity at the next gap mark") };
     assert!(thought.stimulus.0.contains("Changes in your repos"), "{}", thought.stimulus.0);
 
     let model = Fake::saying("I turn it over.", &["Yet another"]);
@@ -772,3 +780,84 @@ fn every_thought_records_the_fixed_parts_of_its_prompts() {
     assert_eq!(model.projected.borrow()[0], rebuilt, "the trace rebuilds the Projector's exact prompt");
 }
 
+
+/// A thought ticked at 900 s and admitted at 1000 s: what landed at 950 s while the models ran is news to the next one.
+#[test]
+fn what_lands_while_she_thinks_is_news_to_the_next_thought() {
+    let mind = TestMind::seeded();
+    mind.admit_at(t(950));
+    mind.committed(vec![memory("m9", MemorySort::Decided, "Landed while the Projector ran", &[])]);
+    mind.admit_at(t(1000));
+    mind.committed(vec![thought_doc(&ticked(900), "A drift that took a hundred seconds")]);
+    let model = Fake::saying("I turn it over.", &["A new drift"]);
+    let persona = persona();
+    let wrap = Wrap::over(&mind, &[0.1]).with_forge(vec![push(2, 1000), push(1, 950), push(0, 899)]);
+    let Outcome::Thought { thought, admitted: true } = tick(&organs(&wrap, &persona, &model), &|| released(), Mode::Live, t(1901)).unwrap() else { panic!("news past the gap") };
+    assert!(thought.stimulus.0.contains("Landed while the Projector ran"), "{}", thought.stimulus.0);
+    assert!(thought.stimulus.0.contains("2 commits pushed by libby"), "the pushes at 950 s and 1000 s, not the one at 899 s: {}", thought.stimulus.0);
+}
+
+#[test]
+fn a_capped_repo_is_judged_by_its_own_events_and_disclosed_when_they_do_not_decide_it() {
+    let mind = mind_with_thought(1000);
+    let read = PersonaRead { state: state(), receipt_id: "r".into() };
+    let cultlib = |id, secs| RepoEvent { repo: "GameCult/CultLib".into(), ..push(id, secs) };
+    let mut two = Wrap::over(&mind, &[]).with_forge(vec![push(2, 1500), cultlib(1, 500)]);
+    two.forge.as_mut().unwrap().capped = vec!["GameCult/Huginn".into(), "GameCult/CultLib".into()];
+    let seed = seed_of(&two, &read, Some(t(1000)), t(1901)).unwrap();
+    assert!(seed.stimulus.to_lowercase().contains("page limit") && seed.stimulus.contains("GameCult/Huginn"), "the repo no event decides stays disclosed: {}", seed.stimulus);
+    assert!(!seed.stimulus.contains("GameCult/CultLib"), "an event before her thought decides CultLib's cap: {}", seed.stimulus);
+    let mut flooded = Wrap::over(&mind, &[]).with_forge(vec![]);
+    flooded.forge.as_mut().unwrap().capped = vec!["GameCult/Huginn".into()];
+    let seed = seed_of(&flooded, &read, Some(t(1000)), t(1901)).unwrap();
+    assert!(seed.news && seed.stimulus.to_lowercase().contains("page limit"), "a cap with no event at all is disclosed: {}", seed.stimulus);
+}
+
+#[test]
+fn a_quiet_hour_reads_the_forge_at_each_gap_mark_only() {
+    let mind = mind_with_thought(1000);
+    let model = Fake::saying("I turn it over.", &["A new drift"]);
+    let persona = persona();
+    let wrap = Wrap::over(&mind, &[0.1]).with_forge(vec![push(1, 500)]);
+    let organs = organs(&wrap, &persona, &model);
+    let mut secs = 1000 + STREAM_POLL_S;
+    while secs < 1000 + STREAM_IDLE_GAP_S {
+        assert_eq!(tick(&organs, &|| released(), Mode::Live, t(secs)).unwrap(), Outcome::NotDue, "at {secs}");
+        secs += STREAM_POLL_S;
+    }
+    assert_eq!(wrap.forge_reads.get(), 3, "one read for each mark crossed at 900, 1800 and 2700 s");
+    assert_eq!(model.calls(), 0);
+}
+
+#[test]
+fn identical_memory_lines_are_each_cited_once() {
+    let mind = TestMind::seeded();
+    mind.admit_at(t(100));
+    mind.committed((0..12).map(|at| memory(&format!("m{at:02}"), MemorySort::Learned, &format!("the same thing {}", "x".repeat(290)), &["@libby:chat"])).collect());
+    let read = PersonaRead { state: state(), receipt_id: "r".into() };
+    let seed = seed_of(&Wrap::over(&mind, &[]), &read, Some(t(300)), t(400)).unwrap();
+    let shown = seed.context.lines().filter(|line| line.contains("the same thing")).count();
+    let cited = seed.reads.iter().filter(|read| read.kind == PipelineKind::Memory).count();
+    assert!(shown >= 1 && cited == shown, "{shown} lines shown, {cited} cited");
+}
+
+/// The environment's proxy settings are read when an agent is built, so the check runs in a child process that has them.
+#[test]
+fn the_stream_agent_ignores_the_proxy_environment() {
+    if std::env::var("STREAM_PROXY_CHILD").is_ok() {
+        let plain: ureq::Agent = ureq::Agent::config_builder().build().into();
+        assert!(plain.config().proxy().is_some(), "the child's environment names a proxy");
+        let stream = OllamaStream::open("http://127.0.0.1:9", "p", "b").unwrap();
+        assert!(stream.agent.config().proxy().is_none(), "the stream's agent took the environment's proxy");
+        assert_eq!(stream.agent.config().max_redirects(), 0);
+        return;
+    }
+    let child = std::process::Command::new(std::env::current_exe().unwrap())
+        .args(["--exact", "stream::tests::the_stream_agent_ignores_the_proxy_environment", "--nocapture"])
+        .env("STREAM_PROXY_CHILD", "1")
+        .env("ALL_PROXY", "http://192.0.2.1:3128")
+        .env("HTTP_PROXY", "http://192.0.2.1:3128")
+        .output()
+        .unwrap();
+    assert!(child.status.success(), "{}{}", String::from_utf8_lossy(&child.stdout), String::from_utf8_lossy(&child.stderr));
+}