diff --git a/src/app.rs b/src/app.rs index 6927a01..f593fc6 100644 --- a/src/app.rs +++ b/src/app.rs @@ -190,6 +190,10 @@ pub struct App { /// task itself, so it can't jump to a neighbor when the list reorders /// (a task exits, or gets tagged into another bucket). pub selected_id: Option, + /// Task id awaiting its first snapshot row. A later `Spawned` event replaces + /// it; a snapshot clears it only after the row appears. This preserves + /// direct-spawn selection across event batching. + pending_select: Option, pub mode: Mode, pub group_mode: GroupMode, pub input: EditBuffer, @@ -375,17 +379,25 @@ impl App { rows: self.pane_rows(), cols: self.cols, }); - self.views.clear(); - self.focused_screen = None; - self.watched = None; - self.selected_id = None; - self.mode = Mode::Dashboard; + self.reset_for_reconnect(); self.status = Some("reconnected".to_string()); } Err(e) => self.status = Some(format!("reconnect failed: {e}")), } } + /// Clear task and selection state owned by the disconnected transport. + /// Task ids are daemon-local and may be reused after a daemon restart, so + /// retaining `pending_select` could select an unrelated task. + fn reset_for_reconnect(&mut self) { + self.views.clear(); + self.focused_screen = None; + self.watched = None; + self.selected_id = None; + self.pending_select = None; + self.mode = Mode::Dashboard; + } + /// `--foreground`: run the core in-process on a thread (no daemon). A /// non-daemon escape hatch, and the deterministic target the UI harnesses use. pub fn new_foreground(rows: u16, cols: u16) -> Self { @@ -421,6 +433,7 @@ impl App { watched: None, daemon_backed: false, selected_id: None, + pending_select: None, mode: Mode::Dashboard, group_mode: GroupMode::State, input: EditBuffer::default(), @@ -722,7 +735,19 @@ impl App { match ev { // The handshake is handled before the transport is created. Event::HelloOk => {} - Event::Tasks(v) => self.views = v, + Event::Tasks(v) => { + self.views = v; + // The acknowledgement precedes its row. Select only after + // the matching snapshot arrives, then rendering keeps that + // row visible. + if let Some(id) = self.pending_select + && self.task_index(id).is_some() + { + self.selected_id = Some(id); + self.pending_select = None; + } + } + Event::Spawned { id } => self.pending_select = Some(id), Event::Screen(s) => self.on_screen(s), Event::Status(s) => { // Mirror attached-mode status messages into the visible notice bar. diff --git a/src/app_tests.rs b/src/app_tests.rs index fabd46d..ec3c3be 100644 --- a/src/app_tests.rs +++ b/src/app_tests.rs @@ -146,6 +146,122 @@ fn selection_follows_task_across_reorder() { assert_eq!(app.views[app.selected_task().unwrap()].id, 1); } +/// Each admitted direct spawn moves selection to its task. +#[test] +fn spawn_moves_selection_to_the_new_task() { + let mut app = App::new_local(30, 100); + let dir = app.invocation_dir.clone(); + app.spawn_in("sleep 30", dir.clone()); + app.resolve_selection(); + let first = app.selected_id.expect("first spawn selected"); + + app.spawn_in("sleep 31", dir); + let second = app.selected_id.expect("second spawn selected"); + assert_ne!(first, second, "selection must move off the prior task"); + assert_eq!( + app.views[app.selected_task().unwrap()].command, + "sleep 31", + "selection must land on the newest spawn" + ); + assert_eq!(app.pending_select, None, "the ack must be consumed"); +} + +/// When two spawn acknowledgements share a snapshot, the later id replaces the +/// earlier pending id and wins selection. +#[test] +fn two_spawns_in_one_sync_select_the_last() { + let mut app = App::new_local(30, 100); + let dir = app.invocation_dir.clone(); + app.transport.send(Command::Spawn { + command: "sleep 30".into(), + cwd: dir.clone(), + group: None, + }); + app.transport.send(Command::Spawn { + command: "sleep 31".into(), + cwd: dir, + group: None, + }); + app.pump(); + assert_eq!( + app.views.len(), + 2, + "both spawns must land: {:?}", + app.status + ); + assert_eq!( + app.views[app.selected_task().unwrap()].command, + "sleep 31", + "the later spawn wins the selection" + ); +} + +/// A refused direct spawn emits no acknowledgement and leaves selection intact. +#[test] +fn refused_spawn_leaves_selection_alone() { + let mut app = App::new_local(30, 100); + let dir = app.invocation_dir.clone(); + app.spawn_in("sleep 30", dir.clone()); + let kept = app.selected_id; + assert!(kept.is_some()); + + // Exceed the direct-spawn command limit by one byte. + app.transport.send(Command::Spawn { + command: "x".repeat(64 * 1024 + 1), + cwd: dir, + group: None, + }); + app.pump(); + assert_eq!( + app.views.len(), + 1, + "the refused spawn must not admit a task" + ); + assert_eq!(app.selected_id, kept, "selection must not move"); + assert_eq!(app.pending_select, None, "a refusal must not leave an ack"); +} + +/// A snapshot without the pending task neither changes selection nor clears the +/// pending id. +#[test] +fn pending_select_ignores_snapshots_without_the_id() { + let mut app = App::new_local(30, 100); + let dir = app.invocation_dir.clone(); + app.spawn_in("sleep 30", dir); + let kept = app.selected_id; + + app.pending_select = Some(9999); + // A state change makes the next pump deliver a fresh snapshot. + let id = kept.unwrap(); + app.transport.send(Command::SetGroup { + id, + group: Some("g".into()), + }); + app.pump(); + assert_eq!( + app.selected_id, kept, + "an absent id must not move selection" + ); + assert_eq!(app.pending_select, Some(9999), "the pending id stays armed"); +} + +/// Reconnect drops a pending spawn id because a replacement daemon may reuse it. +#[test] +fn reconnect_reset_drops_the_pending_spawn_ack() { + let mut app = App::new_local(30, 100); + let dir = app.invocation_dir.clone(); + app.views = vec![view(1, dir, false, None)]; + app.selected_id = Some(1); + app.pending_select = Some(2); + app.mode = Mode::Disconnected; + + app.reset_for_reconnect(); + assert_eq!(app.pending_select, None, "the stale ack must not survive"); + assert_eq!(app.selected_id, None); + assert!(app.views.is_empty()); + assert!(app.mode == Mode::Dashboard); +} + /// Dir mode makes one section per distinct cwd (invocation dir first); state /// mode collapses them back into the state buckets. #[test] @@ -300,8 +416,8 @@ fn custom_mode_selection_survives_group_move() { app.spawn_grouped("sleep 5", inv, "beta"); // id 2 app.pump(); app.group_mode = GroupMode::Custom; - app.resolve_selection(); - assert_eq!(app.selected_id, Some(1)); + // Exercise moving id 1 rather than the selected second spawn. + app.selected_id = Some(1); // Move id 1 from the first section to the last. app.transport.send(Command::SetGroup { @@ -452,8 +568,7 @@ fn selection_follows_task_across_parked_rebucket() { app.spawn_in("sleep 5", dir.clone()); // id 1 app.spawn_in("sleep 5", dir); // id 2 app.pump(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(1)); + app.selected_id = Some(1); // Park id 1 -> it sinks into "Idle", below id 2's "Running". let i = app.views.iter().position(|v| v.id == 1).unwrap(); @@ -1339,14 +1454,13 @@ fn selection_wraps_at_list_edges() { // below crosses a section boundary. app.transport.send(Command::Tag { id: 2, on: true }); app.pump(); - app.resolve_selection(); assert_eq!(app.section_ids().len(), 2, "tag splits the list in two"); let order = app.display_order(); let first = app.views[order[0]].id; let last = app.views[*order.last().unwrap()].id; assert_eq!(first, 2, "tagged task sorts first"); - assert_eq!(app.selected_id, Some(first)); + app.selected_id = Some(first); app.select_up(); assert_eq!( @@ -1540,8 +1654,7 @@ fn app_with_tags_split_across_groups() -> App { #[test] fn cycle_tagged_advances_and_wraps() { let mut app = app_with_tagged_pair(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(2), "first row is the first tag"); + app.selected_id = Some(2); // Start on the first tagged row. app.on_key_dashboard(key(KeyCode::Char('M'))); assert_eq!(app.selected_id, Some(4), "forward to the second tag"); @@ -1577,8 +1690,7 @@ fn cycle_tagged_is_noop_without_tags() { app.spawn_in("sleep 5", inv.clone()); // id 1 app.spawn_in("sleep 5", inv); // id 2 app.pump(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(1)); + app.selected_id = Some(1); app.on_key_dashboard(key(KeyCode::Char('M'))); assert_eq!(app.selected_id, Some(1), "no tags: the selection stands"); @@ -1612,8 +1724,7 @@ fn cycle_tagged_with_one_tag_holds_the_selection() { app.pump(); app.transport.send(Command::Tag { id: 2, on: true }); app.pump(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(2), "the only tag heads the list"); + app.selected_id = Some(2); // Start on the only tagged row. app.on_key_dashboard(key(KeyCode::Char('M'))); app.on_key_dashboard(key(KeyCode::Char('M'))); @@ -1637,7 +1748,7 @@ fn cycle_tagged_without_selection_takes_the_first_tag() { #[test] fn cycle_tagged_mutates_no_task_state() { let mut app = app_with_tags_split_across_groups(); - app.resolve_selection(); + app.selected_id = Some(1); let before: Vec<_> = app .views .iter() @@ -1943,8 +2054,7 @@ fn wheel_moves_dashboard_selection() { app.spawn_in("sleep 5", dir.clone()); // id 1 app.spawn_in("sleep 5", dir); // id 2 app.pump(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(1)); + app.selected_id = Some(1); let wheel = |kind| MouseEvent { kind, @@ -2053,7 +2163,8 @@ fn group_filter_narrows_and_preselects_the_first_match() { app.spawn_grouped("sleep 5", inv.clone(), "alpha"); // id 1 app.spawn_grouped("sleep 5", inv, "beta"); // id 2 app.pump(); - app.resolve_selection(); + // Keep alpha selected so filtered beta carries no "(current)" mark. + app.selected_id = Some(1); app.on_key_dashboard(key(KeyCode::Char('g'))); assert_eq!(app.group_candidates.len(), 3); @@ -2177,7 +2288,8 @@ fn find_palette_opens_on_slash_only_with_tasks() { let inv = app.invocation_dir.clone(); app.spawn_in("sleep 5", inv); app.pump(); - assert_eq!(app.selected_id, None, "nothing selected yet"); + // Exercise opening find without a current selection. + app.selected_id = None; app.on_key_dashboard(key(KeyCode::Char('/'))); assert!(app.mode == Mode::Find); assert_eq!(find_ids(&app), vec![1]); @@ -2322,8 +2434,7 @@ fn find_enter_jumps_the_selection() { app.spawn_in("sleep 5", inv.clone()); // id 2 app.spawn_in("sleep 5", inv); // id 3 app.pump(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(1)); + app.selected_id = Some(1); app.on_key_dashboard(key(KeyCode::Char('/'))); app.on_key_find(key(KeyCode::Down)); @@ -2347,8 +2458,7 @@ fn find_esc_leaves_the_selection_alone() { app.spawn_in("sleep 5", inv.clone()); // id 1 app.spawn_in("true", inv); // id 2 app.pump(); - app.resolve_selection(); - assert_eq!(app.selected_id, Some(1)); + app.selected_id = Some(1); app.on_key_dashboard(key(KeyCode::Char('/'))); find_type(&mut app, "true"); diff --git a/src/protocol.rs b/src/protocol.rs index 739b20f..347be14 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -12,7 +12,7 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; use crate::frame::{KIND_CONTROL, KIND_HELLO, KIND_SCREEN}; /// Wire-protocol version; the handshake rejects mismatched peers. -pub const PROTOCOL_VERSION: u32 = 10; +pub const PROTOCOL_VERSION: u32 = 11; /// Reserved dashboard label for tasks without a custom group. pub const UNASSIGNED: &str = "Unassigned"; @@ -194,6 +194,9 @@ pub enum Event { Screen(ScreenView), /// A one-line notice for the status line (save/load result, spawn error). Status(String), + /// The id assigned to a direct `Spawn`. This event precedes the next task + /// snapshot containing that id. Session and recovery loads do not emit it. + Spawned { id: u64 }, /// The reply to `ListSessions`: saved session-recipe names (sorted) and /// recovery snapshots (newest first). Sessions { @@ -871,6 +874,10 @@ pub fn encode_event(ev: &Event) -> (u8, Vec) { let o = jzon::object! { "t": "status", "msg": msg.as_str() }; (KIND_CONTROL, o.dump().into_bytes()) } + Event::Spawned { id } => { + let o = jzon::object! { "t": "spawned", "id": *id }; + (KIND_CONTROL, o.dump().into_bytes()) + } Event::Sessions { names, recovery } => { let mut rec = jzon::JsonValue::new_array(); for r in recovery { @@ -974,6 +981,9 @@ pub fn decode_event(kind: u8, payload: &[u8]) -> Option { Some(Event::Tasks(views)) } "status" => Some(Event::Status(v["msg"].as_str()?.to_string())), + "spawned" => Some(Event::Spawned { + id: v["id"].as_u64()?, + }), "sessions" => Some(Event::Sessions { names: str_vec(&v["names"])?, recovery: recovery_vec(&v["recovery"]), diff --git a/src/protocol_tests.rs b/src/protocol_tests.rs index 4cdf385..212c90b 100644 --- a/src/protocol_tests.rs +++ b/src/protocol_tests.rs @@ -404,6 +404,17 @@ fn tasks_and_status_round_trip() { assert_eq!(decode_event(k, &p), Some(status)); } +/// A spawn acknowledgement round-trips with an id above `u32::MAX`. +#[test] +fn spawned_round_trips() { + let ev = Event::Spawned { + id: u64::from(u32::MAX) + 7, + }; + let (k, p) = encode_event(&ev); + assert_eq!(k, KIND_CONTROL); + assert_eq!(decode_event(k, &p), Some(ev)); +} + /// `SetGroup` emits `"g"` only for an assignment. A missing or null `"g"` /// decodes as a clear. #[test] diff --git a/src/supervisor.rs b/src/supervisor.rs index fd9791c..1c6b222 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -818,9 +818,9 @@ impl Supervisor { Ok(task) } - /// Spawn under the next id and admit the task to the set, normalizing its - /// labels. The caller owns the `MAX_TASKS` gate and failure reporting, - /// which differ between direct spawns and session loads. + /// Spawn a task under the next id, normalize its labels, and return the id. + /// The caller enforces `MAX_TASKS` and reports failures because direct + /// spawns and session loads handle them differently. fn admit( &mut self, command: &str, @@ -828,13 +828,14 @@ impl Supervisor { env: &[(OsString, OsString)], group: Option, name: Option, - ) -> io::Result<()> { - let mut task = self.spawn_task(self.next_id, 0, command, cwd, env)?; + ) -> io::Result { + let id = self.next_id; + let mut task = self.spawn_task(id, 0, command, cwd, env)?; task.group = normalize_group(group); task.name = normalize_label(name); self.next_id += 1; self.tasks.push(task); - Ok(()) + Ok(id) } fn spawn(&mut self, command: &str, cwd: PathBuf, group: Option) { @@ -853,8 +854,11 @@ impl Supervisor { let Some(launch) = self.launch_or_refuse() else { return; }; - if let Err(e) = self.admit(command, &cwd, &launch.env, group, None) { - self.status(format!("spawn failed: {e}")); + match self.admit(command, &cwd, &launch.env, group, None) { + // Preserve event order: the acknowledgement precedes the next + // `Tasks` snapshot containing this id. + Ok(id) => self.events.push(Event::Spawned { id }), + Err(e) => self.status(format!("spawn failed: {e}")), } } @@ -1021,7 +1025,7 @@ impl Supervisor { entry.group.clone(), entry.name.clone(), ) { - Ok(()) => spawned += 1, + Ok(_) => spawned += 1, // Track spawn failures separately from skipped entries. Err(_) => failed += 1, } diff --git a/src/supervisor_tests.rs b/src/supervisor_tests.rs index 1887304..ce308d5 100644 --- a/src/supervisor_tests.rs +++ b/src/supervisor_tests.rs @@ -98,6 +98,8 @@ fn tick_emits_snapshot_and_watched_screen() { let mut s = sup(24, 80); spawn(&mut s, "sleep 30", here()); + // Discard `Spawned`; this test isolates events queued by `tick`. + let _ = s.drain(); s.tick(); let evs = s.drain(); assert_eq!(evs.len(), 1, "only a Tasks snapshot while unwatched"); @@ -521,13 +523,16 @@ fn spawn_ready(s: &mut Supervisor, command: String, cwd: PathBuf, ready: &Path) first_id(s) } -/// Tick once and return the first snapshotted task's id. +/// Tick once and return the first task id from `Tasks`, ignoring other events. fn first_id(s: &mut Supervisor) -> u64 { s.tick(); - match s.drain().first() { - Some(Event::Tasks(v)) => v[0].id, - _ => panic!("expected a Tasks snapshot"), - } + s.drain() + .iter() + .find_map(|e| match e { + Event::Tasks(v) => Some(v[0].id), + _ => None, + }) + .expect("expected a Tasks snapshot") } /// Tick once and return task `id` from the emitted snapshot. @@ -1017,7 +1022,7 @@ fn spawn_carries_a_normalized_group_from_birth() { let mut s = sup(24, 80); spawn_grouped(&mut s, "sleep 30", here(), " ui\x1b[2J "); s.tick(); - match s.drain().first() { + match s.drain().iter().find(|e| matches!(e, Event::Tasks(_))) { Some(Event::Tasks(v)) => assert_eq!(v[0].group.as_deref(), Some("ui[2J")), _ => panic!("expected a Tasks snapshot"), } @@ -1713,6 +1718,74 @@ fn load_reports_admit_failures_not_clean_success() { ); } +/// A direct spawn queues `Spawned` before `tick` queues the matching `Tasks` +/// snapshot. +#[test] +fn spawn_acks_with_spawned_before_the_snapshot() { + let mut s = sup(24, 80); + spawn(&mut s, "sleep 30", here()); + s.tick(); + let evs = s.drain(); + let ack = evs + .iter() + .position(|e| matches!(e, Event::Spawned { .. })) + .expect("spawn must ack with Spawned"); + let snap = evs + .iter() + .position(|e| matches!(e, Event::Tasks(_))) + .expect("tick must emit a Tasks snapshot"); + assert!(ack < snap, "Spawned must precede the snapshot; got {evs:?}"); + let (Some(Event::Spawned { id }), Some(Event::Tasks(v))) = (evs.get(ack), evs.get(snap)) else { + unreachable!(); + }; + assert_eq!(v.len(), 1); + assert_eq!(*id, v[0].id, "the ack must name the admitted task"); +} + +/// A command-length refusal emits no `Spawned` event. +#[test] +fn refused_spawn_emits_no_spawned() { + let mut s = sup(24, 80); + s.apply(Command::Spawn { + command: "x".repeat(MAX_COMMAND_LEN + 1), + cwd: here(), + group: None, + }); + let evs = s.drain(); + assert!( + !evs.iter().any(|e| matches!(e, Event::Spawned { .. })), + "a refused spawn must not ack; got {evs:?}" + ); +} + +/// Session loads admit tasks without emitting `Spawned` events. +#[test] +fn session_load_emits_no_spawned() { + let dir = scratch("sess_no_ack"); + let config = dir.join("config"); + std::fs::create_dir_all(config.join("sessions")).unwrap(); + std::fs::write( + config.join("sessions").join("fleet.json"), + format!(r#"{{"{}": ["true", "true"]}}"#, dir.display()), + ) + .unwrap(); + let mut s = sup_ctx(config_ctx(&config, dir.to_path_buf(), &[])); + s.apply(Command::LoadSession { + name: "fleet".into(), + }); + s.tick(); + let evs = s.drain(); + assert!( + evs.iter() + .any(|e| matches!(e, Event::Tasks(v) if v.len() == 2)), + "both entries must be admitted; got {evs:?}" + ); + assert!( + !evs.iter().any(|e| matches!(e, Event::Spawned { .. })), + "a session load must not ack; got {evs:?}" + ); +} + /// Direct spawns reject commands above `MAX_COMMAND_LEN` without creating a task. #[test] fn spawn_refuses_over_length_command() { @@ -2228,6 +2301,8 @@ fn recovery_maintenance_writes_detached_and_queues_nothing() { Duration::from_secs(600), ); spawn(&mut s, "sleep 30", dir.to_path_buf()); + // Discard `Spawned`; this test isolates events queued by maintenance. + let _ = s.drain(); // Match the daemon's detached reap-and-maintain loop. assert!( wait_until(Duration::from_secs(5), || { diff --git a/tests/common/mod.rs b/tests/common/mod.rs index da75524..24bd675 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -16,8 +16,8 @@ use std::{ }; /// The protocol version this test suite speaks; must track -/// `protocol::PROTOCOL_VERSION` (drift fails the handshake, loudly). -pub const PROTOCOL_VERSION: u32 = 10; +/// `protocol::PROTOCOL_VERSION` +pub const PROTOCOL_VERSION: u32 = 11; /// One frame of the given kind: `[u32 len][kind][payload]`. pub fn frame(kind: u8, payload: &[u8]) -> Vec {