From dd75ca8a51332871444c9f0b8826c1d9a7ac0349 Mon Sep 17 00:00:00 2001 From: Alex Martin Date: Mon, 14 Sep 2026 14:29:05 -0700 Subject: [PATCH 1/5] feat(tui): add startup session picker Closes #219 --- .../user/getting-started-and-configuration.md | 9 +- docs/user/tui-and-sessions.md | 17 +- src/main.rs | 84 +++- src/session.rs | 149 ++++-- src/tui/mod.rs | 131 +++-- src/tui/startup.rs | 194 ++++++++ src/tui/startup_tests.rs | 162 +++++++ tests/cli.rs | 65 +++ tests/startup_picker.rs | 456 ++++++++++++++++++ 9 files changed, 1157 insertions(+), 110 deletions(-) create mode 100644 src/tui/startup.rs create mode 100644 src/tui/startup_tests.rs create mode 100644 tests/startup_picker.rs 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/session.rs b/src/session.rs index 31efc087..224fe34a 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1398,10 +1398,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 +1413,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 +1444,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 +1469,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 +1477,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 +1666,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 +1706,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 +1912,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 +4181,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 +5112,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 +5164,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 +5202,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 +5239,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/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..c84240d5 --- /dev/null +++ b/src/tui/startup.rs @@ -0,0 +1,194 @@ +//! 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), + Clipboard(ClipboardRoute, ClipboardResult), +} + +/// Pick an existing workspace session without creating or resuming a session. +/// Cancellation and an empty catalog return `None`. +pub async fn pick_session( + root: &Path, + stop: &mut Stop, +) -> Result, Box> { + let root = root + .canonicalize() + .map_err(|error| Failure(format!("{}: {error}", root.display())))?; + let filesystem = + crate::resilient_fs::Fs::new(std::sync::Arc::new(crate::resilient_fs::DiskBackend)); + let Some(entries) = scan_catalog(root.clone(), filesystem, 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 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::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, + } => { + let root = root.clone(); + let updates = updates_tx.clone(); + tokio::spawn(async move { + let id = session_id.clone(); + let name = display_name.clone(); + let result = tokio::task::spawn_blocking(move || { + crate::session::set_display_name_and_title(&root, &id, name.as_deref()) + }) + .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: display_name.map(|name| name.trim().to_string()), + result, + })); + }); + } + 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); + result +} + +// 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, + filesystem: crate::resilient_fs::Fs, + stop: &mut Stop, +) -> Result>, Box> { + let (tx, rx) = tokio::sync::oneshot::channel(); + let worker = std::thread::Builder::new() + .name("session-catalog".into()) + .spawn(move || { + let result = crate::session::catalog_with(&filesystem, &root).map_err(|error| { + format!("could not list sessions for {}: {error}", root.display()) + }); + 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..c9c5b65d --- /dev/null +++ b/src/tui/startup_tests.rs @@ -0,0 +1,162 @@ +//! 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, Fs, + LeaseRequest, +}; +use std::{ + io, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + sync::Arc, + time::{Duration, Instant}, +}; + +struct StalledDisk { + entered: PathBuf, +} +impl Backend for StalledDisk { + fn read_dir(&self, _path: &Path) -> io::Result> { + std::fs::write(&self.entered, b"reading")?; + loop { + std::thread::park(); + } + } + fn open(&self, path: &Path, options: &DiskOpenOptions) -> io::Result> { + DiskBackend.open(path, options) + } + fn metadata(&self, path: &Path, follow: bool) -> io::Result { + DiskBackend.metadata(path, follow) + } + fn read_link(&self, path: &Path) -> io::Result { + DiskBackend.read_link(path) + } + fn canonicalize(&self, path: &Path) -> io::Result { + 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<()> { + 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> { + 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 filesystem = Fs::new(Arc::new(StalledDisk { + entered: directory.join("entered"), + })); + let mut stop = super::Stop::new().unwrap(); + assert!( + super::scan_catalog(directory.clone(), filesystem, &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; + } + for signal in ["-INT", "-TERM", "-HUP"] { + let directory = tempfile::tempdir().unwrap(); + std::fs::create_dir_all(directory.path().join(".kit/sessions")).unwrap(); + let mut child = Reap( + Command::new(std::env::current_exe().unwrap()) + .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" + ); + } +} diff --git a/tests/cli.rs b/tests/cli.rs index 0beb4bab..b8f00561 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -34,6 +34,71 @@ 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(); + for force in [false, true] { + 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"]); + if force { + command.arg("--force"); + } + 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..ea0b4c73 --- /dev/null +++ b/tests/startup_picker.rs @@ -0,0 +1,456 @@ +//! 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 { + let (mut master, mut slave) = (-1, -1); + let mut size = libc::winsize { + ws_row: 40, + ws_col: 120, + 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(), + &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()); + 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 escape_cancels_rename_before_closing_picker() { + let home = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let path = session( + home.path(), + root.path(), + "s-rename-cancel", + "Original title", + ); + let mut picker = Picker::start(home.path(), root.path()); + picker.keys(b"rdraft"); + picker.until("rename: "); + picker.keys(b"\x1b[27u"); + picker.until(BROWSING); + assert!(!path.with_extension("metadata.json").exists()); + picker.keys(b"\x1b[27u"); + assert!(picker.finish().success(), "{}", 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"); + let mut picker = Picker::start(home.path(), root.path()); + 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(); + // Ratatui's diff clears the status text with spaces when the toast expires. + // No key, resize, or clipboard event is sent to provoke this redraw. + picker.until(" "); + picker.keys(b"\x1b[27u"); + picker.until(BROWSING); + picker.keys(b"\x1b[27u"); + assert!(picker.finish().success(), "{}", picker.output); +} From b8d8559ff00f27019dc322b019d6d7bd9d7faff7 Mon Sep 17 00:00:00 2001 From: Alex Martin Date: Tue, 15 Sep 2026 01:04:04 -0700 Subject: [PATCH 2/5] fix(tui): cancel workspace discovery safely Resolve workspace paths inside the isolated discovery worker so stalled canonicalization cannot block signal handling or runtime teardown. Use a portable raw pointer for the PTY window size on Linux and macOS. --- src/tui/startup.rs | 29 +++++++++++------- src/tui/startup_tests.rs | 64 +++++++++++++++++++++++++++++++++++----- tests/startup_picker.rs | 2 +- 3 files changed, 76 insertions(+), 19 deletions(-) diff --git a/src/tui/startup.rs b/src/tui/startup.rs index c84240d5..59910f3c 100644 --- a/src/tui/startup.rs +++ b/src/tui/startup.rs @@ -25,12 +25,8 @@ pub async fn pick_session( root: &Path, stop: &mut Stop, ) -> Result, Box> { - let root = root - .canonicalize() - .map_err(|error| Failure(format!("{}: {error}", root.display())))?; - let filesystem = - crate::resilient_fs::Fs::new(std::sync::Arc::new(crate::resilient_fs::DiskBackend)); - let Some(entries) = scan_catalog(root.clone(), filesystem, stop).await? else { + 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() { @@ -153,16 +149,27 @@ pub async fn pick_session( // unlike spawn_blocking, it does not delay Tokio runtime teardown. async fn scan_catalog( root: std::path::PathBuf, - filesystem: crate::resilient_fs::Fs, + backend: std::sync::Arc, stop: &mut Stop, -) -> Result>, Box> { +) -> 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 = crate::session::catalog_with(&filesystem, &root).map_err(|error| { - format!("could not list sessions for {}: {error}", root.display()) - }); + 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 { diff --git a/src/tui/startup_tests.rs b/src/tui/startup_tests.rs index c9c5b65d..d24fc4c9 100644 --- a/src/tui/startup_tests.rs +++ b/src/tui/startup_tests.rs @@ -9,7 +9,7 @@ )] use crate::resilient_fs::{ - self as fs, Backend, BackendFile, BackendLease, DiskBackend, DiskEntry, DiskOpenOptions, Fs, + self as fs, Backend, BackendFile, BackendLease, DiskBackend, DiskEntry, DiskOpenOptions, LeaseRequest, }; use std::{ @@ -22,9 +22,13 @@ use std::{ struct StalledDisk { entered: PathBuf, + canonicalize: bool, } impl Backend for StalledDisk { - fn read_dir(&self, _path: &Path) -> io::Result> { + fn read_dir(&self, path: &Path) -> io::Result> { + if self.canonicalize { + return DiskBackend.read_dir(path); + } std::fs::write(&self.entered, b"reading")?; loop { std::thread::park(); @@ -40,6 +44,12 @@ impl Backend for StalledDisk { DiskBackend.read_link(path) } fn canonicalize(&self, path: &Path) -> io::Result { + 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<()> { @@ -87,12 +97,13 @@ fn cancelled_catalog_does_not_hold_runtime_teardown() { .unwrap(); runtime.block_on(async { fs::start_recovery_worker(); - let filesystem = Fs::new(Arc::new(StalledDisk { + let backend = Arc::new(StalledDisk { entered: directory.join("entered"), - })); + canonicalize: std::env::var_os("KIT_STALL_CANONICALIZE").is_some(), + }); let mut stop = super::Stop::new().unwrap(); assert!( - super::scan_catalog(directory.clone(), filesystem, &mut stop) + super::scan_catalog(directory.clone(), backend, &mut stop) .await .unwrap() .is_none() @@ -105,11 +116,19 @@ fn cancelled_catalog_does_not_hold_runtime_teardown() { drop(runtime); return; } - for signal in ["-INT", "-TERM", "-HUP"] { + for (signal, canonicalize) in [false, true] + .into_iter() + .flat_map(|canonicalize| ["-INT", "-TERM", "-HUP"].map(|signal| (signal, canonicalize))) + { 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::new(std::env::current_exe().unwrap()) + command .args([ "--exact", "tui::startup::tests::cancelled_catalog_does_not_hold_runtime_teardown", @@ -160,3 +179,34 @@ fn cancelled_catalog_does_not_hold_runtime_teardown() { ); } } + +#[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); + } +} diff --git a/tests/startup_picker.rs b/tests/startup_picker.rs index ea0b4c73..5b3839cc 100644 --- a/tests/startup_picker.rs +++ b/tests/startup_picker.rs @@ -73,7 +73,7 @@ impl Picker { &mut slave, std::ptr::null_mut(), std::ptr::null_mut(), - &mut size, + &raw mut size, ) }, 0 From 54ff21a99233c89c44b8104469325790d163155e Mon Sep 17 00:00:00 2001 From: Alex Martin Date: Tue, 15 Sep 2026 08:57:18 -0700 Subject: [PATCH 3/5] test(tui): assert visible toast expiry behavior Use a narrow footer fixture and wait for restored help instead of requiring an exact run of terminal clearing spaces. --- tests/startup_picker.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/startup_picker.rs b/tests/startup_picker.rs index 5b3839cc..0f0fbd6a 100644 --- a/tests/startup_picker.rs +++ b/tests/startup_picker.rs @@ -58,10 +58,13 @@ impl Picker { 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: 120, + ws_col: width, ws_xpixel: 0, ws_ypixel: 0, }; @@ -425,7 +428,8 @@ 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"); - let mut picker = Picker::start(home.path(), root.path()); + // 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~"); @@ -446,9 +450,9 @@ fn rename_error_is_erased_without_another_input_event() { picker.pump(); } picker.output.clear(); - // Ratatui's diff clears the status text with spaces when the toast expires. + // 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(" "); + picker.until("⏎ send"); picker.keys(b"\x1b[27u"); picker.until(BROWSING); picker.keys(b"\x1b[27u"); From f2772f9b36fec3babeb4ac3e6ee6b47e4354e954 Mon Sep 17 00:00:00 2001 From: Alex Martin Date: Tue, 15 Sep 2026 15:00:43 -0700 Subject: [PATCH 4/5] fix(tui): own rename preparation and commits Isolate cancellable rename reads and preflight from recovery state. Fence prepared replacements by revision before admitting writes. Drain admitted commits after terminal restoration to preserve recovery. --- src/resilient_fs/mod.rs | 254 +++++++++++++++++++++++++++- src/resilient_fs/tests.rs | 283 +++++++++++++++++++++++++++++++ src/session.rs | 140 ++++++++++++---- src/tui/startup.rs | 138 ++++++++++++++-- src/tui/startup_tests.rs | 340 +++++++++++++++++++++++++++++++++++++- 5 files changed, 1099 insertions(+), 56 deletions(-) diff --git a/src/resilient_fs/mod.rs b/src/resilient_fs/mod.rs index 2f57a01f..d5d10232 100644 --- a/src/resilient_fs/mod.rs +++ b/src/resilient_fs/mod.rs @@ -317,11 +317,183 @@ 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, even if it only reads. +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 +506,21 @@ struct Service { max_operations: usize, best_effort: bool, } +impl Service { + // Every acquisition invalidates preparation, including reads and rejected + // operations. 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 +758,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 +775,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)); } @@ -632,7 +878,7 @@ impl Fs { if !self.service.best_effort { return None; } - Some(match lock(&self.service.state) { + Some(match self.service.acquire().map(|(state, _)| state) { Err(_) => BestEffortStatus::Poisoned, Ok(s) if s.dropped => BestEffortStatus::Dropped, Ok(s) if !s.pending.is_empty() => BestEffortStatus::Buffered, @@ -653,7 +899,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); @@ -880,7 +1126,7 @@ impl Fs { /// Use `recover().blocked` for the typed poison error. pub fn status(&self) -> Status { let snapshot = || -> io::Result { - let s = lock(&self.service.state)?; + let s = self.service.acquire().map(|(state, _)| state)?; Ok(Status { pending_operations: s.pending.len(), retained_bytes: Self::retained(&s)?, diff --git a/src/resilient_fs/tests.rs b/src/resilient_fs/tests.rs index 62fb6faf..85980711 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,284 @@ 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"); + #[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(); + assert_eq!( + worker.join().unwrap().unwrap().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"); +} + +#[test] +fn prepared_replacements_can_repeat_without_recovery() { + let t = Fixture::new(); + let path = t.path("legacy/value"); + for value in [b"first".as_slice(), b"second", b"third"] { + t.fs.prepare_private_replace_with_parents(&path, value) + .unwrap() + .commit() + .unwrap(); + // Native inspection does not prune or access the original service. + assert_eq!(native::read(&path).unwrap(), value); + } +} diff --git a/src/session.rs b/src/session.rs index 224fe34a..b228f24f 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1300,14 +1300,89 @@ 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 { + // 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, + &default_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 +1393,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 +1412,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 { diff --git a/src/tui/startup.rs b/src/tui/startup.rs index 59910f3c..ee49a96b 100644 --- a/src/tui/startup.rs +++ b/src/tui/startup.rs @@ -16,11 +16,18 @@ use super::{ 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, @@ -39,6 +46,7 @@ pub async fn pick_session( // 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; @@ -78,6 +86,30 @@ pub async fn pick_session( 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 @@ -104,23 +136,13 @@ pub async fn pick_session( session_id, display_name, } => { - let root = root.clone(); - let updates = updates_tx.clone(); - tokio::spawn(async move { - let id = session_id.clone(); - let name = display_name.clone(); - let result = tokio::task::spawn_blocking(move || { - crate::session::set_display_name_and_title(&root, &id, name.as_deref()) - }) - .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: display_name.map(|name| name.trim().to_string()), - result, - })); - }); + 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 { @@ -140,9 +162,91 @@ pub async fn pick_session( } .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; diff --git a/src/tui/startup_tests.rs b/src/tui/startup_tests.rs index d24fc4c9..701da0b9 100644 --- a/src/tui/startup_tests.rs +++ b/src/tui/startup_tests.rs @@ -23,10 +23,33 @@ use std::{ 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 { + if self.canonicalize || self.rename_stall.is_some() { return DiskBackend.read_dir(path); } std::fs::write(&self.entered, b"reading")?; @@ -35,15 +58,27 @@ impl Backend for StalledDisk { } } 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 { @@ -62,6 +97,7 @@ impl Backend for StalledDisk { 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<()> { @@ -74,6 +110,12 @@ impl Backend for StalledDisk { 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) } } @@ -100,6 +142,7 @@ fn cancelled_catalog_does_not_hold_runtime_teardown() { 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!( @@ -210,3 +253,298 @@ async fn catalog_invalid_root_preserves_diagnostic() { 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"] { + for signal in ["-INT", "-TERM", "-HUP"] { + 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([signal, &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 prepared_rename_commit_persists() { + 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 { + 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"); + }); + drop(runtime); + return; + } + let directory = tempfile::tempdir().unwrap(); + let mut child = rename_child( + "tui::startup::tests::prepared_rename_commit_persists", + directory.path(), + "none", + ); + wait_for_rename_child(&mut child); +} + +#[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 { + 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("changed".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, "changed"); + }); + drop(runtime); + std::fs::write(directory.join("recovered"), b"finished").unwrap(); + return; + } + for signal in ["-INT", "-TERM", "-HUP"] { + 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([signal, &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" + ); + } +} From 724a9a0332206bedb0b8264430ace3319e4b9295 Mon Sep 17 00:00:00 2001 From: Alex Martin Date: Tue, 15 Sep 2026 17:24:25 -0700 Subject: [PATCH 5/5] fix(storage): preserve idle preparation revisions Do not invalidate prepared renames on observational status reads or provably idle recovery. Keep fencing for mutation and cleanup work. Add a regression through the production recovery path and consolidate redundant test scenarios, eliminating 27 process launches while keeping distinct storage, shutdown, and terminal behavior assertions. --- src/resilient_fs/mod.rs | 46 ++++++++++++-- src/resilient_fs/tests.rs | 90 ++++++++++++++++++++++----- src/session.rs | 19 +++++- src/storage_runtime.rs | 52 ++++++++++++++++ src/tui/startup_tests.rs | 128 ++++++++++++++++---------------------- tests/cli.rs | 31 ++++----- tests/startup_picker.rs | 26 +++----- 7 files changed, 258 insertions(+), 134 deletions(-) diff --git a/src/resilient_fs/mod.rs b/src/resilient_fs/mod.rs index d5d10232..99ed584b 100644 --- a/src/resilient_fs/mod.rs +++ b/src/resilient_fs/mod.rs @@ -474,7 +474,8 @@ pub struct Fs { } /// 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, even if it only reads. +/// access conservatively invalidates this preparation, except observational +/// status queries and recovery passes that do no work. pub struct PreparationNamespace { original: Fs, isolated: Fs, @@ -507,8 +508,9 @@ struct Service { best_effort: bool, } impl Service { - // Every acquisition invalidates preparation, including reads and rejected - // operations. Weak tickets cannot keep an obsolete generation alive, and + // 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)?; @@ -878,7 +880,7 @@ impl Fs { if !self.service.best_effort { return None; } - Some(match self.service.acquire().map(|(state, _)| state) { + Some(match lock(&self.service.state) { Err(_) => BestEffortStatus::Poisoned, Ok(s) if s.dropped => BestEffortStatus::Dropped, Ok(s) if !s.pending.is_empty() => BestEffortStatus::Buffered, @@ -1126,7 +1128,7 @@ impl Fs { /// Use `recover().blocked` for the typed poison error. pub fn status(&self) -> Status { let snapshot = || -> io::Result { - let s = self.service.acquire().map(|(state, _)| state)?; + let s = lock(&self.service.state)?; Ok(Status { pending_operations: s.pending.len(), retained_bytes: Self::retained(&s)?, @@ -1665,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 { @@ -1676,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 85980711..068bedcc 100644 --- a/src/resilient_fs/tests.rs +++ b/src/resilient_fs/tests.rs @@ -2418,6 +2418,13 @@ fn prepared_replace_is_effect_free_until_commit_and_preserves_legacy_parents() { .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; @@ -2481,8 +2488,75 @@ fn prepared_replace_blocked_preflight_does_not_hold_original_state() { // 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!( - worker.join().unwrap().unwrap().commit().unwrap_err().kind(), + 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()); @@ -2669,17 +2743,3 @@ fn prepared_replace_preserves_preflight_file_validation() { ); assert_eq!(native::read(&path).unwrap(), b"old"); } - -#[test] -fn prepared_replacements_can_repeat_without_recovery() { - let t = Fixture::new(); - let path = t.path("legacy/value"); - for value in [b"first".as_slice(), b"second", b"third"] { - t.fs.prepare_private_replace_with_parents(&path, value) - .unwrap() - .commit() - .unwrap(); - // Native inspection does not prune or access the original service. - assert_eq!(native::read(&path).unwrap(), value); - } -} diff --git a/src/session.rs b/src/session.rs index b228f24f..32d84e4b 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1341,6 +1341,23 @@ pub(crate) fn prepare_display_name( 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. @@ -1350,7 +1367,7 @@ pub(crate) fn prepare_display_name( let data = display_name_data_in( namespace.filesystem(), root, - &default_directory()?, + global_directory, session_id, display_name, )?; 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/startup_tests.rs b/src/tui/startup_tests.rs index 701da0b9..112d29b0 100644 --- a/src/tui/startup_tests.rs +++ b/src/tui/startup_tests.rs @@ -159,10 +159,13 @@ fn cancelled_catalog_does_not_hold_runtime_teardown() { drop(runtime); return; } - for (signal, canonicalize) in [false, true] - .into_iter() - .flat_map(|canonicalize| ["-INT", "-TERM", "-HUP"].map(|signal| (signal, canonicalize))) - { + // 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()); @@ -388,32 +391,30 @@ fn cancelled_rename_does_not_hold_recovery_or_write_after_shutdown() { return; } for operation in ["canonicalize", "authority", "metadata", "write-open"] { - for signal in ["-INT", "-TERM", "-HUP"] { - 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([signal, &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" - ); - } + 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 prepared_rename_commit_persists() { +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); @@ -422,6 +423,8 @@ fn prepared_rename_commit_persists() { .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( @@ -456,29 +459,6 @@ fn prepared_rename_commit_persists() { assert_eq!(result.unwrap().as_deref(), Some("changed")); fs::finish_best_effort_recovery(&filesystem); assert_rename_title(&root, "changed"); - }); - drop(runtime); - return; - } - let directory = tempfile::tempdir().unwrap(); - let mut child = rename_child( - "tui::startup::tests::prepared_rename_commit_persists", - directory.path(), - "none", - ); - wait_for_rename_child(&mut child); -} - -#[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 { let mut stop = super::Stop::new().unwrap(); let filesystem = fs::Fs::new(Arc::new(StalledDisk { entered: directory.join("entered"), @@ -489,7 +469,7 @@ fn admitted_rename_is_drained_before_recovery() { super::start_rename_preparation( root.clone(), "rename-target".into(), - Some("changed".into()), + Some("drained".into()), filesystem.clone(), tx.clone(), ) @@ -517,34 +497,32 @@ fn admitted_rename_is_drained_before_recovery() { 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, "changed"); + assert_rename_title(&root, "drained"); }); drop(runtime); std::fs::write(directory.join("recovered"), b"finished").unwrap(); return; } - for signal in ["-INT", "-TERM", "-HUP"] { - 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([signal, &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" - ); - } + 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 b8f00561..15165e76 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -39,24 +39,19 @@ fn write_session(home: &Path, root: &Path, id: &str) -> std::path::PathBuf { fn tui_resume_picker_empty_workspace_exits_without_starting_a_session() { let home = tempfile::tempdir().unwrap(); let root = tempfile::tempdir().unwrap(); - for force in [false, true] { - 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"]); - if force { - command.arg("--force"); - } - 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()); - } + 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")] diff --git a/tests/startup_picker.rs b/tests/startup_picker.rs index 0f0fbd6a..e67a2198 100644 --- a/tests/startup_picker.rs +++ b/tests/startup_picker.rs @@ -223,6 +223,13 @@ fn escape_and_control_c_cancel_without_mutating_session() { 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); @@ -235,25 +242,6 @@ fn escape_and_control_c_cancel_without_mutating_session() { } } #[test] -fn escape_cancels_rename_before_closing_picker() { - let home = tempfile::tempdir().unwrap(); - let root = tempfile::tempdir().unwrap(); - let path = session( - home.path(), - root.path(), - "s-rename-cancel", - "Original title", - ); - let mut picker = Picker::start(home.path(), root.path()); - picker.keys(b"rdraft"); - picker.until("rename: "); - picker.keys(b"\x1b[27u"); - picker.until(BROWSING); - assert!(!path.with_extension("metadata.json").exists()); - picker.keys(b"\x1b[27u"); - assert!(picker.finish().success(), "{}", picker.output); -} -#[test] fn renamed_session_is_persisted_then_selected_through_real_lock_error() { let home = tempfile::tempdir().unwrap(); let root = tempfile::tempdir().unwrap();