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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
725facb183
commit
54a89b94bf
7 changed files with 141 additions and 34 deletions
|
|
@ -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<i32> {
|
|||
.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())?;
|
||||
|
|
|
|||
|
|
@ -389,6 +389,8 @@ pub struct Kept {
|
|||
pub session: Uuid,
|
||||
pub transcript: PathBuf,
|
||||
pub warm_until: DateTime<Utc>,
|
||||
/// When the record was written; `warm_until` may not lie more than the longest cache lifetime past it.
|
||||
pub written_at: DateTime<Utc>,
|
||||
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<PathBuf>,
|
||||
/// Her sandbox: with it the child runs under bubblewrap (`sandbox_args`), the Persona's; the Self passes none.
|
||||
sandbox: Option<PathBuf>,
|
||||
/// 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<PathBuf>,
|
||||
/// 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<String>,
|
||||
|
|
@ -826,11 +835,11 @@ impl ClaudeSpawner {
|
|||
/// without it every start is fresh, and the journal says so.
|
||||
pub fn new(unit_environment: impl IntoIterator<Item = (OsString, OsString)>) -> Result<Self> {
|
||||
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<Utc>)>) -> 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")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<DateTime<Utc>>,
|
||||
/// 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<DateTime<Utc>>,
|
||||
/// 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<DateTime<Utc>>,
|
||||
/// 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<DateTime<Utc>> {
|
|||
}
|
||||
|
||||
/// 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<Utc>) -> 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<Utc>) -> (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<Utc>) -> (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<Utc>) -> (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 };
|
||||
|
|
|
|||
|
|
@ -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)));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Kept> {
|
||||
self.files.kept()
|
||||
}
|
||||
|
||||
fn keep(&self, keep: Option<(Uuid, chrono::DateTime<chrono::Utc>)>) -> Result<()> {
|
||||
self.files.keep(keep)
|
||||
}
|
||||
|
||||
fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
|
||||
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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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) > >
|
||||
|
|
|
|||
|
|
@ -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<String> = 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));
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue