fix(shell): a silent running command stays running and stoppable

Refs #6931. The issue's headline case was a running recursive search shown
as "run running · stale · no output 2m 07s". After 60 s without output,
`job_snapshot` flagged a live, non-PTY job `stale`. The inspector, the
/jobs list, the work dock and the Runtime API all read `stale` as "process
gone":
- the Stop action disappeared (shell_job_routing `live = Running && !stale`,
  work_surface `shell_inspector_body`);
- the detail said "stale after restart; the process is no longer running";
- "(no output yet)" was suppressed;
- the API refused resize as "no longer running" and reported no tty.

- A live process is never `stale`. `job_snapshot` sets `stale: false`;
  only `remember_stale_job` (a restart snapshot, status Killed) sets it,
  which is the "known-stale" job docs/RUNTIME_API.md describes. The 60 s
  threshold becomes `QUIET_NO_OUTPUT_AFTER` and only drives presentation.
- The exec card reads "running · no output for 2m 07s" instead of
  "running · stale · no output …". Silence is not shown as hung, which
  acceptance 2 requires.
- `/jobs poll` and `/jobs stdin` no longer consume the model's unread
  output. Both called `poll_delta_for_session`, which advances the model's
  stdout cursor (acceptance 4).
  - `poll` opens the inspector from `inspect_job_for_session`.
  - `/jobs wait` uses the new `wait_job_for_session`, which waits through
    the same `wait_while_running` loop `get_output_delta` uses, then
    inspects.
  - `stdin` shows the job's one-line status and tail.
  - `poll_delta_for_session` has no callers left and is removed.
- Alt+V with no selection prefers the newest running shell only when it is
  visible in the transcript viewport. It no longer takes over the completed
  card the user is looking at (acceptance 1). A click or selection still
  targets that exact card.
