0.10.0 follow-up: user-input deadline under the heartbeat, terminal routes off the runtime workers, sleep inhibitor as a tokio child

crates/tui/src/core/engine/approval.rs

@@ -273,6 +273,11 @@ impl Engine {
         // built-in default; an explicit 0 waits indefinitely.
         let wait = self.config.user_input_timeout.unwrap_or(USER_INPUT_TIMEOUT);
         let started = std::time::Instant::now();
+        // One absolute deadline for the whole wait. `select!` drops the losing
+        // branches whenever the heartbeat wins, so a relative `timeout(wait,
+        // ..)` rebuilt per iteration restarted from zero at every tick and,
+        // with the tick shorter than the timeout, never fired at all.
+        let deadline = (!wait.is_zero()).then(|| tokio::time::Instant::now() + wait);
         let mut heartbeat = tokio::time::interval(WAIT_HEARTBEAT);
         heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         heartbeat.tick().await;
@@ -297,10 +302,11 @@ impl Engine {
                     ));
                 }
                 result = async {
-                    if wait.is_zero() {
-                        Ok(self.rx_user_input.recv().await)
-                    } else {
-                        tokio::time::timeout(wait, self.rx_user_input.recv()).await
+                    match deadline {
+                        None => Ok(self.rx_user_input.recv().await),
+                        Some(deadline) => {
+                            tokio::time::timeout_at(deadline, self.rx_user_input.recv()).await
+                        }
                     }
                 } => {
                     match result {
@@ -560,6 +566,35 @@ mod tests {
         task.abort();
     }
 
+    /// The user-input deadline has to survive the #6184 heartbeat. Under test
+    /// the heartbeat ticks every 50 ms, so a 200 ms timeout that is rebuilt on
+    /// every tick never fires and the turn parks forever; the outer guard here
+    /// is what turns that hang into a failure.
+    #[tokio::test]
+    async fn user_input_deadline_is_not_reset_by_the_wait_heartbeat() {
+        let (mut engine, _handle) = Engine::new(
+            EngineConfig {
+                user_input_timeout: Some(Duration::from_millis(200)),
+                terminal_chrome_enabled: false,
+                ..EngineConfig::default()
+            },
+            &Config::default(),
+        );
+        let request = UserInputRequest {
+            questions: Vec::new(),
+        };
+        let outcome = tokio::time::timeout(
+            Duration::from_secs(3),
+            engine.await_user_input("user-input-deadline", request),
+        )
+        .await
+        .expect("a bounded user-input wait must end at its own deadline");
+        assert!(
+            matches!(outcome, Err(ToolError::Timeout { .. })),
+            "expected the configured timeout, got {outcome:?}"
+        );
+    }
+
     async fn assert_required_fixture(source: ClaimSource, action: HostAction) {
         let tmp = tempfile::tempdir().expect("fixture directory");
         let full_access = matches!(action, HostAction::FullAccess);

crates/tui/src/runtime_api.rs

@@ -593,13 +593,14 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         // SSE journal frames carry their durable `seq` as the event id, and the
         // thread event stream resumes from `Last-Event-ID`.
         event_stream_resume: true,
-        // The terminal family is Unix-only in this build: the owner is
-        // `#[cfg(unix)]` end to end and the Windows routes answer 501. A
-        // client must be able to feature-detect that before it offers a pane.
-        terminal_stream: cfg!(unix),
-        terminal_input: cfg!(unix),
-        terminal_resize: cfg!(unix),
-        terminal_kill: cfg!(unix),
+        // The terminal family follows the routes' own gate: the owner is
+        // `#[cfg(unix)]` end to end, and the Windows and OpenHarmony builds
+        // answer 501. A client must be able to feature-detect that before it
+        // offers a pane, so the flag must never outrun the handler.
+        terminal_stream: cfg!(all(unix, not(target_env = "ohos"))),
+        terminal_input: cfg!(all(unix, not(target_env = "ohos"))),
+        terminal_resize: cfg!(all(unix, not(target_env = "ohos"))),
+        terminal_kill: cfg!(all(unix, not(target_env = "ohos"))),
     }
 }
 

crates/tui/src/runtime_api/terminal.rs

@@ -198,13 +198,30 @@ fn open_session(
         .ok_or_else(|| ApiError::not_found(format!("no live terminal session named '{name}'")))
 }
 
+/// Run one operation against the locked session on the blocking pool.
+///
+/// The session mutex and the PTY behind it are synchronous: the agent's own
+/// tool holds the lock across a whole command, and a write to a child that
+/// stopped reading blocks until the kernel buffer drains. Neither may park a
+/// runtime worker (#6149), so a route never touches the session inline.
 #[cfg(all(unix, not(target_env = "ohos")))]
-fn lock_session(
-    session: &terminal_session::SharedSession,
-) -> Result<std::sync::MutexGuard<'_, terminal_session::TerminalSession>, ApiError> {
-    session
-        .lock()
-        .map_err(|_| ApiError::internal("terminal session lock poisoned"))
+async fn with_session<T>(
+    session: terminal_session::SharedSession,
+    operation: impl FnOnce(&mut terminal_session::TerminalSession) -> Result<T, ApiError>
+    + Send
+    + 'static,
+) -> Result<T, ApiError>
+where
+    T: Send + 'static,
+{
+    tokio::task::spawn_blocking(move || {
+        let mut guard = session
+            .lock()
+            .map_err(|_| ApiError::internal("terminal session lock poisoned"))?;
+        operation(&mut guard)
+    })
+    .await
+    .map_err(|error| ApiError::internal(error.to_string()))?
 }
 
 /// `GET /v1/terminal/{name}/output` — the resumable byte stream.
@@ -221,10 +238,13 @@ pub(super) async fn terminal_output(
     let encoding = chunk_encoding(query.format.as_deref().unwrap_or("base64"))?;
     let max_bytes = bounded_max_bytes(query.max_bytes)?;
     let cursor = query.cursor.unwrap_or(0);
-    let mut guard = lock_session(&session)?;
-    let chunk = terminal_session::read_session_since(&guard, cursor, max_bytes)
-        .map_err(ApiError::internal)?;
-    let exit = terminal_session::session_exit_status(&mut guard).map_err(ApiError::internal)?;
+    let (chunk, exit) = with_session(session, move |session| {
+        let chunk = terminal_session::read_session_since(session, cursor, max_bytes)
+            .map_err(ApiError::internal)?;
+        let exit = terminal_session::session_exit_status(session).map_err(ApiError::internal)?;
+        Ok((chunk, exit))
+    })
+    .await?;
     let running = exit.is_none();
     Ok(Json(TerminalOutputResponse {
         name,
@@ -257,12 +277,12 @@ pub(super) async fn terminal_input(
         &request.data,
         request.encoding.as_deref().unwrap_or("base64"),
     )?;
-    let guard = lock_session(&session)?;
-    terminal_session::write_bytes(&guard, &bytes).map_err(ApiError::internal)?;
-    Ok(Json(TerminalWriteResponse {
-        name,
-        written: bytes.len(),
-    }))
+    let written = bytes.len();
+    with_session(session, move |session| {
+        terminal_session::write_bytes(session, &bytes).map_err(ApiError::internal)
+    })
+    .await?;
+    Ok(Json(TerminalWriteResponse { name, written }))
 }
 
 /// `POST /v1/terminal/{name}/resize` — the window the child should draw for.
@@ -275,8 +295,10 @@ pub(super) async fn terminal_resize(
     let session = open_session(&state, &name)?;
     let rows = bounded_dimension(request.rows, "rows")?;
     let cols = bounded_dimension(request.cols, "cols")?;
-    let guard = lock_session(&session)?;
-    terminal_session::resize_session(&guard, rows, cols).map_err(ApiError::internal)?;
+    with_session(session, move |session| {
+        terminal_session::resize_session(session, rows, cols).map_err(ApiError::internal)
+    })
+    .await?;
     Ok(Json(TerminalResizeResponse { name, rows, cols }))
 }
 
@@ -291,8 +313,10 @@ pub(super) async fn terminal_kill(
     Path(name): Path<String>,
 ) -> Result<Json<TerminalKillResponse>, ApiError> {
     let session = open_session(&state, &name)?;
-    let mut guard = lock_session(&session)?;
-    terminal_session::kill_session(&mut guard).map_err(ApiError::internal)?;
+    with_session(session, |session| {
+        terminal_session::kill_session(session).map_err(ApiError::internal)
+    })
+    .await?;
     Ok(Json(TerminalKillResponse { name, killed: true }))
 }
 

crates/tui/src/runtime_api/tests.rs

@@ -13897,10 +13897,10 @@ async fn runtime_info_advertises_terminal_capabilities() -> Result<()> {
         .json()
         .await?;
     // A GPUI client gates its terminal pane on these. They are true where the
-    // routes serve bytes and false where the owner is Unix-only — the flag
-    // must not claim a capability the build cannot serve, so assert the
-    // platform's truth rather than `true`.
-    let expected = cfg!(unix);
+    // routes serve bytes and false where they answer 501 (Windows, OpenHarmony)
+    // — the flag must not claim a capability the build cannot serve, so assert
+    // the routes' own gate rather than `true`.
+    let expected = cfg!(all(unix, not(target_env = "ohos")));
     for capability in [
         "terminal_stream",
         "terminal_input",

crates/tui/src/sleep_guard.rs

@@ -26,25 +26,35 @@
 //!
 //! Release is `Drop` and never cached: a leaked inhibitor would keep a laptop
 //! awake forever, which is worse than the problem this solves.
+//!
+//! The inhibitor is a `tokio::process` child, because the guard lives inside
+//! `Engine::run_turn` on the runtime: the spawn registers with the runtime,
+//! `kill_on_drop` sends the release signal, and the runtime reaps the child —
+//! no `wait` runs inline on a worker (#6149). `hold` therefore has to be
+//! called from within a Tokio runtime context.
 
 #[cfg(unix)]
-use std::process::Child;
+use tokio::process::Child;
 // Only the macOS and Linux inhibitors spawn anything; every other Unix
 // (Android, the BSDs, illumos) is a no-op and would see these as dead.
 #[cfg(any(target_os = "macos", target_os = "linux"))]
-use std::process::{Command, Stdio};
+use std::process::Stdio;
+#[cfg(any(target_os = "macos", target_os = "linux"))]
+use tokio::process::Command;
 
 /// An idle-sleep assertion held for as long as this value lives.
 pub struct SleepGuard {
     /// The platform inhibitor process, when one was started. `None` means the
     /// platform has no implementation, or the process could not be started —
     /// keeping the host awake is best-effort and must never fail a turn.
+    /// Dropping it is the release: the child is spawned with `kill_on_drop`.
     #[cfg(unix)]
     child: Option<Child>,
 }
 
 impl SleepGuard {
-    /// Hold the host awake until the returned guard drops.
+    /// Hold the host awake until the returned guard drops. Call it from the
+    /// Tokio runtime: the inhibitor is a `tokio::process` child.
     #[must_use]
     pub fn hold() -> Self {
         #[cfg(unix)]
@@ -63,20 +73,17 @@ impl SleepGuard {
     /// platform is a no-op or the process did not start.
     #[cfg(all(test, unix))]
     pub(crate) fn inhibitor_pid(&self) -> Option<u32> {
-        self.child.as_ref().map(Child::id)
+        self.child.as_ref().and_then(Child::id)
     }
 }
 
 #[cfg(unix)]
 impl Drop for SleepGuard {
     fn drop(&mut self) {
-        let Some(child) = self.child.as_mut() else {
-            return;
-        };
-        // Killing the inhibitor is what releases the assertion; reaping it
-        // keeps a zombie out of the process table.
-        let _ = child.kill();
-        let _ = child.wait();
+        // Releasing is dropping the child: `kill_on_drop` sends the signal
+        // here, synchronously, and the runtime reaps the process afterwards.
+        // Explicit so the field's purpose is code rather than a lint waiver.
+        drop(self.child.take());
     }
 }
 
@@ -116,6 +123,9 @@ fn spawn(program: &str, args: &[&str]) -> Option<Child> {
         .stdin(Stdio::null())
         .stdout(Stdio::null())
         .stderr(Stdio::null())
+        // Killing the inhibitor is what releases the assertion; the runtime
+        // reaps the child afterwards, so nothing here waits inline.
+        .kill_on_drop(true)
         .spawn()
         .ok()
 }
@@ -131,9 +141,24 @@ mod tests {
         unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
     }
 
-    #[test]
+    /// The kill is sent on drop and the runtime reaps the child on the next
+    /// `SIGCHLD`, so "gone" is a short poll rather than an instant fact. If
+    /// the pid were reused by a new process in that window the test would be
+    /// racing itself, which is why callers assert on the guard's own child.
+    async fn released(pid: u32) -> bool {
+        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
+        while alive(pid) {
+            if tokio::time::Instant::now() >= deadline {
+                return false;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+        true
+    }
+
+    #[tokio::test]
     #[cfg(any(target_os = "macos", target_os = "linux"))]
-    fn the_inhibitor_lives_exactly_as_long_as_the_guard() {
+    async fn the_inhibitor_lives_exactly_as_long_as_the_guard() {
         let guard = SleepGuard::hold();
         let pid = guard
             .inhibitor_pid()
@@ -142,22 +167,15 @@ mod tests {
 
         drop(guard);
 
-        // Reaping is synchronous in `Drop`, so the pid is gone immediately —
-        // and if it were reused by a new process in this window the test would
-        // be racing itself, which is why we assert on the guard's own child.
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
-        while alive(pid) {
-            assert!(
-                std::time::Instant::now() < deadline,
-                "a released guard must not leave an inhibitor keeping the host awake"
-            );
-            std::thread::sleep(std::time::Duration::from_millis(10));
-        }
+        assert!(
+            released(pid).await,
+            "a released guard must not leave an inhibitor keeping the host awake"
+        );
     }
 
-    #[test]
+    #[tokio::test]
     #[cfg(any(target_os = "macos", target_os = "linux"))]
-    fn holding_twice_holds_two_independent_inhibitors() {
+    async fn holding_twice_holds_two_independent_inhibitors() {
         // Turns are serialized, but nothing here should assume it: two guards
         // must not share one process, or the first drop would release both.
         let first = SleepGuard::hold();
@@ -168,7 +186,7 @@ mod tests {
         );
         assert_ne!(a, b, "each guard owns its own inhibitor process");
         drop(first);
-        assert!(!alive(a), "the first guard released only its own");
+        assert!(released(a).await, "the first guard released only its own");
         assert!(alive(b), "the second guard still holds the host awake");
     }
 }

crates/tui/src/tools/terminal_session.rs

@@ -170,10 +170,14 @@ impl OutputBuffer {
     /// is idempotent. Does not advance `read_cursor` — the consuming
     /// tool-result path is untouched. This is the cursor arithmetic only;
     /// response-size policy belongs to the caller.
+    ///
+    /// A cursor past the head clamps to `total`: nothing is there yet, and
+    /// echoing the future cursor back as `next_cursor` would make every byte
+    /// produced before the stream reached it invisible to that client.
     fn read_since(&self, cursor: u64, max_bytes: usize) -> OutputChunk {
         let dropped = self.oldest();
         let gap = cursor < dropped;
-        let start = cursor.max(dropped);
+        let start = cursor.max(dropped).min(self.total);
         let skip = usize::try_from(start - dropped).unwrap_or(usize::MAX);
         let take = max_bytes.min(self.bytes.len().saturating_sub(skip));
         let bytes = self.bytes.iter().skip(skip).take(take).copied().collect();
@@ -1391,6 +1395,15 @@ mod tests {
         assert!(!current.gap);
         assert!(current.bytes.is_empty());
         assert_eq!(current.next_cursor, BUFFER_LIMIT as u64 + 32);
+
+        // A cursor past the head clamps to it instead of being echoed back:
+        // a client that continues from `next_cursor` must not skip the bytes
+        // the stream produces before it reaches the bogus position.
+        let beyond = wrapped.read_since(BUFFER_LIMIT as u64 + 4096, 8);
+        assert!(!beyond.gap);
+        assert!(beyond.bytes.is_empty());
+        assert_eq!(beyond.offset, BUFFER_LIMIT as u64 + 32);
+        assert_eq!(beyond.next_cursor, BUFFER_LIMIT as u64 + 32);
     }
 
     /// The session-level entry point an Engine byte stream will call: absolute
@@ -1417,12 +1430,13 @@ mod tests {
         assert_eq!(again.bytes, printed.bytes, "a replay read must not consume");
 
         // A cursor ahead of the stream is not a gap: nothing was lost, there
-        // is simply nothing there yet.
+        // is simply nothing there yet — and the answer clamps to the head so
+        // continuing from it cannot skip what arrives next.
         let ahead =
             read_session_since(&session.lock().unwrap(), printed.next_cursor + 4096, 16).unwrap();
         assert!(!ahead.gap);
         assert!(ahead.bytes.is_empty());
-        assert_eq!(ahead.next_cursor, printed.next_cursor + 4096);
+        assert_eq!(ahead.next_cursor, ahead.total);
 
         // More than one response's worth of output proves the clamp.
         let large = fresh("test-read-since-clamp");