From 54a89b94bf87939400a59d3db318ac580dfdfbdc Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 7 Nov 2026 00:23:39 +0000 Subject: [PATCH] self_run resume: the kept conversation lives in the unit second state directory, outside her working directory; a resumed start consumes its record before spawning; the warm window counts from the line before a request Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/bin/mind-self-run.rs | 7 +- crates/mind-body/src/self_run.rs | 26 ++++-- crates/mind-body/src/self_run/decide.rs | 19 +++-- crates/mind-body/src/self_run/decide/tests.rs | 31 +++++-- crates/mind-body/src/self_run/tests/resume.rs | 82 +++++++++++++++++-- crates/mind-body/tests/public_surface.txt | 2 +- crates/mind-body/tests/self_run.rs | 8 +- 7 files changed, 141 insertions(+), 34 deletions(-) diff --git a/crates/mind-body/src/bin/mind-self-run.rs b/crates/mind-body/src/bin/mind-self-run.rs index 3830177..ebeec8c 100644 --- a/crates/mind-body/src/bin/mind-self-run.rs +++ b/crates/mind-body/src/bin/mind-self-run.rs @@ -11,7 +11,8 @@ //! `RUNTIME_DIRECTORY` (the unit's `RuntimeDirectory=`) holds each child's fresh //! configuration; `MIND_SELF_CONFIG_SEED` optionally names a root-owned directory copied into it. //! `STATE_DIRECTORY` (the unit's `StateDirectory=`) keeps her last clean -//! conversation in `session/` for the next process; without it every start is fresh. +//! conversation in its second directory, outside her working directory (the first), for the next +//! process; with only one directory every start is fresh. //! The connector's lease socket is `CLAUDE_LEASE_SOCKET`: the token itself is //! never in this process's environment, only the descriptor the lease grants. @@ -42,8 +43,8 @@ fn run() -> Result { .and_then(|seconds| seconds.parse().ok()) .filter(|seconds| *seconds > 0) .ok_or_else(|| anyhow!("{RUN_CAP_VARIABLE} is not set to the unit's RuntimeMaxSec in seconds"))?; - if std::env::var_os(STATE_DIRECTORY_VARIABLE).is_none() { - println!("{STATE_DIRECTORY_VARIABLE} is not set: no conversation is kept and every start is fresh"); + if std::env::var_os(STATE_DIRECTORY_VARIABLE).is_none_or(|directories| !directories.to_string_lossy().contains(':')) { + println!("{STATE_DIRECTORY_VARIABLE} names no second directory: no conversation is kept and every start is fresh"); } let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT)); let outcome = carry(&mind, &SystemClock, &SocketLender::new(socket), &ClaudeSpawner::new(std::env::vars_os())?, run_cap_s, &mut std::io::stdout())?; diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index a89f431..7d15df3 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -389,6 +389,8 @@ pub struct Kept { pub session: Uuid, pub transcript: PathBuf, pub warm_until: DateTime, + /// When the record was written; `warm_until` may not lie more than the longest cache lifetime past it. + pub written_at: DateTime, pub same_seed: bool, } @@ -614,7 +616,14 @@ fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, s }; // The one start decision: a kept conversation is resumed iff `decide::resumes`; any other is cleared before the child starts. let kept = spawner.kept(); - let resume = kept.clone().filter(|kept| decide::resumes(kept, clock.now())); + let mut resume = kept.clone().filter(|kept| decide::resumes(kept, clock.now())); + if let Some(warm) = &resume { + // A resumed start consumes the record before the child exists: the same conversation, kept as already cold (its transcript stays for `--resume`), + // so a process that never reaches the end keep (a signal, a panic) cannot let the next resume from before the run it was killed in. + if spawner.keep(Some((warm.session, clock.now()))).is_err() { + resume = None; + } + } if resume.is_none() { let why = match &kept { None => "none kept", @@ -814,7 +823,7 @@ pub struct ClaudeSpawner { seed: Option, /// Her sandbox: with it the child runs under bubblewrap (`sandbox_args`), the Persona's; the Self passes none. sandbox: Option, - /// Where her one kept conversation lives (`$STATE_DIRECTORY/session`); none keeps nothing, and every start is fresh. + /// Where her one kept conversation lives (the second of the unit's `STATE_DIRECTORY` entries, outside her working directory, the first); none keeps nothing, and every start is fresh. keep: Option, /// The digest of the seed tree this process copies from (empty without a seed); none when the seed cannot be read, which no kept conversation matches. seed_digest: Option, @@ -826,11 +835,11 @@ impl ClaudeSpawner { /// without it every start is fresh, and the journal says so. pub fn new(unit_environment: impl IntoIterator) -> Result { let unit_environment: Vec<(OsString, OsString)> = unit_environment.into_iter().collect(); - // systemd lists several state directories colon-separated; the unit names one. + // systemd lists several state directories colon-separated: the first is her working directory, the second the kept conversation's own, outside it. let keep = unit_environment .iter() .find(|(name, _)| name == STATE_DIRECTORY_VARIABLE) - .and_then(|(_, value)| value.to_string_lossy().split(':').next().filter(|first| !first.is_empty()).map(|first| Path::new(first).join(KEPT_DIR_NAME))); + .and_then(|(_, value)| value.to_string_lossy().split(':').nth(1).filter(|second| !second.is_empty()).map(PathBuf::from)); let mut spawner = Self::in_sandbox(unit_environment, SEED_VARIABLE, None)?; spawner.seed_digest = match &spawner.seed { Some(seed) => seed_digest(seed).ok(), @@ -851,10 +860,8 @@ impl ClaudeSpawner { } } -/// The unit's `StateDirectory=`, which systemd names in the environment: where her kept conversation lives, in [`KEPT_DIR_NAME`]. +/// The unit's `StateDirectory=`, which systemd names in the environment, colon-separated: her working directory first, then the kept conversation's own directory (ruling `eureka-body:ruling:kept-session-ordinary-separation`). pub const STATE_DIRECTORY_VARIABLE: &str = "STATE_DIRECTORY"; -/// The directory of the kept conversation inside the state directory. -const KEPT_DIR_NAME: &str = "session"; /// The record of the kept conversation beside its transcript: `{"session", "warm_until", "seed"}`. const KEPT_RECORD: &str = "kept.json"; @@ -929,7 +936,8 @@ impl Spawner for ClaudeSpawner { let transcript = dir.join(format!("{session}.jsonl")); transcript.is_file().then_some(())?; let same_seed = self.seed_digest.as_deref() == Some(record.get("seed")?.as_str()?); - Some(Kept { session, transcript, warm_until, same_seed }) + let written_at = DateTime::parse_from_rfc3339(record.get("written_at")?.as_str()?).ok()?.with_timezone(&Utc); + Some(Kept { session, transcript, warm_until, written_at, same_seed }) } fn keep(&self, keep: Option<(Uuid, DateTime)>) -> Result<()> { @@ -940,7 +948,7 @@ impl Spawner for ClaudeSpawner { let _ = std::fs::remove_file(dir.join(KEPT_RECORD)); return clear_kept_dir(dir, &[]).context("the kept conversation could not be cleared"); }; - let record = serde_json::json!({"session": session.to_string(), "warm_until": until.to_rfc3339_opts(SecondsFormat::Secs, true), "seed": seed}); + let record = serde_json::json!({"session": session.to_string(), "warm_until": until.to_rfc3339_opts(SecondsFormat::Secs, true), "written_at": Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), "seed": seed}); write_through_tmp(dir, KEPT_RECORD, record.to_string().as_bytes()).context("the kept conversation's record could not be written")?; clear_kept_dir(dir, &[KEPT_RECORD.to_string(), format!("{session}.jsonl")]).context("the conversation before it could not be removed") } diff --git a/crates/mind-body/src/self_run/decide.rs b/crates/mind-body/src/self_run/decide.rs index 27aee5c..11cab6a 100644 --- a/crates/mind-body/src/self_run/decide.rs +++ b/crates/mind-body/src/self_run/decide.rs @@ -114,8 +114,10 @@ pub(super) struct State { /// An ok result has come: the session holds an answer worth its cache. answered_once: bool, last_result_at: Option>, - /// When the last new request of the main thread was read. The cache's lifetime runs from a request's start (research `cache-pricing`), so `warm_until` counts from here and never from the last result. + /// When the last new request of the main thread began, as far as the stream shows: the line before its first assistant line, never that line's own read time, which comes after the request started. The cache's lifetime runs from a request's start (research `cache-pricing`), so `warm_until` counts from here and never from the last result; a slow response only makes the window shorter than the cache's real life. last_request_at: Option>, + /// When the last line of the stream that matters (an echo, a result, or a request's line) was read: the earliest bound the stream gives on when the next request began, since a request starts after the line before it. + stream_at: Option>, /// The last result was not an error. last_ok: bool, /// The CLI refused a result as over its context limit; once seen, it stays seen. @@ -187,10 +189,13 @@ pub(super) fn warm_until(state: &State) -> Option> { } /// Whether a kept conversation is resumed by a process starting at `now`: its -/// cache is still warm and its seed is the one it was made under. Nothing else -/// decides resume or fresh. +/// cache is still warm, its seed is the one it was made under, and its record +/// claims no more warmth than a cache has: a `warm_until` later than the record's +/// write time plus the longest lifetime (one hour) is forged or wrong, and the +/// conversation is cold (ruling `eureka-body:ruling:kept-session-ordinary-separation`). +/// Nothing else decides resume or fresh. pub(super) fn resumes(kept: &Kept, now: DateTime) -> bool { - kept.same_seed && now < kept.warm_until + kept.same_seed && now < kept.warm_until && kept.warm_until <= kept.written_at + Duration::seconds(TTL_1H_S) } /// How the run ended, by what was done to the child: the strongest signal delivered, else the flush, else nothing. @@ -291,9 +296,11 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta let mut action = Action::None; match event { Event::Request { id, prefix } => { + let began = state.stream_at.or(state.cap.map(|(at, _)| at)).unwrap_or(now); + state.stream_at = Some(now); if state.seen.insert(id) { state.requests += 1; - state.last_request_at = Some(now); + state.last_request_at = Some(began); state.prefix = prefix; if let Some(flush) = state.flush.as_mut() && flush.consumed @@ -307,6 +314,7 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta } } Event::Echo => { + state.stream_at = Some(now); state.consumed += 1; state.turn_open = true; let consumed = state.consumed; @@ -319,6 +327,7 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta Event::Result { ok, blocked, ttl_1h } => { state.turn_open = false; state.last_result_at = Some(now); + state.stream_at = Some(now); state.last_ok = ok; state.blocked |= blocked; state.ttl_s = if ttl_1h { TTL_1H_S } else { TTL_5M_S }; diff --git a/crates/mind-body/src/self_run/decide/tests.rs b/crates/mind-body/src/self_run/decide/tests.rs index 32d8fa1..d01daea 100644 --- a/crates/mind-body/src/self_run/decide/tests.rs +++ b/crates/mind-body/src/self_run/decide/tests.rs @@ -531,18 +531,28 @@ fn near_cap() -> State { fn a_clean_end_is_kept_until_its_last_request_plus_its_cache_lifetime_less_the_margin() { // Two requests and a result at three different times: the base is the last request's. let events = |result: Event| vec![(Event::Echo, 0), (req("a", 10), 10), (req("b", 20), 50), (result, 90)]; - assert_eq!(warm_after(working(), events(OK), Some(0)), Some(t(50 + 3600 - 60))); - assert_eq!(warm_after(working(), events(OK_SHORT), Some(0)), Some(t(50 + 300 - 60))); + assert_eq!(warm_after(working(), events(OK), Some(0)), Some(t(10 + 3600 - 60))); + assert_eq!(warm_after(working(), events(OK_SHORT), Some(0)), Some(t(10 + 300 - 60))); // A repeated request id is not a new request and moves nothing. let repeated = vec![(Event::Echo, 0), (req("a", 10), 10), (req("a", 10), 50), (OK, 90)]; - assert_eq!(warm_after(working(), repeated, Some(0)), Some(t(10 + 3600 - 60))); + assert_eq!(warm_after(working(), repeated, Some(0)), Some(t(3600 - 60))); +} + +#[test] +fn a_requests_start_is_bounded_by_the_line_before_it_never_by_its_own_first_line() { + // The first assistant line arrives 500 s after the echo: the request began somewhere in between, and the earliest bound is the echo. + assert_eq!(warm_after(working(), vec![(Event::Echo, 0), (req("a", 10), 500), (OK, 600)], Some(0)), Some(t(3600 - 60))); + // The next request of the turn begins after the previous request's last line, not after its result. + assert_eq!(warm_after(working(), vec![(Event::Echo, 0), (req("a", 10), 100), (req("a", 10), 200), (req("b", 10), 900), (OK, 950)], Some(0)), Some(t(200 + 3600 - 60))); + // After a result, the next turn's echo is the bound. + assert_eq!(warm_after(working(), vec![(Event::Echo, 0), (req("a", 10), 10), (OK, 20), (Event::Echo, 300), (req("b", 10), 900), (OK, 950)], Some(0)), Some(t(300 + 3600 - 60))); } #[test] fn the_rotation_flush_keeps_its_conversation_and_every_other_end_clears_it() { let clean = vec![(Event::Echo, 0), (req("a", 10), 10), (OK, 20)]; let with = |more: Vec<(Event, i64)>| [clean.clone(), more].concat(); - assert_eq!(warm_after(working(), clean.clone(), Some(0)), Some(t(10 + 3600 - 60))); + assert_eq!(warm_after(working(), clean.clone(), Some(0)), Some(t(3600 - 60))); // The rotation: the flush is asked near the cap, answered, and the end is kept from that answer's request. let rotation = vec![(Event::Tick, 100), (Event::Echo, 101), (req("b", 20), 101), (OK, 102)]; assert_eq!(warm_after(near_cap(), with(rotation.clone()), Some(0)), Some(t(101 + 3600 - 60))); @@ -554,7 +564,7 @@ fn the_rotation_flush_keeps_its_conversation_and_every_other_end_clears_it() { let over_after_rotation = with(vec![(Event::Tick, 100), (Event::Echo, 101), (req("b", CEILING_TOKENS), 101), (OK, 102)]); assert_eq!(warm_after(near_cap(), over_after_rotation, Some(0)), None, "the prefix alone, under a rotation flush"); assert_eq!(warm_after(working(), with(vec![(delivered(Stop::Interrupt), 30)]), Some(0)), None, "a delivered signal"); - assert_eq!(warm_after(working(), with(vec![(Event::Signalled { how: Stop::Interrupt, delivered: false }, 30)]), Some(0)), Some(t(10 + 3600 - 60)), "a signal that never went out ended nothing"); + assert_eq!(warm_after(working(), with(vec![(Event::Signalled { how: Stop::Interrupt, delivered: false }, 30)]), Some(0)), Some(t(3600 - 60)), "a signal that never went out ended nothing"); let blocked_but_ok = vec![(Event::Echo, 0), (req("a", 10), 10), (Event::Result { ok: true, blocked: true, ttl_1h: true }, 20), (OK, 30)]; assert_eq!(warm_after(working(), blocked_but_ok, Some(0)), None, "blocked is sticky"); assert_eq!(warm_after(working(), with(vec![(Event::Echo, 30), (FAILED, 40)]), Some(0)), None, "the last result failed"); @@ -566,9 +576,18 @@ fn the_rotation_flush_keeps_its_conversation_and_every_other_end_clears_it() { #[test] fn a_kept_conversation_resumes_until_its_warm_until_and_not_after_and_never_across_a_seed_change() { - let kept = |same_seed| Kept { session: uuid::Uuid::nil(), transcript: "/state/session/x.jsonl".into(), warm_until: t(100), same_seed }; + let kept = |same_seed| Kept { session: uuid::Uuid::nil(), transcript: "/state/session/x.jsonl".into(), warm_until: t(100), written_at: t(0), same_seed }; assert!(resumes(&kept(true), t(99))); assert!(!resumes(&kept(true), t(100))); assert!(!resumes(&kept(true), t(101))); assert!(!resumes(&kept(false), t(0))); } + +#[test] +fn a_record_claiming_more_warmth_than_a_cache_has_is_cold() { + let written = |warm_until, written_at| Kept { session: uuid::Uuid::nil(), transcript: "/state/session/x.jsonl".into(), warm_until, written_at, same_seed: true }; + // One hour from the write is the longest a cache lives; a record may claim up to it and no more. + assert!(resumes(&written(t(1000 + 3600), t(1000)), t(1100))); + assert!(!resumes(&written(t(1000 + 3600 + 1), t(1000)), t(1100)), "a forged far-future warm_until"); + assert!(!resumes(&written(t(1_000_000), t(1000)), t(1100))); +} diff --git a/crates/mind-body/src/self_run/tests/resume.rs b/crates/mind-body/src/self_run/tests/resume.rs index db95fd6..55e2e53 100644 --- a/crates/mind-body/src/self_run/tests/resume.rs +++ b/crates/mind-body/src/self_run/tests/resume.rs @@ -16,7 +16,7 @@ fn previous() -> Uuid { } fn kept_for(warm_for_s: i64, same_seed: bool) -> Kept { - Kept { session: previous(), transcript: PathBuf::from(format!("/state/session/{PREVIOUS}.jsonl")), warm_until: at_start() + ChronoDuration::seconds(warm_for_s), same_seed } + Kept { session: previous(), transcript: PathBuf::from(format!("/state/session/{PREVIOUS}.jsonl")), warm_until: at_start() + ChronoDuration::seconds(warm_for_s), written_at: at_start(), same_seed } } fn session_a() -> Uuid { @@ -145,7 +145,7 @@ fn a_drive_that_fails_clears_the_conversation() { let ran = carry(&Seen::of(&mind), &*clock, &FakeLender::logging(log.clone(), Answer::Grant), &fake, TEST_CAP_S, &mut Refuses); assert!(ran.is_err()); let keeps: Vec<_> = log.borrow().iter().filter(|event| matches!(event, Event::Kept(_))).cloned().collect(); - assert_eq!(keeps, vec![Event::Kept(None)], "a resumed start keeps nothing, and the failed drive clears"); + assert_eq!(keeps, vec![Event::Kept(Some((previous(), at_start()))), Event::Kept(None)], "a resumed start consumes the record, and the failed drive clears"); } // ---- the files ----------------------------------------------------------------- @@ -192,7 +192,9 @@ fn the_configs_end_moves_the_main_transcripts_out_and_nothing_else_of_it() { } fn spawner_over(state: &Path, scratch: &Path, seed: Option<&Path>) -> ClaudeSpawner { - let mut environment = vec![(SCRATCH_VARIABLE.into(), scratch.into()), (STATE_DIRECTORY_VARIABLE.into(), state.into())]; + // The unit's two state directories: her working directory, and the kept conversation's own beside it. + let directories = format!("{}:{}", state.join("work").display(), state.join("session").display()); + let mut environment = vec![(SCRATCH_VARIABLE.into(), scratch.into()), (STATE_DIRECTORY_VARIABLE.into(), directories.into())]; if let Some(seed) = seed { environment.push((SEED_VARIABLE.into(), seed.into())); } @@ -217,7 +219,9 @@ fn keeping_one_conversation_removes_the_one_before_it_and_clearing_leaves_the_di spawner.keep(Some((newer, until))).unwrap(); assert_eq!(names(&state.path().join("session")), [format!("{newer}.jsonl"), "kept.json".to_string()], "at most one transcript, and no stray"); - assert_eq!(spawner.kept(), Some(Kept { session: newer, transcript: newer_path, warm_until: until, same_seed: true })); + let kept = spawner.kept().unwrap(); + assert_eq!((kept.session, &kept.transcript, kept.warm_until, kept.same_seed), (newer, &newer_path, until, true)); + assert!((chrono::Utc::now() - kept.written_at).num_seconds().abs() < 60, "the record says when it was written"); spawner.keep(None).unwrap(); assert!(names(&state.path().join("session")).is_empty()); @@ -238,11 +242,13 @@ fn a_record_that_is_missing_unreadable_or_names_a_missing_transcript_keeps_nothi assert_eq!(spawner.kept(), None, "a transcript with no record"); write(&dir.join("kept.json"), "not json"); assert_eq!(spawner.kept(), None, "unparseable"); - write(&dir.join("kept.json"), &format!(r#"{{"session":"{PREVIOUS}","warm_until":"yesterday","seed":""}}"#)); + write(&dir.join("kept.json"), &format!(r#"{{"session":"{PREVIOUS}","warm_until":"yesterday","written_at":"2026-10-09T13:00:00Z","seed":""}}"#)); assert_eq!(spawner.kept(), None, "an unreadable instant"); - write(&dir.join("kept.json"), &format!(r#"{{"session":"{}","warm_until":"2026-10-09T13:45:00Z","seed":""}}"#, session_a())); + write(&dir.join("kept.json"), &format!(r#"{{"session":"{}","warm_until":"2026-10-09T13:45:00Z","written_at":"2026-10-09T13:00:00Z","seed":""}}"#, session_a())); assert_eq!(spawner.kept(), None, "names a transcript that is not there"); write(&dir.join("kept.json"), &format!(r#"{{"session":"{PREVIOUS}","warm_until":"2026-10-09T13:45:00Z","seed":""}}"#)); + assert_eq!(spawner.kept(), None, "a record that does not say when it was written"); + write(&dir.join("kept.json"), &format!(r#"{{"session":"{PREVIOUS}","warm_until":"2026-10-09T13:45:00Z","written_at":"2026-10-09T13:00:00Z","seed":""}}"#)); assert!(spawner.kept().is_some_and(|kept| kept.same_seed)); } @@ -299,3 +305,67 @@ fn clearing_a_kept_directory_that_does_not_exist_is_nothing_and_one_that_cannot_ write(&state.path().join("session"), "a file where the directory belongs"); assert!(spawner.keep(None).is_err(), "a conversation that cannot be cleared is told, not swallowed"); } + +/// Her files for `kept`/`keep`, a child that never starts: the process ends where a signal or a panic would end it, before the end keep. +struct FilesThenDies { + files: ClaudeSpawner, + dies: Fake, +} + +impl Spawner for FilesThenDies { + fn kept(&self) -> Option { + self.files.kept() + } + + fn keep(&self, keep: Option<(Uuid, chrono::DateTime)>) -> Result<()> { + self.files.keep(keep) + } + + fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { + self.dies.spawn(args, token) + } +} + +fn start_over_files(state: &Path, seen: &Seen) -> (String, ClaudeSpawner) { + let log = Rc::new(RefCell::new(Vec::new())); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let spawner = FilesThenDies { files: spawner_over(state, state, None), dies: Fake::with(None, true) }; + let mut out = Vec::new(); + let _ = carry(seen, &*clock, &FakeLender::logging(log, Answer::Grant), &spawner, TEST_CAP_S, &mut out); + (String::from_utf8_lossy(&out).into_owned(), spawner.files) +} + +#[test] +fn a_process_killed_after_resuming_leaves_a_record_the_next_start_does_not_resume_from() { + let state = tempfile::tempdir().unwrap(); + let files = spawner_over(state.path(), state.path(), None); + let path = transcript(state.path(), previous()); + files.keep(Some((previous(), at_start() + ChronoDuration::seconds(1800)))).unwrap(); + assert!(files.kept().is_some_and(|kept| decide::resumes(&kept, at_start())), "kept, warm"); + + // The next process resumes it; its child is gone before any end keep could run. + let mind = with_run(); + let (out, files) = start_over_files(state.path(), &Seen::of(&mind)); + assert!(out.contains(&format!("session resumed {PREVIOUS}")), "{out}"); + let after = files.kept().expect("the transcript stays for --resume"); + assert_eq!(after.transcript, path); + assert!(path.is_file()); + assert!(!decide::resumes(&after, at_start() + ChronoDuration::seconds(1)), "the record no longer resumes"); + assert!(!decide::resumes(&after, at_start()), "not even at the instant it was consumed"); +} + +#[test] +fn a_forged_far_future_record_is_cold_and_cleared() { + let state = tempfile::tempdir().unwrap(); + let files = spawner_over(state.path(), state.path(), None); + transcript(state.path(), previous()); + write( + &state.path().join("session/kept.json"), + &format!(r#"{{"session":"{PREVIOUS}","warm_until":"2030-01-01T00:00:00Z","written_at":"2026-10-09T13:00:00Z","seed":""}}"#), + ); + assert!(files.kept().is_some_and(|kept| kept.same_seed), "it parses, and it is the clock that refuses it"); + let (out, files) = start_over_files(state.path(), &Seen::of(&with_run())); + assert!(out.contains("session fresh: cold"), "{out}"); + assert_eq!(files.kept(), None); + assert!(names(&state.path().join("session")).is_empty()); +} diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index cc76419..de65edd 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -268,7 +268,7 @@ mind_body::self_run: mod inbox mind_body::self_run: pub trait Child { fn write_line (& mut self , line : & str) ; fn close_input (& mut self) ; fn next (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; } mind_body::self_run: pub trait Spawner { fn spawn (& self , args : & [String] , token : & TokenFd) -> Result < Box < dyn Child > > ; fn kept (& self) -> Option < Kept > { None } fn keep (& self , _keep : Option < (Uuid , DateTime < Utc >) >) -> Result < () > { Ok (()) } } mind_body::self_run: struct ClaudeSpawner pub-fields[] -mind_body::self_run: struct Kept # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[session: Uuid, transcript: PathBuf, warm_until: DateTime < Utc >, same_seed: bool] +mind_body::self_run: struct Kept # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[session: Uuid, transcript: PathBuf, warm_until: DateTime < Utc >, written_at: DateTime < Utc >, same_seed: bool] mind_body::self_run::decide: impl State mind_body::self_run::inbox: const OWNER : & str mind_body::self_run::inbox: fn backlog (mind : & dyn MindPort) -> Result < Vec < (Brief , String) > > diff --git a/crates/mind-body/tests/self_run.rs b/crates/mind-body/tests/self_run.rs index e6fb358..274b89f 100644 --- a/crates/mind-body/tests/self_run.rs +++ b/crates/mind-body/tests/self_run.rs @@ -226,7 +226,7 @@ mod lane { /// The binary with `bin` first on `PATH`, until the daemon answers (exit 2 before that /// has started nothing). `SELF_RUN_ENV_CANARY` is a variable of the unit's environment. fn ask(endpoint: &str, bin: &Path, socket: &Path) -> Output { - ask_in(endpoint, bin, socket, Some(&bin.join("state"))) + ask_in(endpoint, bin, socket, Some(Path::new(&format!("{0}/state:{0}/kept", bin.display())))) } /// The same, in a unit whose `StateDirectory=` is `state_directory`. @@ -419,9 +419,9 @@ echo '{request}'\necho '{echo}'\necho '{result}'\nexit 0", fake_claude(bin.path(), &body); let connector = fake_connector(); - let first = ask_in(&endpoint, bin.path(), &connector.path, Some(units_state.path())); + let first = ask_in(&endpoint, bin.path(), &connector.path, Some(Path::new(&format!("{0}/work:{0}/kept", units_state.path().display())))); assert_eq!(first.status.code(), Some(0), "{}", all_output(&first)); - let kept = units_state.path().join("session"); + let kept = units_state.path().join("kept"); let mut names: Vec = std::fs::read_dir(&kept).unwrap().map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()).collect(); names.sort(); assert_eq!(names, [format!("{LANE_SESSION}.jsonl"), "kept.json".to_string()], "the first process kept its conversation"); @@ -433,7 +433,7 @@ echo '{request}'\necho '{echo}'\necho '{result}'\nexit 0", run.label = eureka_pipeline::Label("selfrun-lane-two".into()); admit(state.path(), vec![second_run]); let (daemon, endpoint) = daemon_over(state.path()); - let second = ask_in(&endpoint, bin.path(), &connector.path, Some(units_state.path())); + let second = ask_in(&endpoint, bin.path(), &connector.path, Some(Path::new(&format!("{0}/work:{0}/kept", units_state.path().display())))); drop(daemon); assert_eq!(second.status.code(), Some(0), "{}", all_output(&second));