- shell_job_routing.rs records its known limitations. The inspector is a
  snapshot with no follow mode; `/jobs poll` re-reads it. Named PTY
  sessions live in the terminal dock (#6912), not /jobs.
- Tests updated where this change made them wrong:
  - running_job_snapshot_keeps_silent_live_job_live asserts a silent live
    job is not stale and still exposes its quiet time;
  - the stale-running fixture in ui/tests.rs is `stale: false`, and its
    header assertion uses the new label;
  - QUIET_NO_OUTPUT_AFTER replaces the old constant name.

Still open in #6931:
- a follow-mode or auto-refreshing inspector;
- listing delegated agents' shells in the agent detail;
- the PTY journeys on the installed candidate at normal and narrow widths.

Evidence: cargo fmt --all -- --check is clean. The compile and the
tools::shell / tui tests are pending a local test build and hosted CI on
codewhale-hq/Codewhale#6957.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01BsXVvReYca2UxzimQtj7Mj

crates/tui/src/tools/shell.rs

@@ -1003,7 +1003,7 @@ fn shared_output_pipe() -> io::Result<(File, File, File)> {
 }
 
 const SYNC_READER_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
-const STALE_NO_OUTPUT_AFTER: Duration = Duration::from_secs(60);
+pub(crate) const QUIET_NO_OUTPUT_AFTER: Duration = Duration::from_secs(60);
 
 /// Grace between SIGTERM and SIGKILL on the shell kill path (timeout,
 /// cancel, drop). Bounded so a SIGTERM-ignoring command is force-killed
@@ -1717,13 +1717,6 @@ impl BackgroundShell {
             .unwrap_or((0, String::new()));
         let elapsed_since_output_ms = (self.status == ShellStatus::Running)
             .then(|| u64::try_from(self.last_output_at.elapsed().as_millis()).unwrap_or(u64::MAX));
-        // A live PTY can wait indefinitely for input. Silence does not mean it
-        // lost its process owner; marking it stale hides its transport from API
-        // clients and incorrectly prevents reconnect/resize after one minute.
-        let stale = self.terminal_size.is_none()
-            && elapsed_since_output_ms.is_some_and(|elapsed| {
-                elapsed >= u64::try_from(STALE_NO_OUTPUT_AFTER.as_millis()).unwrap_or(u64::MAX)
-            });
         ShellJobSnapshot {
             id: self.id.clone(),
             job_id: self.id.clone(),
@@ -1739,7 +1732,7 @@ impl BackgroundShell {
             stdout_len,
             stderr_len,
             stdin_available: self.stdin.is_some() && self.status == ShellStatus::Running,
-            stale,
+            stale: false,
             elapsed_since_output_ms,
             linked_task_id: self.linked_task_id.clone(),
             owner_agent_id: self
@@ -3127,20 +3120,7 @@ impl ShellManager {
             .processes
             .get_mut(task_id)
             .ok_or_else(|| anyhow!("Task {task_id} not found"))?;
-
-        if wait && shell.status == ShellStatus::Running {
-            let timeout = Duration::from_millis(timeout_ms.clamp(1000, 600_000));
-            let deadline = Instant::now() + timeout;
-
-            while shell.status == ShellStatus::Running && Instant::now() < deadline {
-                if shell.poll() {
-                    break;
-                }
-                std::thread::sleep(Duration::from_millis(100));
-            }
-        } else {
-            shell.poll();
-        }
+        wait_while_running(shell, wait, timeout_ms);
 
         let (
             stdout_delta,
@@ -3351,17 +3331,6 @@ impl ShellManager {
         self.get_output_delta(task_id, wait, timeout_ms)
     }
 
-    pub fn poll_delta_for_session(
-        &mut self,
-        active_session_id: &str,
-        task_id: &str,
-        wait: bool,
-        timeout_ms: u64,
-    ) -> Result<ShellDeltaResult> {
-        self.require_session_owner(task_id, active_session_id)?;
-        self.get_output_delta(task_id, wait, timeout_ms)
-    }
-
     fn get_output_delta_for_session(
         &mut self,
         active_session_id: &str,
@@ -3494,6 +3463,19 @@ impl ShellManager {
         self.inspect_job(task_id)
     }
 
+    pub fn wait_job_for_session(
+        &mut self,
+        active_session_id: &str,
+        task_id: &str,
+        timeout_ms: u64,
+    ) -> Result<ShellJobDetail> {
+        self.require_session_owner(task_id, active_session_id)?;
+        if let Some(shell) = self.processes.get_mut(task_id) {
+            wait_while_running(shell, true, timeout_ms);
+        }
+        self.inspect_job(task_id)
+    }
+
     /// List all live and known-stale background shell jobs for the TUI.
     pub fn list_jobs(&mut self) -> Vec<ShellJobSnapshot> {
         for shell in self.processes.values_mut() {
@@ -3788,6 +3770,22 @@ impl ShellManager {
     }
 }
 
+fn wait_while_running(shell: &mut BackgroundShell, wait: bool, timeout_ms: u64) {
+    if wait && shell.status == ShellStatus::Running {
+        let timeout = Duration::from_millis(timeout_ms.clamp(1000, 600_000));
+        let deadline = Instant::now() + timeout;
+
+        while shell.status == ShellStatus::Running && Instant::now() < deadline {
+            if shell.poll() {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(100));
+        }
+    } else {
+        shell.poll();
+    }
+}
+
 fn job_status_rank(status: &ShellStatus, stale: bool) -> u8 {
     if stale {
         return 4;

crates/tui/src/tools/shell/tests.rs

@@ -2652,7 +2652,7 @@ fn test_job_list_poll_cancel_and_stale_snapshot() {
 }
 
 #[test]
-fn running_job_snapshot_marks_no_output_stale_after_threshold() {
+fn running_job_snapshot_keeps_silent_live_job_live() {
     let tmp = tempdir().expect("tempdir");
     let mut manager = ShellManager::new(tmp.path().to_path_buf());
 
@@ -2662,7 +2662,7 @@ fn running_job_snapshot_marks_no_output_stale_after_threshold() {
 
     {
         let shell = manager.processes.get_mut(&task_id).expect("live shell");
-        shell.last_output_at = Instant::now() - STALE_NO_OUTPUT_AFTER - Duration::from_millis(1);
+        shell.last_output_at = Instant::now() - QUIET_NO_OUTPUT_AFTER - Duration::from_millis(1);
     }
 
     let job = manager
@@ -2672,10 +2672,10 @@ fn running_job_snapshot_marks_no_output_stale_after_threshold() {
         .expect("running job");
 
     assert_eq!(job.status, ShellStatus::Running);
-    assert!(job.stale, "silent running job should be marked stale");
+    assert!(!job.stale, "silence alone must not mark a live job stale");
     assert!(
         job.elapsed_since_output_ms
-            .is_some_and(|elapsed| elapsed >= STALE_NO_OUTPUT_AFTER.as_millis() as u64),
+            .is_some_and(|elapsed| elapsed >= QUIET_NO_OUTPUT_AFTER.as_millis() as u64),
         "elapsed no-output time should be exposed: {job:?}"
     );
 }
@@ -5773,14 +5773,14 @@ fn pty_resize_updates_live_terminal_and_rejects_finished_or_pipe_jobs() {
     // Reopening a client after an idle minute must retain this live PTY's
     // identity and resize eligibility; output silence is normal at a prompt.
     manager.processes.get_mut(&id).unwrap().last_output_at =
-        Instant::now() - STALE_NO_OUTPUT_AFTER - Duration::from_millis(1);
+        Instant::now() - QUIET_NO_OUTPUT_AFTER - Duration::from_millis(1);
     let idle = manager.inspect_job(&id).unwrap().snapshot;
     assert_eq!(idle.status, ShellStatus::Running);
     assert!(idle.stdin_available);
     assert!(!idle.stale, "an idle live PTY must remain reconnectable");
     assert!(
         idle.elapsed_since_output_ms
-            .is_some_and(|ms| ms >= STALE_NO_OUTPUT_AFTER.as_millis() as u64)
+            .is_some_and(|ms| ms >= QUIET_NO_OUTPUT_AFTER.as_millis() as u64)
     );
     let size = PtyDimensions {
         rows: 37,

crates/tui/src/tui/history.rs

@@ -2819,7 +2819,7 @@ pub(crate) fn running_status_label_with_elapsed(elapsed_secs: u64) -> String {
 
 pub(crate) fn stale_shell_status_label(elapsed_since_output_ms: u64) -> String {
     format!(
-        "running · stale · no output {}",
+        "running · no output for {}",
         codewhale_command_contract::elapsed::format_elapsed_ms(elapsed_since_output_ms)
     )
 }

crates/tui/src/tui/shell_job_routing.rs

@@ -1,4 +1,9 @@
 //! Background shell job-center helpers for slash commands and pagers.
+//!
+//! Known limitations: the inspector is a snapshot (`/jobs poll` re-reads it,
+//! without consuming the model's output cursor); there is no follow mode.
+//! Named PTY sessions (`tools/terminal_session.rs`) are not listed here or in
+//! `/jobs`; they live in the terminal dock (#6912).
 
 use crate::tools::shell::{ShellJobDetail, ShellJobSnapshot, ShellResult, ShellStatus};
 use crate::tui::app::App;
@@ -139,13 +144,23 @@ pub(crate) fn shell_job_for_cell(app: &App, cell_index: usize) -> Option<ShellJo
     manager.inspect_job_for_session(session, &job_id).ok()
 }
 
-pub(crate) fn newest_running_shell_cell(app: &App) -> Option<usize> {
-    (0..app.virtual_cell_count()).rev().find(|&index| {
-        matches!(
-            app.cell_at_virtual_index(index),
-            Some(HistoryCell::Tool(ToolCell::Exec(exec))) if exec.status == ToolStatus::Running
-        )
-    })
+pub(crate) fn newest_visible_running_shell_cell(app: &App) -> Option<usize> {
+    let meta = app.viewport.transcript_cache.line_meta();
+    let start = app.viewport.last_transcript_top.min(meta.len());
+    let end = start
+        .saturating_add(app.viewport.last_transcript_visible.max(1))
+        .min(meta.len());
+    meta[start..end]
+        .iter()
+        .rev()
+        .filter_map(|line| line.cell_line())
+        .map(|(rendered, _)| app.original_cell_index_for_rendered(rendered))
+        .find(|&index| {
+            matches!(
+                app.cell_at_virtual_index(index),
+                Some(HistoryCell::Tool(ToolCell::Exec(exec))) if exec.status == ToolStatus::Running
+            ) && shell_job_for_cell(app, index).is_some()
+        })
 }
 
 pub(crate) fn open_shell_job_pager(app: &mut App, detail: &ShellJobDetail) {

crates/tui/src/tui/ui/activity_detail.rs

@@ -283,8 +283,7 @@ pub(super) fn open_tool_details_pager(app: &mut App) -> bool {
     {
         detail_target_cell_index(app)
     } else {
-        crate::tui::shell_job_routing::newest_running_shell_cell(app)
-            .filter(|&idx| crate::tui::shell_job_routing::shell_job_for_cell(app, idx).is_some())
+        crate::tui::shell_job_routing::newest_visible_running_shell_cell(app)
             .or_else(|| detail_target_cell_index(app))
     };
 

crates/tui/src/tui/ui/handlers.rs

@@ -1245,29 +1245,30 @@ pub(crate) fn handle_shell_job_action(app: &mut App, action: crate::tui::app::Sh
             }
         }
         crate::tui::app::ShellJobAction::Poll { id, wait } => {
-            match manager.poll_delta_for_session(
-                &active_session_id,
-                &id,
-                wait,
-                if wait { 5_000 } else { 1_000 },
-            ) {
-                Ok(delta) => add_shell_job_message(app, format_shell_poll(&delta.result)),
+            let detail = if wait {
+                manager.wait_job_for_session(&active_session_id, &id, 5_000)
+            } else {
+                manager.inspect_job_for_session(&active_session_id, &id)
+            };
+            match detail {
+                Ok(detail) => open_shell_job_pager(app, &detail),
                 Err(err) => add_shell_job_message(app, format!("Command poll failed: {err}")),
             }
         }
         crate::tui::app::ShellJobAction::SendStdin { id, input, close } => {
             match manager.write_stdin_for_session(&active_session_id, &id, &input, close) {
-                Ok(()) => {
-                    match manager.poll_delta_for_session(&active_session_id, &id, false, 1_000) {
-                        Ok(delta) => add_shell_job_message(app, format_shell_poll(&delta.result)),
-                        Err(err) => {
-                            add_shell_job_message(
-                                app,
-                                format!("Command input sent; poll failed: {err}"),
-                            );
-                        }
+                Ok(()) => match manager.inspect_job_for_session(&active_session_id, &id) {
+                    Ok(detail) => add_shell_job_message(
+                        app,
+                        format_shell_job_list(std::slice::from_ref(&detail.snapshot)),
+                    ),
+                    Err(err) => {
+                        add_shell_job_message(
+                            app,
+                            format!("Command input sent; poll failed: {err}"),
+                        );
                     }
-                }
+                },
                 Err(err) => add_shell_job_message(app, format!("Command input failed: {err}")),
             }
         }

crates/tui/src/tui/ui/task_projection.rs

@@ -625,8 +625,11 @@ pub(super) fn shell_exec_live_update(
     let next_status = shell_job_tool_status(&job.status);
     let next_live = shell_job_live_output(job).or_else(|| exec.live_output.clone());
     let finalized = !matches!(job.status, ShellStatus::Running);
-    let stale_elapsed_since_output_ms = if matches!(job.status, ShellStatus::Running) && job.stale {
-        Some(job.elapsed_since_output_ms.unwrap_or(0))
+    let quiet_after_ms =
+        u64::try_from(crate::tools::shell::QUIET_NO_OUTPUT_AFTER.as_millis()).unwrap_or(u64::MAX);
+    let stale_elapsed_since_output_ms = if matches!(job.status, ShellStatus::Running) {
+        job.elapsed_since_output_ms
+            .filter(|elapsed| *elapsed >= quiet_after_ms)
     } else {
         None
     };

crates/tui/src/tui/ui/tests.rs

@@ -10185,7 +10185,7 @@ fn shell_live_output_update_marks_stale_running_job_static() {
             stdout_len: 14,
             stderr_len: 0,
             stdin_available: false,
-            stale: true,
+            stale: false,
             elapsed_since_output_ms: Some(61_000),
             linked_task_id: None,
             owner_agent_id: None,
@@ -10228,7 +10228,7 @@ fn shell_live_output_update_marks_stale_running_job_static() {
         .iter()
         .map(|span| span.content.as_ref())
         .collect::<String>();
-    assert!(header_text.contains("running · stale · no output 1m 01s"));
+    assert!(header_text.contains("running · no output for 1m 01s"));
     assert!(!header_text.contains("running ("));
 }