From f6d65f255ca5ec1dc0bfccfada348d62f440304e Mon Sep 17 00:00:00 2001 From: Bastien Gallay Date: Sat, 25 Jul 2026 21:18:17 +0200 Subject: [PATCH 1/3] refactor(app): give the bridge its own serving seam in the shell Answering an external request is not one thing: a read answers off owned state, a mutation needs `&mut` and the effect executor. That fork sat inline in `update`, which already carries a too-many-lines allow, and the wait rung adds a third shape (a reply that lands later, not in this update). Move it to `shell::serve`, unchanged, so there is one auditable place where an external caller meets `core::App`. --- crates/app/src/shell.rs | 17 ++--------------- crates/app/src/shell/serve.rs | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 15 deletions(-) create mode 100644 crates/app/src/shell/serve.rs diff --git a/crates/app/src/shell.rs b/crates/app/src/shell.rs index f07eb6c..1a7c7ff 100644 --- a/crates/app/src/shell.rs +++ b/crates/app/src/shell.rs @@ -43,6 +43,7 @@ mod launch; mod orchestrate; mod record; mod routing; +mod serve; mod session_ops; mod streams; mod terminal; @@ -1158,21 +1159,7 @@ impl Shell { } Message::RecordFrameTick(now) => self.on_record_frame_tick(now), Message::RecordFrame(screenshot) => self.record.on_frame(screenshot), - Message::Bridge { request, reply } => match request { - // Actions mutate, so they can't answer off a `&App`: apply them - // and perform the effects here, where the shell owns both. - bridge::Request::Act(action) => { - let (outcome, task) = self.perform_action(action); - reply.answer(bridge::Reply::Acted(outcome)); - task - } - // The two read requests answer straight from owned state. - read => { - let inputs = self.snapshot_inputs(&read); - reply.answer(bridge::respond(&self.core, &read, &inputs)); - Task::none() - } - }, + Message::Bridge { request, reply } => self.serve(request, reply), } } diff --git a/crates/app/src/shell/serve.rs b/crates/app/src/shell/serve.rs new file mode 100644 index 0000000..17289db --- /dev/null +++ b/crates/app/src/shell/serve.rs @@ -0,0 +1,33 @@ +//! The bridge-serving seam: how the shell answers one external [`Request`]. +//! +//! Split from the shell's state machine because answering is not one thing. A +//! read answers off the state the shell already owns; a mutation needs `&mut` +//! and the effect executor, so it cannot go through the read-only responder. +//! Keeping that fork here — rather than inline in `update` — leaves one +//! auditable place where an external caller meets `core::App`. + +use iced::Task; + +use super::bridge::{self, ReplyPort, Request}; +use super::{Message, Shell}; + +impl Shell { + /// Answer one bridge request and return any async follow-up it needs. + pub(super) fn serve(&mut self, request: Request, reply: ReplyPort) -> Task { + match request { + // Actions mutate, so they can't answer off a `&App`: apply them and + // perform the effects here, where the shell owns both. + Request::Act(action) => { + let (outcome, task) = self.perform_action(action); + reply.answer(bridge::Reply::Acted(outcome)); + task + } + // The read requests answer straight from owned state. + read => { + let inputs = self.snapshot_inputs(&read); + reply.answer(bridge::respond(&self.core, &read, &inputs)); + Task::none() + } + } + } +} From 3ecd0c73c6a50742ce3586f60f74f1fee988ebdf Mon Sep 17 00:00:00 2001 From: Bastien Gallay Date: Sat, 25 Jul 2026 21:29:08 +0200 Subject: [PATCH 2/3] feat(app): wait_for_status + read_terminal on the live bridge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Orchestration could act and perceive, but not synchronise: `run_in_session` writes and returns, so an agent had no way to know when the command finished short of polling `snapshot` and racing the transition it was watching for. `wait_for_status` blocks until a session's OSC-derived activity reaches one of the asked-for statuses, and `read_terminal` returns one pane's visible text — the deep read the light `snapshot` deliberately leaves out. The wait is the first bridge request whose reply lands in a *later* `update`. `ReplyPort` was already built for that (a take-once slot behind an `Arc`), so the shell parks it in a waiter list and settles it from the status change itself — nothing polled, no transition raced. Three rules earn their own tests: - a session already sitting on a target answers at once, or a caller asking "tell me when it is idle" about an idle session would wait out its bound having missed the transition it asked about; - an exit settles every waiter whatever it asked for — a crash records `Exited` without emitting a status change, and a dead session will never reach the target anyway; - a caller that gave up is swept off the list rather than held for a session that may never move again. Timing out is not a failure: the tool answers `{ status, timed_out: true }` with the session's current status, so an agent can choose between waiting again and giving up. The bound is the caller's `timeout_ms` (default 30 s), capped at 5 minutes so no parameterisation can park a caller indefinitely (Q7). `read_terminal` keeps three outcomes distinct that a scoped `snapshot` collapses into one absent key: unknown handle (invalid_params), live but undrawn (`rendered: false`, retry), and text. --- ROADMAP.md | 19 +- crates/app/src/mcp/handler.rs | 457 +++++++++++++++++++++++++++- crates/app/src/shell.rs | 297 ++++++++++++++++++ crates/app/src/shell/bridge.rs | 116 ++++++- crates/app/src/shell/orchestrate.rs | 2 +- crates/app/src/shell/serve.rs | 102 ++++++- 6 files changed, 982 insertions(+), 11 deletions(-) diff --git a/ROADMAP.md b/ROADMAP.md index db0a65f..08e2827 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -374,8 +374,23 @@ geometry) with drag-resize split out to #55 (blocked-by #54; feature-torture click-to-focus (`FocusPane`) only reaches the active one and a silent no-op there would let a close destroy the wrong terminal. Depends on #193; with #212 (perception) this closes the act→observe loop - - [ ] `F-mcp-terminal-sync` (#195) — `wait_for_status` (OSC) + `read_terminal`; - depends on #193 + - [x] `F-mcp-terminal-sync` (#195) — **the wait rung**, closing + act→**wait**→observe. `wait_for_status` blocks until a session's OSC-derived + activity reaches one of the asked-for statuses (default idle-or-attention), + and `read_terminal` returns one pane's visible text — the deep read the + light `snapshot` leaves out. The wait is the first bridge request whose + reply lands in a *later* `update`: the shell parks the reply port in a + waiter list and settles it from the status change, so nothing is polled and + no transition is raced (the feature-torture report cut `wait_for_text` for + exactly that). Three rules earn their tests: a session already on the target + answers at once (never wait for a transition already past); an exit settles + every waiter whatever was asked for (a crash emits no status change, and a + dead session will not reach the target); a caller that gave up is swept off + the list. Timing out is not an error — the tool reports + `{ status, timed_out: true }` with the session's current status, so an agent + can choose between waiting again and giving up. Bounds are the caller's: + `timeout_ms` (default 30 s) capped at 5 min (Q7). Depends on #193; unblocks + #196 - [ ] `F-mcp-agent-loop` (#196) — `type_into_terminal` + prompt→wait→read, opt-in; depends on #195; product-scope question open (may be cut) - [ ] `F-mcp-screenshot` (#215) — expose the window PNG as an MCP tool (async diff --git a/crates/app/src/mcp/handler.rs b/crates/app/src/mcp/handler.rs index e3e84e8..db101a3 100644 --- a/crates/app/src/mcp/handler.rs +++ b/crates/app/src/mcp/handler.rs @@ -17,10 +17,13 @@ use rmcp::{ tool, tool_handler, tool_router, }; use serde::{Deserialize, Serialize}; +use termherd_core::snapshot::DEFAULT_TEXT_LINES; use termherd_core::workspace::SplitDir; -use termherd_core::{Section, SnapshotFilter, TerminalScope}; +use termherd_core::{Section, SessionStatus, SnapshotFilter, TerminalScope}; -use crate::shell::bridge::{Action, BridgeHandle, Reply, Request, SessionInfo, SessionKind}; +use crate::shell::bridge::{ + Action, BridgeHandle, CallError, Reply, Request, SessionInfo, SessionKind, +}; use crate::snapshot_dto::{SnapshotDto, status_str}; /// How long a tool waits for the shell to answer before failing the caller. @@ -28,6 +31,14 @@ use crate::snapshot_dto::{SnapshotDto, status_str}; /// a wedged shell surfaces as a tool error, never a hang (Q7). const CALL_TIMEOUT: Duration = Duration::from_secs(5); +/// Default bound for `wait_for_status` when the caller names none. Long enough +/// for a real command, short enough that a forgotten wait reports back. +const DEFAULT_WAIT_MS: u64 = 30_000; + +/// Hard cap on a wait, whatever the caller asks for — the outer Q7 guarantee +/// that no bridge call can park a caller indefinitely. +const MAX_WAIT_MS: u64 = 300_000; + /// The live-bridge MCP server handler. Holds only the bridge into the shell — /// cloned fresh per session by the transport's service factory. #[derive(Clone)] @@ -233,9 +244,122 @@ impl TermherdMcp { }) .await } + + /// Block until a session's activity reaches one of the given statuses — the + /// synchronisation half of `act → wait → observe`. + #[tool( + name = "wait_for_status", + description = "Wait until a session's activity reaches one of `statuses` \ + (default: idle or attention), then report it. Pair it \ + with `run_in_session`, which returns immediately: run, \ + wait, then `read_terminal`. Args: `session` (handle), \ + `statuses` (any of \"starting\", \"busy\", \"idle\", \ + \"attention\", \"exited\"), `timeout_ms` (default 30000, \ + capped at 300000). A session that exits settles the wait \ + whatever you asked for — it can no longer reach it. \ + Returns `{ status, timed_out }`: on timeout the status is \ + the session's current one, not an error." + )] + async fn wait_for_status( + &self, + Parameters(args): Parameters, + ) -> Result { + let session = parse_handle(&args.session)?; + let targets = parse_statuses(args.statuses)?; + let timeout = wait_timeout(args.timeout_ms); + + match self + .bridge + .call(Request::WaitForStatus { session, targets }, timeout) + .await + { + Ok(Reply::Waited(outcome)) => { + if let Some(reason) = outcome.error { + return Err(ErrorData::invalid_params(reason, None)); + } + Ok(CallToolResult::structured(serde_json::json!({ + "status": outcome.status.map(status_str), + "timed_out": false, + }))) + } + Ok(_) => Err(ErrorData::internal_error( + "bridge answered the wrong reply kind", + None, + )), + // The wait outlived its bound. Report *where the session got to* + // rather than a bare failure: an agent that waited 30s needs the + // current status to decide whether to keep waiting or give up. + Err(CallError::Timeout(_)) => { + let status = self.current_status(session).await?; + Ok(CallToolResult::structured(serde_json::json!({ + "status": status, + "timed_out": true, + }))) + } + Err(error) => Err(ErrorData::internal_error(error.to_string(), None)), + } + } + + /// The visible text of one session's terminal — the deep read `snapshot` + /// deliberately leaves out. + #[tool( + name = "read_terminal", + description = "Read the visible text of one session's terminal. Args: \ + `session` (handle), `lines` (trailing lines to keep, \ + default 40). Returns `{ text, rendered }`; `rendered` is \ + false when the session is live but its screen has not \ + been drawn yet — retry rather than giving up on the \ + handle." + )] + async fn read_terminal( + &self, + Parameters(args): Parameters, + ) -> Result { + let session = parse_handle(&args.session)?; + let lines = args.lines.unwrap_or(DEFAULT_TEXT_LINES); + let reply = self + .bridge + .call(Request::ReadTerminal { session, lines }, CALL_TIMEOUT) + .await + .map_err(|error| ErrorData::internal_error(error.to_string(), None))?; + let Reply::Terminal(read) = reply else { + return Err(ErrorData::internal_error( + "bridge answered the wrong reply kind", + None, + )); + }; + if let Some(reason) = read.error { + return Err(ErrorData::invalid_params(reason, None)); + } + Ok(CallToolResult::structured(serde_json::json!({ + "text": read.text.as_deref().unwrap_or_default(), + "rendered": read.text.is_some(), + }))) + } } impl TermherdMcp { + /// A session's current activity, read back after a wait timed out. A handle + /// that no longer resolves (the session died meanwhile) reports `None` + /// rather than failing — the caller asked about waiting, not existence. + async fn current_status(&self, session: u64) -> Result, ErrorData> { + let reply = self + .bridge + .call(Request::ListSessions, CALL_TIMEOUT) + .await + .map_err(|error| ErrorData::internal_error(error.to_string(), None))?; + let Reply::Sessions(sessions) = reply else { + return Err(ErrorData::internal_error( + "bridge answered the wrong reply kind", + None, + )); + }; + Ok(sessions + .iter() + .find(|info| info.handle == session.to_string()) + .map(|info| status_str(info.status))) + } + /// Send an [`Action`] over the bridge and shape its [`ActionOutcome`] into a /// tool result: a rejection (unknown handle / out-of-range tab) becomes an /// `invalid_params` error, an applied action a `{ focused_handle }` object. @@ -444,6 +568,68 @@ fn parse_optional_handle(handle: Option) -> Result, ErrorDat handle.map(|h| parse_handle(&h)).transpose() } +/// Arguments for `wait_for_status`. +#[derive(Debug, Default, Deserialize, JsonSchema)] +#[schemars(crate = "rmcp::schemars")] +struct WaitArgs { + /// The session handle to watch. + session: String, + /// Statuses that settle the wait. Omit for idle-or-attention. + #[serde(default)] + statuses: Option>, + /// How long to wait before reporting the current status instead. + #[serde(default)] + timeout_ms: Option, +} + +/// Arguments for `read_terminal`. +#[derive(Debug, Default, Deserialize, JsonSchema)] +#[schemars(crate = "rmcp::schemars")] +struct ReadArgs { + /// The session handle to read. + session: String, + /// Trailing lines to keep. + #[serde(default)] + lines: Option, +} + +/// Map the named statuses onto the core enum. An empty or omitted list means +/// idle-or-attention — the two a caller waiting on a command actually wants. +/// An unknown name is rejected rather than dropped: silently waiting on fewer +/// statuses than asked is exactly the silent-catch trap to avoid. +fn parse_statuses(names: Option>) -> Result, ErrorData> { + let Some(names) = names.filter(|names| !names.is_empty()) else { + return Ok(vec![SessionStatus::Idle, SessionStatus::Attention]); + }; + names + .iter() + .map(|name| { + status_from_str(name) + .ok_or_else(|| ErrorData::invalid_params(format!("unknown status {name:?}"), None)) + }) + .collect() +} + +/// The caller's bound for a wait, defaulted and capped. The cap is the outer +/// guarantee behind Q7: no bridge call, however it was parameterised, can park +/// a caller indefinitely. +fn wait_timeout(timeout_ms: Option) -> Duration { + Duration::from_millis(timeout_ms.unwrap_or(DEFAULT_WAIT_MS).min(MAX_WAIT_MS)) +} + +/// The inverse of [`crate::snapshot_dto::status_str`] — the external word back +/// to the core enum. +fn status_from_str(name: &str) -> Option { + match name { + "starting" => Some(SessionStatus::Starting), + "busy" => Some(SessionStatus::Busy), + "idle" => Some(SessionStatus::Idle), + "attention" => Some(SessionStatus::Attention), + "exited" => Some(SessionStatus::Exited), + _ => None, + } +} + #[cfg(test)] mod tests { use super::*; @@ -842,4 +1028,271 @@ mod tests { .expect_err("a closed bridge is a tool error"); assert!(error.message.contains("closed"), "got: {}", error.message); } + + // --- wait_for_status / read_terminal --- + + use crate::shell::bridge::{TerminalRead, WaitOutcome, spawn_test_shell_seq}; + + /// Args for a wait on `session`, otherwise defaulted. + fn wait_args(session: &str) -> WaitArgs { + WaitArgs { + session: session.to_owned(), + ..WaitArgs::default() + } + } + + /// Run `wait_for_status` against a shell answering `reply`, returning the + /// request the tool built. + async fn wait_request(args: WaitArgs, reply: Reply) -> Request { + let (handle, requests) = channel(); + let shell = spawn_test_shell(requests, reply); + let _ = TermherdMcp::new(handle) + .wait_for_status(Parameters(args)) + .await; + shell.await.expect("shell task") + } + + /// A settled wait, as the shell would answer it. + fn settled(status: SessionStatus) -> Reply { + Reply::Waited(WaitOutcome { + status: Some(status), + error: None, + }) + } + + #[tokio::test] + async fn wait_for_status_defaults_to_idle_or_attention() { + // The two a caller waiting on a command actually cares about: finished, + // or stopped to ask something. + let request = wait_request(wait_args("7"), settled(SessionStatus::Idle)).await; + assert_eq!( + request, + Request::WaitForStatus { + session: 7, + targets: vec![SessionStatus::Idle, SessionStatus::Attention], + } + ); + } + + #[tokio::test] + async fn wait_for_status_maps_the_named_statuses() { + let args = WaitArgs { + session: "7".into(), + statuses: Some(vec!["busy".into(), "exited".into()]), + ..WaitArgs::default() + }; + let request = wait_request(args, settled(SessionStatus::Busy)).await; + assert_eq!( + request, + Request::WaitForStatus { + session: 7, + targets: vec![SessionStatus::Busy, SessionStatus::Exited], + } + ); + } + + #[tokio::test] + async fn wait_for_status_rejects_an_unknown_status_name() { + // Dropping the unknown name and waiting on the rest would be the silent + // catch this codebase exists to avoid. + let (handle, requests) = channel(); + let args = WaitArgs { + session: "7".into(), + statuses: Some(vec!["idle".into(), "finished".into()]), + ..WaitArgs::default() + }; + let error = TermherdMcp::new(handle) + .wait_for_status(Parameters(args)) + .await + .expect_err("an unknown status is rejected"); + assert!(error.message.contains("finished"), "got: {}", error.message); + drop(requests); + } + + #[tokio::test] + async fn wait_for_status_reports_the_settled_status() { + let (handle, requests) = channel(); + let _shell = spawn_test_shell(requests, settled(SessionStatus::Attention)); + let result = TermherdMcp::new(handle) + .wait_for_status(Parameters(wait_args("7"))) + .await + .expect("the tool returns a result"); + let value = result.structured_content.expect("structured json content"); + assert_eq!(value["status"], "attention"); + assert_eq!(value["timed_out"], false); + } + + #[tokio::test] + async fn wait_for_status_relays_an_unknown_handle_as_invalid_params() { + let (handle, requests) = channel(); + let _shell = spawn_test_shell( + requests, + Reply::Waited(WaitOutcome { + status: None, + error: Some("no live session with handle 7".into()), + }), + ); + let error = TermherdMcp::new(handle) + .wait_for_status(Parameters(wait_args("7"))) + .await + .expect_err("a rejected wait is a tool error"); + assert!( + error.message.contains("no live session"), + "got: {}", + error.message + ); + } + + #[tokio::test] + async fn wait_for_status_reports_the_current_status_when_it_times_out() { + // Timing out is not a failure: the agent needs to know *where* the + // session got to, to decide between waiting again and giving up. + let (handle, requests) = channel(); + let shell = spawn_test_shell_seq( + requests, + vec![ + // The wait itself: received, never answered. + None, + // The follow-up read of the current status. + Some(Reply::Sessions(vec![SessionInfo { + handle: "7".into(), + title: "proj".into(), + cwd: None, + kind: SessionKind::Shell, + resume_id: None, + status: SessionStatus::Busy, + }])), + ], + ); + let args = WaitArgs { + session: "7".into(), + timeout_ms: Some(50), + ..WaitArgs::default() + }; + let result = TermherdMcp::new(handle) + .wait_for_status(Parameters(args)) + .await + .expect("a timed-out wait still returns a result"); + let value = result.structured_content.expect("structured json content"); + assert_eq!(value["timed_out"], true); + assert_eq!( + value["status"], "busy", + "the session's current status rides back with the timeout" + ); + let seen = shell.await.expect("shell task"); + assert_eq!(seen.len(), 2, "the wait, then the status read-back"); + assert_eq!(seen[1], Request::ListSessions); + } + + #[test] + fn a_wait_is_capped_however_long_the_caller_asks() { + // The outer Q7 guarantee: no parameterisation parks a caller forever. + assert_eq!(wait_timeout(None), Duration::from_millis(DEFAULT_WAIT_MS)); + assert_eq!(wait_timeout(Some(1_000)), Duration::from_millis(1_000)); + assert_eq!( + wait_timeout(Some(u64::MAX)), + Duration::from_millis(MAX_WAIT_MS) + ); + } + + #[tokio::test] + async fn read_terminal_passes_the_handle_and_line_budget() { + let (handle, requests) = channel(); + let shell = spawn_test_shell( + requests, + Reply::Terminal(TerminalRead { + text: Some("$ cargo test".into()), + error: None, + }), + ); + let result = TermherdMcp::new(handle) + .read_terminal(Parameters(ReadArgs { + session: "7".into(), + lines: Some(5), + })) + .await + .expect("the tool returns a result"); + assert_eq!( + shell.await.expect("shell task"), + Request::ReadTerminal { + session: 7, + lines: 5 + } + ); + let value = result.structured_content.expect("structured json content"); + assert_eq!(value["text"], "$ cargo test"); + assert_eq!(value["rendered"], true); + } + + #[tokio::test] + async fn read_terminal_defaults_to_the_snapshot_line_budget() { + let (handle, requests) = channel(); + let shell = spawn_test_shell( + requests, + Reply::Terminal(TerminalRead { + text: None, + error: None, + }), + ); + let _ = TermherdMcp::new(handle) + .read_terminal(Parameters(ReadArgs { + session: "7".into(), + lines: None, + })) + .await; + assert_eq!( + shell.await.expect("shell task"), + Request::ReadTerminal { + session: 7, + lines: DEFAULT_TEXT_LINES + } + ); + } + + #[tokio::test] + async fn read_terminal_marks_an_undrawn_screen_as_not_rendered() { + // Distinct from an unknown handle: the caller should retry, not give up. + let (handle, requests) = channel(); + let _shell = spawn_test_shell( + requests, + Reply::Terminal(TerminalRead { + text: None, + error: None, + }), + ); + let result = TermherdMcp::new(handle) + .read_terminal(Parameters(ReadArgs { + session: "7".into(), + lines: None, + })) + .await + .expect("an undrawn screen is not an error"); + let value = result.structured_content.expect("structured json content"); + assert_eq!(value["rendered"], false); + assert_eq!(value["text"], ""); + } + + #[tokio::test] + async fn read_terminal_relays_an_unknown_handle_as_invalid_params() { + let (handle, requests) = channel(); + let _shell = spawn_test_shell( + requests, + Reply::Terminal(TerminalRead { + text: None, + error: Some("no live session with handle 7".into()), + }), + ); + let error = TermherdMcp::new(handle) + .read_terminal(Parameters(ReadArgs { + session: "7".into(), + lines: None, + })) + .await + .expect_err("a rejected read is a tool error"); + assert!( + error.message.contains("no live session"), + "got: {}", + error.message + ); + } } diff --git a/crates/app/src/shell.rs b/crates/app/src/shell.rs index 1a7c7ff..b7e112d 100644 --- a/crates/app/src/shell.rs +++ b/crates/app/src/shell.rs @@ -255,6 +255,9 @@ struct Shell { config: ConfigInput, /// Latest rendered grid per session. screens: HashMap, + /// Bridge callers parked on a session reaching a target activity — the one + /// request kind whose reply lands in a later `update` (see `shell::serve`). + waiters: Vec, /// Current keyboard target. focus: Focus, /// Last non-empty terminal selection, for the keyboard copy shortcut (FR4). @@ -623,6 +626,7 @@ impl Shell { mcp_tokens, mcp_session_tokens: HashMap::new(), screens: HashMap::new(), + waiters: Vec::new(), focus: Focus::Search, selection: None, theme: startup.theme.to_iced(), @@ -802,6 +806,7 @@ impl Shell { let effects = self .core .apply(termherd_core::Event::StatusChanged { session, status }); + self.settle_waiters(session, status); self.perform(effects) } Message::PtyTitle { session, title } => { @@ -823,6 +828,10 @@ impl Shell { let effects = self .core .apply(termherd_core::Event::PtyExited { session, clean }); + // An exit is the other way a session's activity settles: a crash + // records `Exited` without emitting a status change, and a clean + // exit takes the session out of the registry entirely. + self.settle_waiters(session, SessionStatus::Exited); if effects.is_empty() { // No auto-close: the dead terminal stays on screen. Task::none() @@ -3624,4 +3633,292 @@ mod key_routing { "a draining recorder blocks a new toggle" ); } + + // --- The terminal-sync rung: wait_for_status + read_terminal --- + + use super::bridge::{ + Reply as BridgeReply, ReplyPort, Request as BridgeRequest, TerminalRead, WaitOutcome, + }; + use tokio::sync::oneshot::error::TryRecvError; + + /// Serve `request` through the bridge seam and hand back the caller's reply + /// receiver. A parked request leaves it empty — that emptiness is the + /// assertion a wait test needs. + fn serve( + shell: &mut Shell, + request: BridgeRequest, + ) -> tokio::sync::oneshot::Receiver { + let (tx, rx) = tokio::sync::oneshot::channel(); + let _ = shell.serve(request, ReplyPort::new(tx)); + rx + } + + /// The `WaitOutcome` a served wait answered with, or `None` while it parks. + fn waited(rx: &mut tokio::sync::oneshot::Receiver) -> Option { + match rx.try_recv() { + Ok(BridgeReply::Waited(outcome)) => Some(outcome), + Ok(other) => panic!("expected a Waited reply, got {other:?}"), + Err(TryRecvError::Empty) => None, + Err(error) => panic!("the reply channel closed: {error}"), + } + } + + /// The `TerminalRead` a served read answered with. A read never parks, so an + /// empty channel is a failure, not a state. + fn read_back(rx: &mut tokio::sync::oneshot::Receiver) -> TerminalRead { + match rx.try_recv() { + Ok(BridgeReply::Terminal(read)) => read, + Ok(other) => panic!("expected a Terminal reply, got {other:?}"), + Err(error) => panic!("a read must answer in the same update: {error}"), + } + } + + /// Wait for the two statuses `wait_for_status` exists for. + fn wait_for(session: u64) -> BridgeRequest { + BridgeRequest::WaitForStatus { + session, + targets: vec![SessionStatus::Idle, SessionStatus::Attention], + } + } + + #[test] + fn wait_answers_at_once_when_the_session_already_holds_a_target_status() { + // A session sitting on the target must not park: an agent that asks + // "tell me when it's idle" about an idle session would otherwise hang + // until its own timeout, having already missed the transition. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Idle, + }); + + let mut rx = serve(&mut shell, wait_for(session.0.get())); + assert_eq!( + waited(&mut rx), + Some(WaitOutcome { + status: Some(SessionStatus::Idle), + error: None, + }) + ); + } + + #[test] + fn wait_parks_until_the_status_change_arrives() { + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + + let mut rx = serve(&mut shell, wait_for(session.0.get())); + assert_eq!( + waited(&mut rx), + None, + "a busy session leaves the wait parked" + ); + + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Idle, + }); + assert_eq!( + waited(&mut rx), + Some(WaitOutcome { + status: Some(SessionStatus::Idle), + error: None, + }), + "the status change settles the parked wait" + ); + } + + #[test] + fn wait_stays_parked_through_a_non_target_status() { + // Busy -> Starting is a change, but not the one asked for; waking on it + // would report "done" while the command is still running. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + let mut rx = serve(&mut shell, wait_for(session.0.get())); + + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Starting, + }); + assert_eq!(waited(&mut rx), None); + } + + #[test] + fn wait_ignores_a_status_change_on_another_session() { + let (mut shell, _pty) = shell_with_terminal(); + let watched = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session: watched, + status: SessionStatus::Busy, + }); + let _ = shell.launch("/tmp/other".to_string(), Launch::Shell); + let other = shell.core.workspace.focused_session().expect("focused"); + assert_ne!(other, watched); + + let mut rx = serve(&mut shell, wait_for(watched.0.get())); + let _ = shell.update(Message::PtyStatus { + session: other, + status: SessionStatus::Idle, + }); + assert_eq!( + waited(&mut rx), + None, + "another pane going idle must not settle this wait" + ); + } + + #[test] + fn wait_rejects_an_unknown_handle_without_parking() { + let (mut shell, _pty) = shell_with_terminal(); + let mut rx = serve(&mut shell, wait_for(9_999)); + let outcome = waited(&mut rx).expect("an unknown handle answers immediately"); + assert_eq!(outcome.status, None); + assert!( + outcome.error.is_some(), + "an unknown handle is rejected, not awaited forever" + ); + } + + #[test] + fn wait_is_settled_by_an_unclean_pty_exit() { + // A crashed session emits no status change — only an exit. Without this + // the caller would wait out its whole bound on a dead terminal. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + let mut rx = serve(&mut shell, wait_for(session.0.get())); + + let _ = shell.update(Message::PtyExited { + session, + clean: false, + }); + assert_eq!( + waited(&mut rx), + Some(WaitOutcome { + status: Some(SessionStatus::Exited), + error: None, + }), + "an exit settles the wait with the status it ended on" + ); + } + + #[test] + fn a_waiter_whose_caller_gave_up_is_dropped() { + // The caller timed out and dropped its end; the shell must not keep the + // entry alive for a session that may never reach the target. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + drop(serve(&mut shell, wait_for(session.0.get()))); + assert_eq!(shell.waiters.len(), 1, "the wait parked"); + + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Starting, + }); + assert!( + shell.waiters.is_empty(), + "an abandoned waiter is swept on the next status event" + ); + } + + #[test] + fn read_terminal_returns_the_panes_visible_text() { + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + shell.screens.insert(session, screen_of("$ cargo test")); + + let mut rx = serve( + &mut shell, + BridgeRequest::ReadTerminal { + session: session.0.get(), + lines: 40, + }, + ); + assert_eq!( + read_back(&mut rx), + TerminalRead { + text: Some("$ cargo test".to_owned()), + error: None, + } + ); + } + + #[test] + fn read_terminal_keeps_only_the_requested_trailing_lines() { + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + shell + .screens + .insert(session, screen_of("one\ntwo\nthree\nfour")); + + let mut rx = serve( + &mut shell, + BridgeRequest::ReadTerminal { + session: session.0.get(), + lines: 2, + }, + ); + assert_eq!( + read_back(&mut rx).text.as_deref(), + Some("three\nfour"), + "the read is bounded like a snapshot's text_lines" + ); + } + + #[test] + fn read_terminal_reports_no_text_for_a_pane_that_has_not_rendered() { + // Distinct from an unknown handle: the session is live, its screen just + // hasn't arrived. An agent should retry, not give up on the handle. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + + let mut rx = serve( + &mut shell, + BridgeRequest::ReadTerminal { + session: session.0.get(), + lines: 40, + }, + ); + assert_eq!( + read_back(&mut rx), + TerminalRead { + text: None, + error: None, + } + ); + } + + #[test] + fn read_terminal_rejects_an_unknown_handle() { + let (mut shell, _pty) = shell_with_terminal(); + let mut rx = serve( + &mut shell, + BridgeRequest::ReadTerminal { + session: 9_999, + lines: 40, + }, + ); + let read = read_back(&mut rx); + assert_eq!(read.text, None); + assert!( + read.error.is_some(), + "an unknown handle is an error, not empty text" + ); + } } diff --git a/crates/app/src/shell/bridge.rs b/crates/app/src/shell/bridge.rs index 1668982..a2e3fcb 100644 --- a/crates/app/src/shell/bridge.rs +++ b/crates/app/src/shell/bridge.rs @@ -56,6 +56,45 @@ pub enum Request { /// read requests it is answered off `respond` — the shell applies it through /// `App::apply` and performs the effects, since it needs `&mut` state. Act(Action), + /// Park until a session's activity reaches one of `targets`, for the + /// `wait_for_status` MCP tool. The odd one out on this bridge: its reply + /// lands in a *later* `update` — whichever one carries the status change — + /// so the shell holds the reply port meanwhile. Answers at once when the + /// session already sits on a target, or when the handle is unknown. + WaitForStatus { + session: u64, + targets: Vec, + }, + /// The visible text of one session's terminal, for the `read_terminal` MCP + /// tool — the deep read the light `Snapshot` deliberately leaves out. + ReadTerminal { session: u64, lines: usize }, +} + +/// The result of a [`Request::WaitForStatus`]. `error` is `Some` only when the +/// handle named no live session, in which case nothing was awaited. Otherwise +/// `status` is the activity the session had reached when the wait settled. +/// +/// A caller that gives up first sees [`CallError::Timeout`] instead — the wait +/// is bounded by the caller's own clock, never by one the shell keeps (Q7). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WaitOutcome { + /// The status that settled the wait. + pub status: Option, + /// Why nothing was awaited, or `None` when the wait ran. + pub error: Option, +} + +/// The result of a [`Request::ReadTerminal`]. The three cases stay distinct: an +/// unknown handle (`error`), a live session whose screen has not rendered yet +/// (`text: None`), and real text — a scoped `Snapshot` collapses the first two +/// into one absent key, which an agent cannot act on. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TerminalRead { + /// The visible text, trailing lines only, or `None` when nothing has + /// rendered yet. + pub text: Option, + /// Why nothing was read, or `None` when the read ran. + pub error: Option, } /// A workspace mutation an MCP client asks termherd to perform. Each variant @@ -129,6 +168,8 @@ pub enum Reply { Snapshot(WorkspaceSnapshot), Sessions(Vec), Acted(ActionOutcome), + Waited(WaitOutcome), + Terminal(TerminalRead), } /// The kind of program a session runs, as an MCP client sees it. Distinct from @@ -238,7 +279,7 @@ pub type Requests = TakeOnceSource>; pub struct ReplyPort(Arc>>>); impl ReplyPort { - fn new(tx: oneshot::Sender) -> Self { + pub(super) fn new(tx: oneshot::Sender) -> Self { Self(Arc::new(Mutex::new(Some(tx)))) } @@ -253,6 +294,20 @@ impl ReplyPort { } } +impl ReplyPort { + /// Whether the caller is gone — it timed out and dropped its end, so no + /// answer can land. Lets the shell sweep parked waiters instead of holding + /// them for a session that may never reach its target. + pub(super) fn abandoned(&self) -> bool { + self.0 + .lock() + .ok() + .and_then(|slot| slot.as_ref().map(oneshot::Sender::is_closed)) + // A taken slot means it was already answered; treat it as gone too. + .unwrap_or(true) + } +} + impl fmt::Debug for ReplyPort { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("ReplyPort") @@ -308,6 +363,17 @@ pub fn respond(core: &App, request: &Request, inputs: &SnapshotInputs) -> Reply Request::Act(_) => Reply::Acted(ActionOutcome::rejected( "an action reached the read-only responder; the shell must apply it", )), + // A wait parks and a terminal read needs the `pty` adapter's screens — + // neither is answerable from a bare `&App`, so the shell serves them + // itself. These arms are the defensive default, as above. + Request::WaitForStatus { .. } => Reply::Waited(WaitOutcome { + status: None, + error: Some("a wait reached the read-only responder; the shell must park it".into()), + }), + Request::ReadTerminal { .. } => Reply::Terminal(TerminalRead { + text: None, + error: Some("a terminal read reached the read-only responder".into()), + }), } } @@ -347,14 +413,56 @@ pub(crate) fn spawn_test_shell( requests: Requests, reply: Reply, ) -> tokio::task::JoinHandle { + spawn_test_shell_seq(requests, vec![Some(reply)]).map_into_first() +} + +/// A test shell that serves a *sequence* of requests, answering the nth with +/// `replies[n]` — `None` meaning "receive it and stay silent", which is how a +/// caller-side timeout is exercised. Returns the requests in arrival order. +#[cfg(test)] +pub(crate) fn spawn_test_shell_seq( + requests: Requests, + replies: Vec>, +) -> tokio::task::JoinHandle> { tokio::spawn(async move { let mut rx = requests.take().expect("a receiver on first take"); - let (request, reply_tx) = rx.recv().await.expect("one request"); - let _ = reply_tx.send(reply); - request + let mut seen = Vec::new(); + // Un-answered ports are kept, not dropped: dropping one raises + // `Dropped` at the caller, and it is the *timeout* path we exercise. + let mut held = Vec::new(); + for reply in replies { + let Some((request, reply_tx)) = rx.recv().await else { + break; + }; + seen.push(request); + match reply { + Some(reply) => { + let _ = reply_tx.send(reply); + } + None => held.push(reply_tx), + } + } + seen }) } +/// Adapter so the single-reply helper keeps its one-request return shape. +#[cfg(test)] +trait FirstRequest { + fn map_into_first(self) -> tokio::task::JoinHandle; +} + +#[cfg(test)] +impl FirstRequest for tokio::task::JoinHandle> { + fn map_into_first(self) -> tokio::task::JoinHandle { + tokio::spawn(async move { + let mut seen = self.await.expect("test shell task"); + assert_eq!(seen.len(), 1, "the single-reply shell served one request"); + seen.remove(0) + }) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/app/src/shell/orchestrate.rs b/crates/app/src/shell/orchestrate.rs index 931e739..e7911e2 100644 --- a/crates/app/src/shell/orchestrate.rs +++ b/crates/app/src/shell/orchestrate.rs @@ -149,7 +149,7 @@ impl Shell { /// Resolve a stable handle to a live [`SessionId`], or `None` when no session /// carries it (already closed, or never existed). - fn resolve(&self, handle: u64) -> Option { + pub(super) fn resolve(&self, handle: u64) -> Option { let id = NonZeroU64::new(handle).map(SessionId)?; self.core.sessions.contains_key(&id).then_some(id) } diff --git a/crates/app/src/shell/serve.rs b/crates/app/src/shell/serve.rs index 17289db..2ff0cc2 100644 --- a/crates/app/src/shell/serve.rs +++ b/crates/app/src/shell/serve.rs @@ -8,9 +8,25 @@ use iced::Task; -use super::bridge::{self, ReplyPort, Request}; +use termherd_core::SessionStatus; +use termherd_core::snapshot::tail_lines; +use termherd_core::workspace::SessionId; + +use super::bridge::{self, ReplyPort, Request, TerminalRead, WaitOutcome}; use super::{Message, Shell}; +/// One bridge caller parked on a session reaching a target activity. Held by +/// the shell between the `update` that served the request and the one that +/// carries the status change (or the exit) settling it. +pub(super) struct StatusWaiter { + /// The session being watched. + pub session: SessionId, + /// Any of these activities settles the wait. + pub targets: Vec, + /// Where the answer goes when it does. + pub reply: ReplyPort, +} + impl Shell { /// Answer one bridge request and return any async follow-up it needs. pub(super) fn serve(&mut self, request: Request, reply: ReplyPort) -> Task { @@ -22,7 +38,18 @@ impl Shell { reply.answer(bridge::Reply::Acted(outcome)); task } - // The read requests answer straight from owned state. + // A wait may answer now or park; either way it needs no follow-up. + Request::WaitForStatus { session, targets } => { + self.serve_wait(session, targets, reply); + Task::none() + } + // A terminal read needs the `pty` adapter's screens, which the + // read-only responder over `&App` cannot see. + Request::ReadTerminal { session, lines } => { + reply.answer(bridge::Reply::Terminal(self.read_terminal(session, lines))); + Task::none() + } + // The remaining read requests answer straight from owned state. read => { let inputs = self.snapshot_inputs(&read); reply.answer(bridge::respond(&self.core, &read, &inputs)); @@ -30,4 +57,75 @@ impl Shell { } } } + + /// Answer a wait now when it is already settled — an unknown handle, or a + /// session sitting on a target — otherwise park it. Answering an + /// already-satisfied wait matters: a caller asking "tell me when it is idle" + /// about an idle session would otherwise wait out its own bound having + /// missed the transition it asked about. + fn serve_wait(&mut self, session: u64, targets: Vec, reply: ReplyPort) { + let Some(id) = self.resolve(session) else { + reply.answer(bridge::Reply::Waited(WaitOutcome { + status: None, + error: Some(format!("no live session with handle {session}")), + })); + return; + }; + let status = self.core.sessions.get(&id).map(|live| live.status); + if status.is_some_and(|status| targets.contains(&status)) { + reply.answer(bridge::Reply::Waited(WaitOutcome { + status, + error: None, + })); + return; + } + self.waiters.push(StatusWaiter { + session: id, + targets, + reply, + }); + } + + /// The visible text of one session's terminal, trimmed to its last `lines`. + /// The three outcomes stay distinct — unknown handle, live-but-unrendered, + /// and text — because an agent acts differently on each. + fn read_terminal(&self, session: u64, lines: usize) -> TerminalRead { + let Some(id) = self.resolve(session) else { + return TerminalRead { + text: None, + error: Some(format!("no live session with handle {session}")), + }; + }; + TerminalRead { + // `core` owns the truncation rule; a read borrows it rather than + // growing a second one that could drift from a snapshot's. + text: self + .screens + .get(&id) + .map(|screen| tail_lines(&screen.text(), lines)), + error: None, + } + } + + /// Settle every waiter watching `session` now that it reached `status`, and + /// sweep the ones whose caller gave up. Called from the two updates that can + /// move a session's activity: a status change and a PTY exit — a crash emits + /// no status change, so without the second a caller would wait out its whole + /// bound on a dead terminal. + pub(super) fn settle_waiters(&mut self, session: SessionId, status: SessionStatus) { + self.waiters.retain(|waiter| { + // A target reached settles the wait — and so does an exit, whatever + // was asked for: a dead session will never reach it, so holding the + // caller would only burn its bound to reach the same conclusion. + let settled = waiter.session == session + && (waiter.targets.contains(&status) || status == SessionStatus::Exited); + if settled { + waiter.reply.answer(bridge::Reply::Waited(WaitOutcome { + status: Some(status), + error: None, + })); + } + !settled && !waiter.reply.abandoned() + }); + } } From cc9f4c856f26828a4ec174facff9a1ad743dc807 Mon Sep 17 00:00:00 2001 From: Bastien Gallay Date: Sat, 25 Jul 2026 21:39:01 +0200 Subject: [PATCH 3/3] fix(app): a wait must never park on a session that cannot move again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four defects the review found, all on the same seam — three of them ways a caller ends up parked on a session core has already given up on. `Exited` is terminal in core: `StatusChanged` refuses to overwrite it. So a wait placed *after* an unclean exit parked forever, because the exit that would have settled it had already fired. `serve_wait` now short-circuits on it, the same rule `settle_waiters` applies — extracted to one `settles` predicate rather than two copies that already disagreed. The settling status was taken from the message. A status still in flight when the exit landed is dropped by core, but was reported to the client — a crashed session answering "idle". Read it back from core instead, which also gives the clean-exit case (session gone from the registry) its `Exited` answer for free. Closing a tab drops its sessions with no PTY exit reaching `update`, so nothing settled those waiters, and nothing swept the ones whose caller had timed out on a quiet workspace. `serve` now sweeps first: it is the one path that runs on every external call, however still the sessions are. Finally, the status read-back after a timeout propagated its own failure. The likeliest cause of the first timeout is a wedged shell, which fails the read-back too — turning the answer the caller had earned into an error. It is best-effort now (`status: null`), on a 250ms bound so it barely extends the wait it follows. --- crates/app/src/mcp/handler.rs | 69 ++++++++++++++----- crates/app/src/shell.rs | 120 +++++++++++++++++++++++++++++++++- crates/app/src/shell/serve.rs | 60 ++++++++++++++--- 3 files changed, 221 insertions(+), 28 deletions(-) diff --git a/crates/app/src/mcp/handler.rs b/crates/app/src/mcp/handler.rs index db101a3..e48b168 100644 --- a/crates/app/src/mcp/handler.rs +++ b/crates/app/src/mcp/handler.rs @@ -39,6 +39,11 @@ const DEFAULT_WAIT_MS: u64 = 30_000; /// that no bridge call can park a caller indefinitely. const MAX_WAIT_MS: u64 = 300_000; +/// Bound on the status read-back after a wait timed out. Deliberately short: it +/// is a courtesy on top of an answer the caller has already earned, so it must +/// barely extend the wait it follows. +const READ_BACK_TIMEOUT: Duration = Duration::from_millis(250); + /// The live-bridge MCP server handler. Holds only the bridge into the shell — /// cloned fresh per session by the transport's service factory. #[derive(Clone)] @@ -290,7 +295,7 @@ impl TermherdMcp { // rather than a bare failure: an agent that waited 30s needs the // current status to decide whether to keep waiting or give up. Err(CallError::Timeout(_)) => { - let status = self.current_status(session).await?; + let status = self.current_status(session).await; Ok(CallToolResult::structured(serde_json::json!({ "status": status, "timed_out": true, @@ -339,25 +344,27 @@ impl TermherdMcp { } impl TermherdMcp { - /// A session's current activity, read back after a wait timed out. A handle - /// that no longer resolves (the session died meanwhile) reports `None` - /// rather than failing — the caller asked about waiting, not existence. - async fn current_status(&self, session: u64) -> Result, ErrorData> { - let reply = self + /// A session's current activity, read back after a wait timed out. Best + /// effort by design: `None` covers a handle that no longer resolves *and* a + /// read-back that failed. The likeliest cause of the original timeout is a + /// wedged shell, which would fail this call too — and turning that into an + /// error would swallow the `timed_out` answer the caller is owed. + async fn current_status(&self, session: u64) -> Option<&'static str> { + match self .bridge - .call(Request::ListSessions, CALL_TIMEOUT) + .call(Request::ListSessions, READ_BACK_TIMEOUT) .await - .map_err(|error| ErrorData::internal_error(error.to_string(), None))?; - let Reply::Sessions(sessions) = reply else { - return Err(ErrorData::internal_error( - "bridge answered the wrong reply kind", - None, - )); - }; - Ok(sessions - .iter() - .find(|info| info.handle == session.to_string()) - .map(|info| status_str(info.status))) + { + Ok(Reply::Sessions(sessions)) => sessions + .iter() + .find(|info| info.handle == session.to_string()) + .map(|info| status_str(info.status)), + Ok(_) => None, + Err(error) => { + tracing::debug!(%error, "could not read the status back after a wait timeout"); + None + } + } } /// Send an [`Action`] over the bridge and shape its [`ActionOutcome`] into a @@ -1184,6 +1191,32 @@ mod tests { assert_eq!(seen[1], Request::ListSessions); } + #[tokio::test] + async fn a_wait_that_times_out_still_answers_when_the_read_back_fails() { + // The likeliest cause of the original timeout is a wedged shell, which + // fails the status read-back too. That must not swallow the `timed_out` + // answer the caller already earned — it is the whole point of the path. + let (handle, requests) = channel(); + // Neither the wait nor the read-back is ever answered. + let shell = spawn_test_shell_seq(requests, vec![None, None]); + let args = WaitArgs { + session: "7".into(), + timeout_ms: Some(50), + ..WaitArgs::default() + }; + let result = TermherdMcp::new(handle) + .wait_for_status(Parameters(args)) + .await + .expect("a wedged shell still yields the timeout answer"); + let value = result.structured_content.expect("structured json content"); + assert_eq!(value["timed_out"], true); + assert!( + value["status"].is_null(), + "an unreadable status is null, not an error" + ); + drop(shell); + } + #[test] fn a_wait_is_capped_however_long_the_caller_asks() { // The outer Q7 guarantee: no parameterisation parks a caller forever. diff --git a/crates/app/src/shell.rs b/crates/app/src/shell.rs index b7e112d..c76ecce 100644 --- a/crates/app/src/shell.rs +++ b/crates/app/src/shell.rs @@ -806,7 +806,7 @@ impl Shell { let effects = self .core .apply(termherd_core::Event::StatusChanged { session, status }); - self.settle_waiters(session, status); + self.settle_waiters(session); self.perform(effects) } Message::PtyTitle { session, title } => { @@ -831,7 +831,7 @@ impl Shell { // An exit is the other way a session's activity settles: a crash // records `Exited` without emitting a status change, and a clean // exit takes the session out of the registry entirely. - self.settle_waiters(session, SessionStatus::Exited); + self.settle_waiters(session); if effects.is_empty() { // No auto-close: the dead terminal stays on screen. Task::none() @@ -3837,6 +3837,122 @@ mod key_routing { ); } + #[test] + fn wait_answers_at_once_for_a_session_that_already_exited() { + // `Exited` is terminal in core — it refuses to overwrite it — so a wait + // parked on a dead session could never be woken by a status event. It + // has to be answered on the spot instead. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyExited { + session, + clean: false, + }); + + let mut rx = serve(&mut shell, wait_for(session.0.get())); + assert_eq!( + waited(&mut rx), + Some(WaitOutcome { + status: Some(SessionStatus::Exited), + error: None, + }) + ); + assert!(shell.waiters.is_empty(), "nothing parked"); + } + + #[test] + fn a_status_still_in_flight_when_the_exit_lands_settles_as_exited() { + // The PTY task can emit a status the exit overtakes. Core drops it (a + // dead session stays dead); the wait must report what core recorded, + // not the message, or the client is told a crashed session went idle. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + let mut rx = serve(&mut shell, wait_for(session.0.get())); + + // The exit lands first and settles the waiter... + let _ = shell.update(Message::PtyExited { + session, + clean: false, + }); + assert_eq!( + waited(&mut rx).and_then(|outcome| outcome.status), + Some(SessionStatus::Exited) + ); + + // ...and a late Idle must not resurrect it for a second waiter either. + let mut late = serve(&mut shell, wait_for(session.0.get())); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Idle, + }); + assert_eq!( + waited(&mut late).and_then(|outcome| outcome.status), + Some(SessionStatus::Exited), + "core kept the session dead, so the wait reports dead" + ); + } + + #[test] + fn a_wait_on_a_session_closed_from_the_ui_is_settled_on_the_next_request() { + // Closing a tab drops its sessions without any PTY exit reaching + // `update`, so nothing would settle the waiter. The next served request + // sweeps it rather than leaving the caller parked on a ghost. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + let mut rx = serve(&mut shell, wait_for(session.0.get())); + assert_eq!(waited(&mut rx), None, "parked while busy"); + + let effects = shell.core.apply(termherd_core::Event::CloseTab(0)); + let _ = shell.perform(effects); + assert!( + !shell.core.sessions.contains_key(&session), + "the close took the session out of the registry" + ); + + // Any request re-enters `serve`, which sweeps first. + drop(serve(&mut shell, BridgeRequest::ListSessions)); + assert_eq!( + waited(&mut rx).and_then(|outcome| outcome.status), + Some(SessionStatus::Exited) + ); + assert!(shell.waiters.is_empty()); + } + + #[test] + fn abandoned_waiters_are_swept_on_a_quiet_workspace() { + // Without a sweep on the request path, a quiet workspace never drops + // the waiters of callers that timed out, and the list grows unbounded. + let (mut shell, _pty) = shell_with_terminal(); + let session = shell.core.workspace.focused_session().expect("focused"); + let _ = shell.update(Message::PtyStatus { + session, + status: SessionStatus::Busy, + }); + // Ten timed-out waits in a row on a session that never moves: the list + // must not accumulate them, since each `serve` sweeps before parking. + for _ in 0..10 { + drop(serve(&mut shell, wait_for(session.0.get()))); + assert!( + shell.waiters.len() <= 1, + "at most the wait just parked, never a backlog" + ); + } + + drop(serve(&mut shell, BridgeRequest::ListSessions)); + assert!( + shell.waiters.is_empty(), + "no status event fired, yet the dead waiters are gone" + ); + } + #[test] fn read_terminal_returns_the_panes_visible_text() { let (mut shell, _pty) = shell_with_terminal(); diff --git a/crates/app/src/shell/serve.rs b/crates/app/src/shell/serve.rs index 2ff0cc2..8908fea 100644 --- a/crates/app/src/shell/serve.rs +++ b/crates/app/src/shell/serve.rs @@ -30,6 +30,10 @@ pub(super) struct StatusWaiter { impl Shell { /// Answer one bridge request and return any async follow-up it needs. pub(super) fn serve(&mut self, request: Request, reply: ReplyPort) -> Task { + // Waiters are settled by status events, so a quiet workspace never + // sweeps them. Do it here too: this is the one path that runs on every + // external call, however still the sessions are. + self.sweep_waiters(); match request { // Actions mutate, so they can't answer off a `&App`: apply them and // perform the effects here, where the shell owns both. @@ -72,7 +76,10 @@ impl Shell { return; }; let status = self.core.sessions.get(&id).map(|live| live.status); - if status.is_some_and(|status| targets.contains(&status)) { + // Settled already? Either it sits on a target, or it has exited — and + // `Exited` is terminal in `core`, which refuses to overwrite it, so a + // wait parked on a dead session could never be woken by a status event. + if status.is_some_and(|status| settles(&targets, status)) { reply.answer(bridge::Reply::Waited(WaitOutcome { status, error: None, @@ -107,18 +114,21 @@ impl Shell { } } - /// Settle every waiter watching `session` now that it reached `status`, and + /// Settle every waiter watching `session` now that its activity moved, and /// sweep the ones whose caller gave up. Called from the two updates that can /// move a session's activity: a status change and a PTY exit — a crash emits /// no status change, so without the second a caller would wait out its whole /// bound on a dead terminal. - pub(super) fn settle_waiters(&mut self, session: SessionId, status: SessionStatus) { + /// + /// The settling status is read back from `core`, never taken from the + /// message: `core` refuses to overwrite `Exited`, so a status still in + /// flight when the exit landed would otherwise report a dead session as + /// idle. A session gone from the registry (a clean exit closed its pane) + /// reads as `Exited`. + pub(super) fn settle_waiters(&mut self, session: SessionId) { + let status = self.status_of(session); self.waiters.retain(|waiter| { - // A target reached settles the wait — and so does an exit, whatever - // was asked for: a dead session will never reach it, so holding the - // caller would only burn its bound to reach the same conclusion. - let settled = waiter.session == session - && (waiter.targets.contains(&status) || status == SessionStatus::Exited); + let settled = waiter.session == session && settles(&waiter.targets, status); if settled { waiter.reply.answer(bridge::Reply::Waited(WaitOutcome { status: Some(status), @@ -128,4 +138,38 @@ impl Shell { !settled && !waiter.reply.abandoned() }); } + + /// Drop waiters no one is listening for, and settle any whose session has + /// left the registry — a tab closed from the UI removes its sessions without + /// a PTY exit ever reaching `update`, which would otherwise leave the caller + /// parked on a session that no longer exists. + fn sweep_waiters(&mut self) { + let vanished: Vec = self + .waiters + .iter() + .map(|waiter| waiter.session) + .filter(|id| !self.core.sessions.contains_key(id)) + .collect(); + for session in vanished { + self.settle_waiters(session); + } + self.waiters.retain(|waiter| !waiter.reply.abandoned()); + } + + /// A session's activity as `core` records it, or `Exited` when it is no + /// longer registered — the one place a gone session is read as dead. + fn status_of(&self, session: SessionId) -> SessionStatus { + self.core + .sessions + .get(&session) + .map_or(SessionStatus::Exited, |live| live.status) + } +} + +/// Whether `status` settles a wait on `targets`. An exit always does, whatever +/// was asked for: `Exited` is terminal in `core`, so the session will never +/// reach the target and holding the caller would only burn its bound to reach +/// the same conclusion. +fn settles(targets: &[SessionStatus], status: SessionStatus) -> bool { + targets.contains(&status) || status == SessionStatus::Exited }