From b4e006019662f1060af6c60d4231fc18ab1a4de0 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 17:35:14 -0500 Subject: [PATCH 1/6] Preserve selected legacy checkpoints and refuse blank resident recovery --- core/continuum-core/src/bin/continuum.rs | 481 +++++++++++++++- .../src/cognition/persona_workspace.rs | 540 +++++++++++++++--- .../persona_workspace/checkpoint_adoption.rs | 482 ++++++++++++++++ .../src/cognition/should_respond_module.rs | 43 +- core/continuum-core/src/ipc/vitals_emitter.rs | 61 +- core/continuum-core/src/persona/host.rs | 28 +- .../src/persona/service_loop.rs | 4 +- core/continuum-core/src/persona/supervisor.rs | 362 ++++++++---- .../PERSONA-COGNITION-PIPELINE.md | 6 + docs/personas/CHECKPOINT-RECOVERY.md | 73 +++ 10 files changed, 1796 insertions(+), 284 deletions(-) create mode 100644 core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs create mode 100644 docs/personas/CHECKPOINT-RECOVERY.md diff --git a/core/continuum-core/src/bin/continuum.rs b/core/continuum-core/src/bin/continuum.rs index 374e9bd9c2..6c57df15ad 100644 --- a/core/continuum-core/src/bin/continuum.rs +++ b/core/continuum-core/src/bin/continuum.rs @@ -94,6 +94,7 @@ fn local_help_requested(command: &str, args: &[String]) -> bool { | "orphans" | "deploy-verify" | "verify" + | "checkpoint" ) && args .iter() .any(|arg| matches!(arg.as_str(), "-h" | "--help"))) @@ -110,6 +111,11 @@ async fn run() -> Result<(), CliError> { return Ok(()); } let args = rest.into_iter(); + // Offline recovery must work before a core can start, and inspection must + // not mutate the checkout registry as a side effect. + if first == "checkpoint" { + return checkpoint(CheckpointCommand::parse(args)?).map_err(CliError::from); + } // Every CLI run from inside a repo records that checkout for the core // (repo-card staging reads it); the first deploy after #3706 would otherwise // start with an empty registry until the next `start`/`reboot`. @@ -361,6 +367,185 @@ fn socket_path() -> String { continuum_core::ipc::endpoint_paths::core_socket_path() } +/// Explicit legacy selection, independent of a live core's command registry. +/// Inspect emits a digest-bound plan; adopt consumes that exact plan offline. +#[derive(Debug, PartialEq, Eq)] +enum CheckpointCommand { + Inspect { + source: PathBuf, + persona_id: uuid::Uuid, + plan: PathBuf, + }, + Adopt { + plan: PathBuf, + }, +} + +impl CheckpointCommand { + fn parse(mut args: impl Iterator) -> Result { + let verb = args.next().ok_or("checkpoint requires inspect or adopt")?; + let mut source = None; + let mut persona_id = None; + let mut plan = None; + let mut legacy_writers_stopped = false; + while let Some(flag) = args.next() { + if verb == "adopt" && flag == "--legacy-writers-stopped" && !legacy_writers_stopped { + legacy_writers_stopped = true; + continue; + } + let slot = match (verb.as_str(), flag.as_str()) { + ("inspect", "--source") => &mut source, + ("inspect", "--persona-id") => &mut persona_id, + ("inspect" | "adopt", "--plan") => &mut plan, + _ => { + return Err(format!( + "unknown or repeated checkpoint {verb} option {flag}" + )) + } + }; + if slot.is_some() { + return Err(format!("duplicate checkpoint option {flag}")); + } + *slot = Some( + args.next() + .filter(|value| !value.is_empty() && !value.starts_with('-')) + .ok_or_else(|| format!("{flag} requires a value"))?, + ); + } + let plan = PathBuf::from(plan.ok_or("checkpoint requires --plan ")?); + match verb.as_str() { + "inspect" => Ok(Self::Inspect { + source: PathBuf::from(source.ok_or("inspect requires --source ")?), + persona_id: persona_id.ok_or("inspect requires --persona-id ")? + .parse().map_err(|error| format!("invalid persona UUID: {error}"))?, + plan, + }), + "adopt" if legacy_writers_stopped => Ok(Self::Adopt { plan }), + "adopt" => Err("adopt requires --legacy-writers-stopped: stop legacy cores and their automatic launchers before applying the inspected plan".into()), + _ => Err(format!("unknown checkpoint operation {verb}; use inspect or adopt")), + } + } +} + +fn checkpoint(command: CheckpointCommand) -> Result<(), String> { + use continuum_core::cognition::persona_workspace::checkpoint_adoption::{ + adopt, inspect, AdoptionPlan, + }; + use std::io::{Read, Write}; + + match command { + CheckpointCommand::Inspect { + source, + persona_id, + plan, + } => { + let selection = inspect(&source, persona_id).map_err(|error| error.to_string())?; + let plan = + checkpoint_plan_output(&plan, &selection.source.path, &selection.destination_path)?; + let bytes = serde_json::to_vec_pretty(&selection).map_err(|error| error.to_string())?; + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&plan) + .map_err(|error| format!("create inspection plan {}: {error}", plan.display()))?; + file.write_all(&bytes) + .and_then(|()| file.sync_all()) + .map_err(|error| format!("write inspection plan {}: {error}", plan.display()))?; + println!("{}", String::from_utf8_lossy(&bytes)); + eprintln!( + "checkpoint plan saved to {}; no checkpoint changed", + plan.display() + ); + } + CheckpointCommand::Adopt { plan } => { + // context-budget-exempt: bounds offline plan decoding, not model input. + const MAX_PLAN_BYTES: u64 = 1024 * 1024; + let mut bytes = Vec::new(); + std::fs::File::open(&plan) + .and_then(|file| file.take(MAX_PLAN_BYTES + 1).read_to_end(&mut bytes)) + .map_err(|error| format!("read inspection plan {}: {error}", plan.display()))?; + if bytes.len() as u64 > MAX_PLAN_BYTES { + return Err("checkpoint plan exceeds the offline decoding limit".into()); + } + let selection: AdoptionPlan = serde_json::from_slice(&bytes) + .map_err(|error| format!("invalid checkpoint plan: {error}"))?; + let receipt = + adopt(&selection, ensure_checkpoint_offline).map_err(|error| error.to_string())?; + println!( + "{}", + serde_json::to_string_pretty(&receipt).map_err(|error| error.to_string())? + ); + } + } + Ok(()) +} + +fn ensure_checkpoint_offline() -> std::io::Result<()> { + // A launcher can have written its PID before the executable-name probe + // identifies it. Read failure or a malformed PID is uncertainty, not absence. + let recorded_pid = match std::fs::read_to_string(pidfile_for(&socket_path())) { + Ok(contents) => { + let pid = contents + .trim() + .parse::() + .ok() + .filter(|pid| *pid > 0) + .ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid core PID file") + })?; + Some(pid) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, + Err(error) => return Err(error), + }; + core_process_evidence()?.ensure_offline(recorded_pid) +} + +/// A plan is metadata, never a checkpoint. In particular an absent destination +/// must not be poisoned with plan JSON by an otherwise successful create_new. +fn checkpoint_plan_output( + output: &Path, + source: &Path, + destination: &Path, +) -> Result { + let absolute = std::path::absolute(output).map_err(|error| error.to_string())?; + let parent = absolute + .parent() + .ok_or("plan output has no parent")? + .canonicalize() + .map_err(|error| format!("plan output directory: {error}"))?; + let filename = absolute.file_name().ok_or("plan output has no filename")?; + if ["volatile.json", ".volatile.lock"] + .iter() + .any(|reserved| filename.to_string_lossy().eq_ignore_ascii_case(reserved)) + { + return Err("inspection metadata cannot use a reserved checkpoint filename".into()); + } + #[cfg(windows)] + if parent.components().any(|component| { + matches!(component, + std::path::Component::Prefix(prefix) if matches!(prefix.kind(), + std::path::Prefix::UNC(_, _) | std::path::Prefix::VerbatimUNC(_, _))) + }) { + // An SMB alias can identify the same Persona directory through a + // different namespace, which lexical containment cannot establish. + return Err("write the inspection plan to a local path, not a network share".into()); + } + let output = parent.join(filename); + let native_personas = destination + .parent() + .and_then(Path::parent) + .ok_or("checkpoint destination has no Persona store")?; + let source_personas = source + .parent() + .and_then(Path::parent) + .ok_or("checkpoint source has no Persona store")?; + if output.starts_with(native_personas) || output.starts_with(source_personas) { + return Err("write the inspection plan outside managed Persona checkpoint storage".into()); + } + Ok(output) +} + /// Dispatch through the uniform Connection to an already-running core. Lifecycle is /// explicit: probing a node during a deploy must never launch its pre-swap image. /// This applies to every command, including dynamically registered ML commands; @@ -1471,8 +1656,18 @@ async fn prebuilt_checkout_sha( cmd.kill_on_drop(true); let output = tokio::time::timeout(Duration::from_secs(30), cmd.output()) .await - .map_err(|_| format!("prebuilt checkout HEAD lookup timed out in {}", cwd.display()))? - .map_err(|e| format!("cannot read prebuilt checkout HEAD in {}: {e}", cwd.display()))?; + .map_err(|_| { + format!( + "prebuilt checkout HEAD lookup timed out in {}", + cwd.display() + ) + })? + .map_err(|e| { + format!( + "cannot read prebuilt checkout HEAD in {}: {e}", + cwd.display() + ) + })?; if !output.status.success() { return Err(format!( "cannot verify prebuilt checkout HEAD in {}: git exited {}: {}", @@ -1499,7 +1694,10 @@ fn has_git_checkout(cwd: &Path) -> Result { match std::fs::symlink_metadata(path) { Ok(_) => Ok(true), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), - Err(e) => Err(format!("cannot inspect Git metadata {}: {e}", path.display())), + Err(e) => Err(format!( + "cannot inspect Git metadata {}: {e}", + path.display() + )), } } for ancestor in cwd.ancestors() { @@ -1839,18 +2037,92 @@ fn bound_elsewhere_hint(bound: &[(i32, String)], socket: &str) -> Option /// every platform, so there is one implementation instead of a Unix tool plus an unported gap. /// Matching on the executable NAME rather than `pgrep -f`'s full command line is also more precise /// here: it cannot accidentally match the bash process that is merely launching the core. +struct CoreProcessEvidence { + core_pids: Vec, + observed_pids: std::collections::HashSet, +} + +impl CoreProcessEvidence { + fn from_processes<'a>( + own_pid: i32, + processes: impl IntoIterator)>, + ) -> std::io::Result { + let mut evidence = Self { + core_pids: Vec::new(), + observed_pids: Default::default(), + }; + for (pid, name, exe) in processes { + evidence.observed_pids.insert(pid); + if process_matches_fragment(name, exe, "continuum-core-server") { + evidence.core_pids.push(pid); + } else if exe.is_none() && name == std::ffi::OsStr::new("continuum-core-") { + // Linux comm is truncated. Missing exe evidence is uncertainty; + // it must not count as absence or become a broad kill match. + return Err(std::io::Error::other(format!("cannot identify possible core PID {pid}: truncated name and unavailable executable"))); + } + } + if !evidence.observed_pids.contains(&own_pid) { + return Err(std::io::Error::other( + "cannot establish core absence: this process is missing from the process table", + )); + } + Ok(evidence) + } + + fn ensure_offline(&self, recorded_pid: Option) -> std::io::Result<()> { + if !self.core_pids.is_empty() { + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + format!( + "checkpoint adoption requires stopped cores; running PID(s): {:?}", + self.core_pids + ), + )); + } + if let Some(pid) = recorded_pid.filter(|pid| self.observed_pids.contains(pid)) { + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + format!("checkpoint adoption refused: core PID file names a live process ({pid})"), + )); + } + Ok(()) + } +} + +fn core_process_evidence() -> std::io::Result { + use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind}; + let mut sys = System::new(); + sys.refresh_processes_specifics( + ProcessesToUpdate::All, + true, + ProcessRefreshKind::nothing().with_exe(UpdateKind::OnlyIfNotSet), + ); + CoreProcessEvidence::from_processes( + std::process::id() as i32, + sys.processes() + .values() + .map(|process| (process.pid().as_u32() as i32, process.name(), process.exe())), + ) +} + +fn process_matches_fragment(name: &std::ffi::OsStr, exe: Option<&Path>, fragment: &str) -> bool { + name.to_string_lossy().contains(fragment) + || exe + .and_then(Path::file_name) + .is_some_and(|name| name.to_string_lossy().contains(fragment)) +} + fn processes_named(fragment: &str) -> Vec { - use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System}; + use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind}; let mut sys = System::new(); - sys.refresh_processes_specifics(ProcessesToUpdate::All, true, ProcessRefreshKind::nothing()); + sys.refresh_processes_specifics( + ProcessesToUpdate::All, + true, + ProcessRefreshKind::nothing().with_exe(UpdateKind::OnlyIfNotSet), + ); sys.processes() .values() - .filter(|p| { - p.name().to_string_lossy().contains(fragment) - || p.exe() - .map(|e| e.to_string_lossy().contains(fragment)) - .unwrap_or(false) - }) + .filter(|p| process_matches_fragment(p.name(), p.exe(), fragment)) .map(|p| p.pid().as_u32() as i32) .collect() } @@ -2848,6 +3120,163 @@ fn tail(path: &str, n: usize) -> String { #[cfg(test)] mod tests { + // What this catches (card 9f160b78): missing/truncated process evidence and + // a live PID-file process must never authorize offline memory replacement. + #[test] + fn checkpoint_offline_evidence_keeps_unknown_distinct_from_absent() { + use super::CoreProcessEvidence; + use std::ffi::OsStr; + use std::path::Path; + let own = (41, OsStr::new("continuum"), None); + let other = (42, OsStr::new("launcher"), None); + let evidence = CoreProcessEvidence::from_processes(41, [own, other]).unwrap(); + assert!(evidence.ensure_offline(None).is_ok()); + assert!(evidence.ensure_offline(Some(99)).is_ok()); + assert_eq!( + evidence.ensure_offline(Some(42)).unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + assert!(CoreProcessEvidence::from_processes(41, [other]).is_err()); + let truncated = (43, OsStr::new("continuum-core-"), None); + assert!(CoreProcessEvidence::from_processes(41, [own, truncated]).is_err()); + let resolved = ( + 43, + OsStr::new("continuum-core-"), + Some(Path::new("/bin/continuum-core-server")), + ); + assert!(CoreProcessEvidence::from_processes(41, [own, resolved]) + .unwrap() + .ensure_offline(None) + .is_err()); + // The uncertain truncated name must not broaden process termination. + assert!(!super::process_matches_fragment( + truncated.1, + truncated.2, + "continuum-core-server" + )); + assert!(super::process_matches_fragment( + resolved.1, + resolved.2, + "continuum-core-server" + )); + assert!( + !super::process_matches_fragment( + OsStr::new("python"), + Some(Path::new("/continuum-core-server-fixtures/python")), + "continuum-core-server" + ), + "parent directories must not turn an unrelated executable into a core kill target" + ); + } + + // What this catches (card 9f160b78): inspect --plan + // must not write a valid plan into a Persona's actual memory file. + #[test] + fn inspection_plan_cannot_occupy_checkpoint_or_evidence_storage() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let source_dir = root.join("legacy").join(uuid::Uuid::new_v4().to_string()); + let persona_dir = root.join("personas").join(uuid::Uuid::new_v4().to_string()); + std::fs::create_dir_all(&source_dir).unwrap(); + std::fs::create_dir_all(&persona_dir).unwrap(); + let sibling = root.join("legacy").join(uuid::Uuid::new_v4().to_string()); + std::fs::create_dir_all(&sibling).unwrap(); + let source = source_dir.join("volatile.json"); + let destination = persona_dir.join("volatile.json"); + for invalid in [ + &source, + &destination, + &persona_dir.join(".volatile.lock"), + &source_dir.join("plan.json"), + &sibling.join("volatile.json"), + ] { + assert!(super::checkpoint_plan_output(invalid, &source, &destination).is_err()); + assert!( + !invalid.exists(), + "inspection must leave managed state absent" + ); + } + let outside = root.join("plan.json"); + assert_eq!( + super::checkpoint_plan_output(&outside, &source, &destination).unwrap(), + outside + ); + } + + // What this catches (card 9f160b78): a malformed recovery invocation must + // never select/adopt a checkpoint or imply that legacy writers are stopped. + #[test] + fn checkpoint_recovery_requires_explicit_selection_and_offline_precondition() { + let parse = + |args: &[&str]| super::CheckpointCommand::parse(args.iter().map(|arg| arg.to_string())); + let persona = "68d231fb-1b99-47ea-8615-14538906817a"; + assert!(matches!( + parse(&[ + "inspect", + "--source", + "old/volatile.json", + "--persona-id", + persona, + "--plan", + "plan.json" + ]) + .unwrap(), + super::CheckpointCommand::Inspect { .. } + )); + assert_eq!( + parse(&["adopt", "--plan", "plan.json", "--legacy-writers-stopped"]).unwrap(), + super::CheckpointCommand::Adopt { + plan: "plan.json".into() + } + ); + for args in [ + vec![], + vec!["adopt", "--plan", "plan.json"], + vec!["adopt", "--plan", "plan.json", "--force"], + vec![ + "adopt", + "--plan", + "plan.json", + "--plan", + "other.json", + "--legacy-writers-stopped", + ], + vec![ + "adopt", + "--plan", + "plan.json", + "--legacy-writers-stopped", + "--legacy-writers-stopped", + ], + vec![ + "inspect", + "--source", + "old/volatile.json", + "--plan", + "plan.json", + ], + vec![ + "inspect", + "--source", + "old/volatile.json", + "--persona-id", + "invalid", + "--plan", + "plan.json", + ], + vec![ + "inspect", + "--source", + "--persona-id", + persona, + "--plan", + "plan.json", + ], + ] { + assert!(parse(&args).is_err(), "must refuse {args:?}"); + } + } + // what this catches: card 67f53b63 — a missing/misspelled prebuilt path // must not fall through to a source reboot, and --force only changes leases. #[test] @@ -2964,7 +3393,10 @@ mod tests { super::prebuilt_checkout_sha(&repo, None).await.is_err(), "a checkout with no readable HEAD cannot self-anchor" ); - git(&repo, &["commit", "--allow-empty", "--quiet", "-m", "fixture"]); + git( + &repo, + &["commit", "--allow-empty", "--quiet", "-m", "fixture"], + ); git(&repo, &["checkout", "--detach", "--quiet", "HEAD"]); let head = git(&repo, &["rev-parse", "--short", "HEAD"]); assert_eq!( @@ -2974,7 +3406,14 @@ mod tests { git( &repo, - &["worktree", "add", "--detach", "--quiet", "../linked", "HEAD"], + &[ + "worktree", + "add", + "--detach", + "--quiet", + "../linked", + "HEAD", + ], ); let linked = tmp.path().join("linked"); let nested = linked.join("src"); @@ -3003,11 +3442,12 @@ mod tests { .unwrap(), Some(head), ); - assert!( - super::prebuilt_checkout_sha(&standalone, Some(tmp.path().join("missing").as_os_str())) - .await - .is_err() - ); + assert!(super::prebuilt_checkout_sha( + &standalone, + Some(tmp.path().join("missing").as_os_str()) + ) + .await + .is_err()); std::fs::write(linked.join(".git"), "gitdir: missing-checkout\n").unwrap(); assert!( @@ -3084,6 +3524,7 @@ mod tests { "orphans", "deploy-verify", "verify", + "checkpoint", ] { for flag in ["-h", "--help"] { assert!(super::local_help_requested( @@ -3931,6 +4372,10 @@ fn usage() -> String { continuum stop stop the running core\n \ continuum deploy-verify prove the running core's build SHA matches the deployed source\n\ \n\ + Legacy checkpoint recovery (local; no running core required):\n \ + continuum checkpoint inspect --source --persona-id --plan \n save an explicit digest-bound selection; no checkpoint changed\n \ + continuum checkpoint adopt --plan --legacy-writers-stopped\n preserve both snapshots and adopt the selected bytes offline;\n stop legacy cores and automatic launchers first; no final-flush claim\n\ + \n\ Desktop (the core serves it; no port to remember):\n \ continuum desktop open the desktop in your browser (alias: uu desktop)\n\ \n\ diff --git a/core/continuum-core/src/cognition/persona_workspace.rs b/core/continuum-core/src/cognition/persona_workspace.rs index 6638550eb2..9f181492e0 100644 --- a/core/continuum-core/src/cognition/persona_workspace.rs +++ b/core/continuum-core/src/cognition/persona_workspace.rs @@ -263,18 +263,12 @@ impl GroundingSource { /// routes decisions through the Workspace — without them, that grounding (#1650 / /// #1651) silently falls out of the live path. pub fn build_workspace_cycle(cfg: PersonaBrainConfig) -> WorkspaceCycle { - assemble_workspace_cycle(cfg, WorkspaceLifetime::Ephemeral).0 -} - -#[derive(Clone, Copy)] -enum WorkspaceLifetime { - Resident, - Ephemeral, + assemble_workspace_cycle(cfg, None).0 } fn assemble_workspace_cycle( cfg: PersonaBrainConfig, - lifetime: WorkspaceLifetime, + restored: Option, ) -> (WorkspaceCycle, Arc) { let mut faculties: Vec> = Vec::with_capacity(2 + cfg.grounding_sources.len()); @@ -316,40 +310,38 @@ fn assemble_workspace_cycle( )); // Persistence belongs to resident registration, independently of whether // recall runs inline or deferred. Constructing a fork grants no disk access. - if matches!(lifetime, WorkspaceLifetime::Resident) { - if let Some(persisted) = load_volatile(cfg.persona_id) { - let n = persisted.wm.entries.len(); - working_memory.restore(persisted.wm); - let peer = crate::identity::PeerId::from_uuid(cfg.persona_id); - match &persisted.own_speech { - OwnSpeechPersisted::ByRoom(by_room) => { - for (room, utterances) in by_room { - for utterance in utterances { - super::deliberation_budget::record_own_speech(peer, *room, utterance); - } + if let Some(persisted) = restored { + let n = persisted.wm.entries.len(); + working_memory.restore(persisted.wm); + let peer = crate::identity::PeerId::from_uuid(cfg.persona_id); + match &persisted.own_speech { + OwnSpeechPersisted::ByRoom(by_room) => { + for (room, utterances) in by_room { + for utterance in utterances { + super::deliberation_budget::record_own_speech(peer, *room, utterance); } } - // Pre-room-scoping file: unattributable, so dropped rather than - // mis-filed into a room she may never have spoken in. See - // OwnSpeechPersisted — hydrate_speech_rings re-seeds with rooms. - OwnSpeechPersisted::Legacy(flat) => { - crate::probe!( - class = "persona.volatile.own_speech_legacy_dropped", - persona = %cfg.persona_name, - dropped = flat.len(), - "pre-room-scoping own-speech ring carried no room — dropped, \ - durable-transcript hydration re-seeds it" - ); - } } - crate::probe!( - class = "persona.volatile.restored", - persona = %cfg.persona_name, - entries = n, - ring = persisted.own_speech.len(), - "volatile tier restored — waking mid-work, not blank" - ); + // Pre-room-scoping file: unattributable, so dropped rather than + // mis-filed into a room she may never have spoken in. See + // OwnSpeechPersisted — hydrate_speech_rings re-seeds with rooms. + OwnSpeechPersisted::Legacy(flat) => { + crate::probe!( + class = "persona.volatile.own_speech_legacy_dropped", + persona = %cfg.persona_name, + dropped = flat.len(), + "pre-room-scoping own-speech ring carried no room — dropped, \ + durable-transcript hydration re-seeds it" + ); + } } + crate::probe!( + class = "persona.volatile.restored", + persona = %cfg.persona_name, + entries = n, + ring = persisted.own_speech.len(), + "volatile tier restored — waking mid-work, not blank" + ); } // Async-dispatch listener (LIVE personas only): fold completions of THIS persona's // background dispatches back into working memory by handle, so a compile/train/sentinel @@ -1114,12 +1106,18 @@ impl PersonaWorkspaceRegistry { /// cycle + template: a persona can respawn in the same process (node /// resilience), and the fresh admission + adapter must replace the prior /// lifetime's. This is the production spawn path (see `supervisor.rs`). - pub fn register_from_cfg(&self, cfg: PersonaBrainConfig) -> Arc { + /// Checkpoint lock/read/schema failures refuse this registration and leave + /// an existing resident unchanged. Call from a blocking worker during boot. + pub fn register_from_cfg( + &self, + cfg: PersonaBrainConfig, + ) -> std::io::Result> { let _assembly = self.assembly.lock(); let persona_id = cfg.persona_id; let template = cfg.clone(); // Assembly restores disk state; do it outside the lookup lock. - let (cycle, working_memory) = assemble_workspace_cycle(cfg, WorkspaceLifetime::Resident); + let restored = load_volatile(persona_id)?; + let (cycle, working_memory) = assemble_workspace_cycle(cfg, restored); let cycle = Arc::new(cycle); // cycles THEN templates (the one canonical lock order). let mut cycles = self.cycles.lock(); @@ -1131,24 +1129,25 @@ impl PersonaWorkspaceRegistry { working_memory, }, ); - cycle + Ok(cycle) } /// Get the persona's mind, building + caching it from `cfg` on first access. /// Lazy-init so a persona's cycle is assembled once and reused across every /// room it services (the "one soul" invariant). Also retains the fork-template /// (same as [`register_from_cfg`](Self::register_from_cfg)). - pub fn get_or_build(&self, cfg: PersonaBrainConfig) -> Arc { + pub fn get_or_build(&self, cfg: PersonaBrainConfig) -> std::io::Result> { let persona_id = cfg.persona_id; if let Some(existing) = self.get(&persona_id) { - return existing; + return Ok(existing); } let _assembly = self.assembly.lock(); if let Some(existing) = self.get(&persona_id) { - return existing; + return Ok(existing); } let template = cfg.clone(); - let (cycle, working_memory) = assemble_workspace_cycle(cfg, WorkspaceLifetime::Resident); + let restored = load_volatile(persona_id)?; + let (cycle, working_memory) = assemble_workspace_cycle(cfg, restored); let cycle = Arc::new(cycle); // cycles THEN templates (the one canonical lock order). let mut cycles = self.cycles.lock(); @@ -1160,7 +1159,7 @@ impl PersonaWorkspaceRegistry { working_memory, }, ); - cycle + Ok(cycle) } /// Periodic crash checkpoint. A busy writer coalesces this tick; no queue @@ -1468,6 +1467,10 @@ struct PersistedVolatile { own_speech: OwnSpeechPersisted, } +/// Explicit offline adoption of a selected legacy checkpoint; shares this +/// owner's schema, native path, and cross-process checkpoint exclusion. +pub mod checkpoint_adoption; + /// Her own-speech rings on disk. Room-keyed since 2026-08-14 — a ring restored /// without its room would re-create the cross-room repetition fact the keying /// fix exists to kill. @@ -1515,40 +1518,41 @@ fn save_volatile( persona_id: Uuid, wm: &super::working_memory::WorkingMemory, ) -> std::io::Result<()> { + let path = volatile_path(persona_id)?; + let _checkpoint_lock = checkpoint_adoption::lock_checkpoint(&path, true)?; let persisted = PersistedVolatile { wm: wm.snapshot(), own_speech: OwnSpeechPersisted::ByRoom(super::deliberation_budget::own_speech_by_room( crate::identity::PeerId::from_uuid(persona_id), )), }; - let path = volatile_path(persona_id)?; if let Some(dir) = path.parent() { std::fs::create_dir_all(dir)?; } let tmp = path.with_extension("json.tmp"); - std::fs::write(&tmp, serde_json::to_vec(&persisted)?)?; - std::fs::rename(&tmp, &path) + use std::io::Write; + let mut file = std::fs::File::create(&tmp)?; + file.write_all(&serde_json::to_vec(&persisted)?)?; + file.sync_all()?; + drop(file); + std::fs::rename(&tmp, &path)?; + checkpoint_adoption::sync_parent(&path) } -/// Load the previous life's volatile tier, if any. Unreadable/corrupt files -/// return None LOUDLY (a mind-file that fails to parse must never be silently -/// ignored twice — the warn is the operator's cue to look). -fn load_volatile(persona_id: Uuid) -> Option { - let path = match volatile_path(persona_id) { - Ok(path) => path, - Err(error) => { - tracing::warn!(%persona_id, %error, "volatile-tier root unavailable — previous memory could not be loaded"); - return None; - } +/// Load the previous life's volatile tier under the same owner lock as saves +/// and adoption. Only an absent checkpoint permits a fresh resident; inability +/// to read or decode an existing one must reach the registration caller. +fn load_volatile(persona_id: Uuid) -> std::io::Result> { + let path = volatile_path(persona_id)?; + let _lock = checkpoint_adoption::lock_checkpoint(&path, true)?; + let bytes = match std::fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error), }; - let bytes = std::fs::read(&path).ok()?; - match serde_json::from_slice(&bytes) { - Ok(p) => Some(p), - Err(e) => { - tracing::warn!(persona_id = %persona_id, error = %e, path = %path.display(), "volatile-tier file unreadable — waking blank this once"); - None - } - } + serde_json::from_slice(&bytes) + .map(Some) + .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error)) } #[cfg(test)] @@ -1560,6 +1564,269 @@ mod tests { use tokio::sync::{watch, Notify}; use tokio::time::timeout; + // Uses the real persisted schema and WorkingMemory writer, not a second + // checkpoint model. Explicit legacy root belongs to each temporary fixture. + fn adoption_source( + root: &std::path::Path, + persona: Uuid, + text: &str, + ) -> (std::path::PathBuf, Vec) { + let memory = WorkingMemory::new(8); + memory.record_receipt(text); + let snapshot = PersistedVolatile { + wm: memory.snapshot(), + own_speech: OwnSpeechPersisted::ByRoom(vec![(Uuid::new_v4(), vec![text.into()])]), + }; + let bytes = serde_json::to_vec(&snapshot).expect("test: actual snapshot serializes"); + let path = root.join(persona.to_string()).join("volatile.json"); + std::fs::create_dir_all(path.parent().expect("test: fixture parent")) + .expect("test: fixture directory"); + std::fs::write(&path, &bytes).expect("test: legacy checkpoint"); + (path, bytes) + } + + // 9f160b78: explicit selection preserves WHOLE bytes plus both origins, + // even when the caller chooses an older sequence than the destination. + #[test] + fn legacy_adoption_preserves_selected_bytes_and_is_bound_on_replay() { + use checkpoint_adoption::{adopt, inspect, AdoptionError}; + let dir = tempfile::tempdir().unwrap(); + let _home = crate::paths::NativeHomeOverride::install(&dir.path().join("native")); + let persona = Uuid::new_v4(); + let memory = WorkingMemory::new(8); + memory.record_receipt("different previous lifetime"); + memory.record_receipt("higher sequence is not a selection policy"); + save_volatile(persona, &memory).unwrap(); + let destination = volatile_path(persona).unwrap(); + let original = std::fs::read(&destination).unwrap(); + let (source, selected) = + adoption_source(&dir.path().join("legacy"), persona, "selected work"); + let plan = inspect(&source, persona).unwrap(); + assert!( + plan.source.summary.next_action_seq + < plan + .prior_destination + .as_ref() + .unwrap() + .summary + .next_action_seq + ); + let receipt = adopt(&plan, || Ok(())).unwrap(); + assert!(!receipt.already_applied); + assert!(!receipt.legacy_final_flush_acknowledged); + assert_eq!(std::fs::read(&source).unwrap(), selected); + assert_eq!(std::fs::read(&destination).unwrap(), selected); + assert_eq!(std::fs::read(&receipt.source_archive).unwrap(), selected); + assert_eq!( + std::fs::read(receipt.prior_destination_archive.as_ref().unwrap()).unwrap(), + original + ); + assert!(adopt(&plan, || Ok(())).unwrap().already_applied); + // The source archive is a distinct inode: an in-place future write to + // the destination cannot alter the preserved selection. + std::fs::write(&destination, &original).unwrap(); + assert_eq!(std::fs::read(&receipt.source_archive).unwrap(), selected); + assert!(matches!( + adopt(&plan, || Ok(())), + Err(AdoptionError::DestinationChanged) + )); + std::fs::write(&receipt.source_archive, b"conflicting evidence").unwrap(); + assert!(matches!( + adopt(&plan, || Ok(())), + Err(AdoptionError::EvidenceConflict(_)) + )); + assert_eq!(std::fs::read(&destination).unwrap(), original); + } + + // 9f160b78: both guards are rechecked at the actual publication boundary; + // late writes are preserved, not silently replaced by the inspected plan. + #[test] + fn legacy_adoption_refuses_changed_source_destination_and_offline_state() { + use checkpoint_adoption::{adopt, inspect, AdoptionError}; + for changed_source in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let _home = crate::paths::NativeHomeOverride::install(&dir.path().join("native")); + let persona = Uuid::new_v4(); + let (source, selected) = + adoption_source(&dir.path().join("legacy"), persona, "selected work"); + let plan = inspect(&source, persona).unwrap(); + assert!(plan.prior_destination.is_none()); + let (other, changed) = + adoption_source(&dir.path().join("other"), persona, "concurrent work"); + let mut checks = 0; + let outcome = adopt(&plan, || { + checks += 1; + if checks == 3 { + std::fs::copy( + &other, + if changed_source { + &source + } else { + &plan.destination_path + }, + )?; + } + Ok(()) + }); + assert_eq!( + checks, 3, + "test: mutation reached the pre-publication check" + ); + if changed_source { + assert!(matches!(outcome, Err(AdoptionError::SourceChanged))); + assert_eq!(std::fs::read(&source).unwrap(), changed); + assert!(!plan.destination_path.exists()); + } else { + assert!(matches!(outcome, Err(AdoptionError::DestinationChanged))); + assert_eq!(std::fs::read(&plan.destination_path).unwrap(), changed); + assert_eq!(std::fs::read(&source).unwrap(), selected); + } + } + let dir = tempfile::tempdir().unwrap(); + let _home = crate::paths::NativeHomeOverride::install(&dir.path().join("native")); + let persona = Uuid::new_v4(); + let (source, _) = adoption_source(&dir.path().join("legacy"), persona, "selected work"); + let plan = inspect(&source, persona).unwrap(); + let mut checks = 0; + let outcome = adopt(&plan, || { + checks += 1; + if checks == 2 { + Err(std::io::Error::other( + "core appeared before lock acquisition", + )) + } else { + Ok(()) + } + }); + assert!(matches!(outcome, Err(AdoptionError::Io(_)))); + assert_eq!(checks, 2); + assert!(!plan.destination_path.exists()); + } + + // 9f160b78: inject a real receipt-publication IO failure after the actual + // checkpoint rename. Retry joins preserved intent, never guesses from a + // coincidentally matching destination, and never rewrites source bytes. + #[test] + fn legacy_adoption_recovers_committed_file_without_receipt() { + use checkpoint_adoption::{adopt, inspect}; + let dir = tempfile::tempdir().unwrap(); + let _home = crate::paths::NativeHomeOverride::install(&dir.path().join("native")); + let persona = Uuid::new_v4(); + let (source, selected) = + adoption_source(&dir.path().join("legacy"), persona, "selected work"); + let plan = inspect(&source, persona).unwrap(); + for _ in 0..2 { + let mut checks = 0; + assert!(adopt(&plan, || { + checks += 1; + if checks == 3 { + Err(std::io::Error::other("core appeared before publication")) + } else { + Ok(()) + } + }) + .is_err()); + assert!(!plan.destination_path.exists()); + let evidence_root = plan + .destination_path + .parent() + .unwrap() + .join(".checkpoint-adoptions"); + let evidence = std::fs::read_dir(evidence_root) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + let files = std::fs::read_dir(evidence).unwrap().count(); + assert_eq!( + files, 3, + "source, plan and one staged inode; retry must not accumulate payload copies" + ); + } + let mut checks = 0; + let mut blocked_receipt = None; + let outcome = adopt(&plan, || { + checks += 1; + if checks == 3 { + let archive_root = plan + .destination_path + .parent() + .unwrap() + .join(".checkpoint-adoptions"); + let evidence = std::fs::read_dir(archive_root)?.next().unwrap()?.path(); + let path = evidence.join("receipt.json"); + std::fs::create_dir(&path)?; + blocked_receipt = Some(path); + } + Ok(()) + }); + assert!(outcome.is_err()); + assert_eq!(std::fs::read(&plan.destination_path).unwrap(), selected); + std::fs::remove_dir(blocked_receipt.unwrap()).unwrap(); + let recovered = adopt(&plan, || Ok(())).unwrap(); + assert!(recovered.already_applied); + assert!(recovered.receipt_path.is_file()); + assert_eq!(std::fs::read(&recovered.source_archive).unwrap(), selected); + assert!(adopt(&plan, || Ok(())).unwrap().already_applied); + } + + // 9f160b78: an explicit UUID is a declaration checked against the path, + // not a replacement identity inserted into arbitrary JSON. + #[test] + fn legacy_adoption_inspection_rejects_identity_and_schema_mismatch() { + use checkpoint_adoption::{inspect, AdoptionError}; + let dir = tempfile::tempdir().unwrap(); + let _home = crate::paths::NativeHomeOverride::install(&dir.path().join("native")); + let persona = Uuid::new_v4(); + let (source, _) = adoption_source(&dir.path().join("legacy"), persona, "selected work"); + assert!(matches!( + inspect(&source, Uuid::new_v4()), + Err(AdoptionError::Invalid(_)) + )); + std::fs::write(&source, b"{}").unwrap(); + assert!(matches!( + inspect(&source, persona), + Err(AdoptionError::Invalid(_)) + )); + assert!( + !dir.path().join("native").exists(), + "inspection is read-only" + ); + let (source, selected) = + adoption_source(&dir.path().join("legacy"), persona, "valid selected work"); + let destination = volatile_path(persona).unwrap(); + std::fs::create_dir_all(destination.parent().unwrap()).unwrap(); + std::fs::write(&destination, b"corrupt native checkpoint").unwrap(); + assert!(matches!( + inspect(&source, persona), + Err(AdoptionError::Invalid(_)) + )); + assert_eq!( + std::fs::read(&destination).unwrap(), + b"corrupt native checkpoint" + ); + assert_eq!(std::fs::read(&source).unwrap(), selected); + } + + // 9f160b78: actual cross-process lock primitive, not a simulated busy flag. + #[test] + fn legacy_adoption_refuses_a_checkpoint_owner_already_holding_the_lock() { + use checkpoint_adoption::{adopt, inspect, lock_checkpoint, AdoptionError}; + let dir = tempfile::tempdir().unwrap(); + let _home = crate::paths::NativeHomeOverride::install(&dir.path().join("native")); + let persona = Uuid::new_v4(); + let (source, _) = adoption_source(&dir.path().join("legacy"), persona, "selected work"); + let plan = inspect(&source, persona).unwrap(); + let owner = lock_checkpoint(&plan.destination_path, true).unwrap(); + assert!( + matches!(adopt(&plan, || Ok(())), Err(AdoptionError::Io(error)) if error.kind() == fs2::lock_contended_error().kind()) + ); + assert!(!plan.destination_path.exists()); + drop(owner); + assert!(!adopt(&plan, || Ok(())).unwrap().already_applied); + } + use crate::ai::heuristic_adapter::HeuristicInferenceAdapter; use crate::cognition::workspace::Decision; use crate::persona::engram::{ChatMessageRef, Engram, EngramKind, EngramOrigin, TrustState}; @@ -1583,7 +1850,9 @@ mod tests { .join(persona.to_string()) .join("volatile.json") ); - let saved = load_volatile(persona).expect("read the native checkpoint"); + let saved = load_volatile(persona) + .expect("read the native checkpoint") + .expect("checkpoint exists"); assert!( saved.wm.last_action.is_some(), "the fixture must carry real work" @@ -1606,7 +1875,9 @@ mod tests { let _native = crate::paths::NativeHomeOverride::install(home.path()); let registry = PersonaWorkspaceRegistry::new(); let persona = Uuid::new_v4(); - let old_cycle = registry.register_from_cfg(cfg_for(persona)); + let old_cycle = registry + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); let old_memory = Arc::clone(®istry.cycles.lock()[&persona].working_memory); old_memory.record_receipt("previous resident action"); assert!(registry.checkpoint_volatile_all().remove(0).1.is_ok()); @@ -1614,11 +1885,13 @@ mod tests { old_memory.record_receipt("work after an ordinary explicit save"); assert!(registry.checkpoint_volatile_all().remove(0).1.is_ok()); assert_eq!( - load_volatile(persona).unwrap().wm.last_action, + load_volatile(persona).unwrap().unwrap().wm.last_action, old_memory.snapshot().last_action ); - let current = registry.register_from_cfg(cfg_for(persona)); + let current = registry + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); let current_memory = Arc::clone(®istry.cycles.lock()[&persona].working_memory); // cfg_for uses synchronous recall: restore authority must not depend on // that scheduling flag, or these residents still wake blank. @@ -1652,7 +1925,10 @@ mod tests { old_memory.record_receipt("obsolete cycle must not publish this"); tokio::task::yield_now().await; // the former saver's immediate first tick assert!(registry.stop_volatile_checkpoints().remove(0).1.is_ok()); - assert_eq!(load_volatile(persona).unwrap().wm.last_action, expected); + assert_eq!( + load_volatile(persona).unwrap().unwrap().wm.last_action, + expected + ); let checkpoint = std::fs::read(volatile_path(persona).unwrap()).unwrap(); current_memory.record_receipt("late mutation after shutdown boundary"); assert!(registry.checkpoint_volatile_all().is_empty()); @@ -1702,7 +1978,9 @@ mod tests { let failed = Uuid::new_v4(); let healthy = Uuid::new_v4(); for id in [failed, healthy] { - registry.register_from_cfg(cfg_for(id)); + registry + .register_from_cfg(cfg_for(id)) + .expect("test: resident checkpoint is readable"); registry.cycles.lock()[&id] .working_memory .record_receipt("completed real work"); @@ -1726,7 +2004,12 @@ mod tests { .unwrap() .1 .is_ok()); - assert!(load_volatile(healthy).unwrap().wm.last_action.is_some()); + assert!(load_volatile(healthy) + .unwrap() + .unwrap() + .wm + .last_action + .is_some()); } #[cfg(feature = "stress-tests")] @@ -1742,7 +2025,9 @@ mod tests { let _native = crate::paths::NativeHomeOverride::install(home.path()); let registry = Arc::new(PersonaWorkspaceRegistry::new()); let persona = Uuid::new_v4(); - registry.register_from_cfg(cfg_for(persona)); + registry + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); registry.cycles.lock()[&persona] .working_memory .record_receipt("old resident action"); @@ -1763,7 +2048,9 @@ mod tests { let root = Arc::clone(&home); let replacement = tokio::task::spawn_blocking(move || { let _native = crate::paths::NativeHomeOverride::install(root.path()); - let current = owner.register_from_cfg(cfg_for(persona)); + let current = owner + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); assert!(Arc::ptr_eq(¤t, &owner.get(&persona).unwrap())); let memory = Arc::clone(&owner.cycles.lock()[&persona].working_memory); memory.record_receipt("replacement completed the review"); @@ -1803,7 +2090,10 @@ mod tests { .await?? .remove(0) .1?; - assert_eq!(load_volatile(persona).unwrap().wm.last_action, expected); + assert_eq!( + load_volatile(persona).unwrap().unwrap().wm.last_action, + expected + ); let checkpoint = std::fs::read(volatile_path(persona).unwrap()).unwrap(); current.record_receipt("late action after final ACK"); assert!(registry.checkpoint_volatile_all().is_empty()); @@ -1902,8 +2192,12 @@ mod tests { asha_cfg.persona_name = "Asha".to_string(); let mut atlas_cfg = cfg_for(atlas); atlas_cfg.persona_name = "Atlas".to_string(); - registry.register_from_cfg(asha_cfg); - registry.register_from_cfg(atlas_cfg); + registry + .register_from_cfg(asha_cfg) + .expect("test: resident checkpoint is readable"); + registry + .register_from_cfg(atlas_cfg) + .expect("test: resident checkpoint is readable"); // full UUID assert_eq!( @@ -1950,15 +2244,21 @@ mod tests { async fn registry_keeps_one_mind_per_persona() { let registry = PersonaWorkspaceRegistry::new(); let persona = Uuid::new_v4(); - let first = registry.get_or_build(cfg_for(persona)); - let second = registry.get_or_build(cfg_for(persona)); + let first = registry + .get_or_build(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); + let second = registry + .get_or_build(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); assert!( Arc::ptr_eq(&first, &second), "same persona must resolve to the SAME mind across rooms — not severed per-room" ); assert_eq!(registry.len(), 1); // A different persona is a different mind. - let _ = registry.get_or_build(cfg_for(Uuid::new_v4())); + let _ = registry + .get_or_build(cfg_for(Uuid::new_v4())) + .expect("test: resident checkpoint is readable"); assert_eq!(registry.len(), 2); } @@ -1973,8 +2273,12 @@ mod tests { let persona = Uuid::new_v4(); // register_from_cfg IS the production overwrite path (supervisor.rs spawn); // it builds + caches and returns the fresh Arc. - let first = registry.register_from_cfg(cfg_for(persona)); - let second = registry.register_from_cfg(cfg_for(persona)); + let first = registry + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); + let second = registry + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); let got = registry.get(&persona).expect("registered"); assert!( Arc::ptr_eq(&got, &second), @@ -1987,6 +2291,66 @@ mod tests { assert_eq!(registry.len(), 1); } + // 9f160b78: unreadable existing memory is not permission to register a blank + // replacement. Both the live cycle and its fork template survive refusal. + #[tokio::test(flavor = "current_thread")] + async fn failed_checkpoint_load_preserves_the_registered_resident() { + let home = tempfile::tempdir().unwrap(); + let _native = crate::paths::NativeHomeOverride::install(home.path()); + let registry = PersonaWorkspaceRegistry::new(); + let persona = Uuid::new_v4(); + assert!(load_volatile(persona).unwrap().is_none()); + let current = registry.register_from_cfg(cfg_for(persona)).unwrap(); + let path = volatile_path(persona).unwrap(); + std::fs::write(&path, br#"{"wm":{},"own_speech":[]}"#).unwrap(); + let mut replacement = cfg_for(persona); + replacement.persona_name = "must not replace the resident".into(); + let error = match registry.register_from_cfg(replacement) { + Err(error) => error, + Ok(_) => panic!("test: malformed checkpoint must refuse registration"), + }; + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + assert!(Arc::ptr_eq(¤t, ®istry.get(&persona).unwrap())); + assert_eq!(registry.templates.lock()[&persona].persona_name, "Ivar"); + assert!(Arc::ptr_eq( + ¤t, + ®istry.get_or_build(cfg_for(persona)).unwrap() + )); + assert_eq!( + std::fs::read(&path).unwrap(), + br#"{"wm":{},"own_speech":[]}"# + ); + } + + // 9f160b78: cold lazy registration must preserve lock/read/schema errors; + // none may insert a cycle or fork template as if no file had existed. + #[tokio::test(flavor = "current_thread")] + async fn resident_registration_refuses_checkpoint_and_lock_errors() { + let home = tempfile::tempdir().unwrap(); + let _native = crate::paths::NativeHomeOverride::install(home.path()); + let registry = PersonaWorkspaceRegistry::new(); + for fault in ["schema", "read", "lock"] { + let persona = Uuid::new_v4(); + let path = volatile_path(persona).unwrap(); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + match fault { + "schema" => std::fs::write(&path, b"not a checkpoint").unwrap(), + "read" => std::fs::create_dir(&path).unwrap(), + "lock" => { + std::fs::create_dir(path.parent().unwrap().join(".volatile.lock")).unwrap() + } + _ => unreachable!("test: fixed fault cases"), + } + assert!( + registry.get_or_build(cfg_for(persona)).is_err(), + "{fault} must refuse registration" + ); + assert!(registry.get(&persona).is_none()); + assert!(!registry.templates.lock().contains_key(&persona)); + } + assert_eq!(registry.len(), 0); + } + // THE LIVE BRING-UP: a persona's mind thinks with the REAL local model. // Runs the EXACT production assembly path (build_workspace_cycle → RecallFaculty // + LlmDeliberationFaculty) against the real LlamaCppAdapter (qwen3.5-4b-code- @@ -2267,7 +2631,9 @@ mod tests { // Its gate starts open, so it delivers straight through here. GroundingSource::framing(Arc::new(GatedGrounding::new())), ]; - registry.register_from_cfg(cfg); + registry + .register_from_cfg(cfg) + .expect("test: resident checkpoint is readable"); // Spoken exam (no hands): the workspace-map must not reach her mind. let spoken = registry @@ -2308,7 +2674,9 @@ mod tests { use crate::cognition::workspace::FacultyId; let registry = PersonaWorkspaceRegistry::new(); let persona = Uuid::new_v4(); - registry.register_from_cfg(cfg_for(persona)); + registry + .register_from_cfg(cfg_for(persona)) + .expect("test: resident checkpoint is readable"); let with_recall = registry .fork_eval_cycle(&persona, false, None, false) diff --git a/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs b/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs new file mode 100644 index 0000000000..66267f1680 --- /dev/null +++ b/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs @@ -0,0 +1,482 @@ +//! Explicit selection of one legacy volatile checkpoint, never a merge or a +//! recency heuristic. Whole original bytes survive ordinary restore unchanged. +//! The shared checkpoint lock excludes participating readers/writers. Old +//! binaries cannot retroactively honor it: adoption additionally requires the +//! caller's offline check, and never claims the selected snapshot was flushed +//! at shutdown or authenticated to a process lifetime. + +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::path::{Path, PathBuf}; + +use fs2::FileExt; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use super::{volatile_path, OwnSpeechPersisted, PersistedVolatile}; + +// context-budget-exempt: bounds offline untrusted file decoding, not inference. +const MAX_CHECKPOINT_BYTES: u64 = 64 * 1024 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CheckpointSummary { + pub build_sha: String, + pub saved_at_ms: u64, + pub next_action_seq: u64, + pub entries: usize, + pub result_receipts: usize, + pub own_utterances: usize, + /// Old roomless speech remains in the preserved bytes; ordinary restore + /// cannot attribute it and intentionally does not inject it into a room. + pub unscoped_own_utterances: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CheckpointSelection { + pub path: PathBuf, + pub sha256: String, + pub bytes: u64, + pub summary: CheckpointSummary, +} + +/// Public reviewable preconditions. The UUID is a caller declaration checked +/// against the selected directory; the legacy payload has no embedded UUID. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AdoptionPlan { + pub format_version: u32, + pub declared_persona_id: Uuid, + pub source: CheckpointSelection, + pub destination_path: PathBuf, + pub prior_destination: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum AdoptionAssurance { + /// No old-core final-flush acknowledgment or embedded lifetime identity. + SelectedLegacySnapshotWithDeclaredPersonaBinding, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AdoptionReceipt { + pub plan: AdoptionPlan, + pub assurance: AdoptionAssurance, + /// Always false: adoption cannot manufacture an old-core flush receipt. + pub legacy_final_flush_acknowledged: bool, + pub source_archive: PathBuf, + pub prior_destination_archive: Option, + pub receipt_path: PathBuf, + /// True only when this exact completed plan still matches destination bytes. + pub already_applied: bool, +} + +#[derive(Debug, thiserror::Error)] +pub enum AdoptionError { + #[error(transparent)] + Io(#[from] io::Error), + #[error("invalid checkpoint selection: {0}")] + Invalid(String), + #[error("the selected source changed since inspection")] + SourceChanged, + #[error("the destination changed since inspection; nothing was overwritten")] + DestinationChanged, + #[error("adoption evidence conflicts or needs recovery: {0}")] + EvidenceConflict(String), +} + +fn invalid(detail: impl Into) -> AdoptionError { + AdoptionError::Invalid(detail.into()) +} + +fn digest(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + +fn read_bounded(path: &Path) -> Result, AdoptionError> { + let metadata = fs::symlink_metadata(path)?; + if !metadata.is_file() || metadata.file_type().is_symlink() { + return Err(invalid("checkpoint/evidence must be a regular file")); + } + if metadata.len() > MAX_CHECKPOINT_BYTES { + return Err(invalid("checkpoint exceeds the offline decoding limit")); + } + let mut bytes = Vec::new(); + File::open(path)? + .take(MAX_CHECKPOINT_BYTES + 1) + .read_to_end(&mut bytes)?; + if bytes.len() as u64 > MAX_CHECKPOINT_BYTES { + return Err(invalid("checkpoint grew beyond the offline decoding limit")); + } + Ok(bytes) +} + +fn summary(bytes: &[u8]) -> Result { + // Decode the existing storage schema without restoring, trimming, adding + // resumed facts, or reserializing the selected memory. + let persisted: PersistedVolatile = serde_json::from_slice(bytes) + .map_err(|error| invalid(format!("volatile snapshot schema: {error}")))?; + let wm = &persisted.wm; + if wm.next_action_seq == 0 { + return Err(invalid("next_action_seq must be positive")); + } + if wm.entries.iter().any(|entry| matches!(&entry.kind, super::super::working_memory::WmKind::Receipt { n } if *n == 0 || *n >= wm.next_action_seq)) { + return Err(invalid("working-memory receipt sequence is outside the recorded counter")); + } + if wm + .last_action + .as_ref() + .is_some_and(|(seq, _)| *seq == 0 || *seq >= wm.next_action_seq) + || wm + .recent_results + .iter() + .any(|(seq, _, _, _, _)| *seq == 0 || *seq >= wm.next_action_seq) + { + return Err(invalid("receipt sequence must precede next_action_seq")); + } + let mut previous = None; + for (seq, _, _, _, _) in &wm.recent_results { + if previous.is_some_and(|old| old >= *seq) { + return Err(invalid("result receipt sequences must strictly increase")); + } + previous = Some(*seq); + } + let utterances = match &persisted.own_speech { + OwnSpeechPersisted::ByRoom(rooms) => { + let mut seen = std::collections::HashSet::new(); + for (room, _) in rooms { + if !seen.insert(*room) { + return Err(invalid("duplicate own-speech room")); + } + } + rooms.iter().map(|(_, speech)| speech.len()).sum() + } + OwnSpeechPersisted::Legacy(speech) => speech.len(), + }; + Ok(CheckpointSummary { + build_sha: wm.build_sha.clone(), + saved_at_ms: wm.saved_at_ms, + next_action_seq: wm.next_action_seq, + entries: wm.entries.len(), + result_receipts: wm.recent_results.len(), + own_utterances: utterances, + unscoped_own_utterances: match &persisted.own_speech { + OwnSpeechPersisted::ByRoom(_) => 0, + OwnSpeechPersisted::Legacy(speech) => speech.len(), + }, + }) +} + +fn selection(path: &Path) -> Result<(CheckpointSelection, Vec), AdoptionError> { + let bytes = read_bounded(path)?; + let selected = CheckpointSelection { + path: path.to_path_buf(), + sha256: digest(&bytes), + bytes: bytes.len() as u64, + summary: summary(&bytes)?, + }; + Ok((selected, bytes)) +} + +fn optional_selection(path: &Path) -> Result, AdoptionError> { + match selection(path) { + Ok((selected, _)) => Ok(Some(selected)), + Err(AdoptionError::Io(error)) if error.kind() == io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error), + } +} + +fn resolved_path(path: &Path) -> Result { + match fs::symlink_metadata(path) { + Ok(metadata) => { + if metadata.file_type().is_symlink() { + return Err(invalid( + "symbolic checkpoint/evidence paths are not accepted", + )); + } + Ok(fs::canonicalize(path)?) + } + Err(error) if error.kind() == io::ErrorKind::NotFound => { + let parent = path.parent().ok_or_else(|| invalid("path has no parent"))?; + let name = path + .file_name() + .ok_or_else(|| invalid("path has no filename"))?; + Ok(resolved_path(parent)?.join(name)) + } + Err(error) => Err(error.into()), + } +} + +fn source_path(path: &Path, persona: Uuid) -> Result { + if persona.is_nil() { + return Err(invalid("an explicit non-nil persona UUID is required")); + } + let path = resolved_path(&std::path::absolute(path)?)?; + if path.file_name().and_then(|s| s.to_str()) != Some("volatile.json") + || path + .parent() + .and_then(Path::file_name) + .and_then(|s| s.to_str()) + .and_then(|s| Uuid::parse_str(s).ok()) + != Some(persona) + { + return Err(invalid( + "source must be /volatile.json", + )); + } + Ok(path) +} + +/// Inspect only: no locks/files/directories are created and no winner is chosen +/// by sequence or timestamp. A missing destination is an explicit precondition. +/// Both the selected source and an existing destination must decode as the +/// current checkpoint schema. Recovery OVER a corrupt/incompatible destination +/// is deliberately unsupported here: its original bytes remain untouched and +/// the caller receives the decoding error, never an implicit overwrite policy. +pub fn inspect(source: &Path, persona_id: Uuid) -> Result { + let source = source_path(source, persona_id)?; + let destination_path = resolved_path(&std::path::absolute(volatile_path(persona_id)?)?)?; + if source == destination_path { + return Err(invalid("source and destination must be distinct")); + } + let (source, _) = selection(&source)?; + let prior_destination = optional_selection(&destination_path)?; + Ok(AdoptionPlan { + format_version: 1, + declared_persona_id: persona_id, + source, + destination_path, + prior_destination, + }) +} + +fn exists(path: &Path) -> io::Result { + match fs::symlink_metadata(path) { + Ok(_) => Ok(true), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false), + Err(error) => Err(error), + } +} + +/// Same file lock used by save, load and adoption. Callers keep the returned +/// handle alive until their complete file operation ends. Readers wait rather +/// than interpreting another owner's work as an absent checkpoint. +pub(super) fn lock_checkpoint(path: &Path, wait: bool) -> io::Result { + let parent = path + .parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "checkpoint has no parent"))?; + fs::create_dir_all(parent)?; + let lock_path = parent.join(".volatile.lock"); + match fs::symlink_metadata(&lock_path) { + Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "checkpoint lock is not a regular file", + )); + } + Ok(_) => {} + Err(error) if error.kind() == io::ErrorKind::NotFound => {} + Err(error) => return Err(error), + } + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(lock_path)?; + if wait { + FileExt::lock_exclusive(&file)?; + } else { + FileExt::try_lock_exclusive(&file)?; + } + Ok(file) +} + +pub(super) fn sync_parent(path: &Path) -> io::Result<()> { + // POSIX directory durability. Windows file contents are flushed below; + // this receipt does not promise power-loss directory-journal guarantees. + #[cfg(unix)] + File::open( + path.parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "no parent"))?, + )? + .sync_all()?; + #[cfg(not(unix))] + let _ = path; + Ok(()) +} + +fn preserve(path: &Path, bytes: &[u8]) -> Result<(), AdoptionError> { + if exists(path)? { + return if read_bounded(path)? == bytes { + // A previous publication may have reached its final name before + // parent sync failed; equality alone does not complete that step. + sync_parent(path)?; + Ok(()) + } else { + Err(AdoptionError::EvidenceConflict(path.display().to_string())) + }; + } + let parent = path + .parent() + .ok_or_else(|| invalid("evidence has no parent"))?; + let temporary = parent.join(format!(".preserve-{}.tmp", Uuid::new_v4())); + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary)?; + file.write_all(bytes)?; + file.sync_all()?; + drop(file); + // Publish a complete inode without overwriting existing evidence. Unlike a + // direct create+write, an interrupted write never occupies the final name. + // Both paths are on the same filesystem. Unsupported hard links are an IO + // refusal, not a fallback to a partially visible or destructive write. + let published = fs::hard_link(&temporary, path); + match published { + Ok(()) => {} + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { + if read_bounded(path)? != bytes { + return Err(AdoptionError::EvidenceConflict(path.display().to_string())); + } + } + Err(error) => return Err(error.into()), + } + fs::remove_file(&temporary)?; + sync_parent(path)?; + Ok(()) +} + +/// Adopt the selected bytes under the checkpoint owner's exclusion. The +/// callback MUST reject any running core, including unresponsive/legacy cores; +/// it runs after acquiring the lock and immediately before publication. This +/// condition is essential because legacy binaries do not honor the new lock. +pub fn adopt( + plan: &AdoptionPlan, + mut ensure_offline: impl FnMut() -> io::Result<()>, +) -> Result { + if plan.format_version != 1 { + return Err(invalid("unsupported adoption plan version")); + } + ensure_offline()?; + let destination = resolved_path(&std::path::absolute(volatile_path( + plan.declared_persona_id, + )?)?)?; + if destination != plan.destination_path + || source_path(&plan.source.path, plan.declared_persona_id)? != plan.source.path + { + return Err(invalid( + "plan paths no longer match declared identity and native destination", + )); + } + let _lock = lock_checkpoint(&destination, false)?; + ensure_offline()?; + let encoded_plan = serde_json::to_vec(plan).map_err(|error| invalid(error.to_string()))?; + let parent = destination + .parent() + .ok_or_else(|| invalid("destination has no parent"))?; + let evidence = parent + .join(".checkpoint-adoptions") + .join(digest(&encoded_plan)); + if resolved_path(&evidence)? != evidence { + return Err(invalid("evidence path is redirected")); + } + let archive = evidence.join("source.json"); + let prior = evidence.join("prior.json"); + let intent = evidence.join("plan.json"); + let receipt_path = evidence.join("receipt.json"); + let mut receipt = AdoptionReceipt { + plan: plan.clone(), + assurance: AdoptionAssurance::SelectedLegacySnapshotWithDeclaredPersonaBinding, + legacy_final_flush_acknowledged: false, + source_archive: archive.clone(), + prior_destination_archive: plan.prior_destination.as_ref().map(|_| prior.clone()), + receipt_path: receipt_path.clone(), + already_applied: false, + }; + + let (current_source, source_bytes) = selection(&plan.source.path)?; + if current_source != plan.source { + return Err(AdoptionError::SourceChanged); + } + let current_destination = optional_selection(&destination)?; + // A receipt/intent plus matching archives proves this request's prior + // attempt. Matching destination content alone is not adoption evidence. + let receipt_exists = exists(&receipt_path)?; + if receipt_exists || exists(&intent)? { + if read_bounded(&intent)? != encoded_plan || read_bounded(&archive)? != source_bytes { + return Err(AdoptionError::EvidenceConflict( + "intent/source archive mismatch".into(), + )); + } + if let Some(expected) = &plan.prior_destination { + let bytes = read_bounded(&prior)?; + if digest(&bytes) != expected.sha256 { + return Err(AdoptionError::EvidenceConflict( + "prior destination archive mismatch".into(), + )); + } + } + if current_destination + .as_ref() + .is_some_and(|current| current.sha256 == plan.source.sha256) + { + // Complete a previous rename's directory durability step before + // acknowledging recovery, even if its first attempt failed here. + sync_parent(&destination)?; + let encoded_receipt = + serde_json::to_vec(&receipt).map_err(|error| invalid(error.to_string()))?; + preserve(&receipt_path, &encoded_receipt)?; + receipt.already_applied = true; + return Ok(receipt); + } + if receipt_exists { + return Err(AdoptionError::DestinationChanged); + } + } + if current_destination != plan.prior_destination { + return Err(AdoptionError::DestinationChanged); + } + fs::create_dir_all(&evidence)?; + // Archives before intent: a visible plan always has complete preserved + // inputs. Partial files are a refusal on retry, never silently overwritten. + preserve(&archive, &source_bytes)?; + if let Some(expected) = &plan.prior_destination { + let (current, bytes) = selection(&destination)?; + if ¤t != expected { + return Err(AdoptionError::DestinationChanged); + } + preserve(&prior, &bytes)?; + } + preserve(&intent, &encoded_plan)?; + // Separate inode from immutable evidence, so future in-place writes to the + // live file cannot alter the source archive. + // One staged payload per explicit plan: repeated offline/precondition + // refusals reuse its verified bytes instead of accumulating full copies. + let staged = evidence.join("staged.json"); + preserve(&staged, &source_bytes)?; + ensure_offline()?; + if selection(&plan.source.path)?.0 != plan.source { + return Err(AdoptionError::SourceChanged); + } + if optional_selection(&destination)? != plan.prior_destination { + return Err(AdoptionError::DestinationChanged); + } + fs::rename(&staged, &destination)?; + sync_parent(&destination)?; + let published = optional_selection(&destination)?; + if !published + .as_ref() + .is_some_and(|value| value.sha256 == plan.source.sha256) + { + return Err(AdoptionError::EvidenceConflict( + "destination changed after publication".into(), + )); + } + let encoded_receipt = + serde_json::to_vec(&receipt).map_err(|error| invalid(error.to_string()))?; + preserve(&receipt_path, &encoded_receipt)?; + Ok(receipt) +} diff --git a/core/continuum-core/src/cognition/should_respond_module.rs b/core/continuum-core/src/cognition/should_respond_module.rs index 6dd5b4c906..e70ba77f3d 100644 --- a/core/continuum-core/src/cognition/should_respond_module.rs +++ b/core/continuum-core/src/cognition/should_respond_module.rs @@ -94,10 +94,11 @@ impl ServiceModule for ShouldRespondModule { // Run the persona's continuous mind over the burst. The decision // is the OUTPUT of cognition; `None` (nothing won attention // strongly enough to externalize) is effective silence = Pass. - let room = crate::identity::ActivityRoom::from_uuid(p.room_id) - .map_err(|_| { - format!("{SHOULD_RESPOND_COMMAND}: room_id must be a real (non-nil) room (#425)") - })?; + let room = crate::identity::ActivityRoom::from_uuid(p.room_id).map_err(|_| { + format!( + "{SHOULD_RESPOND_COMMAND}: room_id must be a real (non-nil) room (#425)" + ) + })?; let workspace = cycle .run_framed( crate::cognition::workspace::Burst::raw_in(room, p.burst), @@ -163,22 +164,24 @@ mod tests { fn registry_with_ivar(persona: Uuid) -> Arc { let registry = Arc::new(PersonaWorkspaceRegistry::new()); - registry.get_or_build(PersonaBrainConfig { - persona_id: persona, - persona_name: "Ivar".to_string(), - system_prompt: "You are Ivar, an engineer on the grid.".to_string(), - admission: seed_admission(1_000_000_000), - adapter: Arc::new(HeuristicInferenceAdapter::new()), - capacity: None, - grounding_sources: Vec::new(), - embedder: None, - tool_executor: None, - context_window: crate::cognition::serving_plan::MIN_SERVE_CTX, - // Harness: synchronous perception (deferral is a live-path concern). - defer_recall: false, - defer_grounding: false, - suppress_recall: false, - }); + registry + .get_or_build(PersonaBrainConfig { + persona_id: persona, + persona_name: "Ivar".to_string(), + system_prompt: "You are Ivar, an engineer on the grid.".to_string(), + admission: seed_admission(1_000_000_000), + adapter: Arc::new(HeuristicInferenceAdapter::new()), + capacity: None, + grounding_sources: Vec::new(), + embedder: None, + tool_executor: None, + context_window: crate::cognition::serving_plan::MIN_SERVE_CTX, + // Harness: synchronous perception (deferral is a live-path concern). + defer_recall: false, + defer_grounding: false, + suppress_recall: false, + }) + .expect("test: resident checkpoint is readable"); registry } diff --git a/core/continuum-core/src/ipc/vitals_emitter.rs b/core/continuum-core/src/ipc/vitals_emitter.rs index 4dbd8896b2..fe1070bcf9 100644 --- a/core/continuum-core/src/ipc/vitals_emitter.rs +++ b/core/continuum-core/src/ipc/vitals_emitter.rs @@ -102,7 +102,12 @@ pub fn record_focus(persona: Uuid) { /// Level for a faculty pulse: full while fresh, fading linearly to 0 over the /// window; `None` when nothing fired within it (an honest "awaiting"). -fn faculty_level(persona: Uuid, axis: &'static str, window: Duration, full_scale: u64) -> Option { +fn faculty_level( + persona: Uuid, + axis: &'static str, + window: Duration, + full_scale: u64, +) -> Option { let pulse = FACULTY_PULSE.lock().unwrap_or_else(|e| e.into_inner()); let (n, at) = pulse.get(&(persona, axis))?; let age = at.elapsed(); @@ -244,8 +249,7 @@ pub(crate) fn sample_vitals( let mut vitals = BTreeMap::new(); vitals.insert( "activity".to_string(), - pct_u64(delta, ACT_FULL_SCALE_TICKS) - .max(pct_u64(act_pulse, ACT_PULSE_FULL_SCALE)), + pct_u64(delta, ACT_FULL_SCALE_TICKS).max(pct_u64(act_pulse, ACT_PULSE_FULL_SCALE)), ); vitals.insert( "queue".to_string(), @@ -433,12 +437,21 @@ mod tests { let p = Uuid::from_u128(0x77); assert!(faculty_level(p, "reason", Duration::from_secs(60), 1).is_none()); record_reasoning(p); - assert_eq!(faculty_level(p, "reason", Duration::from_secs(60), 1), Some(100)); + assert_eq!( + faculty_level(p, "reason", Duration::from_secs(60), 1), + Some(100) + ); record_recall(p, 3); - assert_eq!(faculty_level(p, "recall", Duration::from_secs(60), 6), Some(50)); + assert_eq!( + faculty_level(p, "recall", Duration::from_secs(60), 6), + Some(50) + ); assert!(faculty_level(p, "recall", Duration::from_millis(0), 6).is_none()); record_focus(p); - assert_eq!(faculty_level(p, "focus", Duration::from_secs(60), 1), Some(100)); + assert_eq!( + faculty_level(p, "focus", Duration::from_secs(60), 1), + Some(100) + ); } use super::*; @@ -452,23 +465,25 @@ mod tests { /// from `identity.peer_id.as_uuid()`. fn registry_with(peer_id: Uuid) -> std::sync::Arc { let registry = std::sync::Arc::new(PersonaWorkspaceRegistry::new()); - registry.get_or_build(PersonaBrainConfig { - persona_id: peer_id, - persona_name: "Asha".to_string(), - system_prompt: "You are Asha.".to_string(), - admission: std::sync::Arc::new(AdmissionState::new(std::sync::Arc::new( - RecallMetadataRegistry::new(), - ))), - adapter: std::sync::Arc::new(HeuristicInferenceAdapter::new()), - capacity: None, - grounding_sources: Vec::new(), - embedder: None, - tool_executor: None, - context_window: crate::cognition::serving_plan::MIN_SERVE_CTX, - defer_recall: false, - defer_grounding: false, - suppress_recall: false, - }); + registry + .get_or_build(PersonaBrainConfig { + persona_id: peer_id, + persona_name: "Asha".to_string(), + system_prompt: "You are Asha.".to_string(), + admission: std::sync::Arc::new(AdmissionState::new(std::sync::Arc::new( + RecallMetadataRegistry::new(), + ))), + adapter: std::sync::Arc::new(HeuristicInferenceAdapter::new()), + capacity: None, + grounding_sources: Vec::new(), + embedder: None, + tool_executor: None, + context_window: crate::cognition::serving_plan::MIN_SERVE_CTX, + defer_recall: false, + defer_grounding: false, + suppress_recall: false, + }) + .expect("test: resident checkpoint is readable"); registry } diff --git a/core/continuum-core/src/persona/host.rs b/core/continuum-core/src/persona/host.rs index 3c814efb14..bdecdfa484 100644 --- a/core/continuum-core/src/persona/host.rs +++ b/core/continuum-core/src/persona/host.rs @@ -397,8 +397,9 @@ impl PersonaSpawnSupervisor { /// is edge-triggered, keyed by `until_ms` so a renewed hold announces /// itself afresh. fn probe_held_out_once(agent: &str, hold: &crate::persona::roster_hold::RosterHold) { - static PROBED: std::sync::OnceLock>> = - std::sync::OnceLock::new(); + static PROBED: std::sync::OnceLock< + std::sync::Mutex>, + > = std::sync::OnceLock::new(); let set = PROBED.get_or_init(|| std::sync::Mutex::new(std::collections::HashSet::new())); let Ok(mut guard) = set.lock() else { return; // poisoned = a prior panic mid-insert; skip the probe, never the filter @@ -511,12 +512,16 @@ impl PersonaSpawnSupervisor { let plans: Vec = unattended .iter() .zip(profiles) - .map(|(rt, profile)| crate::persona::spawner_module::MaterializedPersonaPlan { - role: desired.role, - instance: - crate::modules::persona_instance_manager::PersonaInstanceInfo::from_runtime(rt), - profile, - }) + .map( + |(rt, profile)| crate::persona::spawner_module::MaterializedPersonaPlan { + role: desired.role, + instance: + crate::modules::persona_instance_manager::PersonaInstanceInfo::from_runtime( + rt, + ), + profile, + }, + ) .collect(); // The reconciler's other entrance passes the SAME operator-intent @@ -705,6 +710,9 @@ fn supervisor_error_facts(err: &SupervisorError) -> (Option, RoleId) { | SupervisorError::AdapterWarmup { slot_index, role, .. } + | SupervisorError::WorkspaceRegistration { + slot_index, role, .. + } | SupervisorError::RuntimeMissing { slot_index, role, .. } => (Some(*slot_index), *role), @@ -775,7 +783,9 @@ mod tests { &self, _profile: &crate::persona::inference_profile::PersonaInferenceProfile, ) -> Result, String> { - panic!("factory must not be consulted when the registry has no unattended citizens"); + panic!( + "factory must not be consulted when the registry has no unattended citizens" + ); } } diff --git a/core/continuum-core/src/persona/service_loop.rs b/core/continuum-core/src/persona/service_loop.rs index 778829c5e2..b8dccc3d8c 100644 --- a/core/continuum-core/src/persona/service_loop.rs +++ b/core/continuum-core/src/persona/service_loop.rs @@ -5048,7 +5048,9 @@ mod tests { defer_grounding: false, suppress_recall: false, }; - crate::cognition::persona_workspace::global().register_from_cfg(cfg); + crate::cognition::persona_workspace::global() + .register_from_cfg(cfg) + .expect("test: resident checkpoint is readable"); // One held (Claimed) card in her hands. A NON-bench title so the // act-question resolves no staged checkout (no hands re-root needed). diff --git a/core/continuum-core/src/persona/supervisor.rs b/core/continuum-core/src/persona/supervisor.rs index 73b45ab064..4a35626111 100644 --- a/core/continuum-core/src/persona/supervisor.rs +++ b/core/continuum-core/src/persona/supervisor.rs @@ -453,6 +453,16 @@ pub enum SupervisorError { role: RoleId, message: String, }, + /// A resident whose checkpoint cannot be loaded must not be hosted with + /// an empty memory. The existing resident, if any, remains registered. + #[error("slot {slot_index} (role {role:?}): workspace registration for {persona_id} failed: {source}")] + WorkspaceRegistration { + slot_index: usize, + role: RoleId, + persona_id: uuid::Uuid, + #[source] + source: std::io::Error, + }, /// The post-bootstrap registry doesn't have a runtime for this /// persona_id. Per [[no-fallbacks-ever]] this is a hard failure — /// the supervisor doesn't fabricate or stub a runtime in @@ -639,7 +649,8 @@ pub async fn materialize_adapters( profile.context_length = floor; } } - } else if profile.tier_category != crate::persona::hw_tier_descriptor::HwTierCategory::Cloud { + } else if profile.tier_category != crate::persona::hw_tier_descriptor::HwTierCategory::Cloud + { let snap = crate::inference::llama_server::current_serving(); // A ready snapshot always carries a real window (the daemon refuses to // publish ready with 0). Guard on both so a not-yet-ready/empty @@ -962,134 +973,153 @@ pub async fn materialize_adapters( // a restart. Build + register replaces it. The retained cfg template is // what lets `cognition/eval` fork an ephemeral measurement copy without // touching this living mind (PersonaWorkspaceRegistry::fork_eval_cycle). - crate::cognition::persona_workspace::global().register_from_cfg( - crate::cognition::persona_workspace::PersonaBrainConfig { - persona_id: identity.peer_id.as_uuid(), - persona_name: identity.agent_name.to_string(), - system_prompt: system_prompt.to_string(), - admission: cognition.admission.clone(), - adapter: adapter.clone(), - capacity: None, - // Neural recall when the embed model serves, lexical otherwise - // — decided once here (process-stable; query + stored vectors - // must share one embedding space). Already cached by the - // resolver (embed-once-per-content, shared across personas). - embedder: Some( - crate::cognition::embedding::resolve_recall_embedder(adapter.clone()).await, - ), - // Roster + doctrine bridged into the brain as STANDING-FRAMING - // grounding faculties (high salience floor). Without these the - // gating cutover routes decisions through the Workspace and the - // #1650/#1651 grounding silently falls out of the live path — - // the persona forgets who is present / what the room is for. - grounding_sources: vec![ - // Roster — WHO is present. SYNCHRONOUS (ColdStartCritical), - // like workspace-map and for the same reason: the deferral - // was earned by the OLD airc-fetch reader this source - // replaced; the ViewState roster is a watch-channel borrow - // (microseconds), and keeping the fallback meant slow turns - // "timed out" into `reproject_to_now` serving a CACHED - // roster from a different presence moment — Benchy's prompt - // flapped between fresh and stale byte layouts at 0.4% - // depth, hit_rate 0.0, and slower turns caused MORE - // reprojections (measured 2026-09-01: both his captures - // carried "[reprojected … held]"). A source this cheap - // never gets a fallback ([[fallbacks-are-illegal-fail-loud]]). - crate::cognition::persona_workspace::GroundingSource::framing(roster_source), - // Doctrine — WHAT the room is for: the PARTICIPATION GATE. This - // one stays SYNCHRONOUS (ColdStartCritical): a cold-start `None` - // would let the persona speak in a room it shouldn't on turn one, - // which is wrong, not merely unenriched. The lone exception to - // "defer almost everything." - crate::cognition::persona_workspace::GroundingSource::framing(doctrine_source), - // The persona's own live work across rooms — enriching framing so - // it knows what it's working on (cross-activity, dynamic, no - // hardcoded card state). Defer-tolerant. - crate::cognition::persona_workspace::GroundingSource::framing( - active_work_source, - ) + let brain_cfg = crate::cognition::persona_workspace::PersonaBrainConfig { + persona_id: identity.peer_id.as_uuid(), + persona_name: identity.agent_name.to_string(), + system_prompt: system_prompt.to_string(), + admission: cognition.admission.clone(), + adapter: adapter.clone(), + capacity: None, + // Neural recall when the embed model serves, lexical otherwise + // — decided once here (process-stable; query + stored vectors + // must share one embedding space). Already cached by the + // resolver (embed-once-per-content, shared across personas). + embedder: Some( + crate::cognition::embedding::resolve_recall_embedder(adapter.clone()).await, + ), + // Roster + doctrine bridged into the brain as STANDING-FRAMING + // grounding faculties (high salience floor). Without these the + // gating cutover routes decisions through the Workspace and the + // #1650/#1651 grounding silently falls out of the live path — + // the persona forgets who is present / what the room is for. + grounding_sources: vec![ + // Roster — WHO is present. SYNCHRONOUS (ColdStartCritical), + // like workspace-map and for the same reason: the deferral + // was earned by the OLD airc-fetch reader this source + // replaced; the ViewState roster is a watch-channel borrow + // (microseconds), and keeping the fallback meant slow turns + // "timed out" into `reproject_to_now` serving a CACHED + // roster from a different presence moment — Benchy's prompt + // flapped between fresh and stale byte layouts at 0.4% + // depth, hit_rate 0.0, and slower turns caused MORE + // reprojections (measured 2026-09-01: both his captures + // carried "[reprojected … held]"). A source this cheap + // never gets a fallback ([[fallbacks-are-illegal-fail-loud]]). + crate::cognition::persona_workspace::GroundingSource::framing(roster_source), + // Doctrine — WHAT the room is for: the PARTICIPATION GATE. This + // one stays SYNCHRONOUS (ColdStartCritical): a cold-start `None` + // would let the persona speak in a room it shouldn't on turn one, + // which is wrong, not merely unenriched. The lone exception to + // "defer almost everything." + crate::cognition::persona_workspace::GroundingSource::framing(doctrine_source), + // The persona's own live work across rooms — enriching framing so + // it knows what it's working on (cross-activity, dynamic, no + // hardcoded card state). Defer-tolerant. + crate::cognition::persona_workspace::GroundingSource::framing(active_work_source) .defer_tolerant() // Claim states flap per turn — floor stays, stable-tier // placement goes (debug/prompt-reuse conviction, 2026-08-22). .volatile_content(), - // WHERE code lives — the real workspace layout as framing, so a - // reasoner can avoid blind globs like `src/**/*.rs` from the - // prompt alone. ColdStartCritical (synchronous, NOT deferred): - // measured 2026-07-13 that the deferred version was ABSENT on - // cold ticks (worst under repeated reboots), so some personas - // acted blind to the layout — a WRONG turn (blind-glob loops), - // not merely unenriched, which is exactly the ColdStartCritical - // bar. It's a cheap local dir listing (unlike the airc-backed - // framing sources that stay deferred), so it earns synchronous - // presence like doctrine. requires_hands: the block SAYS "drill - // in with code/list and code/tree" — it must vanish from a - // tool-stripped cycle (spoken exams) or the RAG lies about her - // affordances. - crate::cognition::persona_workspace::GroundingSource::framing( - workspace_map_source, - ) + // WHERE code lives — the real workspace layout as framing, so a + // reasoner can avoid blind globs like `src/**/*.rs` from the + // prompt alone. ColdStartCritical (synchronous, NOT deferred): + // measured 2026-07-13 that the deferred version was ABSENT on + // cold ticks (worst under repeated reboots), so some personas + // acted blind to the layout — a WRONG turn (blind-glob loops), + // not merely unenriched, which is exactly the ColdStartCritical + // bar. It's a cheap local dir listing (unlike the airc-backed + // framing sources that stay deferred), so it earns synchronous + // presence like doctrine. requires_hands: the block SAYS "drill + // in with code/list and code/tree" — it must vanish from a + // tool-stripped cycle (spoken exams) or the RAG lies about her + // affordances. + crate::cognition::persona_workspace::GroundingSource::framing(workspace_map_source) .requires_hands() // Her OWN writes mutate the map — every productive act. It sat // in the stable tier breaking the KV prefix at ~8k chars of a // 40k prompt (measured 2026-08-23, turn-over-turn capture diff). .volatile_content(), - // The room's pinned shared documents (airc wall) as - // enriching framing — the plan/instructions/recipe that - // shape HOW the persona works here, read from the exact - // rows a teammate or widget pins. Defer-tolerant: a - // first-tick miss costs one under-grounded turn, not a - // wrong one. - crate::cognition::persona_workspace::GroundingSource::framing(wall_source) - .defer_tolerant(), - // The room's WHOLE work board (airc kanban) as enriching - // framing — every card/column/owner, so the persona can - // coordinate against the shared plan, not just its own - // claims. Defer-tolerant: a first-tick miss costs one - // under-grounded turn, not a wrong one. Task #117 O6. - crate::cognition::persona_workspace::GroundingSource::framing( - room_board_source, - ) + // The room's pinned shared documents (airc wall) as + // enriching framing — the plan/instructions/recipe that + // shape HOW the persona works here, read from the exact + // rows a teammate or widget pins. Defer-tolerant: a + // first-tick miss costs one under-grounded turn, not a + // wrong one. + crate::cognition::persona_workspace::GroundingSource::framing(wall_source) + .defer_tolerant(), + // The room's WHOLE work board (airc kanban) as enriching + // framing — every card/column/owner, so the persona can + // coordinate against the shared plan, not just its own + // claims. Defer-tolerant: a first-tick miss costs one + // under-grounded turn, not a wrong one. Task #117 O6. + crate::cognition::persona_workspace::GroundingSource::framing(room_board_source) .defer_tolerant() // Card/column states churn with the round — same conviction. .volatile_content(), - // Live-call perception (#187/#192): WHO is visible on the call + - // what they show, as enriching framing. Defer-tolerant: a - // first-tick miss costs one under-grounded turn, not a wrong one — - // and perception is non-blocking by construction (absent cells are - // simply not present this tick, never awaited). NOT requires_hands: - // seeing is a SENSE, not a tool, so it stays present in a - // tool-stripped (spoken-exam) cycle. Reads only ready cells (O(participants) - // string assembly, no inference) — off the 30fps media plane entirely. - crate::cognition::persona_workspace::GroundingSource::framing( - media_perception_source, - ) - .defer_tolerant(), - ], - // The persona's HANDS — built by the caller for THIS persona's - // identity (None → speak-only). What turns "talks" into "acts". - tool_executor, - // The window the gateway actually serves this persona (task #50: - // single-sourced; Local → ServingPlan.served_context_window). The - // deliberation faculty keeps its prompt inside it so llama-server - // never 500s ("Context size has been exceeded"). - context_window: profile.context_length, - // LIVE mind: recall runs as a speculative prefetch off the hot - // path (Joel's CPU branch-prediction analogy). Turns here are - // seconds apart, so the background worker always catches up and - // the per-turn output reads a warm last-good instead of waiting on - // a neural-embed + vector-search round-trip. Eval forks override - // this to false (faithful synchronous measurement). - defer_recall: true, - // LIVE mind: push the defer-tolerant grounding (roster, active_work, - // workspace_map) off the hot path too — the 90%-async win for the - // enriching framing. Doctrine (ColdStartCritical) stays synchronous - // regardless. Eval/harness override to false. - defer_grounding: true, - // The LIVING persona always keeps her memories — suppression is a - // benchmark-reproducibility knob, never a life-path setting (#207). - suppress_recall: false, - }, - ); + // Live-call perception (#187/#192): WHO is visible on the call + + // what they show, as enriching framing. Defer-tolerant: a + // first-tick miss costs one under-grounded turn, not a wrong one — + // and perception is non-blocking by construction (absent cells are + // simply not present this tick, never awaited). NOT requires_hands: + // seeing is a SENSE, not a tool, so it stays present in a + // tool-stripped (spoken-exam) cycle. Reads only ready cells (O(participants) + // string assembly, no inference) — off the 30fps media plane entirely. + crate::cognition::persona_workspace::GroundingSource::framing( + media_perception_source, + ) + .defer_tolerant(), + ], + // The persona's HANDS — built by the caller for THIS persona's + // identity (None → speak-only). What turns "talks" into "acts". + tool_executor, + // The window the gateway actually serves this persona (task #50: + // single-sourced; Local → ServingPlan.served_context_window). The + // deliberation faculty keeps its prompt inside it so llama-server + // never 500s ("Context size has been exceeded"). + context_window: profile.context_length, + // LIVE mind: recall runs as a speculative prefetch off the hot + // path (Joel's CPU branch-prediction analogy). Turns here are + // seconds apart, so the background worker always catches up and + // the per-turn output reads a warm last-good instead of waiting on + // a neural-embed + vector-search round-trip. Eval forks override + // this to false (faithful synchronous measurement). + defer_recall: true, + // LIVE mind: push the defer-tolerant grounding (roster, active_work, + // workspace_map) off the hot path too — the 90%-async win for the + // enriching framing. Doctrine (ColdStartCritical) stays synchronous + // regardless. Eval/harness override to false. + defer_grounding: true, + // The LIVING persona always keeps her memories — suppression is a + // benchmark-reproducibility knob, never a life-path setting (#207). + suppress_recall: false, + }; + // A checkpoint can be locked by offline adoption. Waiting and reading + // belong off the async runtime workers; registration publishes nothing + // until the existing snapshot has loaded successfully. + #[cfg(test)] + let fixture_home = crate::paths::home_dir(); + let registration = tokio::task::spawn_blocking(move || { + #[cfg(test)] + let _fixture_home = fixture_home + .as_deref() + .map(crate::paths::NativeHomeOverride::install); + crate::cognition::persona_workspace::global().register_from_cfg(brain_cfg) + }) + .await; + let failure = match registration { + Ok(Ok(_)) => None, + Ok(Err(error)) => Some(error), + Err(error) => Some(std::io::Error::other(error)), + }; + if let Some(source) = failure { + out.push(Err(SupervisorError::WorkspaceRegistration { + slot_index, + role: plan.role, + persona_id: identity.peer_id.as_uuid(), + source, + })); + continue; + } out.push(Ok(PersonaContext { role: plan.role, @@ -1324,29 +1354,48 @@ mod tests { let mut with_window = fake_instance("Rin"); with_window.home = homes.path().join("rin"); std::fs::create_dir_all(&with_window.home).expect("home dir"); - crate::persona::model_override::PersonaModelOverride::new_remote("model-a", None, 1, "peer") - .with_context_window(24_832) - .write(&crate::persona::home::PersonaHome::from_root(with_window.home.clone())) - .expect("write override"); + crate::persona::model_override::PersonaModelOverride::new_remote( + "model-a", None, 1, "peer", + ) + .with_context_window(24_832) + .write(&crate::persona::home::PersonaHome::from_root( + with_window.home.clone(), + )) + .expect("write override"); let mut without_window = fake_instance("Sol"); without_window.home = homes.path().join("sol"); std::fs::create_dir_all(&without_window.home).expect("home dir"); - crate::persona::model_override::PersonaModelOverride::new_remote("model-b", None, 1, "peer") - .write(&crate::persona::home::PersonaHome::from_root(without_window.home.clone())) - .expect("write override"); + crate::persona::model_override::PersonaModelOverride::new_remote( + "model-b", None, 1, "peer", + ) + .write(&crate::persona::home::PersonaHome::from_root( + without_window.home.clone(), + )) + .expect("write override"); let mut rin = fake_profile("Rin", "model-a"); rin.context_length = 50_944; let mut sol = fake_profile("Sol", "model-b"); sol.context_length = 50_944; let plans = vec![ - MaterializedPersonaPlan { role: RoleId::Helper, instance: with_window, profile: Ok(rin) }, - MaterializedPersonaPlan { role: RoleId::Coder, instance: without_window, profile: Ok(sol) }, + MaterializedPersonaPlan { + role: RoleId::Helper, + instance: with_window, + profile: Ok(rin), + }, + MaterializedPersonaPlan { + role: RoleId::Coder, + instance: without_window, + profile: Ok(sol), + }, ]; let factory = ScriptedPersonaAdapterFactory::heuristic(); let hosted = materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; let rin = hosted[0].as_ref().expect("Rin hosted"); - assert_eq!(rin.profile.context_length, 24_832, "the recorded responder window wins"); + assert_eq!( + rin.profile.context_length, 24_832, + "the recorded responder window wins" + ); let sol = hosted[1].as_ref().expect("Sol hosted"); assert_eq!( sol.profile.context_length, @@ -1583,6 +1632,65 @@ mod tests { ); } + // 9f160b78: actual boot materialization reports a checkpoint failure for + // that slot, never hosts it blank, and still registers a healthy sibling. + #[tokio::test(flavor = "current_thread")] + async fn checkpoint_failure_refuses_hosting_without_tainting_sibling_slots() { + init_test_registry(); + let home = tempfile::tempdir().unwrap(); + let _native = crate::paths::NativeHomeOverride::install(home.path()); + let mut failed = fake_instance("checkpoint-refused"); + failed.home = home.path().join("refused-persona"); + let failed_id = failed.peer_id.as_uuid(); + let mut healthy = fake_instance("checkpoint-healthy"); + healthy.home = home.path().join("healthy-persona"); + let healthy_id = healthy.peer_id.as_uuid(); + let broken = home + .path() + .join(".continuum/personas") + .join(failed_id.to_string()) + .join("volatile.json"); + std::fs::create_dir_all(broken.parent().unwrap()).unwrap(); + std::fs::write(&broken, b"not a checkpoint").unwrap(); + let plans = vec![ + MaterializedPersonaPlan { + role: RoleId::Helper, + instance: failed, + profile: Ok(fake_profile("checkpoint-refused", "model-a")), + }, + MaterializedPersonaPlan { + role: RoleId::Coder, + instance: healthy, + profile: Ok(fake_profile("checkpoint-healthy", "model-b")), + }, + ]; + let factory = ScriptedPersonaAdapterFactory::heuristic(); + let hosted = + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; + assert_eq!(hosted.len(), 2); + match &hosted[0] { + Err(SupervisorError::WorkspaceRegistration { + slot_index, + role, + persona_id, + source, + }) => { + assert_eq!( + (*slot_index, *role, *persona_id), + (0, RoleId::Helper, failed_id) + ); + assert_eq!(source.kind(), std::io::ErrorKind::InvalidData); + } + Err(other) => panic!("test: expected checkpoint failure, got {other:?}"), + Ok(_) => panic!("test: corrupt checkpoint must not produce a hosted persona"), + } + assert!(hosted[1].is_ok()); + let registry = crate::cognition::persona_workspace::global(); + assert!(registry.get(&failed_id).is_none()); + assert!(registry.get(&healthy_id).is_some()); + assert_eq!(std::fs::read(&broken).unwrap(), b"not a checkpoint"); + } + /// Warmup failure surfaces as `SupervisorError::AdapterWarmup` — /// the persona does NOT reach hosted state. Per [[no-fallbacks-ever]] /// an adapter that refuses to warm gets a typed slot failure; diff --git a/docs/architecture/PERSONA-COGNITION-PIPELINE.md b/docs/architecture/PERSONA-COGNITION-PIPELINE.md index e79bd9515b..edbb81d2da 100644 --- a/docs/architecture/PERSONA-COGNITION-PIPELINE.md +++ b/docs/architecture/PERSONA-COGNITION-PIPELINE.md @@ -27,6 +27,12 @@ Continuum personas are **citizens**, not query handlers. The README has the full **Per-persona means each AI has its own mind.** The cycle runs per-persona. Shared optimizations (the `analyze` single-flight cache) sit underneath, not above. +Resident checkpoint loading must preserve that continuity: an unreadable or +invalid checkpoint refuses registration rather than waking a blank mind. See +[checkpoint recovery](../personas/CHECKPOINT-RECOVERY.md) for the explicit, +digest-bound adoption path when upgrading a legacy core that wrote to a different +data root. A selected legacy snapshot is not an acknowledged final-turn flush. + --- ## 2. The Brain Pipeline — the verbs that exist diff --git a/docs/personas/CHECKPOINT-RECOVERY.md b/docs/personas/CHECKPOINT-RECOVERY.md new file mode 100644 index 0000000000..4e0a3d2657 --- /dev/null +++ b/docs/personas/CHECKPOINT-RECOVERY.md @@ -0,0 +1,73 @@ +# Persona checkpoint recovery + +A resident's working memory and per-room own-speech rings are stored in the +native user data directory at `.continuum/personas//volatile.json`. +Resident registration distinguishes an absent checkpoint from a failed read or +invalid schema. Only absence starts a fresh working memory. A load error refuses +that registration, reports the affected slot, and leaves an existing registered +resident intact; healthy sibling slots can still start. + +Older cores could write this file beneath their working directory when `HOME` +was absent. They also lacked an acknowledged final checkpoint on some shutdown +paths. A newer core cannot retroactively flush those older processes. Their +periodic checkpoint is a recoverable snapshot, not proof of their final turn. + +## Select one legacy snapshot explicitly + +The installed CLI exposes this recovery operation without needing a running +core or a repository checkout: + +```text +continuum checkpoint inspect --source /volatile.json --persona-id --plan +``` + +The source's parent directory must match the declared UUID. The legacy payload +does not embed that UUID or a process-lifetime identity, so this is an explicit +operator declaration, not authenticated provenance. Inspection reports source +and destination paths, exact byte hashes, and snapshot summaries. It changes no +checkpoint and creates the requested plan file exclusively. Put that metadata +file outside both Persona stores, using a local path on Windows. Network plan +paths are refused there because a share can alias the native store through a +different path namespace. Reserved checkpoint filenames cannot name a plan. + +Choose the intended snapshot from its provenance and work history. Sequence +numbers and modification times from different lifetimes do not establish which +mind is newer; recovery never chooses a winner or merges memories by those +numbers. Both selected and existing destination files must decode under the +current storage schema. An incompatible or corrupt destination is preserved and +refused; this command is not a corruption repair tool. + +## Apply while cores are stopped + +Stop the core and any automatic launcher that could restart a legacy binary. +If the source changed since inspection, inspect again into a new plan file. +Then apply the exact plan: + +```text +continuum checkpoint adopt --plan --legacy-writers-stopped +``` + +The flag asserts the operational precondition for old binaries that cannot honor +the new file lock. It does not bypass process checks. The CLI refuses a running +core, a live PID-file process, unreadable process evidence, or an unresolved +truncated core process name. Adoption rechecks this condition under the same +per-persona OS lock used by checkpoint loading and saving, including immediately +before replacement. New cores loading that persona participate in that lock. + +Recovery preserves the exact selected source bytes and previous destination in +`.checkpoint-adoptions//` beside the destination. Complete archives +and an intent record precede replacement; a receipt records the result. The live +file has a separate inode from its archives. Repeating the same plan verifies +the preserved evidence and current bytes, and can finish a receipt interrupted +after replacement. Changed inputs or conflicting evidence cause refusal. + +Files are flushed before publication and parent directories are synchronized on +Unix. Filesystems without the required local hard-link/rename operations are +refused. On Windows, the receipt does not promise directory-journal persistence +across power loss. In every case it explicitly reports +`legacy_final_flush_acknowledged: false`. + +After normal startup, verify the running build and inspect the resident's +restored memory and actual inference requests. Successful adoption proves which +bytes were selected and preserved; it does not prove the last legacy turn was +saved, a teammate message reached an inference request, or learning improved. From 901adbdd1bc8b10430a329de9d623dfdb8669313 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 17:53:33 -0500 Subject: [PATCH 2/6] Refuse resident bootstrap when persistent engram restore fails --- core/continuum-core/src/persona/host.rs | 3 + core/continuum-core/src/persona/supervisor.rs | 143 ++++++++++++++++-- 2 files changed, 136 insertions(+), 10 deletions(-) diff --git a/core/continuum-core/src/persona/host.rs b/core/continuum-core/src/persona/host.rs index bdecdfa484..912d49a8d2 100644 --- a/core/continuum-core/src/persona/host.rs +++ b/core/continuum-core/src/persona/host.rs @@ -713,6 +713,9 @@ fn supervisor_error_facts(err: &SupervisorError) -> (Option, RoleId) { | SupervisorError::WorkspaceRegistration { slot_index, role, .. } + | SupervisorError::AdmissionRestore { + slot_index, role, .. + } | SupervisorError::RuntimeMissing { slot_index, role, .. } => (Some(*slot_index), *role), diff --git a/core/continuum-core/src/persona/supervisor.rs b/core/continuum-core/src/persona/supervisor.rs index 4a35626111..6fdca1e778 100644 --- a/core/continuum-core/src/persona/supervisor.rs +++ b/core/continuum-core/src/persona/supervisor.rs @@ -463,6 +463,18 @@ pub enum SupervisorError { #[source] source: std::io::Error, }, + /// Persistent admission must be loaded before the resident's recall and + /// future admissions can bind to it. A failed load is not an empty store. + #[error( + "slot {slot_index} (role {role:?}): admission restore for {persona_id} failed: {source}" + )] + AdmissionRestore { + slot_index: usize, + role: RoleId, + persona_id: uuid::Uuid, + #[source] + source: crate::orm::OrmStoreError, + }, /// The post-bootstrap registry doesn't have a runtime for this /// persona_id. Per [[no-fallbacks-ever]] this is a hard failure — /// the supervisor doesn't fabricate or stub a runtime in @@ -929,10 +941,11 @@ pub async fn materialize_adapters( // rehydrate prior engrams + recall metadata, so memory SURVIVES restart. // Without this, admission is in-memory only (NoopSink) and the persona is // amnesiac across boots. `identity.home` is the resolved - // /personas/ dir. On disk error we log loud and continue - // in-memory — the persona stays alive; persistence is degraded, not fatal - // (NOT an inference fallback). MUST run before the WorkspaceCycle is - // assembled below, so its RecallFaculty binds the persisted admission. + // /personas/ dir. A storage or schema failure refuses this + // slot before registration: continuing with the fresh NoopSink state + // would hide prior engrams and lose future admissions across restart. + // A new home succeeds with an empty persistent store. MUST run before + // WorkspaceCycle assembly so RecallFaculty binds persisted admission. let home = crate::persona::home::PersonaHome::from_root(identity.home.clone()); let recall_meta = std::sync::Arc::new(crate::persona::recall_metadata::RecallMetadataRegistry::new()); @@ -944,12 +957,14 @@ pub async fn materialize_adapters( std::sync::Arc::new(persisted), ); } - Err(e) => { - tracing::warn!( - persona = %identity.agent_name, - error = %e, - "engram persistence unavailable; running in-memory (memory will NOT survive restart)" - ); + Err(source) => { + out.push(Err(SupervisorError::AdmissionRestore { + slot_index, + role: plan.role, + persona_id: identity.peer_id.as_uuid(), + source, + })); + continue; } } @@ -1632,6 +1647,114 @@ mod tests { ); } + // 6d17695c: a real SQLite initialization failure must not host a blank + // admission store or replace an existing resident. Healthy siblings still + // materialize through the same factory/runtime/registration boundary. + #[tokio::test(flavor = "current_thread")] + async fn admission_restore_failure_refuses_hosting_and_preserves_existing_residents() { + init_test_registry(); + for already_registered in [false, true] { + let home = tempfile::tempdir().unwrap(); + let _native = crate::paths::NativeHomeOverride::install(home.path()); + let factory = ScriptedPersonaAdapterFactory::heuristic(); + let registry = crate::cognition::persona_workspace::global(); + let mut failed = fake_instance("admission-refused"); + failed.home = home.path().join("refused-persona"); + let failed_id = failed.peer_id.as_uuid(); + let previous = if already_registered { + // An already-running life remains valid even if a subsequent + // bootstrap points at storage it cannot open. Its own DB is + // left untouched; only the candidate home is broken below. + let mut previous_identity = failed.clone(); + previous_identity.home = home.path().join("previous-persona"); + let previous_plan = MaterializedPersonaPlan { + role: RoleId::Helper, + instance: previous_identity, + profile: Ok(fake_profile("admission-refused", "model-a")), + }; + let hosted = materialize_adapters( + vec![previous_plan], + &factory, + StubAircCitizen::fresh_lookup(), + |_| None, + ) + .await; + assert!( + hosted[0].is_ok(), + "test: the original resident must be healthy" + ); + Some( + registry + .get(&failed_id) + .expect("test: original resident registered"), + ) + } else { + assert!(registry.get(&failed_id).is_none()); + None + }; + let failed_home = crate::persona::home::PersonaHome::from_root(failed.home.clone()); + failed_home.ensure_exists().unwrap(); + let broken = failed_home.engrams_db(); + let original = b"existing engram database is not valid SQLite"; + std::fs::write(&broken, original).unwrap(); + let mut healthy = fake_instance("admission-healthy"); + healthy.home = home.path().join("healthy-persona"); + let healthy_id = healthy.peer_id.as_uuid(); + let healthy_home = crate::persona::home::PersonaHome::from_root(healthy.home.clone()); + let plans = vec![ + MaterializedPersonaPlan { + role: RoleId::Helper, + instance: failed, + profile: Ok(fake_profile("admission-refused", "model-a")), + }, + MaterializedPersonaPlan { + role: RoleId::Coder, + instance: healthy, + profile: Ok(fake_profile("admission-healthy", "model-b")), + }, + ]; + let hosted = + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None) + .await; + assert_eq!(hosted.len(), 2); + match &hosted[0] { + Err(SupervisorError::AdmissionRestore { + slot_index, + role, + persona_id, + source, + }) => { + assert_eq!( + (*slot_index, *role, *persona_id), + (0, RoleId::Helper, failed_id) + ); + assert!( + matches!(source, crate::orm::OrmStoreError::AdapterFailed { + operation: "initialize", collection, .. + } if collection == "engrams"), + "test: preserve the actual SQLite initialization error: {source}" + ); + } + Err(other) => panic!("test: expected admission restore failure, got {other:?}"), + Ok(_) => { + panic!("test: failed persistent admission must not produce a hosted persona") + } + } + assert!( + hosted[1].is_ok(), + "a new empty home must still initialize persistent admission" + ); + assert!(registry.get(&healthy_id).is_some()); + assert!(healthy_home.engrams_db().is_file()); + if let Some(previous) = previous { + assert!(Arc::ptr_eq(&previous, ®istry.get(&failed_id).unwrap())); + } else { + assert!(registry.get(&failed_id).is_none()); + } + assert_eq!(std::fs::read(&broken).unwrap(), original); + } + } + // 9f160b78: actual boot materialization reports a checkpoint failure for // that slot, never hosts it blank, and still registers a healthy sibling. #[tokio::test(flavor = "current_thread")] From 078c6cdf1c821c6d2d6f5f146b0878736b0e1ac7 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 18:19:32 -0500 Subject: [PATCH 3/6] docs(persona): distinguish engram identity from checkpoint adoption --- docs/personas/CHECKPOINT-RECOVERY.md | 40 ++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/docs/personas/CHECKPOINT-RECOVERY.md b/docs/personas/CHECKPOINT-RECOVERY.md index 4e0a3d2657..3a90ae8c45 100644 --- a/docs/personas/CHECKPOINT-RECOVERY.md +++ b/docs/personas/CHECKPOINT-RECOVERY.md @@ -12,6 +12,46 @@ was absent. They also lacked an acknowledged final checkpoint on some shutdown paths. A newer core cannot retroactively flush those older processes. Their periodic checkpoint is a recoverable snapshot, not proof of their final turn. +## Preserve the persistent engram home separately + +Checkpoint adoption restores only volatile working memory and own-speech rings. +It does not move, restore, or verify the persistent engram database. The native +home correction did not change that database's resident path: + +```text +/citizens/personas//airc/engrams.sqlite +``` + +The resident root comes from `CONTINUUM_ROOT` when set, otherwise +`dirs::home_dir()/.continuum`. The existing +[`citizen_home_path`](../../core/continuum-core/src/context/citizen_path.rs) +resolves the citizen's `airc` directory; +[`PersonaInstanceInfo::from_runtime`](../../core/continuum-core/src/modules/persona_instance_manager.rs) +carries that exact path as `identity.home`. The supervisor passes it through +`PersonaHome::from_root`, and +[`AdmissionState::for_persona`](../../core/continuum-core/src/persona/admission_state.rs) +opens `PersonaHome::engrams_db()` beneath it. This is separate from the +UUID-keyed `volatile.json` destination. + +Before stopping, read the registered resident through the public command: + +```text +continuum persona/instances/get --persona_id +``` + +Record its returned `home`, stored name/peer binding, and existing engram +database. Preserve the same +`CONTINUUM_ROOT` override or native user root and identity binding for the new +launch. After restart, verify that the resident opens that same database and +that previously persisted engrams remain available. A file's presence alone is +insufficient: a different, empty home can initialize a new database successfully. + +Persistent-memory bootstrap refuses a slot when opening or +restoring its admission store fails, preserving an existing registered resident +and allowing healthy siblings to start. It does not establish that a valid but +unrelated database belongs to the intended prior life. A successful checkpoint +adoption receipt is therefore not proof of engram continuity. + ## Select one legacy snapshot explicitly The installed CLI exposes this recovery operation without needing a running From 39087481941b51655f8c28c11a0d10c1a6725862 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 19:13:40 -0500 Subject: [PATCH 4/6] fix(cognition): restore checkpoint evidence without inventing outcomes --- .../src/cognition/working_memory.rs | 143 +++++++++++------- 1 file changed, 89 insertions(+), 54 deletions(-) diff --git a/core/continuum-core/src/cognition/working_memory.rs b/core/continuum-core/src/cognition/working_memory.rs index ba5ecb4db0..c6debdc53e 100644 --- a/core/continuum-core/src/cognition/working_memory.rs +++ b/core/continuum-core/src/cognition/working_memory.rs @@ -297,28 +297,24 @@ pub enum WmKind { /// used to destroy (working memory, freshest full result, act fingerprints, /// the receipt counter). Written to `~/.continuum/personas//volatile.json` /// at shutdown / on tick write-through, restored at spawn. Deliberately -/// EXCLUDES engrams (already durable in sqlite) and dispatched handles (their -/// processes died with the old core — restoring them would fabricate -/// in-flight work). +/// EXCLUDES engrams (already durable in sqlite) and dispatched handles (the +/// checkpoint cannot establish whether their operations are still in flight). #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct VolatileSnapshot { pub entries: Vec, pub last_action: Option<(u64, String)>, pub action_fps: Vec, pub next_action_seq: u64, - /// Wall-clock when this snapshot was written — lets restore render the - /// interruption GAP ("~N minutes ago") as a perceivable fact instead of - /// an invisible discontinuity. `0` on snapshots from before this field - /// (serde default): restore then omits the gap, never guesses it. + /// Wall-clock when this snapshot was written — lets restore render its age, + /// not the time of an interruption. `0` on snapshots from before this field + /// (serde default): restore then reports the save time as unknown. #[serde(default)] pub saved_at_ms: u64, /// LABELS (never handles) of dispatched commands still `Running` at - /// save time. Their processes die with the old core, so the handles are - /// deliberately NOT restored (that would fabricate in-flight work) — - /// but the persona must KNOW what was cut off so she can repeat it in - /// one motion (Joel 2026-07-13: an interruption should be like closing - /// a laptop — reopen, see what didn't finish, redo it easily). Restore - /// renders these into a `[resumed]` fact marked safe-to-repeat. + /// save time. Labels alone establish neither completion nor side effects, + /// so restore reports those outcomes as unknown instead of restoring handles + /// or asserting that repeating the operation is safe. The field name stays + /// unchanged for compatibility with existing checkpoints. #[serde(default)] pub interrupted_dispatches: Vec, /// Build the receipts in this snapshot were RECORDED against. @@ -783,13 +779,10 @@ impl WorkingMemory { /// mid-thought instead of blank. Capacity re-clamps on the way in so a /// snapshot from a larger-capacity life never overflows this one. /// - /// The laptop-lid contract (Joel 2026-07-13): the interruption itself - /// becomes a PERCEIVABLE fact — how long the lid was closed, and exactly - /// which dispatched commands were cut off mid-flight (their processes - /// died with the old core; the work did NOT complete and is safe to - /// repeat). Without this, a killed dispatch is indistinguishable from a - /// finished one, and she either re-does completed work or trusts work - /// that never happened. + /// The checkpoint becomes a PERCEIVABLE fact: its save age and the dispatch + /// labels it recorded as pending. It does not establish when an interruption + /// happened, whether those operations completed, or what side effects they + /// produced. The persona decides what to do with that uncertainty. pub fn restore(&self, snap: VolatileSnapshot) { { let mut e = self.entries.lock(); @@ -836,10 +829,9 @@ impl WorkingMemory { self.next_action_seq .store(snap.next_action_seq.max(1), Ordering::Relaxed); - // Render the interruption as a fact AFTER the entries land, so it is - // the NEWEST thing in her window when she wakes. A fact, never an - // instruction — she decides whether the cut-off work still matters. - let gap = (snap.saved_at_ms > 0) + // Render the checkpoint evidence AFTER the entries land, so it is the + // NEWEST thing in her window when she wakes. A fact, never an instruction. + let checkpoint_age = (snap.saved_at_ms > 0) .then(|| { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -855,23 +847,21 @@ impl WorkingMemory { format!("~{mins} min") } }); - let fact = match (&gap, snap.interrupted_dispatches.is_empty()) { - (Some(g), false) => format!( - "[resumed] your session was interrupted {g} ago and your memory restored. Cut off mid-flight and NOT completed: {} — safe to repeat if still wanted", - snap.interrupted_dispatches.join("; ") - ), - (Some(g), true) => format!( - "[resumed] your session was interrupted {g} ago and your memory restored; nothing was in flight" - ), - (None, false) => format!( - "[resumed] your session was interrupted and your memory restored. Cut off mid-flight and NOT completed: {} — safe to repeat if still wanted", + let saved = match checkpoint_age { + Some(age) => format!("saved {age} ago"), + None => "with an unknown save time".to_string(), + }; + let pending = if snap.interrupted_dispatches.is_empty() { + "No pending dispatches were recorded in that checkpoint.".to_string() + } else { + format!( + "Dispatches recorded as pending at that save: {}. Their completion and side effects are unknown.", snap.interrupted_dispatches.join("; ") - ), - (None, true) => { - "[resumed] your session was interrupted and your memory restored; nothing was in flight".to_string() - } + ) }; - self.record_fact(&fact); + self.record_fact(&format!( + "[resumed] your memory was restored from a checkpoint {saved}. {pending}" + )); // The OTHER discontinuity, and until 2026-08-07 an invisible one: the // substrate itself was rebuilt while she was away (#165). @@ -1729,9 +1719,8 @@ mod tests { // round-trips through the JSON snapshot losslessly: typed entries (kinds // intact), the full last result, fingerprints, and the receipt counter // (so post-restore receipts keep ascending numbers instead of colliding - // with restored ones) — PLUS the laptop-lid contract (Joel 2026-07-13): - // the interruption itself lands as the NEWEST fact, naming the gap and - // any dispatched commands cut off mid-flight as safe to repeat. + // with restored ones). The NEWEST fact names the checkpoint's save age and + // recorded pending labels without inventing interruption time or outcomes. #[test] fn volatile_snapshot_round_trips_and_renders_the_interruption() { let wm = WorkingMemory::new(8); @@ -1740,8 +1729,8 @@ mod tests { wm.record_fact("[unfulfilled] I said I would run commands, but no tool ran"); wm.record_settlement("shared the plan"); wm.note_action_fingerprint("code/list|{\"path\":\".\"}"); - // A dispatched compile still Running at snapshot time — the process - // dies with the old core; only its LABEL must survive. + // A dispatched compile still Running at snapshot time — only its LABEL + // survives, not evidence about later completion or side effects. let handle = Uuid::new_v4(); wm.record_dispatch_event( handle, @@ -1769,14 +1758,27 @@ mod tests { "window identical before the marker" ); assert!( - resumed[0].contains("[resumed]"), - "interruption is perceivable: {resumed:?}" + resumed[0].starts_with("[resumed] your memory was restored from a checkpoint saved ") + && resumed[0].contains(" ago."), + "the age describes the checkpoint save: {resumed:?}" ); assert!( - resumed[0].contains("cargo build (dispatched)") - && resumed[0].contains("safe to repeat"), - "cut-off work named + marked repeatable: {resumed:?}" - ); + resumed[0].contains( + "Dispatches recorded as pending at that save: cargo build (dispatched)." + ) && resumed[0].contains("Their completion and side effects are unknown."), + "pending work is named without inventing its outcome: {resumed:?}" + ); + for unsupported in [ + "was interrupted", + "NOT completed", + "safe to repeat", + "nothing was in flight", + ] { + assert!( + !resumed[0].contains(unsupported), + "unsupported claim {unsupported:?}: {resumed:?}" + ); + } assert!( !restored.iter().any(|l| l.contains("[rebuilt]")), "SAME build across the restart — no rebuild fact, or we cry wolf on every \ @@ -1798,14 +1800,42 @@ mod tests { "counter resumed: {last:?}" ); - // And the quiet path: nothing in flight → the fact says so plainly. + // Missing and zero save times remain unknown even with pending labels. + for missing_saved_at in [false, true] { + let mut legacy: serde_json::Value = serde_json::from_str(&json).expect("snapshot JSON"); + if missing_saved_at { + let _ = legacy + .as_object_mut() + .expect("snapshot object") + .remove("saved_at_ms"); + } else { + legacy["saved_at_ms"] = serde_json::json!(0); + } + let legacy = serde_json::from_value(legacy).expect("legacy snapshot deserializes"); + let restored_legacy = WorkingMemory::new(8); + restored_legacy.restore(legacy); + let recent = restored_legacy.recent(); + let resumed = recent.last().expect("restored checkpoint fact"); + assert!( + resumed.contains("checkpoint with an unknown save time."), + "{resumed}" + ); + assert!(!resumed.contains("ago"), "no fabricated save age: {resumed}"); + assert!(resumed.contains("cargo build (dispatched)"), "{resumed}"); + assert!( + resumed.contains("Their completion and side effects are unknown."), + "{resumed}" + ); + } + + // An empty list describes this checkpoint, not everything that was running. let quiet = WorkingMemory::new(8); quiet.restore(VolatileSnapshot { entries: Vec::new(), last_action: None, action_fps: Vec::new(), next_action_seq: 1, - saved_at_ms: 0, // pre-field snapshot: no gap guessed + saved_at_ms: 0, // pre-field snapshot: no save time guessed interrupted_dispatches: Vec::new(), build_sha: String::new(), // pre-field snapshot: no rebuild guessed either receipt_heads: Vec::new(), @@ -1814,10 +1844,15 @@ mod tests { }); let q = quiet.recent(); assert_eq!(q.len(), 1); - assert!(q[0].contains("nothing was in flight"), "{q:?}"); + assert!( + q[0].contains("No pending dispatches were recorded in that checkpoint."), + "{q:?}" + ); + assert!(!q[0].contains("nothing was in flight"), "{q:?}"); + assert!(q[0].contains("checkpoint with an unknown save time."), "{q:?}"); assert!( !q[0].contains("ago"), - "no fabricated gap on legacy snapshots: {q:?}" + "no fabricated save age on legacy snapshots: {q:?}" ); } From c55720ac9a18191e88a5e7626b069c5ae5a39103 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 19:41:22 -0500 Subject: [PATCH 5/6] fix: resolve checkpoint adoption Clippy findings --- core/continuum-core/src/bin/continuum.rs | 144 +++++++++--------- .../persona_workspace/checkpoint_adoption.rs | 4 +- 2 files changed, 74 insertions(+), 74 deletions(-) diff --git a/core/continuum-core/src/bin/continuum.rs b/core/continuum-core/src/bin/continuum.rs index 6c57df15ad..616e776fd5 100644 --- a/core/continuum-core/src/bin/continuum.rs +++ b/core/continuum-core/src/bin/continuum.rs @@ -3118,6 +3118,78 @@ fn tail(path: &str, n: usize) -> String { lines[start..].join("\n") } +/// The desktop display manager's port — `CONTINUUM_UI_PORT`, else the +/// documented default beside WS 8974 (http::desktop). ONE place; the +/// `desktop` verb and the start/reboot receipt both read it. +fn desktop_port() -> u16 { + std::env::var("CONTINUUM_UI_PORT") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(8975) // unwrap_or: the display manager's documented default +} + +fn desktop_url() -> String { + format!("http://127.0.0.1:{}/", desktop_port()) +} + +/// Bounded (1 s) "is the greeter answering" probe — on a deploy path, so it +/// has a bound and a named outcome, never a hang. +async fn desktop_answering() -> bool { + matches!( + tokio::time::timeout( + std::time::Duration::from_secs(1), + tokio::net::TcpStream::connect(("127.0.0.1", desktop_port())), + ) + .await, + Ok(Ok(_)) + ) +} + +/// The line a verified start/reboot ends with: WHERE the desktop is. A user +/// must never have to know a port (Joel, 2026-09-05: "remembering port is +/// bush league") — the CLI says the address, and `uu desktop` opens it. +async fn desktop_receipt_line() -> String { + if desktop_answering().await { + format!("🖥 desktop: {} (`uu desktop` opens it)", desktop_url()) + } else { + format!( + "🖥 desktop: not serving yet on :{} — the web build lands in the background; \ + `uu desktop` opens it once it does", + desktop_port() + ) + } +} + +fn usage() -> String { + "usage: continuum [json | --key value ...] (uu = continuum)\n\ + \n\ + Lifecycle:\n \ + continuum start build + run the headless Rust core (detached), wait until ready;\n refuses if a core is running but not answering (a second core on\n one socket makes results non-deterministic)\n \ + continuum start --force reclaim those unresponsive core(s) first, then start\n \ + continuum reboot rebuild + relaunch; verifies the RUNNING core's build SHA\n \ + continuum reboot --prebuilt \n validate and launch that core without rebuilding; retains cwd\n and matches checkout HEAD when run in a repository\n \ + continuum stop stop the running core\n \ + continuum deploy-verify prove the running core's build SHA matches the deployed source\n\ + \n\ + Legacy checkpoint recovery (local; no running core required):\n \ + continuum checkpoint inspect --source --persona-id --plan \n save an explicit digest-bound selection; no checkpoint changed\n \ + continuum checkpoint adopt --plan --legacy-writers-stopped\n preserve both snapshots and adopt the selected bytes offline;\n stop legacy cores and automatic launchers first; no final-flush claim\n\ + \n\ + Desktop (the core serves it; no port to remember):\n \ + continuum desktop open the desktop in your browser (alias: uu desktop)\n\ + \n\ + Commands (dispatch to the running core):\n \ + continuum ping\n \ + continuum ping --message hi # --key value, coerced + camelCased automatically\n \ + continuum ping '{\"message\":\"hi\"}' # or a single JSON object (AI / power-user path)\n \ + continuum commands/list # discover commands dynamically (single source)\n \ + continuum commands/list --filter data/\n\ + \n\ + Env: CONTINUUM_CORE_SOCKET (default /tmp/continuum-core.sock)\n \ + CONTINUUM_START_SCRIPT (override the start script path)" + .to_string() +} + #[cfg(test)] mod tests { // What this catches (card 9f160b78): missing/truncated process evidence and @@ -4318,75 +4390,3 @@ mod tests { ); } } - -/// The desktop display manager's port — `CONTINUUM_UI_PORT`, else the -/// documented default beside WS 8974 (http::desktop). ONE place; the -/// `desktop` verb and the start/reboot receipt both read it. -fn desktop_port() -> u16 { - std::env::var("CONTINUUM_UI_PORT") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(8975) // unwrap_or: the display manager's documented default -} - -fn desktop_url() -> String { - format!("http://127.0.0.1:{}/", desktop_port()) -} - -/// Bounded (1 s) "is the greeter answering" probe — on a deploy path, so it -/// has a bound and a named outcome, never a hang. -async fn desktop_answering() -> bool { - matches!( - tokio::time::timeout( - std::time::Duration::from_secs(1), - tokio::net::TcpStream::connect(("127.0.0.1", desktop_port())), - ) - .await, - Ok(Ok(_)) - ) -} - -/// The line a verified start/reboot ends with: WHERE the desktop is. A user -/// must never have to know a port (Joel, 2026-09-05: "remembering port is -/// bush league") — the CLI says the address, and `uu desktop` opens it. -async fn desktop_receipt_line() -> String { - if desktop_answering().await { - format!("🖥 desktop: {} (`uu desktop` opens it)", desktop_url()) - } else { - format!( - "🖥 desktop: not serving yet on :{} — the web build lands in the background; \ - `uu desktop` opens it once it does", - desktop_port() - ) - } -} - -fn usage() -> String { - "usage: continuum [json | --key value ...] (uu = continuum)\n\ - \n\ - Lifecycle:\n \ - continuum start build + run the headless Rust core (detached), wait until ready;\n refuses if a core is running but not answering (a second core on\n one socket makes results non-deterministic)\n \ - continuum start --force reclaim those unresponsive core(s) first, then start\n \ - continuum reboot rebuild + relaunch; verifies the RUNNING core's build SHA\n \ - continuum reboot --prebuilt \n validate and launch that core without rebuilding; retains cwd\n and matches checkout HEAD when run in a repository\n \ - continuum stop stop the running core\n \ - continuum deploy-verify prove the running core's build SHA matches the deployed source\n\ - \n\ - Legacy checkpoint recovery (local; no running core required):\n \ - continuum checkpoint inspect --source --persona-id --plan \n save an explicit digest-bound selection; no checkpoint changed\n \ - continuum checkpoint adopt --plan --legacy-writers-stopped\n preserve both snapshots and adopt the selected bytes offline;\n stop legacy cores and automatic launchers first; no final-flush claim\n\ - \n\ - Desktop (the core serves it; no port to remember):\n \ - continuum desktop open the desktop in your browser (alias: uu desktop)\n\ - \n\ - Commands (dispatch to the running core):\n \ - continuum ping\n \ - continuum ping --message hi # --key value, coerced + camelCased automatically\n \ - continuum ping '{\"message\":\"hi\"}' # or a single JSON object (AI / power-user path)\n \ - continuum commands/list # discover commands dynamically (single source)\n \ - continuum commands/list --filter data/\n\ - \n\ - Env: CONTINUUM_CORE_SOCKET (default /tmp/continuum-core.sock)\n \ - CONTINUUM_START_SCRIPT (override the start script path)" - .to_string() -} diff --git a/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs b/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs index 66267f1680..cccb5ab3ba 100644 --- a/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs +++ b/core/continuum-core/src/cognition/persona_workspace/checkpoint_adoption.rs @@ -467,9 +467,9 @@ pub fn adopt( fs::rename(&staged, &destination)?; sync_parent(&destination)?; let published = optional_selection(&destination)?; - if !published + if published .as_ref() - .is_some_and(|value| value.sha256 == plan.source.sha256) + .is_none_or(|value| value.sha256 != plan.source.sha256) { return Err(AdoptionError::EvidenceConflict( "destination changed after publication".into(), From 7dd3bfffbc7de8d03d702c6e3de2398afc5e2279 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 19:42:45 -0500 Subject: [PATCH 6/6] test(persona): check held-work fixture registration --- core/continuum-core/src/persona/service_loop.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/core/continuum-core/src/persona/service_loop.rs b/core/continuum-core/src/persona/service_loop.rs index adb5fd24bf..6a6ce7d689 100644 --- a/core/continuum-core/src/persona/service_loop.rs +++ b/core/continuum-core/src/persona/service_loop.rs @@ -5136,7 +5136,9 @@ mod tests { defer_grounding: false, suppress_recall: false, }; - crate::cognition::persona_workspace::global().register_from_cfg(cfg); + crate::cognition::persona_workspace::global() + .register_from_cfg(cfg) + .expect("test: register the held-work publication fixture"); let stub = StubAircCitizen::new(peer).with_claims(vec![held_card(peer)]); let mut conversation = ScriptedConversation::new().with_citizen(Arc::new(stub)); if refuse_publication {