diff --git a/docs/user/getting-started-and-configuration.md b/docs/user/getting-started-and-configuration.md index 37b0da4c..a3d8bd6f 100644 --- a/docs/user/getting-started-and-configuration.md +++ b/docs/user/getting-started-and-configuration.md @@ -122,13 +122,18 @@ Start the ACP-backed terminal client at a project root: kit tui --root /path/to/project ``` -Resume a persisted conversation by its displayed session ID: +Choose a persisted conversation in the startup picker, or resume directly by ID: ```sh +kit tui --root /path/to/project --resume kit tui --root /path/to/project --resume ``` -If a dead process left a stale session lock, `--force` can accompany `--resume`. It is not a general overwrite option and Clap rejects it without a resume argument. +Without an ID, `--resume` opens the same workspace-scoped, newest-first picker as `/sessions`, including session names and inline rename. Workspace selection honors `--root` and configured root defaults as in normal startup. No new persisted session is created just to show the picker. Plain `kit tui` is unchanged. + +`Esc` at the top level or `Ctrl+C` cancels startup successfully without creating or resuming a session; `Esc` while renaming only cancels the rename. An empty catalog reports no resumable sessions and exits successfully. Catalog read failures exit unsuccessfully with an actionable error; a selected session that cannot be resumed reports its resume error instead of starting a new session. + +If a dead process left a stale session lock, `--force` can accompany either form of `--resume`. With the picker, it applies only to the selected session and never takes over a lock held by a live process. It is not a general overwrite option and Clap rejects it without `--resume`. See [session locks and recovery](tui-and-sessions.md#session-locks---resume-and---force). ### One-shot automation with `kit prompt` diff --git a/docs/user/tui-and-sessions.md b/docs/user/tui-and-sessions.md index 55feeafb..0b1c22f9 100644 --- a/docs/user/tui-and-sessions.md +++ b/docs/user/tui-and-sessions.md @@ -10,7 +10,18 @@ Start an interactive session at a project root with the installed binary: kit tui --root /path/to/project ``` -List sessions for the workspace, then resume the ID shown in the header or catalog: +Open the session picker at startup, or resume a known ID directly: + +```sh +kit tui --root /path/to/project --resume +kit tui --root /path/to/project --resume +``` + +Without an ID, `--resume` opens the same workspace-scoped, newest-first picker as `/sessions`, with the same session names, selection, and inline rename interactions. It resolves `--root` and configured root defaults just like normal startup. No new persisted session is created to display the picker; a session is resumed only after selection. Plain `kit tui` still starts normally. + +At the top-level picker, `Esc` or `Ctrl+C` cancels startup and exits successfully without creating or resuming a session; `Esc` during inline rename only cancels the rename. An empty catalog reports that the workspace has no resumable sessions and exits successfully. Catalog read failures report an actionable error and exit unsuccessfully. If the selected session disappears, becomes invalid, or is locked before resume, Kit reports the resume error rather than starting a new session. + +You can also list and rename sessions from the command line, then resume the ID shown in the header or catalog: ```sh kit sessions --root /path/to/project @@ -188,7 +199,7 @@ ACP v1 clients restore a closed durable session with `session/load` and discover Session discovery and restoration are isolated to the server's canonical workspace root. The primary `cwd` must match that root. ACP `session/new`, `session/load`, `session/resume`, and supported `session/fork` requests can also supply `additionalDirectories`: absolute paths to existing project directories. Kit canonicalizes and deduplicates these paths, loads their ancestor `AGENTS.md` instructions as session context, and reports the roots in `SessionInfo`. Additional roots are not a filesystem allowlist and do not change the default tool cwd, configuration, or durable session namespace. Each attachment supplies its complete additional-root set; an empty list clears prior extra roots. These roots are persisted as extensible transcript metadata, so old sessions without that metadata report an empty list. Legacy transcripts under a project-local `.kit/sessions` directory follow the same migration and root checks as CLI resume; they do not make a same-named session visible from another workspace. Old global transcripts without workspace metadata are excluded from discovery in every workspace, but an explicit resume by ID remains supported and binds the transcript to that workspace. An individually malformed or concurrently incomplete transcript is omitted from catalog results without preventing valid sessions from being listed; explicit resume remains strict and reports its error. -An arbitrary ACP load or resume never applies the server process's configured `--force` setting. The one exception is the initial resume requested by `kit tui --resume --force`: only that matching configured session may use the explicit stale-lock override. If another live Kit instance owns the session lock, restoration fails instead of taking over the session. A missing or invalid ID also fails normally. After the session closes and releases its lock, an ACP client can restore it again. +An arbitrary ACP load or resume never applies the server process's configured `--force` setting. The one exception is the initial resume requested by `kit tui --resume [] --force`: only the explicitly named or picker-selected session may use the stale-lock override. If another live Kit instance owns the session lock, restoration fails instead of taking over the session. A missing or invalid ID also fails normally. After the session closes and releases its lock, an ACP client can restore it again. Before the restoration response, Kit replays the canonical transcript as ordered ACP updates for representable user text and attachments, assistant text and thoughts, and tool calls and results. Internal instructions, ambient context, notifications, and provider-specific content are not replayed to the client, but remain in the model transcript. Because compaction replaces the canonical transcript, restoring a compacted session replays its canonical summary history rather than the superseded pre-compaction items. @@ -206,7 +217,7 @@ first confirm that no Kit process is still using the session. Then retry the res kit tui --root /path/to/project --resume --force ``` -`--force` is only for a stale lock left by an exited or crashed process, and the CLI accepts it only with `--resume`. It does not steal a lock held by a live process: the OS-level lock check instead reports `session is actively locked by another Kit instance (...)`. Do not manually remove a lock belonging to a running Kit process. +`--force` is only for a stale lock left by an exited or crashed process, and the CLI accepts it only with `--resume`. With `kit tui --resume --force`, the override applies only to the session selected in the startup picker. It does not steal a lock held by a live process: the OS-level lock check instead reports `session is actively locked by another Kit instance (...)`. Do not manually remove a lock belonging to a running Kit process. A new session ID that already exists reports `session ... already exists; use --resume`; a missing resume target reports `session ... does not exist`. Use the correct ID and mode rather than `--force` for either error. diff --git a/src/main.rs b/src/main.rs index f121849a..5483a73d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1092,7 +1092,8 @@ impl Command { .value_name("RESUME") .value_parser(clap::value_parser!(String)) .action(clap::ArgAction::Set) - .help("Resume this persisted session id"), + .num_args(0..=1) + .help("Resume a persisted session by ID, or open the session picker without an ID"), ); command.arg( clap::Arg::new("force") @@ -1175,7 +1176,10 @@ impl Command { reasoning_effort: optional_arg(matches, "reasoning_effort")?, a2a: optional_arg(matches, "a2a")?, mcp: McpArgs::from_matches(matches)?, - resume: optional_arg(matches, "resume")?, + resume: matches + .contains_id("resume") + .then(|| optional_arg(matches, "resume")) + .transpose()?, force: required_arg(matches, "force")?, }), _ => Err(clap::Error::raw( @@ -1709,8 +1713,8 @@ enum Command { a2a: Option, mcp: McpArgs, - /// Resume this persisted session id. - resume: Option, + /// Resume by ID, or open the workspace session picker without an ID. + resume: Option>, /// Override the resumed session's stale lock. force: bool, }, @@ -2377,22 +2381,40 @@ async fn run_cli(cli: Cli) -> Result<(), Box> { // the selected reference before starting that subprocess. let _ = config.harnesses()?; let root = config.root(root); + let mut stop = kit::tui::Stop::new()?; + let resume = match resume { + Some(None) => match kit::tui::pick_session(&root, &mut stop).await? { + Some(id) => Some(id), + None => return Ok(()), + }, + Some(Some(id)) => Some(id), + None => None, + }; let model = config.model(model); let provider = config.provider(provider); let reasoning_effort = config.reasoning_effort(reasoning_effort); let a2a = config.a2a(a2a); let credential_storage = mcp.credentials.storage(&config)?; let (_, explicit_mcp) = mcp.config_paths(&config)?; - let _ = config.plugin_runtime(&root).await?; let voice_enabled = config.experimental.voice; - let config_path = config.config_path.clone(); - tokio::task::spawn_blocking(move || { - if let Some(path) = config_path { - fs::global().require_disk(path)?; - } - Ok::<_, io::Error>(()) - }) - .await??; + let prepared = stop + .until(async { + let _ = config.plugin_runtime(&root).await?; + let config_path = config.config_path.clone(); + tokio::task::spawn_blocking(move || { + if let Some(path) = config_path { + fs::global().require_disk(path)?; + } + Ok::<_, io::Error>(()) + }) + .await??; + Ok::<_, Box>(()) + }) + .await; + let Some(prepared) = prepared else { + return Ok(()); + }; + prepared?; kit::tui::run_with_reasoning_effort_and_openrouter_key( &root, &model, @@ -2406,6 +2428,7 @@ async fn run_cli(cli: Cli) -> Result<(), Box> { resume.as_deref(), force, voice_enabled, + &mut stop, ) .await? } @@ -2439,6 +2462,41 @@ mod tests { supervise_serve_with_trigger, validate_auth_storage, }; + #[cfg(feature = "tui")] + #[test] + fn tui_resume_forms_preserve_direct_and_normal_startup() { + for (args, expected, expected_force) in [ + (vec!["kit", "tui"], None, false), + (vec!["kit", "tui", "--resume"], Some(None), false), + (vec!["kit", "tui", "--resume", "--force"], Some(None), true), + ( + vec!["kit", "tui", "--resume", "s-example"], + Some(Some("s-example".to_string())), + false, + ), + ( + vec!["kit", "tui", "--resume", "s-example", "--force"], + Some(Some("s-example".to_string())), + true, + ), + ] { + let cli = Cli::try_parse_from(args).unwrap(); + let Command::Tui { resume, force, .. } = cli.command else { + panic!("expected TUI") + }; + assert_eq!(resume, expected); + assert_eq!(force, expected_force); + } + assert!(Cli::try_parse_from(["kit", "tui", "--force"]).is_err()); + assert!(Cli::try_parse_from(["kit", "prompt", "hello", "--resume"]).is_err()); + let help = Cli::try_parse_from(["kit", "tui", "--help"]) + .err() + .unwrap() + .to_string(); + assert!(help.contains("--resume []")); + assert!(help.contains("session picker without an ID")); + } + #[cfg(feature = "tui")] #[test] fn config_experimental_voice_defaults_and_strict_boolean() { diff --git a/src/resilient_fs/mod.rs b/src/resilient_fs/mod.rs index 2f57a01f..99ed584b 100644 --- a/src/resilient_fs/mod.rs +++ b/src/resilient_fs/mod.rs @@ -317,11 +317,184 @@ struct Entry { path: PathBuf, object: Option, } +impl PreparationNamespace { + /// Trusted crate callers must use this namespace only for read operations. + /// Writes here would bypass the original service's ordering and admission. + #[cfg(any(feature = "tui", test))] + pub(crate) fn filesystem(&self) -> &Fs { + &self.isolated + } + + pub fn prepare_private_replace_with_parents>( + self, + path: P, + contents: &[u8], + ) -> io::Result { + let fs = &self.isolated; + let path = fs.norm(path.as_ref())?; + let mut state = fs.state()?; + let mut parents = Vec::new(); + let mut current = PathBuf::new(); + for component in path + .parent() + .ok_or_else(|| error(io::ErrorKind::InvalidInput, "no parent"))? + .components() + { + current.push(component.as_os_str()); + match fs.lookup(&state, ¤t) { + Ok(meta) if meta.is_dir() => continue, + Ok(_) => { + return Err(error( + io::ErrorKind::NotADirectory, + "ancestor is not a directory", + )); + } + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => return Err(e), + } + fs.secure_path(&state, ¤t, false)?; + fs.parent(&state, ¤t)?; + Fs::prepare(&mut state, 1)?; + parents.try_reserve(1).map_err(|_| allocation_oom())?; + let meta = Metadata { + identity: next_identity(), + disk_identity: None, + disk: None, + kind: FileType { + file: false, + dir: true, + symlink: false, + }, + len: 0, + permissions: fs.new_permissions(&state, ¤t, false, true)?, + modified: SystemTime::now(), + }; + let object = Arc::new(Mutex::new(Object::memory( + Arc::new(Zeroizing::new(Vec::new())), + meta, + ))); + // lookup exposes entries only while an action owns their overlay. + // This isolated queue is a virtual plan, never submitted/recovered; + // after taking this guard we use only direct read-only helpers. + state.pending.push_back(Pending { + action: Action::Mkdir { + path: current.clone(), + private: false, + stage: 0, + }, + lease: None, + }); + Fs::entry(&mut state, current.clone(), Some(object.clone()))?; + parents.push((current.clone(), object)); + } + // Existing preflight opens an existing file without create/truncate to + // validate write permission; it never writes bytes or creates a probe. + fs.preflight(&state, &path)?; + let permissions = fs.new_permissions(&state, &path, true, false)?; + let data = Arc::new(bytes(contents)?); + let image = Image::memory(data.clone()); + let object = Arc::new(Mutex::new(Object::memory( + data, + Metadata { + identity: next_identity(), + disk_identity: None, + disk: None, + kind: FileType { + file: true, + dir: false, + symlink: false, + }, + len: image.len, + permissions: permissions.clone(), + modified: SystemTime::now(), + }, + ))); + drop(state); + Ok(PreparedPrivateReplace { + original: self.original, + revision: self.revision, + parents, + path, + object, + image, + permissions, + }) + } +} + +impl PreparedPrivateReplace { + /// Validate without I/O, then submit parent-first under the same state guard. + /// No recovery or global preflight precedes submission. A stale plan is + /// rejected with WouldBlock and must be prepared again, never replayed. + pub fn commit(self) -> io::Result<()> { + let fs = &self.original; + let (mut state, previous) = fs.service.acquire()?; + if !self.revision.ptr_eq(&Arc::downgrade(&previous)) { + return Err(error( + io::ErrorKind::WouldBlock, + "stale preparation; retry the operation", + )); + } + fs.preparation_quiescent(&state)?; + Fs::prepare(&mut state, self.parents.len() + 1)?; + let result = (|| { + for (path, object) in self.parents { + fs.submit( + &mut state, + Action::Mkdir { + path: path.clone(), + private: false, + stage: 0, + }, + )?; + Fs::entry(&mut state, path, Some(object))?; + } + let action = Fs::put_action(&mut state, &self.path, self.image, self.permissions); + fs.submit(&mut state, action)?; + // Quiescence excludes old live handles; acquisitions creating one + // invalidate the ticket. Therefore replacement cannot rebind a handle. + Fs::entry(&mut state, self.path, Some(self.object))?; + fs.rebase(&mut state)?; + Ok(()) + })(); + // Remove completed plan bookkeeping on success AND ordinary rejection. + // This is still owned commit work; preparation must not drop arbitrary + // backend descriptors under the original state lock just to prune. + // Pending overlays remain intact for process-owned recovery. An unwind + // instead poisons the service, as in ordinary mutation paths. + Fs::prune(&mut state); + result + } +} + #[derive(Clone)] pub struct Fs { service: Arc, lease: Option>, } +/// Isolated namespace protected by an original-service revision ticket. +/// No backend reads hold the original service mutex. Concurrent original-service +/// access conservatively invalidates this preparation, except observational +/// status queries and recovery passes that do no work. +pub struct PreparationNamespace { + original: Fs, + isolated: Fs, + revision: std::sync::Weak<()>, +} + +/// A validated private replacement. Dropping it has no filesystem effects. +/// Commit uses the original service; mkdir plus replace is not atomic: accepted +/// directories can remain if a later action is rejected. +pub struct PreparedPrivateReplace { + original: Fs, + revision: std::sync::Weak<()>, + parents: Vec<(PathBuf, Obj)>, + path: PathBuf, + object: Obj, + image: Image, + permissions: Permissions, +} + // Lock order: service state -> cursor (when present) -> object -> native file. // Lease fencing is acquired under service state, or independently by observers; // it never acquires service state. No guard crosses an await. File and namespace @@ -334,10 +507,22 @@ struct Service { max_operations: usize, best_effort: bool, } +impl Service { + // Ordinary acquisitions invalidate preparation, including reads and rejected + // operations. Only status queries and provably idle recovery bypass this. + // Weak tickets cannot keep an obsolete generation alive, and + // allocation identity cannot wrap or be reused while its Weak survives. + fn acquire(&self) -> io::Result<(std::sync::MutexGuard<'_, State>, Arc<()>)> { + let mut state = lock(&self.state)?; + let previous = std::mem::replace(&mut state.revision, Arc::new(())); + Ok((state, previous)) + } +} // Fs and File transitions coordinate namespace entries/redirects, live object // images, budget accounting, replay stages, and retained lease authority here. // Disk effects cannot be rolled back generically after an unwind. struct State { + revision: Arc<()>, entries: Vec, redirects: Vec<(PathBuf, PathBuf)>, objects: Vec>>, @@ -575,6 +760,7 @@ impl Fs { service: Arc::new(Service { backend, state: Mutex::new(State { + revision: Arc::new(()), entries: Vec::new(), redirects: Vec::new(), objects: Vec::new(), @@ -591,10 +777,72 @@ impl Fs { lease: None, } } + fn preparation_quiescent(&self, state: &State) -> io::Result<()> { + if state.dropped { + return Err(io::Error::other(DroppedScope)); + } + if state.exhausted || ALLOCATION_EXHAUSTED.load(std::sync::atomic::Ordering::Acquire) { + return Err(oom()); + } + // Native lease validation is I/O, so leased namespaces cannot use this + // optimistic path. Ordinary guarded operations retain their semantics. + if self.lease.is_some() + || state + .leases + .iter() + .any(|(_, lease, _)| lease.strong_count() > 0) + { + return Err(error( + io::ErrorKind::PermissionDenied, + "leased preparation unsupported", + )); + } + if !state.pending.is_empty() + || !state.entries.is_empty() + || !state.redirects.is_empty() + || state.objects.iter().any(|object| object.strong_count() > 0) + { + return Err(error( + io::ErrorKind::WouldBlock, + "storage is busy; retry the operation", + )); + } + Ok(()) + } + + /// Capture a quiescent revision BEFORE any normalization or backend reads. + /// The returned namespace shares only the backend, never the original queue, + /// locks, objects, or lease authority. Retry preparation on WouldBlock. + pub fn prepare_namespace(&self) -> io::Result { + let state = self.state()?; + self.preparation_quiescent(&state)?; + let revision = Arc::downgrade(&state.revision); + drop(state); + Ok(PreparationNamespace { + original: self.clone(), + isolated: Self::with_policy( + self.service.backend.clone(), + self.service.max_bytes, + self.service.max_operations, + self.service.best_effort, + ), + revision, + }) + } + + pub fn prepare_private_replace_with_parents>( + &self, + path: P, + contents: &[u8], + ) -> io::Result { + self.prepare_namespace()? + .prepare_private_replace_with_parents(path, contents) + } + // Fence IO entry points, not just queue submission: reads // must not mistake an abandoned image for a complete transcript either. fn state(&self) -> io::Result> { - let s = lock(&self.service.state)?; + let s = self.service.acquire().map(|(state, _)| state)?; if s.dropped { return Err(io::Error::other(DroppedScope)); } @@ -653,7 +901,7 @@ impl Fs { "not a best-effort service", )); } - let mut s = lock(&self.service.state)?; + let mut s = self.service.acquire().map(|(state, _)| state)?; s.dropped = true; let pending = std::mem::take(&mut s.pending); let entries = std::mem::take(&mut s.entries); @@ -1419,7 +1667,7 @@ impl Fs { } } pub fn recover(&self) -> RecoveryReport { - let mut s = match self.state() { + let mut s = match lock(&self.service.state) { Ok(s) => s, Err(e) => { return RecoveryReport { @@ -1430,6 +1678,38 @@ impl Fs { }; } }; + // An empty queue alone is insufficient: rebase can change live objects + // and prune can remove overlays/registrations. Classify under the same + // guard, and bypass all recovery work only for a demonstrably idle pass. + if !s.dropped + && s.pending.is_empty() + && s.entries.is_empty() + && s.redirects.is_empty() + && s.objects.is_empty() + && self.lease.is_none() + && !s + .leases + .iter() + .any(|(_, lease, _)| lease.strong_count() > 0) + { + return RecoveryReport { + completed_operations: 0, + remaining_operations: 0, + blocked: None, + lease_blocked: false, + }; + } + // Fence before any attempted work, including failed replay/rebase and + // rejected recovery on an abandoned scope. Never restore a generation. + s.revision = Arc::new(()); + if s.dropped { + return RecoveryReport { + completed_operations: 0, + remaining_operations: usize::MAX, + blocked: Some(io::Error::other(DroppedScope)), + lease_blocked: false, + }; + } let mut report = self.recover_locked(&mut s); if let Err(e) = self.rebase(&mut s) { report.blocked = Some(e); diff --git a/src/resilient_fs/tests.rs b/src/resilient_fs/tests.rs index 62fb6faf..068bedcc 100644 --- a/src/resilient_fs/tests.rs +++ b/src/resilient_fs/tests.rs @@ -188,6 +188,8 @@ impl Backend for Injected { })) } fn metadata(&self, p: &Path, follow: bool) -> io::Result { + self.faults.panic_at(Point::Metadata); + self.faults.pause_at(Point::Metadata); self.disk.metadata(p, follow) } fn read_dir(&self, p: &Path) -> io::Result> { @@ -2400,3 +2402,344 @@ fn best_effort_post_publication_payload_cannot_bypass_retained_budget() { } } } + +#[test] +fn prepared_replace_is_effect_free_until_commit_and_preserves_legacy_parents() { + let t = Fixture::new(); + let path = t.path("legacy/metadata/session.json"); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + assert!(!t.path("legacy").exists()); + drop(prepared); + assert!(!t.path("legacy").exists()); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + prepared.commit().unwrap(); + assert_eq!(native::read(&path).unwrap(), b"new"); + // Native inspection above does not prune or access the original service. + // A second commit must work without an intervening service read or recovery. + t.fs.prepare_private_replace_with_parents(&path, b"second") + .unwrap() + .commit() + .unwrap(); + assert_eq!(native::read(&path).unwrap(), b"second"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + native::metadata(&path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + // Compare with ordinary create_dir_all under the same process umask. + native::create_dir_all(t.path("ordinary/child")).unwrap(); + assert_eq!( + native::metadata(t.path("legacy")) + .unwrap() + .permissions() + .mode() + & 0o777, + native::metadata(t.path("ordinary")) + .unwrap() + .permissions() + .mode() + & 0o777 + ); + } +} + +#[test] +fn prepared_replace_reads_share_ticket_and_aba_access_invalidates_it() { + let t = Fixture::new(); + let path = t.path("value"); + native::write(&path, b"old").unwrap(); + let namespace = t.fs.prepare_namespace().unwrap(); + assert_eq!(namespace.filesystem().read(&path).unwrap(), b"old"); + // Open and drop leaves no live object, but must still invalidate the ticket. + drop(t.fs.open(&path).unwrap()); + let prepared = namespace + .prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + t.faults.arm_panic(Point::Metadata); + assert_eq!( + prepared.commit().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!(native::read(&path).unwrap(), b"old"); + t.faults.2.store(0, Ordering::SeqCst); + t.fs.prepare_private_replace_with_parents(&path, b"retry") + .unwrap() + .commit() + .unwrap(); + assert_eq!(native::read(&path).unwrap(), b"retry"); +} + +#[test] +fn prepared_replace_blocked_preflight_does_not_hold_original_state() { + let t = Fixture::new(); + let entered = Arc::new(std::sync::Barrier::new(2)); + let resume = Arc::new(std::sync::Barrier::new(2)); + *t.faults.3.lock().unwrap() = Some((Point::Metadata, entered.clone(), resume.clone())); + let fs = t.fs.clone(); + let path = t.path("value"); + let worker = std::thread::spawn(move || fs.prepare_private_replace_with_parents(path, b"new")); + entered.wait(); + // A synchronous acquisition completes while isolated backend I/O is paused. + assert_eq!(t.fs.status().pending_operations, 0); + resume.wait(); + worker.join().unwrap().unwrap().commit().unwrap(); + assert_eq!(native::read(t.path("value")).unwrap(), b"new"); +} + +#[test] +fn prepared_replace_survives_observational_status_and_idle_recovery() { + for best_effort in [false, true] { + let t = Fixture::new(); + let fs = if best_effort { + t.fs.best_effort(1024, 16) + } else { + t.fs.clone() + }; + let prepared = fs + .prepare_private_replace_with_parents(t.path("value"), b"new") + .unwrap(); + assert_eq!(fs.status().pending_operations, 0); + assert_eq!( + fs.best_effort_status(), + best_effort.then_some(BestEffortStatus::Ready) + ); + let report = fs.recover(); + assert_eq!(report.completed_operations, 0); + assert_eq!(report.remaining_operations, 0); + assert!(report.blocked.is_none()); + prepared.commit().unwrap(); + assert_eq!(native::read(t.path("value")).unwrap(), b"new"); + } +} + +#[test] +fn prepared_replace_empty_queue_recovery_still_fences_object_cleanup() { + let t = Fixture::new(); + let path = t.path("value"); + native::write(&path, b"old").unwrap(); + drop(t.fs.open(&path).unwrap()); + // Acquire the ticket AFTER handle access, so only recovery can stale it. + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + assert_eq!(t.fs.status().pending_operations, 0); + assert!(t.fs.recover().blocked.is_none()); + assert_eq!( + prepared.commit().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!(native::read(&path).unwrap(), b"old"); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + assert!(t.fs.recover().blocked.is_none()); + prepared.commit().unwrap(); + assert_eq!(native::read(&path).unwrap(), b"new"); +} + +#[test] +fn prepared_replace_failed_mutation_stays_stale_after_idle_recovery() { + let t = Fixture::new(); + let prepared = + t.fs.prepare_private_replace_with_parents(t.path("value"), b"new") + .unwrap(); + assert_eq!( + t.fs.remove_file(t.path("missing")).unwrap_err().kind(), + io::ErrorKind::NotFound + ); + assert!(t.fs.recover().blocked.is_none()); + assert_eq!(t.fs.status().pending_operations, 0); + assert_eq!( + prepared.commit().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + assert!(!t.path("value").exists()); +} + +#[test] +fn prepared_replace_preflight_panic_is_isolated_but_commit_panic_poisons() { + let t = Fixture::new(); + let path = t.path("value"); + t.faults.arm_panic(Point::Metadata); + assert!( + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _ = t.fs.prepare_private_replace_with_parents(&path, b"new"); + })) + .is_err() + ); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + t.faults.arm_panic(Point::AfterRename); + assert!(std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| prepared.commit())).is_err()); + assert_eq!(native::read(&path).unwrap(), b"new"); + let err = t.fs.prepare_namespace().err().unwrap(); + assert!(err.get_ref().is_some_and(|e| e.is::())); +} + +#[test] +fn prepared_replace_rejects_live_handles_and_never_rebinds_old_handle() { + let t = Fixture::new(); + let path = t.path("value"); + native::write(&path, b"old").unwrap(); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + let mut handle = t.fs.open(&path).unwrap(); + assert_eq!( + t.fs.prepare_namespace().err().unwrap().kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!( + prepared.commit().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + let mut contents = String::new(); + handle.read_to_string(&mut contents).unwrap(); + assert_eq!(contents, "old"); +} + +#[test] +#[cfg(unix)] +fn prepared_replace_queues_parent_first_and_rejects_existing_overlay() { + let t = Fixture::new(); + let path = t.path("legacy/metadata/value"); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + t.faults.arm(Point::Mkdir, libc::ENOSPC); + prepared.commit().unwrap(); + assert!(!t.path("legacy").exists()); + assert_eq!(t.fs.read(&path).unwrap(), b"new"); + t.faults.clear(); + // No preparation-side recovery may flush accepted obligations. + assert_eq!( + t.fs.prepare_namespace().err().unwrap().kind(), + io::ErrorKind::WouldBlock + ); + assert!(!path.exists()); + t.settle(); + assert_eq!(native::read(&path).unwrap(), b"new"); +} + +#[test] +#[cfg(unix)] +fn prepared_replace_partial_plan_rejection_keeps_accepted_directory() { + let t = Fixture::new(); + let path = t.path("legacy/value"); + let prepared = + t.fs.prepare_private_replace_with_parents(&path, b"new") + .unwrap(); + t.faults.arm(Point::Open, libc::EACCES); + assert_eq!( + prepared.commit().unwrap_err().kind(), + io::ErrorKind::PermissionDenied + ); + assert!(t.path("legacy").is_dir()); + assert!(!path.exists()); + t.faults.clear(); + t.fs.prepare_private_replace_with_parents(&path, b"retry") + .unwrap() + .commit() + .unwrap(); + assert_eq!(native::read(&path).unwrap(), b"retry"); +} + +#[test] +fn prepared_replace_fences_dropped_scopes_and_invalidates_sibling_tickets() { + let t = Fixture::new(); + let fs = t.fs.best_effort(1024, 16); + let first = fs + .prepare_private_replace_with_parents(t.path("first"), b"one") + .unwrap(); + let second = fs + .prepare_private_replace_with_parents(t.path("second"), b"two") + .unwrap(); + assert_eq!( + first.commit().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + // Even a rejected commit invalidates competing preparations. + assert_eq!( + second.commit().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + let prepared = fs + .prepare_private_replace_with_parents(t.path("value"), b"new") + .unwrap(); + fs.abandon_best_effort().unwrap(); + assert!(prepared.commit().is_err()); + let err = fs.prepare_namespace().err().unwrap(); + assert!(err.get_ref().is_some_and(|e| e.is::())); + assert!(!t.path("value").exists()); +} + +#[test] +#[cfg(unix)] +fn prepared_replace_rejects_exhausted_and_leased_services_without_preflight() { + let t = Fixture::budget(0, 0); + let prepared = + t.fs.prepare_private_replace_with_parents(t.path("value"), b"new") + .unwrap(); + t.faults.arm(Point::Open, libc::ENOSPC); + assert_eq!( + prepared.commit().unwrap_err().kind(), + io::ErrorKind::OutOfMemory + ); + t.faults.clear(); + t.faults.arm_panic(Point::Metadata); + assert_eq!( + t.fs.prepare_namespace().err().unwrap().kind(), + io::ErrorKind::OutOfMemory + ); + assert!(!t.path("value").exists()); + + let t = Fixture::new(); + let lease = + t.fs.acquire_lease(t.path("lock"), &t.root, LeaseMode::CreateNew) + .unwrap(); + let guarded = t.fs.guarded(&lease).unwrap(); + t.faults.arm_panic(Point::LeaseCheck); + assert_eq!( + guarded.prepare_namespace().err().unwrap().kind(), + io::ErrorKind::PermissionDenied + ); + assert_eq!( + t.fs.prepare_namespace().err().unwrap().kind(), + io::ErrorKind::PermissionDenied + ); + t.faults.2.store(0, Ordering::SeqCst); +} + +#[test] +#[cfg(unix)] +fn prepared_replace_preserves_preflight_file_validation() { + use std::os::unix::fs::PermissionsExt; + let t = Fixture::new(); + let path = t.path("value"); + native::write(&path, b"old").unwrap(); + native::set_permissions(&path, Permissions::from_mode(0o400)).unwrap(); + assert_eq!( + t.fs.prepare_private_replace_with_parents(&path, b"new") + .err() + .unwrap() + .kind(), + io::ErrorKind::PermissionDenied + ); + native::set_permissions(&path, Permissions::from_mode(0o600)).unwrap(); + native::hard_link(&path, t.path("other")).unwrap(); + assert_eq!( + t.fs.prepare_private_replace_with_parents(&path, b"new") + .err() + .unwrap() + .kind(), + io::ErrorKind::Unsupported + ); + assert_eq!(native::read(&path).unwrap(), b"old"); +} diff --git a/src/session.rs b/src/session.rs index 31efc087..32d84e4b 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1300,14 +1300,106 @@ fn set_display_name_in( session_id: &str, display_name: Option<&str>, ) -> Result, String> { + display_name_data_in( + fs::best_effort_global(), + root, + global_directory, + session_id, + display_name, + )? + .commit() +} + +struct DisplayNameData { + directory: PathBuf, + path: PathBuf, + output: Vec, + effective_title: Option, +} + +/// A prepared metadata replacement owns no accepted writes. Only the picker +/// can consume it at admission; dropping it (including a late reply) is inert. +#[cfg(feature = "tui")] +pub(crate) struct PreparedDisplayName { + replacement: fs::PreparedPrivateReplace, + effective_title: Option, +} + +#[cfg(feature = "tui")] +impl PreparedDisplayName { + pub(crate) fn commit(self) -> Result, String> { + self.replacement + .commit() + .map_err(|error| format!("could not commit session rename: {error}"))?; + Ok(self.effective_title) + } +} + +#[cfg(feature = "tui")] +pub(crate) fn prepare_display_name( + filesystem: &Fs, + root: &Path, + session_id: &str, + display_name: Option<&str>, +) -> Result { + prepare_display_name_in( + filesystem, + root, + &default_directory()?, + session_id, + display_name, + ) +} + +#[cfg(feature = "tui")] +pub(crate) fn prepare_display_name_in( + filesystem: &Fs, + root: &Path, + global_directory: &Path, + session_id: &str, + display_name: Option<&str>, +) -> Result { + // Capture the original namespace revision before any workspace/authority + // reads. Preparation uses only the independent read view, never global Fs. + let namespace = filesystem + .prepare_namespace() + .map_err(|error| format!("could not prepare session rename: {error}"))?; + let data = display_name_data_in( + namespace.filesystem(), + root, + global_directory, + session_id, + display_name, + )?; + let replacement = namespace + .prepare_private_replace_with_parents(&data.path, &data.output) + .map_err(|error| { + format!( + "could not prepare session metadata {}: {error}", + data.path.display() + ) + })?; + Ok(PreparedDisplayName { + replacement, + effective_title: data.effective_title, + }) +} + +fn display_name_data_in( + filesystem: &Fs, + root: &Path, + global_directory: &Path, + session_id: &str, + display_name: Option<&str>, +) -> Result { validate_id(session_id)?; - let root = fs::canonicalize(root).map_err(|error| { + let root = filesystem.canonicalize(root).map_err(|error| { format!( "could not resolve workspace root {}: {error}", root.display() ) })?; - if !fs::best_effort_global() + if !filesystem .metadata(&root) .map(|metadata| metadata.is_dir()) .unwrap_or(false) @@ -1318,8 +1410,9 @@ fn set_display_name_in( )); } - let authority = select_authority_for_rename(global_directory, &root, session_id)? - .ok_or_else(|| format!("session {session_id} was not found in {}", root.display()))?; + let authority = + select_authority_with(filesystem, global_directory, &root, session_id, false, true)? + .ok_or_else(|| format!("session {session_id} was not found in {}", root.display()))?; if catalog_is_subagent(&authority.historical_items, &authority.items) { return Err(format!( "session {session_id} was not found in {}", @@ -1336,39 +1429,35 @@ fn set_display_name_in( output.push(b'\n'); let directory = workspace_storage_directory(global_directory, &root); - fs::best_effort_global() - .create_dir_all(&directory) - .map_err(|error| { - format!( - "could not create session directory {}: {error}", - directory.display() - ) - })?; let path = metadata_path(&directory, session_id); - fs::best_effort_global() - .replace_private(&path, &output) - .map_err(|error| { - format!( - "could not replace session metadata {}: {error}", - path.display() - ) - })?; - Ok(effective_title) + Ok(DisplayNameData { + directory, + path, + output, + effective_title, + }) } -fn select_authority_for_rename( - directory: &Path, - root: &Path, - session_id: &str, -) -> Result, String> { - select_authority_with( - fs::best_effort_global(), - directory, - root, - session_id, - false, - true, - ) +impl DisplayNameData { + fn commit(self) -> Result, String> { + fs::best_effort_global() + .create_dir_all(&self.directory) + .map_err(|error| { + format!( + "could not create session directory {}: {error}", + self.directory.display() + ) + })?; + fs::best_effort_global() + .replace_private(&self.path, &self.output) + .map_err(|error| { + format!( + "could not replace session metadata {}: {error}", + self.path.display() + ) + })?; + Ok(self.effective_title) + } } pub(crate) fn is_safe_display_name_character(character: char) -> bool { @@ -1398,10 +1487,8 @@ fn validate_display_name(value: &str) -> Result { Ok(value.to_string()) } -fn read_display_name(directory: &Path, session_id: &str) -> Option { - let input = fs::best_effort_global() - .read(metadata_path(directory, session_id)) - .ok()?; +fn read_display_name_with(filesystem: &Fs, directory: &Path, session_id: &str) -> Option { + let input = filesystem.read(metadata_path(directory, session_id)).ok()?; let metadata: SessionMetadata = serde_json::from_slice(&input).ok()?; metadata .display_name @@ -1415,13 +1502,28 @@ fn catalog_for_workspace( root: &Path, global_directory: &Path, ) -> Result, String> { - let root = fs::canonicalize(root).map_err(|error| { + catalog_for_workspace_with(fs::best_effort_global(), root, global_directory) +} + +/// Read a disk catalog using an isolated filesystem with no pending writes. +/// Startup may abandon this read without sharing locks with storage recovery. +#[cfg(feature = "tui")] +pub(crate) fn catalog_with(filesystem: &Fs, root: &Path) -> Result, String> { + catalog_for_workspace_with(filesystem, root, &default_directory()?) +} + +fn catalog_for_workspace_with( + filesystem: &Fs, + root: &Path, + global_directory: &Path, +) -> Result, String> { + let root = filesystem.canonicalize(root).map_err(|error| { format!( "could not resolve workspace root {}: {error}", root.display() ) })?; - if !fs::best_effort_global() + if !filesystem .metadata(&root) .map(|metadata| metadata.is_dir()) .unwrap_or(false) @@ -1431,19 +1533,14 @@ fn catalog_for_workspace( root.display() )); } - let ids = list_ids_for_workspace(&root, global_directory)?; + let ids = list_ids_for_workspace_with(filesystem, &root, global_directory)?; let mut entries = Vec::with_capacity(ids.len()); for id in ids { // Discovery is best-effort per transcript: a damaged file or one caught // mid-append must not hide every other session in the workspace. - let Ok(Some(authority)) = select_authority_with( - fs::best_effort_global(), - global_directory, - &root, - &id, - false, - false, - ) else { + let Ok(Some(authority)) = + select_authority_with(filesystem, global_directory, &root, &id, false, false) + else { continue; }; if catalog_is_subagent(&authority.historical_items, &authority.items) { @@ -1461,7 +1558,7 @@ fn catalog_for_workspace( }) .max() .unwrap_or(0); - let file_updated = fs::best_effort_global() + let file_updated = filesystem .metadata(&authority.path) .and_then(|metadata| metadata.modified()) .ok() @@ -1469,7 +1566,7 @@ fn catalog_for_workspace( .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)) .unwrap_or(0); let directory = workspace_storage_directory(global_directory, &root); - let title = read_display_name(&directory, &id).or(title); + let title = read_display_name_with(filesystem, &directory, &id).or(title); entries.push(CatalogEntry { id, title, @@ -1658,23 +1755,27 @@ fn civil_date(days: i64) -> (i64, i64, i64) { (year, month, day) } -fn list_ids_for_workspace(root: &Path, global_directory: &Path) -> Result, String> { - let root = canonical_workspace(root); - let scoped_directory = workspace_storage_directory(global_directory, &root); - let legacy_directory = workspace_directory(&root); +fn list_ids_for_workspace_with( + filesystem: &Fs, + root: &Path, + global_directory: &Path, +) -> Result, String> { + let scoped_directory = workspace_storage_directory(global_directory, root); + let legacy_directory = workspace_directory(root); let mut ids = Vec::new(); - ids.extend(list_ids_in(&scoped_directory)?); - for id in list_ids_in(global_directory)? { + ids.extend(list_ids_in_with(filesystem, &scoped_directory)?); + for id in list_ids_in_with(filesystem, global_directory)? { let path = transcript_path(global_directory, &id); - if matches!( - transcript_workspace(&path, &id), - Ok(Some(stored)) if stored == root - ) { + let workspace = filesystem + .read(&path) + .map_err(|error| error.to_string()) + .and_then(|bytes| transcript_workspace_bytes(&path, &id, &bytes)); + if matches!(workspace, Ok(Some(stored)) if stored == root) { ids.push(id); } } // Its location scopes this pre-global-layout directory to the workspace. - ids.extend(list_ids_in(&legacy_directory)?); + ids.extend(list_ids_in_with(filesystem, &legacy_directory)?); ids.sort(); ids.dedup(); Ok(ids) @@ -1694,8 +1795,8 @@ fn belongs_to_workspace_in( Ok(select_authority(global_directory, &root, session_id)?.is_some()) } -pub(crate) fn list_ids_in(directory: &Path) -> Result, String> { - let entries = match fs::best_effort_global().read_dir(directory) { +fn list_ids_in_with(filesystem: &Fs, directory: &Path) -> Result, String> { + let entries = match filesystem.read_dir(directory) { Ok(entries) => entries, Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), Err(error) => { @@ -1900,8 +2001,12 @@ fn select_authority_with( } } StoredTranscript::Redirect(target) => { - let target = normalized_absolute(&target)?; - let scoped_target = normalized_absolute(&scoped)?; + let target = filesystem.canonicalize(&target).map_err(|error| { + format!("could not normalize {}: {error}", target.display()) + })?; + let scoped_target = filesystem.canonicalize(&scoped).map_err(|error| { + format!("could not normalize {}: {error}", scoped.display()) + })?; if target != scoped_target { return Err(format!("invalid session redirect in {}", path.display())); } @@ -4165,11 +4270,14 @@ mod tests { fs::write(directory.path().join("bad id.jsonl"), "invalid").unwrap(); fs::create_dir(directory.path().join("nested.jsonl")).unwrap(); - assert_eq!(list_ids_in(directory.path()).unwrap(), ["alpha", "zeta"]); + assert_eq!( + list_ids_in_with(fs::best_effort_global(), directory.path()).unwrap(), + ["alpha", "zeta"] + ); let not_directory = directory.path().join("plain-file"); fs::write(¬_directory, "not a directory").unwrap(); assert!( - list_ids_in(¬_directory) + list_ids_in_with(fs::best_effort_global(), ¬_directory) .unwrap_err() .contains("could not list session directory") ); @@ -5093,11 +5201,21 @@ mod tests { assert_eq!(item_text(&first_resumed.transcript[1]), "first-only"); assert_eq!(item_text(&second_resumed.transcript[1]), "second-only"); assert_eq!( - list_ids_for_workspace(&first, storage.path()).unwrap(), + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&first), + storage.path() + ) + .unwrap(), ["shared"] ); assert_eq!( - list_ids_for_workspace(&second, storage.path()).unwrap(), + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&second), + storage.path() + ) + .unwrap(), ["shared"] ); } @@ -5135,13 +5253,22 @@ mod tests { ); assert!(load_in(&second, storage.path(), "legacy-id").is_err()); assert_eq!( - list_ids_for_workspace(&first, storage.path()).unwrap(), + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&first), + storage.path() + ) + .unwrap(), ["legacy-id"] ); assert!( - list_ids_for_workspace(&second, storage.path()) - .unwrap() - .is_empty() + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&second), + storage.path() + ) + .unwrap() + .is_empty() ); let migrated = open_in(&first, storage.path(), "legacy-id", true, false, Vec::new()).unwrap(); @@ -5164,11 +5291,21 @@ mod tests { .unwrap(); assert_eq!(item_text(&second_open.transcript[0]), "second"); assert_eq!( - list_ids_for_workspace(&first, storage.path()).unwrap(), + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&first), + storage.path() + ) + .unwrap(), ["legacy-id"] ); assert_eq!( - list_ids_for_workspace(&second, storage.path()).unwrap(), + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&second), + storage.path() + ) + .unwrap(), ["legacy-id"] ); } @@ -5191,13 +5328,22 @@ mod tests { fs::write(legacy.join("legacy.jsonl"), format!("{record}\n")).unwrap(); assert_eq!( - list_ids_for_workspace(&first, storage.path()).unwrap(), + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&first), + storage.path() + ) + .unwrap(), ["legacy"] ); assert!( - list_ids_for_workspace(&second, storage.path()) - .unwrap() - .is_empty() + list_ids_for_workspace_with( + fs::best_effort_global(), + &canonical_workspace(&second), + storage.path() + ) + .unwrap() + .is_empty() ); assert!(belongs_to_workspace_in(&first, storage.path(), "legacy").unwrap()); diff --git a/src/storage_runtime.rs b/src/storage_runtime.rs index 364a8685..cba163c5 100644 --- a/src/storage_runtime.rs +++ b/src/storage_runtime.rs @@ -241,6 +241,58 @@ mod tests { ); } + #[test] + #[cfg(feature = "tui")] + fn idle_recovery_pass_preserves_prepared_session_rename() { + use crate::resilient_fs::{BestEffortStatus, DiskBackend, Fs}; + let root = tempfile::tempdir().unwrap(); + let directory = root.path().join(".kit/sessions"); + std::fs::create_dir_all(&directory).unwrap(); + let id = crate::session::new_id(); + let transcript = serde_json::json!({ + "schema_version": 1, + "session_id": id, + "generation": 1, + "item": agentkit_core::Item::text(agentkit_core::ItemKind::User, "Original title"), + }) + .to_string(); + std::fs::write(directory.join(format!("{id}.jsonl")), &transcript).unwrap(); + // Match the worker's distinct strict and best-effort services without + // touching process-global queues or relying on a timer/thread race. + let strict = Fs::new(std::sync::Arc::new(DiskBackend)); + let optional = strict.best_effort(1024 * 1024, 1024); + let storage = tempfile::tempdir().unwrap(); + let prepared = crate::session::prepare_display_name_in( + &optional, + root.path(), + storage.path(), + &id, + Some("Renamed"), + ) + .unwrap(); + assert_eq!( + super::recover_best_effort(&optional), + Some(BestEffortStatus::Ready) + ); + assert_eq!(strict.status().pending_operations, 0); + assert_eq!(prepared.commit().unwrap().as_deref(), Some("Renamed")); + let metadata_directory = std::fs::read_dir(storage.path()) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + assert_eq!( + std::fs::read_to_string(metadata_directory.join(format!("{id}.metadata.json"))) + .unwrap(), + "{\"display_name\":\"Renamed\"}\n" + ); + assert_eq!( + std::fs::read_to_string(directory.join(format!("{id}.jsonl"))).unwrap(), + transcript + ); + } + #[test] fn optional_recovery_does_not_operate_on_a_strict_service() { let strict = diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 26c862a8..018aa80b 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -16,6 +16,9 @@ mod image; mod keyboard_tests; mod markdown; mod source; +mod startup; + +pub use startup::pick_session; mod theme; mod ui; mod wrap; @@ -1082,23 +1085,6 @@ async fn bounded_agent_request( } } -async fn bounded_startup_request( - request: impl std::future::Future, - exit: &mut oneshot::Receiver>, - timeout: Duration, -) -> Result { - // One-shot ties prefer request, exit, then timeout (all formerly allowed). - let request = pin!(request); - let timeout = pin!(tokio::time::sleep(timeout)); - match select(request, select(&mut *exit, timeout)).await { - Either::Left((output, _)) => Ok(output), - Either::Right((Either::Left((status, _)), _)) => Err(RequestFailure::AgentExited( - status.ok().and_then(Result::ok), - )), - Either::Right((Either::Right(_), _)) => Err(RequestFailure::TimedOut), - } -} - async fn bounded_cancellable_request( request: impl std::future::Future, exit: &mut oneshot::Receiver>, @@ -1338,6 +1324,7 @@ pub async fn run_with_reasoning_effort( resume, force, false, + &mut Stop::new()?, ) .await } @@ -1357,6 +1344,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( resume: Option<&str>, force: bool, voice_enabled: bool, + stop: &mut Stop, ) -> Result<(), Box> { // The agent fixes itself to the canonical root, so the client resolves it // up front: the header names a real directory and the ACP session opens on @@ -1400,17 +1388,25 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( force, )?; let child_mcp_config = mcp_config.clone(); - tokio::task::spawn_blocking(move || { - if let Some(home) = std::env::var_os("HOME").filter(|home| !home.is_empty()) { - crate::resilient_fs::global() - .require_disk(PathBuf::from(home).join(".kit/config.toml"))?; - } - if let Some(path) = child_mcp_config { - crate::resilient_fs::global().require_disk(path)?; - } - Ok::<_, std::io::Error>(()) - }) - .await??; + let prepared = stop + .until(async { + tokio::task::spawn_blocking(move || { + if let Some(home) = std::env::var_os("HOME").filter(|home| !home.is_empty()) { + crate::resilient_fs::global() + .require_disk(PathBuf::from(home).join(".kit/config.toml"))?; + } + if let Some(path) = child_mcp_config { + crate::resilient_fs::global().require_disk(path)?; + } + Ok::<_, std::io::Error>(()) + }) + .await + }) + .await; + let Some(prepared) = prepared else { + return Ok(()); + }; + prepared??; let auth_invocation = AgentInvocation::from_command(command.as_std()); detach_from_controlling_terminal(&mut command); let mut child = command @@ -1542,8 +1538,9 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( ) .block_task(); let initialized = - match bounded_startup_request(initialize, &mut exit_rx, HANDSHAKE).await { - Ok(initialized) => initialized?, + match bounded_cancellable_request(initialize, &mut exit_rx, stop.requested(), HANDSHAKE).await { + Ok(Some(initialized)) => initialized?, + Ok(None) => return Ok(()), Err(failure) => { return Err(request_failure( failure, @@ -1573,18 +1570,20 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( }) }); - let initial = match bounded_startup_request( + let initial = match bounded_cancellable_request( request_initial_session( &connection, resume_session_id.as_deref(), &root, ), &mut exit_rx, + stop.requested(), HANDSHAKE, ) .await { - Ok(session) => session, + Ok(Some(session)) => session, + Ok(None) => return Ok(()), Err(failure) => { return Err(request_failure( failure, @@ -1609,10 +1608,6 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( result => result, }; - // Install every fallible signal handler before changing terminal modes so - // an installation failure cannot leave the caller's terminal altered. - let mut stop = - Stop::new().map_err(agent_client_protocol::Error::into_internal_error)?; let (mut terminal, mut images) = enter().map_err(agent_client_protocol::Error::into_internal_error)?; let mut native_hyperlinks = hyperlinks::HyperlinkRenderer::default(); @@ -1712,7 +1707,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( &auth_invocation, &root, &method, - &mut stop, + stop, ); leave(&mut terminal); println!("Starting {}…", method.name); @@ -2390,7 +2385,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( &auth_invocation, &root, &method, - &mut stop, + stop, ); leave(&mut terminal); println!("Starting {}…", method.name); @@ -3307,7 +3302,7 @@ fn restore_modes() { /// the shell in raw mode with mouse reporting on — every later mouse move /// arrives at the prompt as garbage. Holding the signal streams for the whole /// session and returning through the normal exit keeps that from happening. -struct Stop { +pub struct Stop { #[cfg(unix)] interrupt: tokio::signal::unix::Signal, #[cfg(unix)] @@ -3317,8 +3312,18 @@ struct Stop { } impl Stop { + /// Cancel preparation before launching any detached child tasks. + pub async fn until(&mut self, work: impl std::future::Future) -> Option { + let stopped = pin!(self.requested()); + let work = pin!(work); + match select(stopped, work).await { + Either::Left(_) => None, + Either::Right((result, _)) => Some(result), + } + } + #[cfg(unix)] - fn new() -> std::io::Result { + pub fn new() -> std::io::Result { use tokio::signal::unix::{SignalKind, signal}; Ok(Self { interrupt: signal(SignalKind::interrupt())?, @@ -3328,7 +3333,7 @@ impl Stop { } #[cfg(not(unix))] - fn new() -> std::io::Result { + pub fn new() -> std::io::Result { Ok(Self {}) } @@ -7209,14 +7214,20 @@ mod signal_tests { async fn startup_requests_preserve_success_timeout_and_agent_exit() { let (exit_tx, mut exit_rx) = tokio::sync::oneshot::channel(); assert!(matches!( - super::bounded_startup_request(future::ready(7), &mut exit_rx, Duration::from_secs(30)) - .await, - Ok(7) + super::bounded_cancellable_request( + future::ready(7), + &mut exit_rx, + future::pending(), + Duration::from_secs(30) + ) + .await, + Ok(Some(7)) )); assert!(matches!( - super::bounded_startup_request( + super::bounded_cancellable_request( future::pending::<()>(), &mut exit_rx, + future::pending(), Duration::from_secs(30) ) .await, @@ -7224,9 +7235,10 @@ mod signal_tests { )); drop(exit_tx); assert!(matches!( - super::bounded_startup_request( + super::bounded_cancellable_request( future::pending::<()>(), &mut exit_rx, + future::pending(), Duration::from_secs(30) ) .await, @@ -7360,6 +7372,33 @@ mod signal_tests { assert!(closed.is_none()); } + #[tokio::test] + async fn startup_stop_survives_handoff_and_cancels_requests() { + for signal in ["-INT", "-TERM", "-HUP"] { + let mut stop = Stop::new().unwrap(); + // A successful preparation leaves the same streams alive, including + // signals received before the next startup future begins polling. + assert_eq!(stop.until(future::ready(7)).await, Some(7)); + std::process::Command::new("kill") + .args([signal, &std::process::id().to_string()]) + .status() + .unwrap(); + let (_exit_tx, mut exit_rx) = tokio::sync::oneshot::channel(); + let result = tokio::time::timeout( + Duration::from_secs(3), + super::bounded_cancellable_request( + future::pending::<()>(), + &mut exit_rx, + stop.requested(), + Duration::from_secs(30), + ), + ) + .await + .unwrap(); + assert!(matches!(result, Ok(None))); + } + } + /// A client killed from outside must still reach its restore path, or it /// leaves the shell in raw mode with mouse reporting on. #[tokio::test] diff --git a/src/tui/startup.rs b/src/tui/startup.rs new file mode 100644 index 00000000..ee49a96b --- /dev/null +++ b/src/tui/startup.rs @@ -0,0 +1,305 @@ +//! Session selection before launching an agent or opening a persisted session. + +use std::path::Path; + +use crossterm::event::{Event, EventStream, KeyCode, KeyEventKind, KeyModifiers}; +use futures_util::{ + StreamExt, + future::{Either, select}, +}; + +use super::{ + Action, App, ClipboardResult, ClipboardRoute, Failure, Stop, Update, enter, handle, leave, + read_clipboard, ui, +}; + +enum PickerUpdate { + Tick, + Session(Update), + RenamePrepared { + session_id: String, + display_name: Option, + result: Result, + }, + Clipboard(ClipboardRoute, ClipboardResult), +} + +/// Pick an existing workspace session without creating or resuming a session. +/// Cancellation and an empty catalog return `None`. +/// The caller must await this future to completion and signal cancellation via +/// `Stop`: dropping it cannot asynchronously drain admitted storage commits. +pub async fn pick_session( + root: &Path, + stop: &mut Stop, +) -> Result, Box> { + let backend = std::sync::Arc::new(crate::resilient_fs::DiskBackend); + let Some((root, entries)) = scan_catalog(root.to_path_buf(), backend, stop).await? else { + return Ok(None); + }; + if entries.is_empty() { + eprintln!("no resumable sessions for workspace {}", root.display()); + return Ok(None); + } + + let mut app = App::new(root.clone(), String::new(), String::new(), String::new()); + app.apply(Update::SessionCatalog(Ok(entries))); + // The caller retains signal handlers across selection and startup. Restore the + // terminal on both successful cancellation and fallible reads/draws. + let (mut terminal, mut images) = enter()?; + let mut renames = RenameCommits::default(); + let result = async { + let mut events = EventStream::new(); + let mut clipboard_pending = false; + let mut ticker = tokio::time::interval(super::TICK); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let (updates_tx, mut updates_rx) = tokio::sync::mpsc::unbounded_channel(); + loop { + terminal.draw(|frame| ui::draw(frame, &mut app, &mut images))?; + let action = { + let event = std::pin::pin!(events.next()); + let redraw = app.needs_redraw_tick() || images.pending(); + let update = std::pin::pin!(async { + if !redraw { + return updates_rx.recv().await; + } + let update = std::pin::pin!(updates_rx.recv()); + let tick = std::pin::pin!(ticker.tick()); + match select(tick, update).await { + Either::Left(_) => Some(PickerUpdate::Tick), + Either::Right((update, _)) => update, + } + }); + let stopped = std::pin::pin!(stop.requested()); + match select(stopped, select(event, update)).await { + Either::Left(_) => return Ok(None), + Either::Right((Either::Left((event, _)), _)) => match event { + Some(Ok(event)) => handle_picker_event(&mut app, event), + Some(Err(error)) => return Err(error.into()), + None => return Ok(None), + }, + Either::Right((Either::Right((update, _)), _)) => match update { + Some(PickerUpdate::Tick) => { + app.tick(); + Action::None + } + Some(PickerUpdate::Session(update)) => { + app.apply(update); + Action::None + } + Some(PickerUpdate::RenamePrepared { + session_id, + display_name, + result, + }) => { + match result { + Ok(prepared) => { + // This branch is the write-admission boundary. Only the + // picker owns prepared values; a stopped picker drops them. + renames.admit( + prepared, + session_id, + display_name, + updates_tx.clone(), + ); + } + Err(error) => app.apply(Update::SessionRenamed { + session_id, + display_name, + result: Err(error), + }), + } + Action::None + } + Some(PickerUpdate::Clipboard(route, result)) => { + clipboard_pending = false; + // A cancelled or submitted rename must not receive an + // earlier clipboard read when another dialog opens. + if app.clipboard_route() == route { + match result { + ClipboardResult::Text(text) => { + super::handle_paste(&mut app, &text); + } + ClipboardResult::Error(error) => app.note(error), + _ => {} + } + } + Action::None + } + None => return Ok(None), + }, + } + }; + match action { + Action::Quit => return Ok(None), + Action::Resume(id) => return Ok(Some(id)), + Action::RenameSession { + session_id, + display_name, + } => { + start_rename_preparation( + root.clone(), + session_id, + display_name, + crate::resilient_fs::best_effort_global().clone(), + updates_tx.clone(), + )?; + } + Action::ReadClipboard(route, mode) => { + if clipboard_pending { + app.toast("clipboard is busy; try again"); + continue; + } + clipboard_pending = true; + let updates = updates_tx.clone(); + tokio::task::spawn_blocking(move || { + let result = read_clipboard(&route, mode); + let _ = updates.send(PickerUpdate::Clipboard(route, result)); + }); + } + _ => {} + } + } + } + .await; + leave(&mut terminal); + // No preparation can submit writes after the receiver is dropped above. + // Accepted commits remain owned: drain them before the caller can recover + // global storage. An indefinitely stalled accepted write is not cancellable. + renames.drain().await?; + result +} + +/// Admission is owned by the picker, not the preparation thread. This owner is +/// drained after terminal restoration on every ordinary picker exit/error path. +#[derive(Default)] +struct RenameCommits { + workers: Vec, String>>>, +} +impl RenameCommits { + fn admit( + &mut self, + prepared: crate::session::PreparedDisplayName, + session_id: String, + display_name: Option, + updates: tokio::sync::mpsc::UnboundedSender, + ) { + self.workers.push(tokio::spawn(async move { + let result = tokio::task::spawn_blocking(move || prepared.commit()) + .await + .map_err(|error| format!("session rename worker failed: {error}")) + .and_then(|result| result); + let _ = updates.send(PickerUpdate::Session(Update::SessionRenamed { + session_id, + display_name, + result: result.clone(), + })); + result + })); + } + + async fn drain(self) -> Result<(), String> { + let mut failure = None; + for worker in self.workers { + match worker.await { + Ok(Ok(_)) => {} + Ok(Err(error)) => { + // A closed stderr must not unwind past still-owned commits. + use std::io::Write; + let _ = writeln!(std::io::stderr(), "{error}"); + } + Err(error) => { + failure.get_or_insert_with(|| format!("session rename worker failed: {error}")); + } + } + } + failure.map_or(Ok(()), Err) + } +} + +fn start_rename_preparation( + root: std::path::PathBuf, + session_id: String, + display_name: Option, + filesystem: crate::resilient_fs::Fs, + updates: tokio::sync::mpsc::UnboundedSender, +) -> std::io::Result<()> { + let (tx, rx) = tokio::sync::oneshot::channel(); + let id = session_id.clone(); + let name = display_name.clone(); + std::thread::Builder::new() + .name("session-rename-prepare".into()) + .spawn(move || { + let result = + crate::session::prepare_display_name(&filesystem, &root, &id, name.as_deref()); + let _ = tx.send(result); + })?; + tokio::spawn(async move { + let result = rx + .await + .map_err(|error| format!("session rename preparation worker failed: {error}")) + .and_then(|result| result); + let _ = updates.send(PickerUpdate::RenamePrepared { + session_id, + display_name: display_name.map(|name| name.trim().to_string()), + result, + }); + }); + Ok(()) +} + +// This worker only reads disk, with its own empty filesystem namespace: it +// cannot migrate sessions, recover accepted writes, or hold global recovery +// locks. Dropping its JoinHandle intentionally detaches it on cancellation; +// unlike spawn_blocking, it does not delay Tokio runtime teardown. +async fn scan_catalog( + root: std::path::PathBuf, + backend: std::sync::Arc, + stop: &mut Stop, +) -> Result< + Option<(std::path::PathBuf, Vec)>, + Box, +> { + let (tx, rx) = tokio::sync::oneshot::channel(); + let worker = std::thread::Builder::new() + .name("session-catalog".into()) + .spawn(move || { + let result = (|| { + let root = backend + .canonicalize(&root) + .map_err(|error| format!("{}: {error}", root.display()))?; + let filesystem = crate::resilient_fs::Fs::new(backend); + let entries = + crate::session::catalog_with(&filesystem, &root).map_err(|error| { + format!("could not list sessions for {}: {error}", root.display()) + })?; + Ok::<_, String>((root, entries)) + })(); + let _ = tx.send(result); + })?; + let Some(result) = stop.until(rx).await else { + return Ok(None); + }; + // The sender has finished scanning; joining here cannot wait on disk I/O. + worker + .join() + .map_err(|_| Failure("session catalog worker panicked".into()))?; + Ok(Some(result?.map_err(Failure)?)) +} + +fn handle_picker_event(app: &mut App, event: Event) -> Action { + if matches!(&event, Event::Key(key) if key.kind == KeyEventKind::Press + && key.code == KeyCode::Char('c') && key.modifiers.contains(KeyModifiers::CONTROL)) + { + return Action::Quit; + } + let action = handle(app, event); + if app.session_dialog.is_none() && !matches!(action, Action::Resume(_)) { + Action::Quit + } else { + action + } +} + +#[cfg(all(test, unix))] +#[path = "startup_tests.rs"] +mod tests; diff --git a/src/tui/startup_tests.rs b/src/tui/startup_tests.rs new file mode 100644 index 00000000..112d29b0 --- /dev/null +++ b/src/tui/startup_tests.rs @@ -0,0 +1,528 @@ +//! Exercise cancellation through actual signals, recovery, and runtime teardown +//! in a separate process. The filesystem boundary models a permanently stuck read. +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] + +use crate::resilient_fs::{ + self as fs, Backend, BackendFile, BackendLease, DiskBackend, DiskEntry, DiskOpenOptions, + LeaseRequest, +}; +use std::{ + io, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + sync::Arc, + time::{Duration, Instant}, +}; + +struct StalledDisk { + entered: PathBuf, + canonicalize: bool, + rename_stall: Option, +} +impl StalledDisk { + fn stall_rename(&self, operation: &str) -> io::Result<()> { + if self.rename_stall.as_deref() == Some(operation) { + std::fs::write(&self.entered, operation)?; + let release = self.entered.with_file_name("release"); + while !release.exists() { + std::thread::sleep(Duration::from_millis(10)); + } + } + Ok(()) + } +} + +impl Drop for StalledDisk { + fn drop(&mut self) { + if self.rename_stall.is_some() { + // The isolated worker has relinquished its last backend reference. + std::fs::write(self.entered.with_file_name("backend-dropped"), b"done").unwrap(); + } + } +} + +impl Backend for StalledDisk { + fn read_dir(&self, path: &Path) -> io::Result> { + if self.canonicalize || self.rename_stall.is_some() { + return DiskBackend.read_dir(path); + } + std::fs::write(&self.entered, b"reading")?; + loop { + std::thread::park(); + } + } + fn open(&self, path: &Path, options: &DiskOpenOptions) -> io::Result> { + if options.write || options.append { + self.stall_rename("write-open")?; + } else if path + .extension() + .is_some_and(|extension| extension == "jsonl") + { + self.stall_rename("authority")?; + } + DiskBackend.open(path, options) + } + fn metadata(&self, path: &Path, follow: bool) -> io::Result { + if path.to_string_lossy().ends_with(".metadata.json") { + self.stall_rename("metadata")?; + } + DiskBackend.metadata(path, follow) + } + fn read_link(&self, path: &Path) -> io::Result { + DiskBackend.read_link(path) + } + fn canonicalize(&self, path: &Path) -> io::Result { + self.stall_rename("canonicalize")?; + if self.canonicalize { + std::fs::write(&self.entered, b"canonicalizing")?; + loop { + std::thread::park(); + } + } + DiskBackend.canonicalize(path) + } + fn create_dir(&self, path: &Path, private: bool) -> io::Result<()> { + DiskBackend.create_dir(path, private) + } + fn remove_file(&self, path: &Path) -> io::Result<()> { + DiskBackend.remove_file(path) + } + fn remove_dir(&self, path: &Path) -> io::Result<()> { + DiskBackend.remove_dir(path) + } + fn rename(&self, from: &Path, to: &Path) -> io::Result<()> { + self.stall_rename("commit")?; + DiskBackend.rename(from, to) + } + fn set_permissions(&self, path: &Path, permissions: std::fs::Permissions) -> io::Result<()> { + DiskBackend.set_permissions(path, permissions) + } + fn sync_directory(&self, path: &Path) -> io::Result<()> { + DiskBackend.sync_directory(path) + } + fn acquire_lease(&self, request: &LeaseRequest) -> io::Result> { + DiskBackend.acquire_lease(request) + } + fn open_beneath(&self, root: &Path, relative: &Path) -> io::Result> { + if relative + .extension() + .is_some_and(|extension| extension == "jsonl") + { + self.stall_rename("authority")?; + } + DiskBackend.open_beneath(root, relative) + } +} + +struct Reap(Child); +impl Drop for Reap { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[test] +fn cancelled_catalog_does_not_hold_runtime_teardown() { + const CHILD: &str = "KIT_CATALOG_TEARDOWN_CHILD"; + if let Some(directory) = std::env::var_os(CHILD) { + let directory = PathBuf::from(directory); + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + fs::start_recovery_worker(); + let backend = Arc::new(StalledDisk { + entered: directory.join("entered"), + canonicalize: std::env::var_os("KIT_STALL_CANONICALIZE").is_some(), + rename_stall: None, + }); + let mut stop = super::Stop::new().unwrap(); + assert!( + super::scan_catalog(directory.clone(), backend, &mut stop) + .await + .unwrap() + .is_none() + ); + // Keep precisely the storage finish sequence used by CLI main. + fs::finish_best_effort_recovery(fs::best_effort_global()); + fs::finish_recovery(fs::global()).unwrap(); + std::fs::write(directory.join("recovered"), b"finished").unwrap(); + }); + drop(runtime); + return; + } + // Exercise every signal on one stall, and each other stall with SIGINT. + for (signal, canonicalize) in [ + ("-INT", false), + ("-TERM", false), + ("-HUP", false), + ("-INT", true), + ] { + let directory = tempfile::tempdir().unwrap(); + std::fs::create_dir_all(directory.path().join(".kit/sessions")).unwrap(); + let mut command = Command::new(std::env::current_exe().unwrap()); + command.env_remove("KIT_STALL_CANONICALIZE"); + if canonicalize { + command.env("KIT_STALL_CANONICALIZE", "1"); + } + let mut child = Reap( + command + .args([ + "--exact", + "tui::startup::tests::cancelled_catalog_does_not_hold_runtime_teardown", + "--nocapture", + ]) + .env(CHILD, directory.path()) + .env("HOME", directory.path()) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ); + let deadline = Instant::now() + Duration::from_secs(10); + while !directory.path().join("entered").exists() { + assert!( + child.0.try_wait().unwrap().is_none(), + "child exited before scanning" + ); + assert!( + Instant::now() < deadline, + "child did not enter catalog read" + ); + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + Command::new("kill") + .args([signal, &child.0.id().to_string()]) + .status() + .unwrap() + .success() + ); + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + if let Some(status) = child.0.try_wait().unwrap() { + break status; + } + assert!( + Instant::now() < deadline, + "cancelled scan held recovery/runtime teardown" + ); + std::thread::sleep(Duration::from_millis(10)); + }; + assert!(status.success(), "child failed: {status}"); + assert_eq!( + std::fs::read(directory.path().join("recovered")).unwrap(), + b"finished" + ); + } +} + +#[tokio::test] +async fn catalog_returns_canonical_workspace_root() { + let directory = tempfile::tempdir().unwrap(); + let workspace = directory.path().join("workspace"); + std::fs::create_dir(&workspace).unwrap(); + let alias = directory.path().join("alias"); + std::os::unix::fs::symlink(&workspace, &alias).unwrap(); + let backend = Arc::new(DiskBackend); + let mut stop = super::Stop::new().unwrap(); + let (root, entries) = super::scan_catalog(alias, backend, &mut stop) + .await + .unwrap() + .unwrap(); + assert_eq!(root, workspace.canonicalize().unwrap()); + assert!(entries.is_empty()); +} + +#[tokio::test] +async fn catalog_invalid_root_preserves_diagnostic() { + let directory = tempfile::tempdir().unwrap(); + for root in [directory.path().join("missing"), PathBuf::new()] { + let expected = format!("{}: {}", root.display(), root.canonicalize().unwrap_err()); + let backend = Arc::new(DiskBackend); + let mut stop = super::Stop::new().unwrap(); + let error = super::scan_catalog(root, backend, &mut stop) + .await + .unwrap_err(); + assert_eq!(error.to_string(), expected); + } +} + +// Each subprocess owns HOME and global recovery state. Never mutate HOME in the +// test runner: other session tests can be creating transcripts concurrently. +fn rename_fixture(directory: &Path) -> PathBuf { + let root = directory.join("workspace"); + std::fs::create_dir(&root).unwrap(); + drop( + crate::session::open( + &root, + "rename-target", + false, + false, + vec![agentkit_core::Item::text( + agentkit_core::ItemKind::System, + "startup rename fixture", + )], + ) + .unwrap(), + ); + crate::session::set_display_name(&root, "rename-target", Some("original")).unwrap(); + assert_rename_title(&root, "original"); + root +} + +fn assert_rename_title(root: &Path, expected: &str) { + // A new namespace observes disk rather than a successful in-memory overlay. + let filesystem = fs::Fs::new(Arc::new(DiskBackend)); + let entries = crate::session::catalog_with(&filesystem, root).unwrap(); + let entry = entries + .iter() + .find(|entry| entry.id == "rename-target") + .unwrap(); + assert_eq!(entry.title.as_deref(), Some(expected)); +} + +fn wait_for_marker(path: &Path, child: &mut Reap) { + let deadline = Instant::now() + Duration::from_secs(10); + while !path.exists() { + assert!( + child.0.try_wait().unwrap().is_none(), + "child exited before {}", + path.display() + ); + assert!( + Instant::now() < deadline, + "child did not reach {}", + path.display() + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +fn wait_for_rename_child(child: &mut Reap) { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Some(status) = child.0.try_wait().unwrap() { + assert!(status.success(), "rename child failed: {status}"); + return; + } + assert!( + Instant::now() < deadline, + "rename held recovery/runtime teardown" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +fn rename_child(test: &str, directory: &Path, operation: &str) -> Reap { + Reap( + Command::new(std::env::current_exe().unwrap()) + .args(["--exact", test, "--nocapture"]) + .env("KIT_RENAME_TEARDOWN_CHILD", directory) + .env("KIT_RENAME_STALL", operation) + .env("HOME", directory) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ) +} + +#[test] +fn cancelled_rename_does_not_hold_recovery_or_write_after_shutdown() { + if let Some(directory) = std::env::var_os("KIT_RENAME_TEARDOWN_CHILD") { + let directory = PathBuf::from(directory); + let root = rename_fixture(&directory); + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + fs::start_recovery_worker(); + let mut stop = super::Stop::new().unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let filesystem = fs::Fs::new(Arc::new(StalledDisk { + entered: directory.join("entered"), + canonicalize: false, + rename_stall: Some(std::env::var("KIT_RENAME_STALL").unwrap()), + })); + super::start_rename_preparation( + root.clone(), + "rename-target".into(), + Some("changed".into()), + filesystem.clone(), + tx, + ) + .unwrap(); + assert!( + stop.until(rx.recv()).await.is_none(), + "preparation completed before cancellation" + ); + drop(rx); + // Recovery must finish while preparation is still inside the backend. + fs::finish_best_effort_recovery(&filesystem); + fs::finish_best_effort_recovery(fs::best_effort_global()); + fs::finish_recovery(fs::global()).unwrap(); + }); + drop(runtime); + std::fs::write(directory.join("recovered"), b"finished").unwrap(); + // Runtime teardown also cancels the forwarding task. Let the detached + // worker finish, so absence of a late write is not a timing assertion. + std::fs::write(directory.join("release"), b"go").unwrap(); + let deadline = Instant::now() + Duration::from_secs(10); + while !directory.join("backend-dropped").exists() { + assert!( + Instant::now() < deadline, + "released preparation did not finish" + ); + std::thread::sleep(Duration::from_millis(10)); + } + assert_rename_title(&root, "original"); + return; + } + for operation in ["canonicalize", "authority", "metadata", "write-open"] { + let directory = tempfile::tempdir().unwrap(); + let mut child = rename_child( + "tui::startup::tests::cancelled_rename_does_not_hold_recovery_or_write_after_shutdown", + directory.path(), + operation, + ); + wait_for_marker(&directory.path().join("entered"), &mut child); + assert!( + Command::new("kill") + .args(["-INT", &child.0.id().to_string()]) + .status() + .unwrap() + .success() + ); + wait_for_rename_child(&mut child); + assert_eq!( + std::fs::read(directory.path().join("recovered")).unwrap(), + b"finished" + ); + } +} + +#[test] +fn admitted_rename_is_drained_before_recovery() { + if let Some(directory) = std::env::var_os("KIT_RENAME_TEARDOWN_CHILD") { + let directory = PathBuf::from(directory); + let root = rename_fixture(&directory); + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + // Observe successful preparation and completion in this same child, + // then exercise shutdown with the picker's receiver gone. + let filesystem = fs::Fs::new(Arc::new(DiskBackend)); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + super::start_rename_preparation( + root.clone(), + "rename-target".into(), + Some(" changed ".into()), + filesystem.clone(), + tx, + ) + .unwrap(); + let super::PickerUpdate::RenamePrepared { + session_id, + display_name, + result, + } = rx.recv().await.unwrap() + else { + panic!("expected prepared rename"); + }; + assert_eq!(session_id, "rename-target"); + assert_eq!(display_name.as_deref(), Some("changed")); + let prepared = result.unwrap(); + assert_rename_title(&root, "original"); + let (updates, mut completed) = tokio::sync::mpsc::unbounded_channel(); + let mut commits = super::RenameCommits::default(); + commits.admit(prepared, session_id, display_name, updates); + commits.drain().await.unwrap(); + let super::PickerUpdate::Session(super::Update::SessionRenamed { result, .. }) = + completed.recv().await.unwrap() + else { + panic!("expected completed rename"); + }; + assert_eq!(result.unwrap().as_deref(), Some("changed")); + fs::finish_best_effort_recovery(&filesystem); + assert_rename_title(&root, "changed"); + let mut stop = super::Stop::new().unwrap(); + let filesystem = fs::Fs::new(Arc::new(StalledDisk { + entered: directory.join("entered"), + canonicalize: false, + rename_stall: Some("commit".into()), + })); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + super::start_rename_preparation( + root.clone(), + "rename-target".into(), + Some("drained".into()), + filesystem.clone(), + tx.clone(), + ) + .unwrap(); + let super::PickerUpdate::RenamePrepared { + session_id, + display_name, + result, + } = rx.recv().await.unwrap() + else { + panic!("expected prepared rename"); + }; + let mut commits = super::RenameCommits::default(); + commits.admit(result.unwrap(), session_id, display_name, tx); + // The parent sends a real signal only after the admitted write stalls. + assert!(stop.until(std::future::pending::<()>()).await.is_none()); + drop(rx); + let mut drain = std::pin::pin!(commits.drain()); + assert!( + futures_util::poll!(drain.as_mut()).is_pending(), + "admitted write escaped its owner" + ); + std::fs::write(directory.join("draining"), b"waiting").unwrap(); + drain.await.unwrap(); + fs::finish_best_effort_recovery(&filesystem); + fs::finish_best_effort_recovery(fs::best_effort_global()); + fs::finish_recovery(fs::global()).unwrap(); + assert_rename_title(&root, "drained"); + }); + drop(runtime); + std::fs::write(directory.join("recovered"), b"finished").unwrap(); + return; + } + let directory = tempfile::tempdir().unwrap(); + let mut child = rename_child( + "tui::startup::tests::admitted_rename_is_drained_before_recovery", + directory.path(), + "commit", + ); + wait_for_marker(&directory.path().join("entered"), &mut child); + assert!( + Command::new("kill") + .args(["-INT", &child.0.id().to_string()]) + .status() + .unwrap() + .success() + ); + wait_for_marker(&directory.path().join("draining"), &mut child); + assert!(!directory.path().join("recovered").exists()); + std::fs::write(directory.path().join("release"), b"go").unwrap(); + wait_for_rename_child(&mut child); + assert_eq!( + std::fs::read(directory.path().join("recovered")).unwrap(), + b"finished" + ); +} diff --git a/tests/cli.rs b/tests/cli.rs index 0beb4bab..15165e76 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -34,6 +34,66 @@ fn write_session(home: &Path, root: &Path, id: &str) -> std::path::PathBuf { directory.join(format!("{id}.metadata.json")) } +#[cfg(feature = "tui")] +#[test] +fn tui_resume_picker_empty_workspace_exits_without_starting_a_session() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let mut command = Command::new(env!("CARGO_BIN_EXE_kit")); + command + .env("HOME", home.path()) + .args(["tui", "--resume", "--root"]) + .arg(root.path()) + .args(["--credential-store", "memory"]); + let output = command.output().unwrap(); + assert!(output.status.success(), "{output:?}"); + assert!( + String::from_utf8_lossy(&output.stderr).contains("no resumable sessions for workspace"), + "{output:?}" + ); + assert!(!home.path().join(".kit/sessions").exists()); +} + +#[cfg(feature = "tui")] +#[test] +fn tui_resume_picker_uses_configured_root_and_reports_catalog_errors() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let config_dir = home.path().join(".kit"); + fs::create_dir_all(&config_dir).unwrap(); + fs::write( + config_dir.join("config.toml"), + format!("root = {:?}\n", root.path()), + ) + .unwrap(); + let run = || { + Command::new(env!("CARGO_BIN_EXE_kit")) + .env("HOME", home.path()) + .args(["tui", "--resume", "--credential-store", "memory"]) + .output() + .unwrap() + }; + let output = run(); + assert!(output.status.success(), "{output:?}"); + assert!( + String::from_utf8_lossy(&output.stderr) + .contains(root.path().canonicalize().unwrap().to_str().unwrap()), + "{output:?}" + ); + assert!(!config_dir.join("sessions").exists()); + fs::write(config_dir.join("sessions"), "not a directory").unwrap(); + let output = run(); + assert!(!output.status.success(), "{output:?}"); + assert!( + String::from_utf8_lossy(&output.stderr).contains("could not list sessions for"), + "{output:?}" + ); + assert_eq!( + fs::read_to_string(config_dir.join("sessions")).unwrap(), + "not a directory" + ); +} + #[test] fn terminal_auth_arguments_run_login_from_acp_server_invocations() { let home = tempfile::tempdir().unwrap(); diff --git a/tests/startup_picker.rs b/tests/startup_picker.rs new file mode 100644 index 00000000..e67a2198 --- /dev/null +++ b/tests/startup_picker.rs @@ -0,0 +1,448 @@ +//! Real startup-picker interactions through a controlling terminal. +#![cfg(all(unix, feature = "tui"))] +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +use agentkit_core::{Item, ItemKind}; +use std::{ + fs::{self, File}, + io::{Read, Write}, + os::{ + fd::{AsRawFd, FromRawFd}, + unix::process::CommandExt, + }, + path::{Path, PathBuf}, + process::{Child, Command, ExitStatus, Stdio}, + time::{Duration, Instant}, +}; +const BROWSING: &str = "r rename"; +fn session(home: &Path, root: &Path, id: &str, title: &str) -> PathBuf { + let root = root.canonicalize().unwrap(); + let identity = blake3::hash(root.as_os_str().as_encoded_bytes()); + let directory = home + .join(".kit/sessions") + .join(format!("w-{}", identity.to_hex())); + fs::create_dir_all(&directory).unwrap(); + let record = serde_json::json!({"schema_version": 3, "session_id": id, "generation": 1, "workspace_root": root, "item": Item::text(ItemKind::User, title)}); + let path = directory.join(format!("{id}.jsonl")); + fs::write(&path, format!("{record}\n")).unwrap(); + path +} +struct Picker { + child: Child, + master: File, + output: String, + queries: String, +} +impl Drop for Picker { + fn drop(&mut self) { + // SAFETY: the child leads its own session; reap any CLI agent descendants too. + unsafe { + libc::kill(-(self.child.id() as i32), libc::SIGKILL); + } + let _ = self.child.kill(); + // Close the PTY before reaping: macOS may still be unwinding terminal I/O. + drop(std::mem::replace( + &mut self.master, + File::open("/dev/null").unwrap(), + )); + let _ = self.child.wait(); + } +} +impl Picker { + fn start(home: &Path, root: &Path) -> Self { + Self::with_args(home, root, &[]) + } + fn with_args(home: &Path, root: &Path, args: &[&str]) -> Self { + Self::with_width(home, root, args, 120) + } + fn with_width(home: &Path, root: &Path, args: &[&str], width: u16) -> Self { + let (mut master, mut slave) = (-1, -1); + let mut size = libc::winsize { + ws_row: 40, + ws_col: width, + ws_xpixel: 0, + ws_ypixel: 0, + }; + // SAFETY: openpty initializes both descriptors; size is valid for the call. + assert_eq!( + unsafe { + libc::openpty( + &mut master, + &mut slave, + std::ptr::null_mut(), + std::ptr::null_mut(), + &raw mut size, + ) + }, + 0 + ); + // SAFETY: each successfully opened descriptor is owned exactly once. + let (master, slave) = unsafe { (File::from_raw_fd(master), File::from_raw_fd(slave)) }; + // SAFETY: master is a valid descriptor; nonblocking reads preserve deadlines. + assert_ne!( + unsafe { libc::fcntl(master.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK) }, + -1 + ); + let mut command = Command::new(env!("CARGO_BIN_EXE_kit")); + command + .env("HOME", home) + .env("TERM", "xterm-256color") + .args(["tui", "--resume", "--root"]) + .arg(root) + .args(["--credential-store", "memory"]) + .args(args) + .stdin(Stdio::from(slave.try_clone().unwrap())) + .stdout(Stdio::from(slave.try_clone().unwrap())) + .stderr(Stdio::from(slave)); + // SAFETY: only async-signal-safe calls run between fork and exec; the + // controlling terminal prevents queries reaching the developer's terminal. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 || libc::ioctl(0, libc::TIOCSCTTY as _, 0) == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + let mut picker = Self { + child: command.spawn().unwrap(), + master, + output: String::new(), + queries: String::new(), + }; + picker.until(BROWSING); + // Wait for the completed draw, not the footer emitted mid-frame. Terminal + // initialization can still be consuming query replies until drawing ends. + let deadline = Instant::now() + Duration::from_secs(15); + while !picker + .output + .split_once(BROWSING) + .unwrap() + .1 + .contains("\x1b[?25h") + { + assert!(Instant::now() < deadline, "initial draw did not finish"); + picker.pump(); + } + picker + } + fn pump(&mut self) { + let mut fd = libc::pollfd { + fd: self.master.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + // SAFETY: fd points to one initialized pollfd. + let ready = unsafe { libc::poll(&mut fd, 1, 50) }; + if ready < 0 { + assert_eq!( + std::io::Error::last_os_error().kind(), + std::io::ErrorKind::Interrupted + ); + return; + } + if ready > 0 { + let mut bytes = [0; 8192]; + match self.master.read(&mut bytes) { + Ok(n) => { + let text = String::from_utf8_lossy(&bytes[..n]); + self.output.push_str(&text); + self.queries.push_str(&text); + } + // Linux PTYs report EIO after slave closure; macOS returns EOF. + Err(error) + if error.raw_os_error() == Some(libc::EIO) + || error.kind() == std::io::ErrorKind::WouldBlock => {} + Err(error) => panic!("PTY read: {error}"), + } + } + // Answer terminal capability probes in addition to keyboard negotiation. + // Otherwise the image probe can consume a later keyboard response. + for (query, reply) in [ + ( + "\x1b_Gi=31,s=1,v=1,a=q,t=d,f=24;AAAA\x1b\\", + "\x1b_Gi=31;ENOTSUP\x1b\\", + ), + ("\x1b[16t", "\x1b[6;16;8t"), + ("\x1b[5n", "\x1b[0n"), + ("\x1b[?u", "\x1b[?0u"), + ("\x1b[c", "\x1b[?1;2c"), + ] { + while self.queries.contains(query) { + self.master.write_all(reply.as_bytes()).unwrap(); + self.queries = self.queries.replacen(query, "", 1); + } + } + } + fn until(&mut self, text: &str) { + let deadline = Instant::now() + Duration::from_secs(15); + while !self.output.contains(text) { + assert!( + Instant::now() < deadline, + "waiting for {text:?}: {:?}", + self.output + ); + self.pump(); + assert!( + self.child.try_wait().unwrap().is_none(), + "picker exited waiting for {text:?}: {:?}", + self.output + ); + } + } + fn keys(&mut self, keys: &[u8]) { + self.output.clear(); + self.master.write_all(keys).unwrap(); + } + fn finish(&mut self) -> ExitStatus { + let deadline = Instant::now() + Duration::from_secs(15); + loop { + self.pump(); + if let Some(status) = self.child.try_wait().unwrap() { + return status; + } + assert!( + Instant::now() < deadline, + "picker did not exit: {:?}", + self.output + ); + } + } +} +#[test] +fn escape_and_control_c_cancel_without_mutating_session() { + // Use the negotiated keyboard protocol, avoiding ambiguous bare Escape prefixes. + for key in [b"\x1b[27u".as_slice(), b"\x1b[99;5u"] { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session(home.path(), root.path(), "s-cancel", "Cancel fixture"); + let original = fs::read(&path).unwrap(); + let mut picker = Picker::start(home.path(), root.path()); + if key == b"\x1b[27u" { + picker.keys(b"rdraft"); + picker.until("rename: "); + picker.keys(key); + picker.until(BROWSING); + assert!(!path.with_extension("metadata.json").exists()); + } + picker.keys(key); + assert!(picker.finish().success(), "{}", picker.output); + assert_eq!(fs::read(&path).unwrap(), original); + assert_eq!(fs::read_dir(path.parent().unwrap()).unwrap().count(), 1); + assert!( + picker.output.contains("\x1b[?1049l"), + "terminal not restored: {:?}", + picker.output + ); + } +} +#[test] +fn renamed_session_is_persisted_then_selected_through_real_lock_error() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session(home.path(), root.path(), "s-renamed", "Original title"); + let mut picker = Picker::start(home.path(), root.path()); + picker.keys(b"r"); + picker.until("rename: "); + // Bracketed paste is a real terminal event, not rapid typing mistaken for paste. + picker.keys(b"\x1b[200~PersistedName\x1b[201~"); + picker.until("PersistedName"); + picker.keys(b"\r"); + picker.until(BROWSING); + let metadata: serde_json::Value = + serde_json::from_slice(&fs::read(path.with_extension("metadata.json")).unwrap()).unwrap(); + assert_eq!(metadata["display_name"], "PersistedName"); + // A real stale lock exercises resume validation without an authenticated agent. + fs::write(path.with_extension("lock"), "abandoned").unwrap(); + picker.keys(b"\r"); + assert!(!picker.finish().success(), "{}", picker.output); + assert!( + picker + .output + .contains("session is locked by another Kit instance"), + "{}", + picker.output + ); + assert!( + picker.output.contains("s-renamed.lock"), + "{}", + picker.output + ); +} +#[test] +fn newest_workspace_session_is_selected_and_revalidated() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let other = tempfile::tempdir().unwrap(); + session(home.path(), root.path(), "s-older", "OlderWorkspaceSession"); + let newest = session( + home.path(), + root.path(), + "s-newest", + "NewestWorkspaceSession", + ); + session( + home.path(), + other.path(), + "s-foreign", + "ForeignWorkspaceSession", + ); + // Explicit mtime avoids sleeps and filesystem timestamp-resolution races. + File::options() + .write(true) + .open(&newest) + .unwrap() + .set_times( + fs::FileTimes::new() + .set_modified(std::time::SystemTime::now() + Duration::from_secs(3600)), + ) + .unwrap(); + let mut picker = Picker::start(home.path(), root.path()); + assert!(picker.output.contains("NewestWorkspaceSession")); + assert!(picker.output.contains("OlderWorkspaceSession")); + assert!(!picker.output.contains("ForeignWorkspaceSession")); + // The catalog is a snapshot: resume must still validate the selected file. + fs::write(&newest, "not json\n").unwrap(); + picker.keys(b"\r"); + assert!(!picker.finish().success(), "{}", picker.output); + assert!(picker.output.contains("line 1"), "{}", picker.output); +} + +#[test] +fn force_cannot_override_active_selected_lock_or_touch_unrelated_stale_lock() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session(home.path(), root.path(), "s-active", "ActiveSession"); + let original = fs::read(&path).unwrap(); + let lock_path = path.with_extension("lock"); + let active = File::options() + .read(true) + .write(true) + .create_new(true) + .open(&lock_path) + .unwrap(); + active.lock().unwrap(); + let unrelated = path.parent().unwrap().join("s-unrelated.lock"); + fs::write(&unrelated, "unrelated stale lock").unwrap(); + let mut picker = Picker::with_args(home.path(), root.path(), &["--force"]); + picker.keys(b"\r"); + assert!(!picker.finish().success(), "{}", picker.output); + assert!( + picker + .output + .contains("session is actively locked by another Kit instance"), + "{}", + picker.output + ); + assert!(picker.output.contains("s-active.lock"), "{}", picker.output); + assert_eq!(fs::read(&path).unwrap(), original); + assert!(lock_path.exists()); + assert_eq!( + fs::read_to_string(&unrelated).unwrap(), + "unrelated stale lock" + ); + active.unlock().unwrap(); +} + +#[test] +fn force_overrides_only_selected_stale_lock_then_reaches_auth_validation() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session(home.path(), root.path(), "s-stale", "StaleSession"); + fs::write(path.with_extension("lock"), "selected stale lock").unwrap(); + let unrelated = path.parent().unwrap().join("s-unrelated.lock"); + fs::write(&unrelated, "unrelated stale lock").unwrap(); + // Memory credential storage and an explicit provider make authentication + // fail locally, before any request, independently of developer credentials. + let mut picker = Picker::with_args( + home.path(), + root.path(), + &[ + "--force", + "--provider", + "openai-subscription", + "--model", + "gpt-5", + ], + ); + picker.keys(b"\r"); + assert!(!picker.finish().success(), "{}", picker.output); + assert!( + picker.output.contains("Authentication required"), + "{}", + picker.output + ); + assert!( + !picker.output.contains("session is locked"), + "{}", + picker.output + ); + assert_eq!( + fs::read_to_string(&unrelated).unwrap(), + "unrelated stale lock" + ); +} + +#[test] +fn selected_session_disappearing_is_not_recreated() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session( + home.path(), + root.path(), + "s-disappeared", + "DisappearingSession", + ); + let mut picker = Picker::start(home.path(), root.path()); + fs::remove_file(&path).unwrap(); + picker.keys(b"\r"); + assert!(!picker.finish().success(), "{}", picker.output); + assert!( + picker.output.contains("s-disappeared") && picker.output.contains("does not exist"), + "{}", + picker.output + ); + assert!(!path.exists()); + assert!(!path.with_extension("lock").exists()); +} + +#[test] +fn rename_error_is_erased_without_another_input_event() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session(home.path(), root.path(), "s-toast", "Toast fixture"); + // Keep the footer narrow enough that the toast displaces its help text. + let mut picker = Picker::with_width(home.path(), root.path(), &[], 60); + picker.keys(b"r"); + picker.until("rename: "); + picker.keys(b"\x1b[200~ChangedName\x1b[201~"); + picker.until("ChangedName"); + fs::remove_file(path).unwrap(); + picker.keys(b"\r"); + picker.until("could not rename session"); + // Finish reading the error frame before observing the idle expiry redraw. + let deadline = Instant::now() + Duration::from_secs(15); + while !picker + .output + .split_once("could not rename session") + .unwrap() + .1 + .contains("\x1b[?25h") + { + assert!(Instant::now() < deadline, "error frame did not finish"); + picker.pump(); + } + picker.output.clear(); + // Observe restored help, not the renderer's choice of clearing spaces. + // No key, resize, or clipboard event is sent to provoke this redraw. + picker.until("⏎ send"); + picker.keys(b"\x1b[27u"); + picker.until(BROWSING); + picker.keys(b"\x1b[27u"); + assert!(picker.finish().success(), "{}", picker.output); +}