From 2397130388212f8018899518e47f74e577f5ca34 Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 18 Oct 2026 17:21:52 -0700 Subject: [PATCH 1/5] Split Task into identity and a Process run part --- src/supervisor.rs | 115 ++++++++------ src/supervisor_capture_tests.rs | 108 +++++++++----- src/supervisor_tests.rs | 22 +-- src/task.rs | 255 ++++++++++++++++++-------------- src/task_tests.rs | 184 ++++++++++++----------- src/transport.rs | 4 +- 6 files changed, 399 insertions(+), 289 deletions(-) diff --git a/src/supervisor.rs b/src/supervisor.rs index 68d1fdc..a13b3b2 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -23,7 +23,7 @@ use crate::{ UNASSIGNED, env_get, }, session::{self, EntryKind, SessionConfig, SessionEntry}, - task::{Exec, Task, WriteRefused}, + task::{Exec, Process, Task, WriteRefused}, }; /// Quiet period after which a live task becomes idle. @@ -289,14 +289,15 @@ impl Recovery { pub struct Supervisor { tasks: Vec, - /// Removed tasks whose process groups may still be winding down: `TERMed` at - /// removal, escalated to KILL by `reap` at grace end, and dropped once the - /// leader's zombie is collected. Invisible to `tick` snapshots, so the row - /// disappears instantly while the sweep runs behind it. + /// Runs of removed or rerun tasks whose process groups may still be winding + /// down: `TERMed` at retirement, escalated to KILL by `reap` at grace end, and + /// dropped once the leader's zombie is collected. Invisible to `tick` + /// snapshots, so the row disappears (or shows its replacement) instantly + /// while the sweep runs behind it. /// /// Retain each leader's zombie until SIGKILL has been sent: collecting it earlier /// would release the process-group ID. - graveyard: Vec, + graveyard: Vec, next_id: u64, /// PTY content size (rows already minus the client's status bar). Every task /// runs at this size, so attach never reflows. @@ -320,7 +321,7 @@ pub struct Supervisor { /// Spawning is refused until one is installed. launch: Option, events: Vec, - /// Handed to every `Task` so its reader thread can wake the core loop when the PTY + /// Handed to every `Process` so its reader thread can wake the core loop when the PTY /// produces output. Install the serving loop's sender on connect (`set_waker`) and /// drop it on disconnect (`clear_waker`). Between clients it is `None`; task output /// is parsed without waking a serving loop. @@ -436,11 +437,11 @@ impl Supervisor { group, } => self.spawn(&command, &cwd, group), Command::SpawnAgent { agent, cwd, group } => self.spawn_agent(&agent, &cwd, group), - Command::Kill { id } => self.with_task(id, Task::terminate), + Command::Kill { id } => self.with_task(id, |t| t.process.terminate()), Command::Remove { id } => { if let Some(i) = self.index_of(id) { let t = self.tasks.remove(i); - self.retire(t); + self.retire(t.process); } } Command::Restart { id } => self.rerun(id), @@ -461,7 +462,7 @@ impl Supervisor { self.cols = (MAX_CELLS / u32::from(self.rows)) as u16; } for t in &mut self.tasks { - let _ = t.resize(self.rows, self.cols); + let _ = t.process.resize(self.rows, self.cols); } } Command::Watch { id, attached } => { @@ -470,14 +471,14 @@ impl Supervisor { && let Some(old) = self.watched && let Some(t) = self.by_id_mut(old) { - t.scroll_view(ScrollAction::Live); + t.process.scroll_view(ScrollAction::Live); } if id != self.watched || attached != self.watch_attached { // Discard stores captured before this watch state took effect. if let Some(new) = id && let Some(t) = self.by_id_mut(new) { - let _ = t.drain_clipboard(); + let _ = t.process.drain_clipboard(); } self.last_screen = None; } @@ -486,15 +487,19 @@ impl Supervisor { } // Paste and mouse encoding depend on the child's terminal modes, // which only the core's emulator can see. - Command::Paste { id, bytes } => self.deliver(id, "paste", |t| t.send_paste(&bytes)), + Command::Paste { id, bytes } => { + self.deliver(id, "paste", |t| t.process.send_paste(&bytes)); + } Command::Mouse { id, kind, col, row } => { - self.deliver(id, "mouse input", |t| t.send_mouse(kind, col, row)); + self.deliver(id, "mouse input", |t| t.process.send_mouse(kind, col, row)); } // Encode keys here because the child's cursor-key mode is core-side. Command::Key { id, code, mods } => { - self.deliver(id, "key input", |t| t.send_key(code, mods)); + self.deliver(id, "key input", |t| t.process.send_key(code, mods)); + } + Command::Scrollback { id, action } => { + self.with_task(id, |t| t.process.scroll_view(action)); } - Command::Scrollback { id, action } => self.with_task(id, |t| t.scroll_view(action)), Command::SaveSession { name } => self.save_session(&name), Command::LoadSession { name } => self.load_session(&name), Command::LoadRecovery { stem } => self.load_recovery(&stem), @@ -510,14 +515,14 @@ impl Supervisor { // Swallow a poll error rather than propagate: the task just isn't // latched this pass and is retried next. waitid failing is rare and // must not take down the loop. - let _ = t.poll_exit(); + let _ = t.process.poll_exit(); // Freeze the preview from the complete output and final screen. t.finalize_preview(); - if t.overdue(now, self.kill_grace) { - t.force_kill(); + if t.process.overdue(now, self.kill_grace) { + t.process.force_kill(); } } - // Removed tasks still need exit handling and escalation. + // Retired runs still need exit handling and escalation. for t in &mut self.graveyard { let _ = t.poll_exit(); if t.overdue(now, self.kill_grace) { @@ -544,7 +549,7 @@ impl Supervisor { fn flagship_holder(&self) -> Option { self.flagship.filter(|&id| { self.index_of(id) - .is_some_and(|i| self.tasks[i].finished.is_none()) + .is_some_and(|i| self.tasks[i].process.finished.is_none()) }) } @@ -555,7 +560,7 @@ impl Supervisor { /// Collection and PTY teardown never block on live processes or workers. fn shutdown_all(&mut self) { for t in &mut self.tasks { - t.terminate(); + t.process.terminate(); } let deadline = Instant::now() + self.kill_grace; while (!self.tasks.is_empty() || !self.graveyard.is_empty()) && Instant::now() < deadline { @@ -586,9 +591,9 @@ impl Supervisor { .tasks .iter_mut() .map(|t| { - t.flush_expired_sync(); + t.process.flush_expired_sync(); // Drain all tasks so stores from inactive tasks cannot be forwarded later. - let stores = t.drain_clipboard(); + let stores = t.process.drain_clipboard(); if forwarding == Some(t.id) { clipboard = Some((t.id, stores)); } @@ -601,11 +606,15 @@ impl Supervisor { managed: t.harness.is_some(), group: t.group.clone(), name: t.name.clone(), - lifecycle: t.lifecycle(now, IDLE_AFTER), + lifecycle: t.process.lifecycle(now, IDLE_AFTER), preview: t.resolve_preview(now), - started_ago: now.duration_since(t.started), - quiet_ago: t.finished.is_none().then(|| t.quiet_for(now)), - finished_ago: t.finished.map(|f| now.duration_since(f)), + started_ago: now.duration_since(t.process.started), + quiet_ago: t + .process + .finished + .is_none() + .then(|| t.process.quiet_for(now)), + finished_ago: t.process.finished.map(|f| now.duration_since(f)), } }) .collect(); @@ -627,9 +636,9 @@ impl Supervisor { if let Some(id) = self.watched && let Some(t) = self.tasks.iter().find(|t| t.id == id) { - let (formatted, cursor, hide) = t.formatted(); - let (wants_mouse, alt_screen, alt_scroll) = t.input_hints(); - let sb = t.scroll_offset(); + let (formatted, cursor, hide) = t.process.formatted(); + let (wants_mouse, alt_screen, alt_scroll) = t.process.input_hints(); + let sb = t.process.scroll_offset(); let mut view = ScreenView { id, // `lines` stays empty on both the stored and candidate copies @@ -648,7 +657,7 @@ impl Supervisor { // Send only when rendering or input-policy state changes. if self.last_screen.as_ref() != Some(&view) { self.last_screen = Some(view.clone()); - view.lines = t.screen_lines(); + view.lines = t.process.screen_lines(); self.events.push(Event::Screen(view)); } } @@ -744,14 +753,14 @@ impl Supervisor { } } - /// Terminate a removed task, unlink its capture, and retain it for + /// Terminate a displaced run, unlink its capture, and retain it for /// escalation and reaping. - fn retire(&mut self, mut t: Task) { - t.terminate(); - if let Some(cap) = &t.capture_file { + fn retire(&mut self, mut p: Process) { + p.terminate(); + if let Some(cap) = &p.capture_file { let _ = std::fs::remove_file(cap); } - self.graveyard.push(t); + self.graveyard.push(p); } /// Route one input send to task `id`, reporting a bounded-queue refusal. @@ -885,8 +894,8 @@ impl Supervisor { if let Some((h, home, capture_file, resume_id)) = meta { task.harness = Some(h); // Registry reads must keep using the launch-time configuration. - task.harness_home = home; - task.capture_file = Some(capture_file); + task.process.harness_home = home; + task.process.capture_file = Some(capture_file); task.resume_id = resume_id; } task.run = run; @@ -989,8 +998,8 @@ impl Supervisor { return; }; // A task can exit between the last reap tick and this request. - let _ = self.tasks[i].poll_exit(); - if self.tasks[i].finished.is_none() { + let _ = self.tasks[i].process.poll_exit(); + if self.tasks[i].process.finished.is_none() { self.status("rerun failed: task is still running"); return; } @@ -1023,16 +1032,26 @@ impl Supervisor { // run number to isolate the replacement's capture file. let run = self.tasks[i].run + 1; match self.spawn_task(id, run, &relaunch, &cwd, &launch.env) { - Ok((mut fresh, notice)) => { + Ok((fresh, notice)) => { if let Some(notice) = notice { self.status(format!("task {id}: {notice}")); } - fresh.tagged = self.tasks[i].tagged; - fresh.group.clone_from(&self.tasks[i].group); - fresh.name.clone_from(&self.tasks[i].name); - // Read the resume ID before retiring the displaced run and removing its - // capture file. - let old = std::mem::replace(&mut self.tasks[i], fresh); + // Keep the row's identity; take from the fresh spawn only what a + // new launch sets: its harness, run number, planned resume ID, + // and the process. The resume ID was read above, before the + // displaced run and its capture file go. + let Task { + harness, + run, + resume_id, + process, + .. + } = fresh; + let t = &mut self.tasks[i]; + t.harness = harness; + t.run = run; + t.resume_id = resume_id; + let old = std::mem::replace(&mut t.process, process); self.retire(old); // Reset the fingerprint for the replacement task's screen. if self.watched == Some(id) { diff --git a/src/supervisor_capture_tests.rs b/src/supervisor_capture_tests.rs index 7fbca5e..92fd372 100644 --- a/src/supervisor_capture_tests.rs +++ b/src/supervisor_capture_tests.rs @@ -101,7 +101,7 @@ fn install_status_record(home: &Path, pid: u32, cwd: &Path, status: &str) { fn stamped(task: &Task, json: &str) -> String { format!( "{}\n{json}\n", - task.pid().expect("a spawned task has a pid") + task.process.pid().expect("a spawned task has a pid") ) } @@ -238,7 +238,7 @@ fn spawn_claude_pins_an_id_and_layers_settings() { ); let t = &s.tasks[0]; - let cap = t.capture_file.clone().expect("capture file set"); + let cap = t.process.capture_file.clone().expect("capture file set"); // An explicit runtime directory is used without a discriminator; // capture files sit in this incarnation's `-` namespace // directly under it. @@ -284,7 +284,7 @@ fn spawn_non_agent_command_is_not_instrumented() { spawn(&mut s, "printf ok", dir.to_path_buf()); let t = &s.tasks[0]; assert!(t.harness.is_none()); - assert!(t.capture_file.is_none()); + assert!(t.process.capture_file.is_none()); assert!(t.resume_id.is_none()); assert!( s.capture.is_empty(), @@ -328,7 +328,7 @@ fn typed_agent_word_is_literal_runs_verbatim_and_saves_as_text() { ); let t = &s.tasks[0]; assert!(t.harness.is_none(), "a literal task has no harness channel"); - assert!(t.capture_file.is_none()); + assert!(t.process.capture_file.is_none()); assert!(t.resume_id.is_none()); assert_eq!(t.command, "claude"); assert!( @@ -340,12 +340,13 @@ fn typed_agent_word_is_literal_runs_verbatim_and_saves_as_text() { "no assets are installed for a literal" ); assert!(!runtime.exists()); - let pid = t.pid().expect("a spawned task has a pid"); + let pid = t.process.pid().expect("a spawned task has a pid"); // Install a namespace by launching a managed sibling. Write a capture for task 1 there, // stamped with task 1's leader PID, and create a registry record keyed to that PID. s.spawn_agent("claude", &dir, None); let ns = s.tasks[1] + .process .capture_file .as_deref() .and_then(Path::parent) @@ -392,7 +393,11 @@ fn rerun_cannot_read_the_old_runs_stale_capture() { s.spawn_agent("claude", &dir, None); let id = s.tasks[0].id; // The old run's final capture becomes the new run's launch ID. - let old_cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let old_cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); std::fs::write( &old_cap, stamped(&s.tasks[0], &format!(r#"{{"session_id":"{CAP_ID}"}}"#)), @@ -401,7 +406,11 @@ fn rerun_cannot_read_the_old_runs_stale_capture() { wait_exited(&mut s, id); s.apply(Command::Restart { id }); - let new_cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let new_cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); assert_ne!(new_cap, old_cap, "the fresh run needs its own capture file"); assert_eq!(s.tasks[0].resume_id.as_deref(), Some(CAP_ID)); assert!( @@ -429,7 +438,11 @@ fn reconnect_with_unchanged_root_preserves_capture_files() { install_stub(&bin, "claude", &dir); let mut s = sup_ctx(agent_ctx(&bin, &runtime, dir.to_path_buf())); s.spawn_agent("claude", &dir, None); - let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); std::fs::write(&cap, "{}").unwrap(); // The client reconnects with an identical env and spawns again. @@ -450,7 +463,11 @@ fn returning_to_a_prior_root_preserves_its_live_captures() { install_stub(&bin, "claude", &dir); let mut s = sup_ctx(agent_ctx(&bin, &root_a, dir.to_path_buf())); s.spawn_agent("claude", &dir, None); - let cap_a = s.tasks[0].capture_file.clone().expect("capture file set"); + let cap_a = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); std::fs::write(&cap_a, "{}").unwrap(); // The client reconnects under root B, spawns, then returns to A and @@ -466,6 +483,7 @@ fn returning_to_a_prior_root_preserves_its_live_captures() { "returning to a known root must not disturb its live captures" ); let ns_b = s.tasks[1] + .process .capture_file .as_deref() .and_then(|c| c.parent()) @@ -489,7 +507,7 @@ fn remove_deletes_the_capture_file_under_the_spawn_root() { let _ = wait_argv(&mut s, &dir.join("argv")); let id = s.tasks[0].id; wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok); - let cap = s.tasks[0].capture_file.clone().unwrap(); + let cap = s.tasks[0].process.capture_file.clone().unwrap(); std::fs::write(&cap, "{}").unwrap(); // Root B is installed by a newer spawn; a same-id file under it must @@ -497,6 +515,7 @@ fn remove_deletes_the_capture_file_under_the_spawn_root() { s.set_launch_context(agent_ctx(&bin, &root_b, dir.to_path_buf())); s.spawn_agent("claude", &dir, None); let decoy = s.tasks[1] + .process .capture_file .as_deref() .and_then(|c| c.parent()) @@ -560,7 +579,7 @@ fn spawn_codex_installs_the_notify_and_embedded_overrides() { let t = &s.tasks[0]; assert_eq!(t.command, "codex"); assert!(t.resume_id.is_none(), "codex cannot pin an id at launch"); - assert!(t.capture_file.is_some()); + assert!(t.process.capture_file.is_some()); } /// A `grok` spawn receives exactly the pinned ID: no settings overlay, @@ -634,7 +653,7 @@ fn spawn_omp_loads_the_capture_extension() { "the module must write to the capture env: {text:?}" ); let t = &s.tasks[0]; - let cap = t.capture_file.clone().expect("capture file set"); + let cap = t.process.capture_file.clone().expect("capture file set"); assert_eq!( module.parent(), cap.parent(), @@ -703,7 +722,7 @@ fn printed_resume_hints_do_not_change_saved_recovery_or_rerun_targets() { let expected_id = match tool { "claude" => { std::fs::write( - s.tasks[0].capture_file.as_ref().unwrap(), + s.tasks[0].process.capture_file.as_ref().unwrap(), stamped(&s.tasks[0], &format!(r#"{{"session_id":"{CAP_OTHER}"}}"#)), ) .unwrap(); @@ -715,7 +734,7 @@ fn printed_resume_hints_do_not_change_saved_recovery_or_rerun_targets() { assert_ne!(expected_id.as_deref(), Some(CAP_ID)); assert!( reap_until(&mut s, Duration::from_secs(5), |s| { - s.tasks[0].finished.is_some() && s.tasks[0].reader_done() + s.tasks[0].process.finished.is_some() && s.tasks[0].process.reader_done() }), "{tool}: output never completed" ); @@ -723,6 +742,7 @@ fn printed_resume_hints_do_not_change_saved_recovery_or_rerun_targets() { s.reap(); assert!( s.tasks[0] + .process .screen_lines() .iter() .any(|line| line.contains(CAP_ID)), @@ -785,7 +805,7 @@ fn rerun_latches_exit_without_reap_and_preserves_the_launch_id() { assert!(s.tasks[0].harness.is_some()); assert_eq!(s.tasks[0].resume_id.as_deref(), Some(CAP_OTHER)); let id = s.tasks[0].id; - let pid = Pid::from_raw(s.tasks[0].pid().unwrap() as i32).unwrap(); + let pid = Pid::from_raw(s.tasks[0].process.pid().unwrap() as i32).unwrap(); assert!( wait_until(Duration::from_secs(5), || { let exited = waitid( @@ -794,12 +814,12 @@ fn rerun_latches_exit_without_reap_and_preserves_the_launch_id() { ) .unwrap() .is_some(); - exited && s.tasks[0].reader_done() + exited && s.tasks[0].process.reader_done() }), "the stub never exited and drained its output" ); assert!( - s.tasks[0].finished.is_none(), + s.tasks[0].process.finished.is_none(), "no exit latch may have run yet" ); std::fs::remove_file(dir.join("argv")).unwrap(); @@ -842,7 +862,7 @@ fn resume_id_precedence_registry_over_spawn_under_capture() { .expect("a fresh claude launch pins an id"); assert_ne!(injected.as_str(), CAP_ID); // Key the registry fixture to the spawned task. - let pid = s.tasks[0].pid().expect("a live task has a pid"); + let pid = s.tasks[0].process.pid().expect("a live task has a pid"); install_status_record(&claude_home, pid, &dir, r#""status":"idle""#); assert_eq!( @@ -852,7 +872,11 @@ fn resume_id_precedence_registry_over_spawn_under_capture() { ); // A capture-file ID outranks the registry ID. - let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); std::fs::write(&cap, stamped(&s.tasks[0], &hook_json(CAP_OTHER, "clear"))).unwrap(); assert_eq!( saved_resume(&mut s, &config, "capture").as_deref(), @@ -903,8 +927,12 @@ until [ -e '{d}/done' ]; do sleep 0.05; done"#, .resume_id .clone() .expect("a fresh claude launch pins an id"); - let pid = s.tasks[0].pid().expect("a live task has a pid"); - let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let pid = s.tasks[0].process.pid().expect("a live task has a pid"); + let cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); // The foreign stage runs the hook command from the installed overlay. let overlay = std::fs::read_to_string(cap.with_file_name("claude-settings.json")).unwrap(); @@ -1006,13 +1034,14 @@ fn silent_codex_tasks_save_and_recover_without_a_resume_id() { assert!(reap_until(&mut s, Duration::from_secs(5), |s| s .tasks .iter() - .all(|t| t.finished.is_some()))); + .all(|t| t.process.finished.is_some()))); for task in &s.tasks { assert!( task.current_resume_id().is_none(), "no capture channel fired" ); let spawn_ms = task + .process .spawned_at .duration_since(std::time::SystemTime::UNIX_EPOCH) .unwrap() @@ -1055,7 +1084,11 @@ fn title_turn() -> String { /// `codex_home` as the environment. Return the written root, as for a binary invocation /// from the injected script. fn arrive(s: &Supervisor, codex_home: &Path, payload: &str) -> Option { - let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); crate::harness::record_arrival(payload, &|key| match key { crate::harness::CAPTURE_ENV => Some(cap.clone()), "CODEX_HOME" => Some(codex_home.to_path_buf()), @@ -1250,7 +1283,7 @@ fn home_only_launch_env_targets_the_clients_dot_codex() { )); s.spawn_agent("codex", &dir, None); assert_eq!( - s.tasks[0].harness_home.as_deref(), + s.tasks[0].process.harness_home.as_deref(), Some(codex_home.as_path()), "HOME alone must resolve the harness home" ); @@ -1310,7 +1343,11 @@ fn stale_inherited_notify_chain_is_never_executed() { s.spawn_agent("codex", &dir, None); let id = s.tasks[0].id; wait_exited(&mut s, id); - let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); assert_eq!( std::fs::read_to_string(&cap).unwrap(), CAP_ID, @@ -1421,7 +1458,7 @@ fn config_toml_notify_chains_through_the_injected_script() { format!("{}\nturn-ended", notifier.display()), "the child env must carry the displaced argv, newline-joined" ); - let cap = s.tasks[0].capture_file.clone().unwrap(); + let cap = s.tasks[0].process.capture_file.clone().unwrap(); assert_eq!( std::fs::read_to_string(&cap).unwrap(), CAP_ID, @@ -1541,7 +1578,11 @@ fn recovery_cadence_rewrites_on_capture_drift_and_skips_when_static() { assert!(!snapshot(&rec).unwrap().1.contains(CAP_OTHER)); // Change the capture ID without a recipe mutation. - let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + let cap = s.tasks[0] + .process + .capture_file + .clone() + .expect("capture file set"); std::fs::write(&cap, stamped(&s.tasks[0], &hook_json(CAP_OTHER, "clear"))).unwrap(); assert!( wait_until(Duration::from_secs(5), || { @@ -1571,6 +1612,7 @@ fn recovery_cadence_rewrites_on_capture_drift_and_skips_when_static() { /// The settings overlay beside the sole task's capture file. fn settings_beside(s: &Supervisor) -> String { s.tasks[0] + .process .capture_file .as_deref() .and_then(Path::parent) @@ -1625,7 +1667,7 @@ fn managed_claude_accepts_its_own_capture_and_refuses_a_foreign_stamp() { ); // Reject the same payload when stamped with another process's PID. - let cap = s.tasks[0].capture_file.clone().unwrap(); + let cap = s.tasks[0].process.capture_file.clone().unwrap(); let text = std::fs::read_to_string(&cap).unwrap(); let (_, json) = text.split_once('\n').unwrap(); std::fs::write(&cap, format!("{}\n{json}", dead_pid())).unwrap(); @@ -1684,13 +1726,13 @@ fn managed_rerun_resumes_the_captured_id_from_the_binary_path_finds_now() { s.tasks[0].name = Some("pilot".into()); wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok); std::fs::write( - s.tasks[0].capture_file.as_ref().unwrap(), + s.tasks[0].process.capture_file.as_ref().unwrap(), stamped(&s.tasks[0], &hook_json(CAP_OTHER, "clear")), ) .unwrap(); std::fs::remove_file(dir.join("argv")).unwrap(); - let old_cap = s.tasks[0].capture_file.clone().unwrap(); + let old_cap = s.tasks[0].process.capture_file.clone().unwrap(); s.apply(Command::Restart { id }); let argv = wait_argv(&mut s, &dir.join("argv")); assert_eq!( @@ -1788,7 +1830,7 @@ fn managed_task_saves_as_a_managed_entry_and_reloads_managed() { let claude = s.tasks[0].id; wait_for_lifecycle(&mut s, claude, |l| l == Lifecycle::Ok); std::fs::write( - s.tasks[0].capture_file.as_ref().unwrap(), + s.tasks[0].process.capture_file.as_ref().unwrap(), stamped(&s.tasks[0], &format!(r#"{{"session_id":"{CAP_ID}"}}"#)), ) .unwrap(); @@ -1929,7 +1971,7 @@ fn managed_codex_reports_the_capture_notice_and_leads_a_rerun_with_resume() { wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok); // The v1 slot holds one bare root UUID. - std::fs::write(s.tasks[0].capture_file.as_ref().unwrap(), CAP_ID).unwrap(); + std::fs::write(s.tasks[0].process.capture_file.as_ref().unwrap(), CAP_ID).unwrap(); std::fs::remove_file(dir.join("argv")).unwrap(); s.apply(Command::Restart { id }); assert_sole_notice(s.drain(), "capture unavailable"); @@ -1980,7 +2022,7 @@ fn registry_waiting_status_reaches_the_dashboard_preview() { )); s.spawn_agent("claude", &dir, None); // Key the record to the task's leader PID. - let pid = s.tasks[0].pid().expect("a live task has a pid"); + let pid = s.tasks[0].process.pid().expect("a live task has a pid"); // A non-waiting status must not anchor the preview. install_status_record(&claude_home, pid, &dir, r#""status":"idle""#); diff --git a/src/supervisor_tests.rs b/src/supervisor_tests.rs index dbecd2b..4d58cd5 100644 --- a/src/supervisor_tests.rs +++ b/src/supervisor_tests.rs @@ -516,7 +516,7 @@ fn ticks_without_copies(s: &mut Supervisor, why: &str) -> bool { /// Search the first task's formatted grid for `marker`. fn grid_shows(s: &Supervisor, marker: &str) -> bool { s.tasks.first().is_some_and(|t| { - let (formatted, _, _) = t.formatted(); + let (formatted, _, _) = t.process.formatted(); String::from_utf8_lossy(&formatted).contains(marker) }) } @@ -536,7 +536,7 @@ fn scrolled_task() -> (Supervisor, u64) { id, action: ScrollAction::Up(3), }); - s.tasks[0].scroll_offset() > 0 + s.tasks[0].process.scroll_offset() > 0 }); assert!(scrolled, "the task never accrued scrollback"); (s, id) @@ -742,7 +742,7 @@ fn clear_watch_snaps_the_watched_task_live() { let (mut s, _) = scrolled_task(); s.clear_watch(); assert_eq!( - s.tasks[0].scroll_offset(), + s.tasks[0].process.scroll_offset(), 0, "disconnect must return the watched task's viewport to live" ); @@ -875,7 +875,7 @@ fn kill_with_an_unknown_id_leaves_the_live_task_alone() { s.apply(Command::Kill { id: 999 }); assert!(s.drain().is_empty(), "unknown-id Kill must stay silent"); assert!( - !s.tasks[0].overdue(Instant::now(), Duration::ZERO), + !s.tasks[0].process.overdue(Instant::now(), Duration::ZERO), "unknown-id Kill must not signal the live task" ); assert_eq!(view_of(&mut s, id).lifecycle, before); @@ -902,7 +902,7 @@ fn tag_with_an_unknown_id_leaves_the_live_task_alone() { #[test] fn scrollback_with_an_unknown_id_leaves_the_live_task_alone() { let (mut s, _) = scrolled_task(); - let offset = s.tasks[0].scroll_offset(); + let offset = s.tasks[0].process.scroll_offset(); s.apply(Command::Scrollback { id: 999, @@ -913,7 +913,7 @@ fn scrollback_with_an_unknown_id_leaves_the_live_task_alone() { "unknown-id Scrollback must stay silent" ); assert_eq!( - s.tasks[0].scroll_offset(), + s.tasks[0].process.scroll_offset(), offset, "unknown-id Scrollback must not snap the live task's viewport" ); @@ -1160,7 +1160,9 @@ fn reap_until( fn wait_exited(s: &mut Supervisor, id: u64) { assert!( reap_until(s, Duration::from_secs(5), |s| { - s.tasks.iter().all(|t| t.id != id || t.finished.is_some()) + s.tasks + .iter() + .all(|t| t.id != id || t.process.finished.is_some()) }), "task {id} never exited" ); @@ -1240,7 +1242,7 @@ fn rerun_sweeps_stragglers_of_the_old_run() { "rerun never swept the old run's straggler" ); // The fresh run exists under the same id; its own straggler dies with - // the supervisor (Task::drop backstop). + // the supervisor (Process::drop backstop). assert!(s.tasks.iter().any(|t| t.id == id)); } @@ -1342,7 +1344,7 @@ fn shutdown_waits_the_grace_when_every_task_has_finished() { spawn(&mut s, "true", here()); } assert!(reap_until(&mut s, Duration::from_secs(5), |s| { - s.tasks.len() == 2 && s.tasks.iter().all(|t| t.finished.is_some()) + s.tasks.len() == 2 && s.tasks.iter().all(|t| t.process.finished.is_some()) })); let t0 = Instant::now(); s.apply(Command::Shutdown); @@ -1774,7 +1776,7 @@ fn key_command_encodes_against_live_cursor_mode() { s.tasks .iter() .find(|t| t.id == id) - .is_some_and(|t| t.input_hints().1) + .is_some_and(|t| t.process.input_hints().1) }; assert!( wait_until(Duration::from_secs(5), || app_cursor_on(&s)), diff --git a/src/task.rs b/src/task.rs index f4eff93..6807337 100644 --- a/src/task.rs +++ b/src/task.rs @@ -62,12 +62,44 @@ pub fn positive_pid(field: &str) -> Option { field.parse::().ok().filter(|p| *p > 0) } +/// A dashboard row: the identity a rerun preserves, plus the [`Process`] of +/// its current run. Every field here survives `r`; everything bound to one +/// launch, from the PTY to the exit latch, sits in `process` and is swapped +/// whole on rerun. pub struct Task { pub id: u64, pub command: String, /// Working directory the command was launched in: the grouping key for /// "by dir" mode and the label shown when it differs from the default. pub cwd: PathBuf, + pub tagged: bool, + /// Dashboard group stored with the task; `None` means unassigned. + pub group: Option, + /// Custom display name; `None` means unnamed. + pub name: Option, + /// Harness for a managed task's session ID and blocked-status reads: the harness built + /// the argv, no shell sits between, and a rerun goes through it again. `None` for + /// literal tasks: never read a harness channel for those tasks, and never edit their + /// command text. + pub harness: Option<&'static dyn crate::harness::Harness>, + /// Display-only summary adapter selected from the command's first word, + /// for literal and managed tasks alike. + pub summary_adapter: Option<&'static dyn crate::preview::SummaryAdapter>, + /// Run number used to give each rerun a distinct capture path. + pub run: u32, + /// Session ID pinned or resumed at spawn. Prefer a captured ID or matching live + /// registry ID when available. + pub resume_id: Option, + /// The current run. `Task` has no `Drop` of its own so the run can be moved + /// out and retired on its own; the kill backstop travels with it. + pub process: Process, +} + +/// One run of a task: the PTY, writer queue, reader thread, emulator, the +/// harness paths resolved at launch, preview state, and exit bookkeeping for a +/// single process group. Dropping it is the kill backstop: SIGKILL to the group +/// and a best-effort reap, however the run leaves scope. +pub struct Process { /// Keep the PTY master open for `resize` (`TIOCSWINSZ`). Split the reader and /// writer handles from it at spawn. master: Box, @@ -90,30 +122,12 @@ pub struct Task { parser: Arc>, last_activity: Arc>, handle: Option>, - pub tagged: bool, - /// Dashboard group stored with the task; `None` means unassigned. - pub group: Option, - /// Custom display name; `None` means unnamed. - pub name: Option, - /// Harness for a managed task's session ID and blocked-status reads: the harness built - /// the argv, no shell sits between, and a rerun goes through it again. `None` for - /// literal tasks: never read a harness channel for those tasks, and never edit their - /// command text. - pub harness: Option<&'static dyn crate::harness::Harness>, /// Harness home resolved from this run's launch environment. pub harness_home: Option, - /// Display-only summary adapter selected from the command's first word, - /// for literal and managed tasks alike. - pub summary_adapter: Option<&'static dyn crate::preview::SummaryAdapter>, - /// Run number used to give each rerun a distinct capture path. - pub run: u32, - /// Session ID pinned or resumed at spawn. Prefer a captured ID or matching live - /// registry ID when available. - pub resume_id: Option, - /// Capture path allocated for this task run. + /// Capture path allocated for this run. pub capture_file: Option, - /// Dashboard-preview resolution state; resets with the task on rerun - /// because a rerun replaces the whole `Task`. + /// Dashboard-preview resolution state; resets on rerun because a rerun + /// replaces the whole `Process`. preview: PreviewState, /// Cached blocked-on-user status from the harness registry. blocked: Option<(String, &'static str)>, @@ -150,7 +164,7 @@ fn signal(waker: &Waker) { /// Admit one whole message to a writer queue bounded by `MAX_PENDING_WRITE`, /// or refuse it whole. The cap check and the `fetch_add` are separate /// operations, so racing admitters can overshoot the cap by one message (see -/// `Task::pending_write`). A failed send means the worker exited; the +/// `Process::pending_write`). A failed send means the worker exited; the /// compensating `fetch_sub` removes that admission so the count never leaks. fn admit_write( tx: &Sender>, @@ -231,6 +245,103 @@ impl Task { scrollback: usize, env: &[(OsString, OsString)], waker: &Waker, + ) -> io::Result { + let process = Process::spawn(command, exec, cwd, rows, cols, scrollback, env, waker)?; + Ok(Self { + id, + command: command.to_string(), + cwd: cwd.to_path_buf(), + tagged: false, + group: None, + name: None, + harness: None, + summary_adapter: crate::harness::select(command), + run: 0, + resume_id: None, + process, + }) + } + + /// Check the capture file, then the live registry, then the spawn-time ID. A + /// managed CLI can select a session after launch, so validate captured IDs + /// through the harness with the task leader's PID. If a source has no valid ID, + /// check the next source. For literal tasks, return `None`. + pub fn current_resume_id(&self) -> Option { + let p = &self.process; + if let (Some(h), Some(path)) = (self.harness, &p.capture_file) + && let Ok(payload) = std::fs::read_to_string(path) + && let Some(id) = h.parse_capture(&payload, p.pid()) + { + return Some(id); + } + if let (Some(h), Some(pid)) = (self.harness, p.pid()) + && let Some(id) = + h.live_session_id(pid, &self.cwd, p.spawned_at, p.harness_home.as_deref()) + { + return Some(id); + } + self.resume_id.clone() + } + + /// The dashboard preview, resolved through the provenance cascade under + /// the grid lock (see [`crate::preview`]). `now` is the caller's tick + /// instant so every task in one snapshot resolves against the same clock. + pub fn resolve_preview(&mut self, now: Instant) -> Preview { + self.refresh_blocked(now); + let p = &mut self.process; + let emu = p.parser.lock(); + let blocked = p.blocked.as_ref().map(|(text, rule)| (&**text, *rule)); + p.preview + .resolve(now, &*emu, self.summary_adapter, blocked) + .clone() + } + + /// Refresh the harness's blocked-on-user status at most once per + /// [`BLOCKED_PROBE_INTERVAL`]. Tasks without a harness or PID do not probe; + /// finished tasks clear the cached status. + fn refresh_blocked(&mut self, now: Instant) { + let p = &mut self.process; + let (Some(h), Some(pid), None) = (self.harness, p.pid, p.finished) else { + p.blocked = None; + return; + }; + if p.blocked_probed + .is_some_and(|t| now.duration_since(t) < BLOCKED_PROBE_INTERVAL) + { + return; + } + p.blocked_probed = Some(now); + p.blocked = h.live_blocked_status(pid, &self.cwd, p.spawned_at, p.harness_home.as_deref()); + } + + /// Freeze the preview once output is complete. Apply any open `?2026` frame + /// first so the preview includes its buffered output. + pub fn finalize_preview(&mut self) { + let p = &mut self.process; + if p.preview.finalized() || !p.output_complete() { + return; + } + let mut emu = p.parser.lock(); + let _ = emu.finish_output(); + p.preview.finalize(&*emu, self.summary_adapter); + } +} + +impl Process { + /// Spawn `exec` in a fresh `rows`×`cols` PTY with `scrollback` history + /// rows; a literal `exec` runs `command` under the launch shell. Pass + /// exactly `env` to the child and notify the core through `waker` on + /// terminal output. + #[allow(clippy::too_many_arguments)] // All arguments define launch state. + pub fn spawn( + command: &str, + exec: Exec, + cwd: &Path, + rows: u16, + cols: u16, + scrollback: usize, + env: &[(OsString, OsString)], + waker: &Waker, ) -> io::Result { let pair = native_pty_system() .openpty(PtySize { @@ -338,9 +449,6 @@ impl Task { let pid = child.process_id(); drop(child); Ok(Self { - id, - command: command.to_string(), - cwd: cwd.to_path_buf(), master: pair.master, input_tx: Some(input_tx), pending_write, @@ -348,14 +456,7 @@ impl Task { parser, last_activity, handle: Some(handle), - tagged: false, - group: None, - name: None, - harness: None, harness_home: None, - summary_adapter: crate::harness::select(command), - run: 0, - resume_id: None, capture_file: None, preview: PreviewState::new(), blocked: None, @@ -370,39 +471,15 @@ impl Task { }) } - /// Return the task's session-leader PID. + /// Return the session-leader PID. pub fn pid(&self) -> Option { self.pid } - /// Check the capture file, then the live registry, then the spawn-time ID. A - /// managed CLI can select a session after launch, so validate captured IDs - /// through the harness with the task leader's PID. If a source has no valid ID, - /// check the next source. For literal tasks, return `None`. - pub fn current_resume_id(&self) -> Option { - if let (Some(h), Some(path)) = (self.harness, &self.capture_file) - && let Ok(payload) = std::fs::read_to_string(path) - && let Some(id) = h.parse_capture(&payload, self.pid()) - { - return Some(id); - } - if let (Some(h), Some(pid)) = (self.harness, self.pid()) - && let Some(id) = h.live_session_id( - pid, - &self.cwd, - self.spawned_at, - self.harness_home.as_deref(), - ) - { - return Some(id); - } - self.resume_id.clone() - } - /// Latch the exit code and finish time if the leader has exited, without /// reaping it. `WNOWAIT` leaves the zombie in place, which is what keeps /// the pid (and therefore the pgid) reserved so the group stays signalable - /// for the task's whole life; see the `reaped` field. The zombie is + /// for the run's whole life; see the `reaped` field. The zombie is /// collected at teardown, after SIGKILL. pub fn poll_exit(&mut self) -> io::Result<()> { if self.finished.is_some() || self.reaped { @@ -492,7 +569,7 @@ impl Task { } /// Time since the reader thread last saw PTY output. Zero when the - /// activity lock is poisoned, so a task whose reader died mid-update + /// activity lock is poisoned, so a run whose reader died mid-update /// reads as just-active, never as stuck-idle. pub fn quiet_for(&self, now: Instant) -> Duration { self.last_activity @@ -514,57 +591,11 @@ impl Task { } } - /// Drain OSC 52 clipboard stores captured by this task's emulator. + /// Drain OSC 52 clipboard stores captured by this run's emulator. pub fn drain_clipboard(&self) -> ClipboardStores { self.parser.lock().drain_clipboard() } - /// The dashboard preview, resolved through the provenance cascade under - /// the grid lock (see [`crate::preview`]). `now` is the caller's tick - /// instant so every task in one snapshot resolves against the same clock. - pub fn resolve_preview(&mut self, now: Instant) -> Preview { - self.refresh_blocked(now); - let emu = self.parser.lock(); - let blocked = self.blocked.as_ref().map(|(text, rule)| (&**text, *rule)); - self.preview - .resolve(now, &*emu, self.summary_adapter, blocked) - .clone() - } - - /// Refresh the harness's blocked-on-user status at most once per - /// [`BLOCKED_PROBE_INTERVAL`]. Tasks without a harness or PID do not probe; - /// finished tasks clear the cached status. - fn refresh_blocked(&mut self, now: Instant) { - let (Some(h), Some(pid), None) = (self.harness, self.pid, self.finished) else { - self.blocked = None; - return; - }; - if self - .blocked_probed - .is_some_and(|t| now.duration_since(t) < BLOCKED_PROBE_INTERVAL) - { - return; - } - self.blocked_probed = Some(now); - self.blocked = h.live_blocked_status( - pid, - &self.cwd, - self.spawned_at, - self.harness_home.as_deref(), - ); - } - - /// Freeze the preview once output is complete. Apply any open `?2026` frame - /// first so the preview includes its buffered output. - pub fn finalize_preview(&mut self) { - if self.preview.finalized() || !self.output_complete() { - return; - } - let mut emu = self.parser.lock(); - let _ = emu.finish_output(); - self.preview.finalize(&*emu, self.summary_adapter); - } - /// Full screen as ANSI bytes for attached mode, plus cursor state so we can /// place the real cursor where the child put it. pub fn formatted(&self) -> (Vec, (u16, u16), bool) { @@ -607,7 +638,7 @@ impl Task { /// Admit one whole message to the writer queue, or refuse it whole. fn queue_write(&self, msg: Vec) -> Result<(), WriteRefused> { - // A force-killed task has no writer queue; discard subsequent input. + // A force-killed run has no writer queue; discard subsequent input. let Some(tx) = &self.input_tx else { return Ok(()); }; @@ -662,10 +693,10 @@ impl Task { ) } - /// Ask the whole task to exit: SIGTERM to the process *group*, not just the + /// Ask the whole run to exit: SIGTERM to the process *group*, not just the /// direct child, so every group member gets it, including background /// children a `cmd &` left behind (a non-interactive shell's `&` creates no - /// new group, so they never leave this one). TERM, not KILL: the task gets a + /// new group, so they never leave this one). TERM, not KILL: the child gets a /// chance to flush and clean up. The supervisor owns the escalation: /// `overdue` turns true once the grace elapses, and `force_kill` finishes it. /// @@ -689,7 +720,7 @@ impl Task { .is_some_and(|t| now.duration_since(t) >= grace) } - /// Send SIGKILL to the task's process group without waiting for it to exit. + /// Send SIGKILL to the process group without waiting for it to exit. pub fn force_kill(&mut self) { if !self.reaped && let Some(pid) = self.pid @@ -706,10 +737,10 @@ impl Task { } } -impl Drop for Task { +impl Drop for Process { fn drop(&mut self) { // The last-resort backstop, not the policy point: guarantees no - // orphaned task tree regardless of how a Task leaves scope. Graceful + // orphaned process tree regardless of how a run leaves scope. Graceful // TERM-first teardown happens above this, in the supervisor. The // collect is best-effort: an already-exited leader reaps instantly; one // still dying from the KILL reparents to init, which collects it. diff --git a/src/task_tests.rs b/src/task_tests.rs index 75f63f1..30d0f63 100644 --- a/src/task_tests.rs +++ b/src/task_tests.rs @@ -34,8 +34,8 @@ fn spawn_sh(id: u64, command: &str) -> Task { fn wait_finished(t: &mut Task) { assert!( wait_until(Duration::from_secs(5), || { - t.poll_exit().unwrap(); - t.finished.is_some() + t.process.poll_exit().unwrap(); + t.process.finished.is_some() }), "task never finished" ); @@ -45,8 +45,8 @@ fn wait_finished(t: &mut Task) { fn wait_complete(t: &mut Task) { assert!( wait_until(Duration::from_secs(60), || { - t.poll_exit().unwrap(); - t.output_complete() + t.process.poll_exit().unwrap(); + t.process.output_complete() }), "child never completed" ); @@ -59,25 +59,25 @@ fn spawn_reads_output_and_exits_zero() { let mut t = spawn(1, "printf 'alpha\\nomega\\n'"); let mut preview = String::new(); wait_until(Duration::from_secs(5), || { - t.poll_exit().unwrap(); + t.process.poll_exit().unwrap(); preview = t.resolve_preview(Instant::now()).text; - t.finished.is_some() && preview.contains("omega") + t.process.finished.is_some() && preview.contains("omega") }); - assert_eq!(t.exit_code, Some(0)); + assert_eq!(t.process.exit_code, Some(0)); assert!(preview.contains("omega"), "preview was {preview:?}"); - t.terminate(); + t.process.terminate(); } #[test] fn nonzero_exit_is_recorded() { let mut t = spawn(2, "exit 3"); wait_finished(&mut t); - assert_eq!(t.exit_code, Some(3)); + assert_eq!(t.process.exit_code, Some(3)); assert_eq!( - t.lifecycle(Instant::now(), Duration::from_secs(10)), + t.process.lifecycle(Instant::now(), Duration::from_secs(10)), Lifecycle::Failed ); - t.terminate(); + t.process.terminate(); } /// A live task becomes idle only after the quiet threshold; output resets it. @@ -86,19 +86,22 @@ fn lifecycle_crosses_idle_threshold_and_resets_on_activity() { let mut t = spawn(5, "sleep 5"); // `sleep` writes nothing, so `last_activity` keeps its spawn value // and the injected `now`s measure against a fixed instant. - let quiet_since = *t.last_activity.lock().unwrap(); + let quiet_since = *t.process.last_activity.lock().unwrap(); let window = Duration::from_secs(10); let inside = quiet_since + Duration::from_secs(9); - assert_eq!(t.lifecycle(inside, window), Lifecycle::Active); + assert_eq!(t.process.lifecycle(inside, window), Lifecycle::Active); - assert_eq!(t.lifecycle(quiet_since + window, window), Lifecycle::Active); + assert_eq!( + t.process.lifecycle(quiet_since + window, window), + Lifecycle::Active + ); let past = quiet_since + Duration::from_secs(11); - assert_eq!(t.lifecycle(past, window), Lifecycle::Idle); - *t.last_activity.lock().unwrap() = past; - assert_eq!(t.lifecycle(past, window), Lifecycle::Active); - t.terminate(); + assert_eq!(t.process.lifecycle(past, window), Lifecycle::Idle); + *t.process.last_activity.lock().unwrap() = past; + assert_eq!(t.process.lifecycle(past, window), Lifecycle::Active); + t.process.terminate(); } /// Keep a task active on repeated output before the quiet threshold. @@ -106,14 +109,14 @@ fn lifecycle_crosses_idle_threshold_and_resets_on_activity() { fn sub_window_quiet_gaps_never_read_as_idle() { let mut t = spawn(7, "sleep 5"); let window = Duration::from_secs(10); - let start = *t.last_activity.lock().unwrap(); + let start = *t.process.last_activity.lock().unwrap(); for gaps in 1..=4u32 { let probe = start + Duration::from_secs(9) * gaps; - assert_eq!(t.lifecycle(probe, window), Lifecycle::Active); + assert_eq!(t.process.lifecycle(probe, window), Lifecycle::Active); // Simulate output at the end of each quiet gap. - *t.last_activity.lock().unwrap() = probe; + *t.process.last_activity.lock().unwrap() = probe; } - t.terminate(); + t.process.terminate(); } /// Exit status takes precedence over elapsed quiet time. @@ -122,18 +125,18 @@ fn finished_lifecycle_ignores_quiet_time() { for (command, expected) in [("exit 0", Lifecycle::Ok), ("exit 3", Lifecycle::Failed)] { let mut t = spawn(6, command); wait_finished(&mut t); - let now = *t.last_activity.lock().unwrap() + Duration::from_secs(11); - assert_eq!(t.lifecycle(now, Duration::from_secs(10)), expected); - t.terminate(); + let now = *t.process.last_activity.lock().unwrap() + Duration::from_secs(11); + assert_eq!(t.process.lifecycle(now, Duration::from_secs(10)), expected); + t.process.terminate(); } } #[test] fn resize_is_reflected_in_the_grid() { let mut t = spawn(3, "sleep 5"); - t.resize(30, 100).unwrap(); - assert_eq!(t.parser.lock().size(), (30, 100)); - t.terminate(); + t.process.resize(30, 100).unwrap(); + assert_eq!(t.process.parser.lock().size(), (30, 100)); + t.process.terminate(); } /// The exit latch must not reap: after `finished` latches, the leader is @@ -143,8 +146,8 @@ fn resize_is_reflected_in_the_grid() { fn exited_leader_stays_a_zombie_until_drop() { let mut t = spawn(4, "exit 7"); wait_finished(&mut t); - assert_eq!(t.exit_code, Some(7)); - let pid = Pid::from_raw(t.pid.expect("spawn always yields a pid") as i32); + assert_eq!(t.process.exit_code, Some(7)); + let pid = Pid::from_raw(t.process.pid.expect("spawn always yields a pid") as i32); // Signal 0 = existence check; a zombie still exists. assert!( kill(pid, None).is_ok(), @@ -181,7 +184,7 @@ fn managed_launch_makes_the_agent_the_task_leader_under_any_shell() { wait_finished(&mut t); assert_eq!( read_pid(&pid_file).as_raw() as u32, - t.pid().unwrap(), + t.process.pid().unwrap(), "the agent must be the task leader with no shell in between" ); @@ -191,7 +194,7 @@ fn managed_launch_makes_the_agent_the_task_leader_under_any_shell() { wait_finished(&mut t); assert_ne!( read_pid(&pid_file).as_raw() as u32, - t.pid().unwrap(), + t.process.pid().unwrap(), "premise: under a resident shell the literal agent is a child, not the leader" ); } @@ -210,7 +213,7 @@ fn terminate_reaches_stragglers_after_leader_exit() { let straggler = read_pid(&spid); assert!(kill(straggler, None).is_ok(), "straggler should be alive"); - t.terminate(); // leader already finished: the group signal must still fire + t.process.terminate(); // leader already finished: the group signal must still fire assert!( wait_until(Duration::from_secs(5), || kill(straggler, None).is_err()), "TERM after leader exit never reached the straggler" @@ -221,12 +224,12 @@ fn terminate_reaches_stragglers_after_leader_exit() { #[test] fn killed_leader_latches_137_via_collect() { let mut t = spawn(8, "sleep 300"); - t.force_kill(); // sets kill_sent, so try_collect may reap + t.process.force_kill(); // sets kill_sent, so try_collect may reap assert!( - wait_until(Duration::from_secs(5), || t.try_collect()), + wait_until(Duration::from_secs(5), || t.process.try_collect()), "KILLed leader was never collected" ); - assert_eq!(t.exit_code, Some(137)); + assert_eq!(t.process.exit_code, Some(137)); } /// Collection must retain the group reservation until escalation, including @@ -235,19 +238,19 @@ fn killed_leader_latches_137_via_collect() { fn exited_leader_is_collectible_only_after_kill() { let mut t = spawn(30, "exit 0"); wait_finished(&mut t); - assert!(!t.try_collect()); - t.terminate(); - let term_sent = t.term_sent.unwrap(); - t.terminate(); + assert!(!t.process.try_collect()); + t.process.terminate(); + let term_sent = t.process.term_sent.unwrap(); + t.process.terminate(); assert_eq!( - t.term_sent, + t.process.term_sent, Some(term_sent), "repeated TERM reset the grace" ); - assert!(!t.try_collect()); - assert!(kill(Pid::from_raw(t.pid.unwrap() as i32), None).is_ok()); - t.force_kill(); - assert!(t.try_collect()); + assert!(!t.process.try_collect()); + assert!(kill(Pid::from_raw(t.process.pid.unwrap() as i32), None).is_ok()); + t.process.force_kill(); + assert!(t.process.try_collect()); } /// Clamp scrollback at both ends; return to live output on input. @@ -256,53 +259,56 @@ fn viewport_scrolls_and_snaps_live_on_input() { let mut t = spawn(9, "cat"); // Feed enough rows to create scrollback. for i in 0..50 { - t.parser.lock().process(format!("line{i}\r\n").as_bytes()); + t.process + .parser + .lock() + .process(format!("line{i}\r\n").as_bytes()); } - assert_eq!(t.scroll_offset(), 0); - t.scroll_view(ScrollAction::Up(10)); - assert_eq!(t.scroll_offset(), 10); - t.scroll_view(ScrollAction::Down(4)); - assert_eq!(t.scroll_offset(), 6); - t.scroll_view(ScrollAction::Top); - let top = t.scroll_offset(); + assert_eq!(t.process.scroll_offset(), 0); + t.process.scroll_view(ScrollAction::Up(10)); + assert_eq!(t.process.scroll_offset(), 10); + t.process.scroll_view(ScrollAction::Down(4)); + assert_eq!(t.process.scroll_offset(), 6); + t.process.scroll_view(ScrollAction::Top); + let top = t.process.scroll_offset(); assert!(top > 0); assert!( - t.screen_lines()[0].starts_with("line0"), + t.process.screen_lines()[0].starts_with("line0"), "Top must show the oldest stored row, got {:?}", - t.screen_lines()[0] + t.process.screen_lines()[0] ); // Large upward movement clamps at the oldest row. - t.scroll_view(ScrollAction::Live); - t.scroll_view(ScrollAction::Up(10_000)); - assert_eq!(t.scroll_offset(), top); + t.process.scroll_view(ScrollAction::Live); + t.process.scroll_view(ScrollAction::Up(10_000)); + assert_eq!(t.process.scroll_offset(), top); // Return the viewport to live output on input. - t.send_input(b"x".to_vec()).unwrap(); - assert_eq!(t.scroll_offset(), 0); - t.terminate(); + t.process.send_input(b"x".to_vec()).unwrap(); + assert_eq!(t.process.scroll_offset(), 0); + t.process.terminate(); } /// `screen_lines` returns one entry per grid row, including a blank final row. #[test] fn screen_lines_yields_one_entry_per_grid_row() { let mut t = spawn(60, "sleep 300"); - t.parser.lock().process(b"top"); - let lines = t.screen_lines(); + t.process.parser.lock().process(b"top"); + let lines = t.process.screen_lines(); // The spawn helper's grid is 24x80. assert_eq!(lines.len(), 24, "one entry per grid row"); assert_eq!(lines[0], "top"); assert_eq!(lines[23], "", "the blank bottom row keeps its slot"); - t.terminate(); + t.process.terminate(); } /// The per-task writer worker delivers queued messages in FIFO order. #[test] fn queued_writes_reach_the_child_in_order() { let mut t = spawn(10, "cat"); - t.send_input(b"zqfirstqz\n".to_vec()).unwrap(); - t.send_input(b"zqsecondqz\n".to_vec()).unwrap(); + t.process.send_input(b"zqfirstqz\n".to_vec()).unwrap(); + t.process.send_input(b"zqsecondqz\n".to_vec()).unwrap(); let mut contents = String::new(); wait_until(Duration::from_secs(5), || { - contents = t.parser.lock().contents(); + contents = t.process.parser.lock().contents(); contents.contains("zqsecondqz") }); let first = contents @@ -312,7 +318,7 @@ fn queued_writes_reach_the_child_in_order() { .find("zqsecondqz") .expect("second message never echoed"); assert!(first < second, "queued writes reordered: {contents:?}"); - t.terminate(); + t.process.terminate(); } /// Test writer that accepts writes within `limit` bytes, then fails. @@ -372,14 +378,14 @@ fn write_error_keeps_draining_the_pending_counter() { #[test] fn input_hints_track_child_modes() { let mut t = spawn(8, "sleep 5"); - assert_eq!(t.input_hints(), (false, false, false)); - t.parser.lock().process(b"\x1b[?1000h"); - assert_eq!(t.input_hints(), (true, false, false)); - t.parser.lock().process(b"\x1b[?1000l\x1b[?1049h"); - assert_eq!(t.input_hints(), (false, true, true)); - t.parser.lock().process(b"\x1b[?1007l"); - assert_eq!(t.input_hints(), (false, true, false)); - t.terminate(); + assert_eq!(t.process.input_hints(), (false, false, false)); + t.process.parser.lock().process(b"\x1b[?1000h"); + assert_eq!(t.process.input_hints(), (true, false, false)); + t.process.parser.lock().process(b"\x1b[?1000l\x1b[?1049h"); + assert_eq!(t.process.input_hints(), (false, true, true)); + t.process.parser.lock().process(b"\x1b[?1007l"); + assert_eq!(t.process.input_hints(), (false, true, false)); + t.process.terminate(); } /// A live reader may still hold bytes that have not reached the grid, so @@ -392,7 +398,7 @@ fn finalize_preview_waits_for_reader_eof() { // Reopen the reader gate: a thread outliving the child's exit stands in // for a reader still working through bytes the grid has not seen. let (release, parked) = channel::<()>(); - t.handle = Some(thread::spawn(move || { + t.process.handle = Some(thread::spawn(move || { let _ = parked.recv(); })); t.finalize_preview(); @@ -400,12 +406,12 @@ fn finalize_preview_waits_for_reader_eof() { !t.resolve_preview(Instant::now()).frozen, "finalization must wait for reader EOF" ); - t.parser.lock().process(b"late output\r\n"); + t.process.parser.lock().process(b"late output\r\n"); // Dropping the sender ends the stand-in: the reader reached EOF. drop(release); assert!( - wait_until(Duration::from_secs(60), || t.reader_done()), + wait_until(Duration::from_secs(60), || t.process.reader_done()), "the stand-in reader never stopped" ); t.finalize_preview(); @@ -420,11 +426,21 @@ fn finalize_preview_lands_an_open_sync_frame() { let mut t = spawn_sh(21, "printf '\\033[?2026htest result: ok\\n'"); wait_complete(&mut t); assert!( - !t.parser.lock().contents().contains("test result: ok"), + !t.process + .parser + .lock() + .contents() + .contains("test result: ok"), "premise: the unclosed frame still buffers the final output" ); t.finalize_preview(); - assert!(t.parser.lock().contents().contains("test result: ok")); + assert!( + t.process + .parser + .lock() + .contents() + .contains("test result: ok") + ); let p = t.resolve_preview(Instant::now()); assert_eq!((p.text.as_str(), p.frozen), ("test result: ok", true)); } @@ -479,7 +495,7 @@ fn finalize_preview_keeps_the_last_render_across_alt_teardown() { std::fs::write(&teardown, b"").unwrap(); assert!( wait_until(Duration::from_secs(60), || { - !t.parser.lock().alternate_screen() + !t.process.parser.lock().alternate_screen() }), "teardown never reached the grid" ); @@ -494,7 +510,7 @@ fn finalize_preview_keeps_the_last_render_across_alt_teardown() { wait_complete(&mut t); t.finalize_preview(); assert_eq!( - t.parser.lock().live_floor(), + t.process.parser.lock().live_floor(), "prelaunch junk", "premise: 1049l restored the pre-launch primary screen" ); @@ -615,5 +631,5 @@ fn probe_replies_reach_the_child_through_the_allowlist() { got.ends_with(b"R"), "CPR reply must follow the DA reply; got {got:?}" ); - t.terminate(); + t.process.terminate(); } diff --git a/src/transport.rs b/src/transport.rs index b6b442d..16132a8 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -105,7 +105,7 @@ impl ThreadTransport { true }); // Loop returned (Shutdown or client gone): `sup` drops here, and with - // it every Task (Task::drop → killpg), so no task outlives the core. + // it every Task's Process (Process::drop → killpg), so no task outlives the core. }); Self { wake_tx, @@ -118,7 +118,7 @@ impl ThreadTransport { fn stop(&mut self) { // Tell the core to kill tasks and exit, then wait for it. The join is what // guarantees the SIGKILLs have been sent before we return. The core - // clears its tasks (Task::drop → killpg) as `run_loop` returns. + // clears its tasks (Process::drop → killpg) as `run_loop` returns. let _ = self.wake_tx.send(Wake::Cmd(Command::Shutdown)); if let Some(h) = self.handle.take() { let _ = h.join(); From 31ecb9daa9994c44b85f6e6a12d7c180c4edc66c Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 18 Oct 2026 17:57:01 -0700 Subject: [PATCH 2/5] Restore stopped session entries as rows without a process --- src/app.rs | 19 +- src/app_readme_tests.rs | 44 ++--- src/app_tests.rs | 67 ++++++- src/protocol.rs | 30 +++- src/protocol_tests.rs | 44 ++++- src/session.rs | 86 +++++++-- src/supervisor.rs | 299 ++++++++++++++++++++---------- src/supervisor_capture_tests.rs | 219 ++++++++++++++++++---- src/supervisor_tests.rs | 309 ++++++++++++++++++++++++++++++-- src/task.rs | 86 ++++++--- src/task_tests.rs | 266 ++++++++++++++++++--------- src/ui.rs | 57 ++++-- tests/common/mod.rs | 2 +- tests/daemon_resume.rs | 72 ++++++-- 14 files changed, 1275 insertions(+), 325 deletions(-) diff --git a/src/app.rs b/src/app.rs index 2379d2b..2276786 100644 --- a/src/app.rs +++ b/src/app.rs @@ -364,12 +364,12 @@ fn step_down(sel: usize, len: usize) -> usize { fn state_section(v: &TaskView) -> (u8, &'static str) { if v.tagged { (0, "In use") + } else if v.lifecycle.finished() { + (3, "Completed") + } else if v.lifecycle == Lifecycle::Idle { + (2, "Idle") } else { - match v.lifecycle { - Lifecycle::Active => (1, "Running"), - Lifecycle::Idle => (2, "Idle"), - Lifecycle::Ok | Lifecycle::Failed => (3, "Completed"), - } + (1, "Running") } } @@ -379,7 +379,7 @@ fn state_section(v: &TaskView) -> (u8, &'static str) { fn row_rank(v: &TaskView) -> u8 { if v.tagged { 0 - } else if matches!(v.lifecycle, Lifecycle::Ok | Lifecycle::Failed) { + } else if v.lifecycle.finished() { 2 } else { 1 @@ -1327,7 +1327,7 @@ impl App { // next daemon tick, before inclusion in a snapshot. KeyCode::Char(']') => { if let Some(i) = self.selected_task() - && !matches!(self.views[i].lifecycle, Lifecycle::Ok | Lifecycle::Failed) + && !self.views[i].lifecycle.finished() { let v = &self.views[i]; let id = (!v.flagship).then_some(v.id); @@ -1961,7 +1961,7 @@ impl App { /// Restart the selected task only when the local snapshot marks it finished. fn rerun_selected(&mut self) { if let Some(i) = self.selected_task() - && matches!(self.views[i].lifecycle, Lifecycle::Ok | Lifecycle::Failed) + && self.views[i].lifecycle.finished() { self.transport.send(Command::Restart { id: self.views[i].id, @@ -1974,8 +1974,7 @@ impl App { return; }; let id = self.views[i].id; - let finished = matches!(self.views[i].lifecycle, Lifecycle::Ok | Lifecycle::Failed); - if finished { + if self.views[i].lifecycle.finished() { // Drop it, landing selection on the neighbor (not the top). The // removal reflects next tick, so pick the neighbor id from the // *current* order now and pin selection to it. diff --git a/src/app_readme_tests.rs b/src/app_readme_tests.rs index d51bc55..e2ebb00 100644 --- a/src/app_readme_tests.rs +++ b/src/app_readme_tests.rs @@ -144,7 +144,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { "Scope small fixes for dashboard and CLI", "claude:action-row", ), - started_ago: mins(2), + started_ago: Some(mins(2)), quiet_ago: Some(secs(3)), finished_ago: None, }, @@ -162,7 +162,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { "Fable 5.1 (high) · Inferring… · thinking with high effort", "claude:spinner", ), - started_ago: mins(5), + started_ago: Some(mins(5)), quiet_ago: Some(secs(8)), finished_ago: None, }, @@ -177,7 +177,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: Some("Grok Language".to_string()), lifecycle: Lifecycle::Active, preview: working("Grok 4.7 (xhigh) · Responding…", "grok:spinner"), - started_ago: mins(12), + started_ago: Some(mins(12)), quiet_ago: Some(secs(4)), finished_ago: None, }, @@ -192,7 +192,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: Some("Codex Language".to_string()), lifecycle: Lifecycle::Active, preview: anchor(CODEX_LANGUAGE, "codex:working"), - started_ago: mins(18), + started_ago: Some(mins(18)), quiet_ago: Some(secs(2)), finished_ago: None, }, @@ -207,7 +207,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: Some("Codex Review".to_string()), lifecycle: Lifecycle::Active, preview: anchor(CODEX_REVIEW, "codex:working"), - started_ago: mins(24), + started_ago: Some(mins(24)), quiet_ago: Some(secs(6)), finished_ago: None, }, @@ -222,7 +222,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: None, lifecycle: Lifecycle::Ok, preview: frozen(FLEETCOM_TESTS), - started_ago: mins(2), + started_ago: Some(mins(2)), quiet_ago: None, finished_ago: Some(secs(12)), }, @@ -237,7 +237,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: None, lifecycle: Lifecycle::Failed, preview: frozen(FLEETCOM_CLIPPY), - started_ago: mins(5), + started_ago: Some(mins(5)), quiet_ago: None, finished_ago: Some(mins(3)), }, @@ -252,7 +252,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: Some("claude agents".to_string()), lifecycle: Lifecycle::Active, preview: title("2 awaiting input · claude agents"), - started_ago: mins(63), + started_ago: Some(mins(63)), quiet_ago: Some(secs(9)), finished_ago: None, }, @@ -267,7 +267,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: Some("Zellij".to_string()), lifecycle: Lifecycle::Active, preview: title("Desktop ¦ Utility"), - started_ago: mins(126), + started_ago: Some(mins(126)), quiet_ago: Some(secs(4)), finished_ago: None, }, @@ -282,7 +282,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: None, lifecycle: Lifecycle::Idle, preview: floor(">>>"), - started_ago: mins(48), + started_ago: Some(mins(48)), quiet_ago: Some(mins(41)), finished_ago: None, }, @@ -297,7 +297,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: None, lifecycle: Lifecycle::Ok, preview: frozen("Already up-to-date."), - started_ago: mins(14), + started_ago: Some(mins(14)), quiet_ago: None, finished_ago: Some(mins(13)), }, @@ -312,7 +312,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { name: Some("Game Infra Review".to_string()), lifecycle: Lifecycle::Active, preview: title("Turret Game Codebase Organization and Ex… - grok"), - started_ago: mins(8), + started_ago: Some(mins(8)), quiet_ago: Some(secs(5)), finished_ago: None, }, @@ -329,7 +329,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { // omp has no anchor tier: the primary-screen title tier renders // the conversation label omp announces, with no model prefix. preview: title("Tuning missile damage falloff"), - started_ago: mins(33), + started_ago: Some(mins(33)), quiet_ago: Some(secs(7)), finished_ago: None, }, @@ -346,7 +346,7 @@ fn live_fleet(dirs: &Dirs) -> Vec { // At its prompt: the primary-screen title tier renders the // conversation label omp announces as `π >