Merge PR #6361 (work/0.9.14-core-unblocks) into the 0.10.0 readiness branch

Brings the owner's Core unblocks in as themselves: the Engine terminal
byte-stream routes (output/input/resize/kill, Unix-only, 501 elsewhere and
on ohos), Last-Event-ID stream resume plus idempotent-replay reporting on
turn submission, the pet agent-count contract pin, the turn-scoped host
idle-sleep guard, and the #6184 approval/user-input wait heartbeat.

Conflicts (3, all generated or measured files) resolved to main's newer
versions, which the branch's own commits never intended to change:
docs/public-surface-facts.json and web/lib/facts.generated.ts (0.10.0
capture facts), and scripts/runtime-contract-budget.json (main's 2026-09-19
Linux lock-in already carries the agent cwd / execute_tools ceilings the
branch measured; every metric value is identical, only the _comment
history differed).

Gates on the merged tree: cargo fmt --check clean; blocking-call budget
603 sites within budget; dead-code budget 174 at budget; OHOS dependency
graph OK. Rust tests for the merged code run in the next verification
pass on this branch.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_0134iUMxmGuXiG1LzPgfZVnv

crates/protocol/src/runtime/mod.rs

@@ -118,6 +118,24 @@ pub struct RuntimeCapabilities {
     /// Durable, workspace-scoped cross-task Agent Mail endpoints and events.
     #[serde(default)]
     pub agent_mail: bool,
+    /// `GET /v1/terminal/{name}/output` — the resumable byte stream over a
+    /// persistent Engine-owned terminal session, with absolute cursors.
+    #[serde(default)]
+    pub terminal_stream: bool,
+    /// `POST /v1/terminal/{name}/input` — bytes into the live session.
+    #[serde(default)]
+    pub terminal_input: bool,
+    /// `POST /v1/terminal/{name}/resize` — the window the child draws for.
+    #[serde(default)]
+    pub terminal_resize: bool,
+    /// `POST /v1/terminal/{name}/kill` — end the live session.
+    #[serde(default)]
+    pub terminal_kill: bool,
+    /// `GET /v1/threads/{id}/events` puts the durable `seq` on every journal
+    /// frame as the SSE `id:` and resumes from a `Last-Event-ID` header, so a
+    /// browser `EventSource` reconnects without a cursor in the query string.
+    #[serde(default)]
+    pub event_stream_resume: bool,
 }
 
 /// Experimental opt-in flags advertised by `GET /v1/runtime/info`.
@@ -425,6 +443,11 @@ mod tests {
             skill_lifecycle: false,
             plugin_management: false,
             agent_mail: true,
+            terminal_stream: false,
+            terminal_input: false,
+            terminal_resize: false,
+            terminal_kill: false,
+            event_stream_resume: true,
         };
         let value = serde_json::to_value(&caps).unwrap();
         let obj = value.as_object().unwrap();

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

@@ -14,6 +14,27 @@ use crate::tools::user_input::{UserInputRequest, UserInputResponse};
 
 const USER_INPUT_TIMEOUT: Duration = Duration::from_secs(300);
 
+/// How often a parked wait says it is still parked.
+///
+/// A wait with no deadline and no periodic line is indistinguishable from a
+/// freeze (#6184): the approval card may never expire (only a top-of-stack view
+/// ticks), the turn wall clock is paused across this wait, and nothing else
+/// reports. This is the line that gives a stall a name. Tests drive it at a
+/// tiny interval so the real path can be observed without waiting a minute.
+#[cfg(not(test))]
+const WAIT_HEARTBEAT: Duration = Duration::from_secs(60);
+#[cfg(test)]
+const WAIT_HEARTBEAT: Duration = Duration::from_millis(50);
+
+/// The announcement a parked wait makes, in one place so the log line and the
+/// status event cannot drift apart.
+fn wait_announcement(what: &str, tool_id: &str, waited: Duration) -> String {
+    format!(
+        "Still waiting for {what} on `{tool_id}` after {}s — the turn is parked here until it is answered",
+        waited.as_secs()
+    )
+}
+
 use super::Engine;
 
 #[derive(Debug, Clone)]
@@ -166,8 +187,26 @@ impl Engine {
         &mut self,
         tool_id: &str,
     ) -> Result<ApprovalResult, ToolError> {
+        let started = std::time::Instant::now();
+        let mut heartbeat = tokio::time::interval(WAIT_HEARTBEAT);
+        heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        // The first tick completes immediately; consume it so the first
+        // announcement is a heartbeat later, not at the gate itself.
+        heartbeat.tick().await;
+        let mut announced = false;
         loop {
             tokio::select! {
+                _ = heartbeat.tick() => {
+                    let waited = started.elapsed();
+                    let message = wait_announcement("tool approval", tool_id, waited);
+                    // Log every heartbeat; tell the user once, so a long park
+                    // leaves a trail without filling the transcript.
+                    tracing::warn!(tool_id, waited_secs = waited.as_secs(), "{message}");
+                    if !announced {
+                        announced = true;
+                        let _ = self.tx_event.send(Event::Status { message }).await;
+                    }
+                }
                 _ = self.cancel_token.cancelled() => {
                     let suffix = self.cancel_reason_suffix();
                     self.commit_approval_outcome(tool_id, ApprovalOutcome::Cancelled).await?;
@@ -233,8 +272,24 @@ impl Engine {
         // #6003: `[tools] user_input_timeout_seconds` — absent uses the
         // 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();
+        let mut heartbeat = tokio::time::interval(WAIT_HEARTBEAT);
+        heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        heartbeat.tick().await;
+        let mut announced = false;
         loop {
             tokio::select! {
+                _ = heartbeat.tick() => {
+                    // An indefinite wait (`user_input_timeout_seconds = 0`) is
+                    // the case that needs this most: nothing else bounds it.
+                    let waited = started.elapsed();
+                    let message = wait_announcement("user input", tool_id, waited);
+                    tracing::warn!(tool_id, waited_secs = waited.as_secs(), "{message}");
+                    if !announced {
+                        announced = true;
+                        let _ = self.tx_event.send(Event::Status { message }).await;
+                    }
+                }
                 _ = self.cancel_token.cancelled() => {
                     let suffix = self.cancel_reason_suffix();
                     return Err(ToolError::cancelled(
@@ -424,6 +479,87 @@ mod tests {
         .expect("required approval event deadline")
     }
 
+    /// #6184: a turn parked on an approval must say so. Before this the wait
+    /// had no engine-side deadline, no periodic line and no event, so a stalled
+    /// turn was indistinguishable from a working one until the user gave up.
+    #[tokio::test]
+    async fn a_parked_approval_announces_the_wait_instead_of_hanging_silently() {
+        let tmp = tempfile::tempdir().expect("fixture directory");
+        let mock = Arc::new(MockLlmClient::new(vec![counter_request(
+            false,
+            CURRENT_CALL,
+        )]));
+        let (mut engine, handle) = Engine::new_with_model_client(
+            EngineConfig {
+                workspace: tmp.path().to_path_buf(),
+                snapshots_enabled: false,
+                subagents_enabled: false,
+                terminal_chrome_enabled: false,
+                ..EngineConfig::default()
+            },
+            &Config::default(),
+            mock.clone(),
+        );
+        engine.session.approval_mode = ApprovalMode::Suggest;
+        engine.session.add_message(Message {
+            role: Role::User,
+            content: vec![ContentBlock::Text {
+                text: "Park on the approval gate.".into(),
+                cache_control: None,
+            }],
+        });
+        let mut registry = crate::tools::ToolRegistry::new(ToolContext::new(tmp.path()));
+        registry.register(Arc::new(ApprovalFixtureTool {
+            executions: Arc::new(AtomicUsize::new(0)),
+            claim_only: false,
+        }));
+        let catalog = registry.to_api_tools_with_cache(true);
+        let surface = ToolSurfacePolicy::new(
+            registry,
+            Some(catalog),
+            AppMode::Agent,
+            &engine.config.tools_always_load,
+            &[],
+            false,
+            None,
+            None,
+            Some(4),
+            engine.session.approval_mode,
+            crate::core::engine::tool_catalog::ToolMode::Direct,
+        );
+
+        let events = handle.rx_event.clone();
+        let task = tokio::spawn(async move {
+            engine
+                .run_turn(&mut TurnContext::new(8), surface, None, None)
+                .await
+        });
+
+        // Reach the gate and answer nothing: this is the park.
+        let _ = wait_for_fixture_approval(&events, CURRENT_CALL).await;
+
+        let announced = tokio::time::timeout(Duration::from_secs(5), async {
+            let mut rx = events.write().await;
+            while let Some(event) = rx.recv().await {
+                if let Event::Status { message } = &event
+                    && message.contains("Still waiting for tool approval")
+                    && message.contains(CURRENT_CALL)
+                {
+                    return true;
+                }
+            }
+            false
+        })
+        .await
+        .expect("a parked approval must announce itself before anything else happens");
+        assert!(
+            announced,
+            "the announcement must name the wait and the tool it waits on"
+        );
+
+        task.abort();
+    }
+
     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/core/engine/turn_loop.rs

@@ -688,6 +688,15 @@ impl Engine {
 
         // Only interactive TUI hosts own terminal chrome. Headless exec,
         // app-server, and stream-json stdout must remain byte-clean.
+        //
+        // The sleep guard rides the same gate: a turn that outlives the host's
+        // idle timer is lost work, and an interactive host is the only one
+        // that owns a human's machine. Bound to this function, so it releases
+        // on every return path. See `crate::sleep_guard` for its limits.
+        let _sleep_guard = self
+            .config
+            .terminal_chrome_enabled
+            .then(crate::sleep_guard::SleepGuard::hold);
         if self.config.terminal_chrome_enabled {
             crate::tui::notifications::set_taskbar_progress_busy();
             crate::tui::notifications::start_title_animation("codewhale");

crates/tui/src/lib.rs

@@ -132,6 +132,7 @@ mod settings;
 mod shell_dispatcher;
 mod skill_state;
 mod skills;
+mod sleep_guard;
 mod snapshot;
 mod startup_trace;
 mod task_manager;

crates/tui/src/runtime_api.rs

@@ -12,7 +12,7 @@ use anyhow::{Context, Result, anyhow, bail};
 use async_stream::stream;
 use axum::extract::{ConnectInfo, DefaultBodyLimit, Path, Query, Request, State};
 use axum::http::header;
-use axum::http::{HeaderName, HeaderValue, Method, StatusCode};
+use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
 use axum::middleware;
 use axum::response::Html;
 use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
@@ -109,6 +109,7 @@ mod plugins;
 mod secrets;
 mod sessions;
 mod targets;
+mod terminal;
 mod voice;
 mod web;
 mod workspace;
@@ -589,6 +590,16 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         skill_lifecycle: true,
         plugin_management: true,
         agent_mail: true,
+        // 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),
     }
 }
 
@@ -872,6 +883,16 @@ struct FleetEventsQuery {
 struct StartTurnResponse {
     thread: ThreadRecord,
     turn: TurnRecord,
+    /// Present only when the durable `operation_key` made this submission a
+    /// replay of one already accepted: the turn is the original and nothing
+    /// new was admitted. Omitted otherwise so every existing response stays
+    /// byte-identical — a client that never sends a key sees no change.
+    #[serde(skip_serializing_if = "replay_flag_is_absent")]
+    idempotent_replay: bool,
+}
+
+fn replay_flag_is_absent(replayed: &bool) -> bool {
+    !*replayed
 }
 
 fn install_runtime_server_workshop_budgets(
@@ -1153,6 +1174,15 @@ pub fn build_router(state: RuntimeApiState) -> Router {
             get(read_session_artifact),
         )
         .route("/v1/workspace/status", get(workspace_status))
+        // The Engine's terminal byte stream (#34). Auth is the route layer's,
+        // not this module's; these never create a session — see terminal.rs.
+        .route("/v1/terminal/{name}/output", get(terminal::terminal_output))
+        .route("/v1/terminal/{name}/input", post(terminal::terminal_input))
+        .route(
+            "/v1/terminal/{name}/resize",
+            post(terminal::terminal_resize),
+        )
+        .route("/v1/terminal/{name}/kill", post(terminal::terminal_kill))
         .route("/v1/workspace/files/search", get(workspace_file_search))
         .route(
             "/v1/workspace/files",
@@ -5676,9 +5706,9 @@ async fn start_thread_turn(
     Path(id): Path<String>,
     Json(req): Json<StartTurnRequest>,
 ) -> Result<(StatusCode, Json<StartTurnResponse>), ApiError> {
-    let turn = state
+    let (turn, replayed) = state
         .runtime_threads
-        .start_turn(&id, req)
+        .start_turn_reporting_replay(&id, req)
         .await
         .map_err(map_thread_err)?;
     let thread = state
@@ -5686,9 +5716,22 @@ async fn start_thread_turn(
         .get_thread(&id)
         .await
         .map_err(map_thread_err)?;
+    // A replay acknowledges work already accepted rather than admitting new
+    // work: 200 tells the client "this is the turn I already started", which
+    // is what lets an ambiguous submit resolve without duplicate messages or
+    // tools. A fresh admission stays 201.
+    let status = if replayed {
+        StatusCode::OK
+    } else {
+        StatusCode::CREATED
+    };
     Ok((
-        StatusCode::CREATED,
-        Json(StartTurnResponse { thread, turn }),
+        status,
+        Json(StartTurnResponse {
+            thread,
+            turn,
+            idempotent_replay: replayed,
+        }),
     ))
 }
 
@@ -5871,7 +5914,11 @@ async fn compact_thread(
         .map_err(map_thread_err)?;
     Ok((
         StatusCode::ACCEPTED,
-        Json(StartTurnResponse { thread, turn }),
+        Json(StartTurnResponse {
+            thread,
+            turn,
+            idempotent_replay: false,
+        }),
     ))
 }
 
@@ -6152,6 +6199,7 @@ async fn stream_thread_events(
     State(state): State<RuntimeApiState>,
     Path(id): Path<String>,
     Query(query): Query<ThreadEventsQuery>,
+    headers: HeaderMap,
 ) -> Result<Response, ApiError> {
     let _ = state
         .runtime_threads
@@ -6159,6 +6207,14 @@ async fn stream_thread_events(
         .await
         .map_err(map_thread_err)?;
 
+    // Two clients, two cursors. A browser `EventSource` can only replay through
+    // the `Last-Event-ID` header it sets on reconnect (the ids now ride the
+    // journal frames below); every other client passes `since_seq`. An explicit
+    // query cursor wins over the header, so a deliberate replay-from-zero is
+    // never silently overridden by a stale header — the header is the fallback
+    // when no cursor was asked for.
+    let since_seq = query.since_seq.or_else(|| last_event_id(&headers));
+
     // Subscribe before reading durable history. An event emitted while replay
     // is loaded is then present in both places (and deduped below) or queued
     // live, never in an uncovered handoff window.
@@ -6173,7 +6229,7 @@ async fn stream_thread_events(
     }
     let replay = state
         .runtime_threads
-        .replay_events(&id, query.since_seq, query.replay_limit)
+        .replay_events(&id, since_seq, query.replay_limit)
         .await
         .map_err(|e| ApiError::internal(e.to_string()))?;
 
@@ -6246,7 +6302,8 @@ fn replay_live_thread_events(
                 yield Ok(sse_json(
                     &event_name,
                     runtime_event_payload_with_previous(event, previous_seq),
-                ));
+                )
+                .id(last_seq.to_string()));
             }
         }
 
@@ -6280,7 +6337,8 @@ fn replay_live_thread_events(
                     yield Ok(sse_json(
                         &event_name,
                         runtime_event_payload_with_previous(event, previous_seq),
-                    ));
+                    )
+                    .id(last_seq.to_string()));
                 }
                 Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                     if progress {
@@ -6330,7 +6388,8 @@ fn replay_live_thread_events(
                             yield Ok(sse_json(
                                 &event_name,
                                 runtime_event_payload_with_previous(event, previous_seq),
-                            ));
+                            )
+                            .id(last_seq.to_string()));
                         }
                     }
                 }
@@ -6866,6 +6925,19 @@ fn sse_json(event: &str, payload: serde_json::Value) -> SseEvent {
     SseEvent::default().event(event).data(data)
 }
 
+/// Read a `Last-Event-ID` cursor off the request.
+///
+/// Only a decimal sequence number is ours. Anything else is ignored rather
+/// than rejected: an opaque id from a proxy or an older client should start
+/// the stream from the durable head, not fail to open it — a refused stream
+/// looks like an outage to a reconnecting client.
+fn last_event_id(headers: &HeaderMap) -> Option<u64> {
+    headers
+        .get("last-event-id")
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.trim().parse::<u64>().ok())
+}
+
 fn truncate_text(text: &str, max_chars: usize) -> String {
     let char_count = text.chars().count();
     if char_count <= max_chars {

crates/tui/src/runtime_api/terminal.rs

@@ -0,0 +1,394 @@
+//! `/v1/terminal/{name}` — the Engine's terminal byte stream.
+//!
+//! The owner is [`crate::tools::terminal_session`]: the same PTY-backed shell
+//! the agent's terminal tools drive, so a client attaching here sees the
+//! session the model is already working in rather than a second shell beside
+//! it. These routes never create a session — a name with no live session is a
+//! 404, because conjuring a shell from an HTTP request would give the app a
+//! terminal the Engine does not know about.
+//!
+//! Authentication is the `/v1` route layer's bearer token (see
+//! [`super::auth`]); nothing here re-implements or bypasses it.
+//!
+//! Wire shape deliberately follows `/v1/threads/{id}/jobs/{job_id}/output`:
+//! `cursor` / `max_bytes` / `format` in, `offset` / `next_cursor` / `total` /
+//! `dropped` out. Two byte streams in one product should not speak two
+//! dialects.
+//!
+//! Known limitations, recorded because a reader will otherwise assume them:
+//!
+//! - **No long poll.** `wait_ms` is not accepted; a client polls the cursor.
+//!   The jobs route can block because a job owns a notification; a terminal
+//!   session's ring has no wake-up channel yet, and inventing one here would
+//!   be a second mechanism rather than a reuse.
+//! - **No scrollback recovery.** `dropped` reports what the 512 KiB ring
+//!   discarded; those bytes are gone with the process, not on disk.
+//! - **Live sessions only.** Persistence is identity and lifecycle, never
+//!   output, so a restarted Engine reports no session rather than pretending to
+//!   reattach (#34 acceptance: "Restart truthfully reports lost live PTYs").
+//! - **Unix only.** The owner is `#[cfg(all(unix, not(target_env = "ohos")))]` end to end; on Windows these
+//!   routes do not exist yet. ConPTY qualification is its own slice.
+
+use axum::Json;
+use axum::extract::{Path, Query, State};
+use base64::Engine as _;
+use serde::{Deserialize, Serialize};
+
+// The owner does not exist on ohos (`tools/mod.rs`), so neither does any
+// handler that drives it; the stubs below answer there instead.
+#[cfg(all(unix, not(target_env = "ohos")))]
+use crate::tools::terminal_session;
+
+use super::{ApiError, RuntimeApiState};
+
+/// Default per-response ceiling; the owner clamps to its own `READ_LIMIT`.
+const TERMINAL_CHUNK_DEFAULT: usize = 64 * 1024;
+/// Session names come from the agent's tools; this only bounds the echo.
+const TERMINAL_NAME_MAX_BYTES: usize = 128;
+/// One input frame. Interactive typing is bytes, not uploads.
+const TERMINAL_INPUT_MAX_BYTES: usize = 64 * 1024;
+const TERMINAL_DIMENSION_MAX: u16 = 1000;
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct TerminalOutputQuery {
+    /// Absolute byte offset into the session's lifetime output.
+    #[serde(default)]
+    cursor: Option<u64>,
+    /// Per-response byte ceiling, default 64 KiB, clamped by the owner.
+    #[serde(default)]
+    max_bytes: Option<usize>,
+    /// `base64` (default, exact bytes) or `text` (lossy UTF-8).
+    #[serde(default)]
+    format: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+pub(super) struct TerminalOutputResponse {
+    name: String,
+    /// Absolute offset of `data[0]`; exceeds `cursor` when the ring already
+    /// discarded that prefix (`dropped` reports the cutoff).
+    offset: u64,
+    /// Pass back as `cursor` to continue.
+    next_cursor: u64,
+    /// Everything the session has produced, including discarded bytes.
+    total: u64,
+    /// Leading bytes the bounded ring permanently discarded.
+    dropped: u64,
+    encoding: &'static str,
+    data: String,
+    /// False once the shell has exited and no bytes remain past `next_cursor`.
+    running: bool,
+    exit_code: Option<i64>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct TerminalInputRequest {
+    data: String,
+    /// `base64` (default, exact bytes) or `text`.
+    #[serde(default)]
+    encoding: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+pub(super) struct TerminalWriteResponse {
+    name: String,
+    written: usize,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct TerminalResizeRequest {
+    rows: u16,
+    cols: u16,
+}
+
+#[derive(Debug, Serialize)]
+pub(super) struct TerminalResizeResponse {
+    name: String,
+    rows: u16,
+    cols: u16,
+}
+
+#[derive(Debug, Serialize)]
+pub(super) struct TerminalKillResponse {
+    name: String,
+    killed: bool,
+}
+
+/// `base64` keeps bytes exact; `text` is the lossy convenience form.
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn chunk_encoding(format: &str) -> Result<&'static str, ApiError> {
+    match format {
+        "base64" => Ok("base64"),
+        "text" => Ok("text"),
+        _ => Err(ApiError::bad_request("format must be base64 or text")),
+    }
+}
+
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn encode_bytes(bytes: &[u8], encoding: &str) -> String {
+    if encoding == "base64" {
+        base64::engine::general_purpose::STANDARD.encode(bytes)
+    } else {
+        String::from_utf8_lossy(bytes).into_owned()
+    }
+}
+
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn decode_bytes(data: &str, encoding: &str) -> Result<Vec<u8>, ApiError> {
+    let bytes = match encoding {
+        "base64" => base64::engine::general_purpose::STANDARD
+            .decode(data)
+            .map_err(|_| ApiError::bad_request("data is not valid base64"))?,
+        "text" => data.as_bytes().to_vec(),
+        _ => return Err(ApiError::bad_request("encoding must be base64 or text")),
+    };
+    if bytes.len() > TERMINAL_INPUT_MAX_BYTES {
+        return Err(ApiError::bad_request(format!(
+            "input exceeds {TERMINAL_INPUT_MAX_BYTES} bytes"
+        )));
+    }
+    Ok(bytes)
+}
+
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn bounded_max_bytes(requested: Option<usize>) -> Result<usize, ApiError> {
+    let max_bytes = requested.unwrap_or(TERMINAL_CHUNK_DEFAULT);
+    if !(1..=terminal_session::READ_LIMIT).contains(&max_bytes) {
+        return Err(ApiError::bad_request(format!(
+            "max_bytes must be between 1 and {}",
+            terminal_session::READ_LIMIT
+        )));
+    }
+    Ok(max_bytes)
+}
+
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn bounded_dimension(value: u16, field: &str) -> Result<u16, ApiError> {
+    if !(1..=TERMINAL_DIMENSION_MAX).contains(&value) {
+        return Err(ApiError::bad_request(format!(
+            "{field} must be between 1 and {TERMINAL_DIMENSION_MAX}"
+        )));
+    }
+    Ok(value)
+}
+
+/// Resolve a live session or 404. Never creates one — see the module docs.
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn open_session(
+    state: &RuntimeApiState,
+    name: &str,
+) -> Result<terminal_session::SharedSession, ApiError> {
+    if name.is_empty() || name.len() > TERMINAL_NAME_MAX_BYTES {
+        return Err(ApiError::not_found("terminal session not found"));
+    }
+    terminal_session::lookup(name, &state.workspace)
+        .ok_or_else(|| ApiError::not_found(format!("no live terminal session named '{name}'")))
+}
+
+#[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"))
+}
+
+/// `GET /v1/terminal/{name}/output` — the resumable byte stream.
+///
+/// Reads are non-consuming: several clients may hold independent cursors, and
+/// polling here never steals output from the agent's own consuming read.
+#[cfg(all(unix, not(target_env = "ohos")))]
+pub(super) async fn terminal_output(
+    State(state): State<RuntimeApiState>,
+    Path(name): Path<String>,
+    Query(query): Query<TerminalOutputQuery>,
+) -> Result<Json<TerminalOutputResponse>, ApiError> {
+    let session = open_session(&state, &name)?;
+    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 running = exit.is_none();
+    Ok(Json(TerminalOutputResponse {
+        name,
+        offset: chunk.offset,
+        next_cursor: chunk.next_cursor,
+        total: chunk.total,
+        dropped: chunk.dropped,
+        encoding,
+        data: encode_bytes(&chunk.bytes, encoding),
+        // A gap means bytes were lost; `running` alone must not imply there is
+        // nothing behind us, so drain state is reported independently.
+        running,
+        exit_code: exit.map(|status| i64::from(status.exit_code())),
+    }))
+}
+
+/// `POST /v1/terminal/{name}/input` — bytes into the live shell.
+///
+/// Input attribution is the caller's: this route is the client's writer, and
+/// the agent's writer is `terminal_send`. Nothing here re-labels one as the
+/// other.
+#[cfg(all(unix, not(target_env = "ohos")))]
+pub(super) async fn terminal_input(
+    State(state): State<RuntimeApiState>,
+    Path(name): Path<String>,
+    Json(request): Json<TerminalInputRequest>,
+) -> Result<Json<TerminalWriteResponse>, ApiError> {
+    let session = open_session(&state, &name)?;
+    let bytes = decode_bytes(
+        &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(),
+    }))
+}
+
+/// `POST /v1/terminal/{name}/resize` — the window the child should draw for.
+#[cfg(all(unix, not(target_env = "ohos")))]
+pub(super) async fn terminal_resize(
+    State(state): State<RuntimeApiState>,
+    Path(name): Path<String>,
+    Json(request): Json<TerminalResizeRequest>,
+) -> Result<Json<TerminalResizeResponse>, ApiError> {
+    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)?;
+    Ok(Json(TerminalResizeResponse { name, rows, cols }))
+}
+
+/// `POST /v1/terminal/{name}/kill` — end the shell.
+///
+/// The exit itself is observed through `output` (`running` / `exit_code`),
+/// so a client that kills and then polls learns the truth instead of an
+/// optimistic acknowledgement.
+#[cfg(all(unix, not(target_env = "ohos")))]
+pub(super) async fn terminal_kill(
+    State(state): State<RuntimeApiState>,
+    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)?;
+    Ok(Json(TerminalKillResponse { name, killed: true }))
+}
+
+/// Windows build: the owner is `#[cfg(all(unix, not(target_env = "ohos")))]` end to end, so the contract
+/// exists but cannot be served. These answer 501 rather than 404 so a client
+/// can tell "this Engine build cannot do terminals" apart from "that session
+/// is gone" — and so the ConPTY slice has one place to replace.
+#[cfg(any(not(unix), target_env = "ohos"))]
+mod platform {
+    use super::*;
+
+    fn unsupported() -> ApiError {
+        ApiError::not_implemented(
+            "terminal sessions are Unix-only in this build; native Windows PTY support is not implemented yet",
+        )
+    }
+
+    pub(super) async fn terminal_output(
+        State(_): State<RuntimeApiState>,
+        Path(_): Path<String>,
+        Query(_): Query<TerminalOutputQuery>,
+    ) -> Result<Json<TerminalOutputResponse>, ApiError> {
+        Err(unsupported())
+    }
+
+    pub(super) async fn terminal_input(
+        State(_): State<RuntimeApiState>,
+        Path(_): Path<String>,
+        Json(_): Json<TerminalInputRequest>,
+    ) -> Result<Json<TerminalWriteResponse>, ApiError> {
+        Err(unsupported())
+    }
+
+    pub(super) async fn terminal_resize(
+        State(_): State<RuntimeApiState>,
+        Path(_): Path<String>,
+        Json(_): Json<TerminalResizeRequest>,
+    ) -> Result<Json<TerminalResizeResponse>, ApiError> {
+        Err(unsupported())
+    }
+
+    pub(super) async fn terminal_kill(
+        State(_): State<RuntimeApiState>,
+        Path(_): Path<String>,
+    ) -> Result<Json<TerminalKillResponse>, ApiError> {
+        Err(unsupported())
+    }
+}
+
+#[cfg(any(not(unix), target_env = "ohos"))]
+pub(super) use platform::{terminal_input, terminal_kill, terminal_output, terminal_resize};
+
+#[cfg(all(test, unix, not(target_env = "ohos")))]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn encodings_round_trip_exact_bytes_and_stay_lossy_only_on_request() {
+        // Non-UTF-8 bytes survive base64 and are the reason it is the default.
+        let raw = [0xf0, 0x9f, 0x90, 0x8b, 0x00, 0xff];
+        let encoded = encode_bytes(&raw, "base64");
+        assert_eq!(decode_bytes(&encoded, "base64").unwrap(), raw);
+        // The lossy form is explicit and cannot be mistaken for fidelity.
+        let text = encode_bytes(&raw, "text");
+        assert!(text.contains('\u{fffd}'));
+        assert_eq!(decode_bytes(&text, "text").unwrap(), text.as_bytes());
+    }
+
+    #[test]
+    fn encoding_names_are_closed_sets() {
+        for good in ["base64", "text"] {
+            assert_eq!(chunk_encoding(good).unwrap(), good);
+        }
+        for bad in ["utf8", "raw", "Base64", ""] {
+            assert!(chunk_encoding(bad).is_err(), "{bad} must not be accepted");
+            assert!(decode_bytes("", bad).is_err(), "{bad} must not decode");
+        }
+        // A base64 decoder that ignores padding would accept junk bytes.
+        assert!(decode_bytes("not base64!!", "base64").is_err());
+    }
+
+    #[test]
+    fn chunk_and_dimension_bounds_reject_the_edges() {
+        assert_eq!(bounded_max_bytes(None).unwrap(), TERMINAL_CHUNK_DEFAULT);
+        assert_eq!(
+            bounded_max_bytes(Some(terminal_session::READ_LIMIT)).unwrap(),
+            terminal_session::READ_LIMIT
+        );
+        assert!(bounded_max_bytes(Some(0)).is_err());
+        assert!(bounded_max_bytes(Some(terminal_session::READ_LIMIT + 1)).is_err());
+        assert_eq!(bounded_dimension(24, "rows").unwrap(), 24);
+        assert!(bounded_dimension(0, "rows").is_err());
+        assert!(bounded_dimension(TERMINAL_DIMENSION_MAX + 1, "cols").is_err());
+    }
+
+    #[test]
+    fn input_is_bounded_before_it_reaches_the_pty() {
+        let too_much =
+            base64::engine::general_purpose::STANDARD
+                .encode(vec![b'a'; TERMINAL_INPUT_MAX_BYTES + 1]);
+        assert!(decode_bytes(&too_much, "base64").is_err());
+        let at_limit =
+            base64::engine::general_purpose::STANDARD.encode(vec![b'a'; TERMINAL_INPUT_MAX_BYTES]);
+        assert_eq!(
+            decode_bytes(&at_limit, "base64").unwrap().len(),
+            TERMINAL_INPUT_MAX_BYTES
+        );
+    }
+}

crates/tui/src/runtime_api/tests.rs

@@ -3766,9 +3766,19 @@ async fn turn_endpoint_operation_key_returns_original_and_conflicts_on_mismatch(
     assert!(!serde_json::to_string(&first)?.contains("cwc-http-operation-1"));
 
     let replay_response = client.post(&url).json(&request).send().await?;
-    assert_eq!(replay_response.status(), StatusCode::CREATED);
+    // A replay acknowledges work already accepted rather than admitting new
+    // work: 200 plus an explicit flag, so a client that retried an ambiguous
+    // submit can tell it is looking at the turn it already started (#76).
+    assert_eq!(replay_response.status(), StatusCode::OK);
     let replay: serde_json::Value = replay_response.json().await?;
     assert_eq!(replay["turn"]["id"], first_turn_id);
+    assert_eq!(replay["idempotent_replay"], true);
+    // A fresh admission carries no flag, so the response every existing client
+    // already parses is byte-identical to before.
+    assert!(
+        first.get("idempotent_replay").is_none(),
+        "only a replay is marked as one: {first}"
+    );
 
     let mismatch = client
         .post(&url)
@@ -4151,6 +4161,114 @@ async fn events_endpoint_respects_since_seq_cursor() -> Result<()> {
     Ok(())
 }
 
+/// The SSE `id:` a browser `EventSource` resumes from, and the `Last-Event-ID`
+/// header it replays with, against the same durable cursor `since_seq` uses.
+#[tokio::test]
+async fn thread_event_frames_carry_the_id_a_reconnect_resumes_from() -> Result<()> {
+    let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+    let thread = runtime_threads
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+
+    // Every journal frame carries its durable seq as the SSE id.
+    let first = client
+        .get(format!(
+            "http://{addr}/v1/threads/{}/events?since_seq=0",
+            thread.id
+        ))
+        .send()
+        .await?
+        .error_for_status()?;
+    let frame = read_first_sse_frame(first).await?;
+    let (_event, payload) = parse_sse_frame(&frame)?;
+    let first_seq = payload
+        .get("seq")
+        .and_then(Value::as_u64)
+        .context("missing seq in first frame")?;
+    let id = frame
+        .lines()
+        .find_map(|line| line.strip_prefix("id:"))
+        .map(str::trim)
+        .context("SSE frames must carry an id, or nothing can resume")?
+        .to_string();
+    assert_eq!(
+        id.parse::<u64>()?,
+        first_seq,
+        "the id is the durable seq, not a frame counter"
+    );
+
+    // A second durable event, so a resume has somewhere to land.
+    let second_seq = runtime_threads
+        .emit_event_for_test(
+            &thread.id,
+            None,
+            "approval.required",
+            json!({"approval_id": "resume-proof", "tool_name": "exec_command"}),
+        )
+        .await?
+        .seq;
+    assert!(second_seq > first_seq, "the second event is later");
+
+    // The header alone is enough: a browser cannot set a query cursor.
+    let resumed = client
+        .get(format!("http://{addr}/v1/threads/{}/events", thread.id))
+        .header("Last-Event-ID", &id)
+        .send()
+        .await?
+        .error_for_status()?;
+    let frame = read_first_sse_frame(resumed).await?;
+    let (_event, payload) = parse_sse_frame(&frame)?;
+    assert_eq!(
+        payload.get("seq").and_then(Value::as_u64),
+        Some(second_seq),
+        "Last-Event-ID must resume past the acknowledged frame"
+    );
+
+    // An explicit `since_seq` outranks the header, so a deliberate
+    // replay-from-zero is never silently overridden by a stale id.
+    let explicit = client
+        .get(format!(
+            "http://{addr}/v1/threads/{}/events?since_seq=0",
+            thread.id
+        ))
+        .header("Last-Event-ID", &id)
+        .send()
+        .await?
+        .error_for_status()?;
+    let frame = read_first_sse_frame(explicit).await?;
+    let (_event, payload) = parse_sse_frame(&frame)?;
+    assert_eq!(
+        payload.get("seq").and_then(Value::as_u64),
+        Some(first_seq),
+        "the query cursor wins over the header"
+    );
+
+    handle.abort();
+    Ok(())
+}
+
+#[test]
+fn last_event_id_accepts_only_decimal_cursors() {
+    use super::last_event_id;
+
+    let mut headers = axum::http::HeaderMap::new();
+    assert_eq!(last_event_id(&headers), None, "absent header is no cursor");
+
+    headers.insert("last-event-id", "42".parse().unwrap());
+    assert_eq!(last_event_id(&headers), Some(42));
+
+    headers.insert("last-event-id", " 7 ".parse().unwrap());
+    assert_eq!(last_event_id(&headers), Some(7), "whitespace is trimmed");
+
+    // An opaque id from a proxy or an older client opens the stream from the
+    // durable head instead of refusing to open it at all.
+    headers.insert("last-event-id", "fev1_abcdef".parse().unwrap());
+    assert_eq!(last_event_id(&headers), None);
+}
+
 #[tokio::test]
 async fn event_handoff_replays_and_dedupes_interaction_prompts_without_a_gap() -> Result<()> {
     let Some((_addr, runtime_threads, handle)) = spawn_test_server().await? else {
@@ -13764,6 +13882,209 @@ async fn runtime_info_advertises_plugin_management_capability() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn runtime_info_advertises_terminal_capabilities() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    let info: serde_json::Value = client
+        .get(format!("http://{addr}/v1/runtime/info"))
+        .send()
+        .await?
+        .error_for_status()?
+        .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);
+    for capability in [
+        "terminal_stream",
+        "terminal_input",
+        "terminal_resize",
+        "terminal_kill",
+    ] {
+        assert_eq!(
+            info["capabilities"][capability], expected,
+            "runtime/info must advertise {capability}={expected}"
+        );
+    }
+
+    handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+#[cfg(unix)]
+async fn terminal_routes_serve_a_live_engine_session_over_http() -> Result<()> {
+    let tmp = tempfile::tempdir()?;
+    let root = tmp.path().join("runtime");
+    let workspace = tmp.path().join("ws");
+    fs::create_dir_all(&root)?;
+    fs::create_dir_all(&workspace)?;
+    let Some((addr, _runtime_threads, handle)) =
+        spawn_test_server_with_root_token_mobile_workspace(
+            root.clone(),
+            root.join("sessions"),
+            None,
+            false,
+            workspace.clone(),
+        )
+        .await?
+    else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+    let base = format!("http://{addr}/v1/terminal/pane");
+
+    // The agent's terminal tools own session creation; stand in for that
+    // producer on the same workspace the server was started with, which is
+    // what makes this the Engine's own shell rather than a second one.
+    let _session = crate::tools::terminal_session::get_or_create(
+        "pane",
+        &workspace,
+        crate::sandbox::SandboxPolicy::DangerFullAccess,
+    )
+    .map_err(anyhow::Error::msg)?;
+
+    // Input through the route, then the shell's own echo back through the
+    // route. Bytes in, bytes out, no direct access to the session object.
+    let write: serde_json::Value = client
+        .post(format!("{base}/input"))
+        .json(&serde_json::json!({
+            "data": "printf 'terminal-route-proof\\n'\n",
+            "encoding": "text"
+        }))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert!(write["written"].as_u64().unwrap_or_default() > 0);
+
+    let read_chunk = |base: String, client: reqwest::Client| async move {
+        let chunk: serde_json::Value = client
+            .get(format!("{base}/output?cursor=0&format=text"))
+            .send()
+            .await
+            .ok()?
+            .error_for_status()
+            .ok()?
+            .json()
+            .await
+            .ok()?;
+        Some(chunk)
+    };
+
+    let deadline = std::time::Instant::now() + ci_scaled(Duration::from_secs(10));
+    loop {
+        let chunk = read_chunk(base.clone(), client.clone())
+            .await
+            .expect("terminal output route answers");
+        let data = chunk["data"].as_str().unwrap_or_default();
+        if data.contains("terminal-route-proof") {
+            // Reads are non-consuming: the same cursor returns the same bytes.
+            let again = read_chunk(base.clone(), client.clone())
+                .await
+                .expect("terminal output route answers");
+            assert_eq!(again["data"], chunk["data"]);
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "route never delivered the shell's output: {data}"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+
+    // Resize is checked through the shell, not the handler: `stty size` reads
+    // the kernel's window, so a handler that only stored the numbers fails.
+    client
+        .post(format!("{base}/resize"))
+        .json(&serde_json::json!({"rows": 40, "cols": 100}))
+        .send()
+        .await?
+        .error_for_status()?;
+    client
+        .post(format!("{base}/input"))
+        .json(&serde_json::json!({"data": "stty size\n", "encoding": "text"}))
+        .send()
+        .await?
+        .error_for_status()?;
+    let deadline = std::time::Instant::now() + ci_scaled(Duration::from_secs(10));
+    loop {
+        let chunk = read_chunk(base.clone(), client.clone())
+            .await
+            .expect("terminal output route answers");
+        let data = chunk["data"].as_str().unwrap_or_default();
+        if data.contains("40 100") {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "resize never reached the shell: {data}"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+
+    // Kill, then learn the truth from the stream rather than the ack.
+    client
+        .post(format!("{base}/kill"))
+        .send()
+        .await?
+        .error_for_status()?;
+    let deadline = std::time::Instant::now() + ci_scaled(Duration::from_secs(10));
+    loop {
+        let chunk = read_chunk(base.clone(), client.clone())
+            .await
+            .expect("terminal output route answers");
+        if chunk["running"] == serde_json::json!(false) {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "killed session still reports running: {chunk}"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+
+    handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+#[cfg(unix)]
+async fn terminal_output_for_an_unknown_session_is_not_found_and_creates_nothing() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+    let base = format!("http://{addr}/v1/terminal");
+
+    // The route exists and answers for a name that has no live session: the
+    // Engine attaches to shells it owns, it does not conjure one per request.
+    let missing = client
+        .get(format!("{base}/no-such-session/output"))
+        .send()
+        .await?;
+    assert_eq!(missing.status(), reqwest::StatusCode::NOT_FOUND);
+
+    // An over-long name is rejected as a miss too, so the registry is never
+    // asked to allocate for it.
+    let long_name = "n".repeat(200);
+    let oversized = client
+        .get(format!("{base}/{long_name}/output"))
+        .send()
+        .await?;
+    assert_eq!(oversized.status(), reqwest::StatusCode::NOT_FOUND);
+
+    handle.abort();
+    Ok(())
+}
+
 #[tokio::test]
 async fn plugin_lifecycle_over_http_installs_reviews_enables_and_uninstalls() -> Result<()> {
     let tmp = tempfile::tempdir()?;
@@ -17540,14 +17861,30 @@ async fn threads_running_lists_active_turns_and_clears_on_settle() -> Result<()>
         sleep(Duration::from_millis(20)).await;
     }
 
-    let settled: serde_json::Value = client
-        .get(format!("{base}/v1/threads/running"))
-        .send()
-        .await?
-        .error_for_status()?
-        .json()
-        .await?;
-    assert_eq!(settled, serde_json::json!([]));
+    // The listing above is read from the durable store, and the engine still
+    // owns this record: it can persist its own status after the write above,
+    // which puts the turn back in flight and made a single read flaky on a
+    // loaded macOS runner. That is not a defect — the engine is entitled to
+    // finish its turn. What must hold is that a settled turn stops being
+    // listed, so poll for that instead of assuming the first read is final.
+    let deadline = std::time::Instant::now() + ci_scaled(Duration::from_secs(5));
+    loop {
+        let settled: serde_json::Value = client
+            .get(format!("{base}/v1/threads/running"))
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        if settled == serde_json::json!([]) {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "a settled turn must leave the running list: {settled}"
+        );
+        sleep(Duration::from_millis(50)).await;
+    }
 
     handle.abort();
     Ok(())

crates/tui/src/runtime_threads.rs

@@ -5983,6 +5983,7 @@ impl RuntimeThreadManager {
             false,
         )
         .await
+        .map(|(turn, _replayed)| turn)
     }
 
     /// Terminal goal settlement for one finished turn.
@@ -6506,7 +6507,7 @@ impl RuntimeThreadManager {
             .await;
 
         match turn_result {
-            Ok(turn) => {
+            Ok((turn, _replayed)) => {
                 let delivered = {
                     let _mail_mutation = self.store.mail_mutation.lock();
                     let mut envelope = self.store.load_agent_mail(message_id)?;
@@ -9433,6 +9434,22 @@ impl RuntimeThreadManager {
         self.start_turn_inner(thread_id, req, None).await
     }
 
+    /// Start a turn and report whether the durable `operation_key` made it a
+    /// replay of an already-accepted submission.
+    ///
+    /// The distinction belongs to admission, not to the route: a client that
+    /// retried an ambiguous submit needs to be told "this is the turn you
+    /// already started" so it does not render a duplicate, and only the
+    /// admission path knows that for certain.
+    pub async fn start_turn_reporting_replay(
+        &self,
+        thread_id: &str,
+        req: StartTurnRequest,
+    ) -> Result<(TurnRecord, bool)> {
+        self.start_turn_inner_reporting_replay(thread_id, req, None)
+            .await
+    }
+
     pub(crate) async fn start_turn_with_reserved_id(
         &self,
         thread_id: &str,
@@ -9455,6 +9472,17 @@ impl RuntimeThreadManager {
         req: StartTurnRequest,
         reserved_turn_id: Option<&str>,
     ) -> Result<TurnRecord> {
+        self.start_turn_inner_reporting_replay(thread_id, req, reserved_turn_id)
+            .await
+            .map(|(turn, _replayed)| turn)
+    }
+
+    async fn start_turn_inner_reporting_replay(
+        &self,
+        thread_id: &str,
+        req: StartTurnRequest,
+        reserved_turn_id: Option<&str>,
+    ) -> Result<(TurnRecord, bool)> {
         if reserved_turn_id.is_some() && req.operation_key.is_none() {
             bail!("a reserved turn id requires an operation key");
         }
@@ -9484,8 +9512,11 @@ impl RuntimeThreadManager {
             true,
         )
         .await
+        .map(|(turn, _replayed)| turn)
     }
 
+    /// Returns the turn and whether the durable operation key made this a
+    /// replay of an already-accepted submission rather than a new admission.
     async fn start_turn_with_source(
         &self,
         thread_id: &str,
@@ -9493,7 +9524,7 @@ impl RuntimeThreadManager {
         input_source: RuntimeTurnInputSource,
         reserved_turn_id: Option<&str>,
         stored_image_bytes: bool,
-    ) -> Result<TurnRecord> {
+    ) -> Result<(TurnRecord, bool)> {
         // Heap-allocate the turn-start state machine. Its future holds two full
         // Config clones plus ThreadRecord/EngineHandle/TurnRecord/TurnItemRecord
         // and the Op::SendMessage, and inlines the large ensure_engine_loaded
@@ -9605,7 +9636,7 @@ impl RuntimeThreadManager {
         if let Some(operation) = operation.as_ref()
             && let Some(original_turn) = self.replay_turn_for_operation(operation)?
         {
-            return Ok(original_turn);
+            return Ok((original_turn, true));
         }
         if !image_blocks.is_empty() || req.max_output_tokens.is_some() {
             let identity = self.provider_identity_for_thread(&cfg_snapshot, &thread)?;
@@ -9927,7 +9958,7 @@ impl RuntimeThreadManager {
             if let Some(operation) = operation.as_ref()
                 && let Some(original_turn) = self.replay_turn_for_operation(operation)?
             {
-                return Ok(original_turn);
+                return Ok((original_turn, true));
             }
             let Some(state) = active.engines.get_mut(thread_id) else {
                 bail!("Thread engine not loaded");
@@ -10020,10 +10051,11 @@ impl RuntimeThreadManager {
             )
         };
 
-        acceptance_rx
+        let turn = acceptance_rx
             .await
             .map_err(|_| anyhow!("Turn lifecycle task ended before acknowledgement"))?
-            .map_err(anyhow::Error::msg)
+            .map_err(anyhow::Error::msg)?;
+        Ok((turn, false))
         })
         .await
     }

crates/tui/src/sleep_guard.rs

@@ -0,0 +1,169 @@
+//! Keep the host from idling into sleep while a turn is in flight.
+//!
+//! A suspended host cannot run the engine, so nothing here survives a real
+//! suspend — `core::engine::streaming::sleep_gap_detected` already reports that
+//! case and the engine re-issues the request (issue #2990). What this module
+//! prevents is the avoidable one: an unattended machine idling into sleep in
+//! the middle of a turn, which is how a long turn gets lost with no error at
+//! all.
+//!
+//! Scope, stated so nobody expects more than it does:
+//!
+//! - It holds the platform's *idle-sleep* assertion only. An explicit `sleep`/
+//!   `pmset sleepnow`, a closed lid, or a low battery still wins — refusing
+//!   those is the machine owner's call, not a running task's.
+//! - It is held for the duration of a turn and released on drop, so the host's
+//!   power behaviour outside a turn is untouched.
+//! - It follows the same gate as the rest of the host-facing chrome
+//!   (`EngineConfig::terminal_chrome_enabled`): an interactive TUI turn holds
+//!   it, while headless hosts — `exec`, app-server, CI — never do. A dedicated
+//!   `[tui]` opt-out key is not implemented yet; the headless gate is the
+//!   escape hatch today.
+//! - Windows is not implemented. `SetThreadExecutionState` is thread-affine —
+//!   the release has to happen on the thread that set it, which a guard that
+//!   travels with a turn cannot promise. Rather than ship an untested holder
+//!   that might silently never release, this is a no-op there for now.
+//!
+//! Release is `Drop` and never cached: a leaked inhibitor would keep a laptop
+//! awake forever, which is worse than the problem this solves.
+
+use std::process::{Child, Command, Stdio};
+
+/// 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.
+    #[cfg(unix)]
+    child: Option<Child>,
+}
+
+impl SleepGuard {
+    /// Hold the host awake until the returned guard drops.
+    #[must_use]
+    pub fn hold() -> Self {
+        #[cfg(unix)]
+        {
+            Self {
+                child: start_inhibitor(),
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            Self {}
+        }
+    }
+
+    /// The inhibitor's process id, for diagnostics and tests. Absent when the
+    /// 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)
+    }
+}
+
+#[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();
+    }
+}
+
+/// `-i` prevents idle sleep. Without `-t` caffeinate runs until it is killed,
+/// which is what `Drop` does; macOS releases the assertion with the process.
+#[cfg(target_os = "macos")]
+fn start_inhibitor() -> Option<Child> {
+    spawn("caffeinate", &["-i"])
+}
+
+/// `--what=idle` only: an explicit suspend or a closed lid is still honoured.
+/// `sleep infinity` is the command whose lifetime holds the block open.
+#[cfg(target_os = "linux")]
+fn start_inhibitor() -> Option<Child> {
+    spawn(
+        "systemd-inhibit",
+        &[
+            "--what=idle",
+            "--why=Codewhale turn in flight",
+            "--mode=block",
+            "sleep",
+            "infinity",
+        ],
+    )
+}
+
+/// Everything else Unix (BSD, illumos, …) has no inhibitor this module knows.
+#[cfg(all(unix, not(any(target_os = "macos", target_os = "linux"))))]
+fn start_inhibitor() -> Option<Child> {
+    None
+}
+
+#[cfg(unix)]
+fn spawn(program: &str, args: &[&str]) -> Option<Child> {
+    Command::new(program)
+        .args(args)
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .ok()
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+
+    /// Whether the process is still there. `kill(pid, 0)` asks the kernel
+    /// without touching the process, so this cannot perturb the guard.
+    fn alive(pid: u32) -> bool {
+        // SAFETY: signal 0 performs the permission/existence check only.
+        unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
+    }
+
+    #[test]
+    #[cfg(any(target_os = "macos", target_os = "linux"))]
+    fn the_inhibitor_lives_exactly_as_long_as_the_guard() {
+        let guard = SleepGuard::hold();
+        let pid = guard
+            .inhibitor_pid()
+            .expect("this platform starts an inhibitor");
+        assert!(alive(pid), "the inhibitor must be running while held");
+
+        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));
+        }
+    }
+
+    #[test]
+    #[cfg(any(target_os = "macos", target_os = "linux"))]
+    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();
+        let second = SleepGuard::hold();
+        let (a, b) = (
+            first.inhibitor_pid().expect("first inhibitor"),
+            second.inhibitor_pid().expect("second inhibitor"),
+        );
+        assert_ne!(a, b, "each guard owns its own inhibitor process");
+        drop(first);
+        assert!(!alive(a), "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

@@ -35,6 +35,11 @@ use super::spec::{optional_u64, required_str};
 const BUFFER_LIMIT: usize = 512 * 1024;
 #[cfg(unix)]
 const OUTPUT_LIMIT: usize = 12 * 1024;
+/// Ceiling for one [`OutputBuffer::read_since`] response. The ring is already
+/// bounded at [`BUFFER_LIMIT`]; this bounds a single frame so a client cannot
+/// ask for the whole window at once.
+#[cfg(unix)]
+pub(crate) const READ_LIMIT: usize = 64 * 1024;
 #[cfg(unix)]
 const DEFAULT_TIMEOUT_SECS: u64 = 120;
 #[cfg(unix)]
@@ -45,8 +50,12 @@ const CANCEL_CONFIRM_TIMEOUT: Duration = Duration::from_secs(2);
 const CANCEL_SENTINEL_RETRY_INTERVAL: Duration = Duration::from_millis(50);
 
 #[cfg(unix)]
-struct TerminalSession {
+pub(crate) struct TerminalSession {
     writer: Arc<Mutex<Box<dyn Write + Send>>>,
+    /// The pty master is retained after the reader and writer clones are taken
+    /// so [`TerminalSession::resize`] can reach the kernel's window size. The
+    /// clones keep the pty alive; nothing else clones the master.
+    master: Box<dyn portable_pty::MasterPty + Send>,
     child: Box<dyn portable_pty::Child + Send>,
     output: Arc<Mutex<OutputBuffer>>,
     read_cursor: u64,
@@ -110,6 +119,33 @@ struct OutputBuffer {
     total: u64,
 }
 
+/// One absolute-offset slice of a session's output.
+///
+/// Offsets are absolute for the life of the live PTY: they count every byte
+/// the reader thread ever appended, not the bytes still retained. That is what
+/// lets a client resume from a cursor it stored earlier, and what lets this
+/// type tell it the truth when the retained window has moved past it.
+///
+/// Field names match the `/v1/terminal/{name}/output` wire so the payload is a
+/// projection, not a translation (the same vocabulary the jobs byte stream
+/// uses).
+#[cfg(unix)]
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct OutputChunk {
+    pub(crate) bytes: Vec<u8>,
+    /// Absolute offset of `bytes[0]` in the session's lifetime output.
+    pub(crate) offset: u64,
+    /// Offset to pass as the next `cursor`.
+    pub(crate) next_cursor: u64,
+    /// Every byte the session has produced, retained or not.
+    pub(crate) total: u64,
+    /// Leading bytes the ring has permanently discarded.
+    pub(crate) dropped: u64,
+    /// True when the requested cursor predates the retained window. The bytes
+    /// in between cannot be recovered from this process — report, not repair.
+    pub(crate) gap: bool,
+}
+
 #[cfg(unix)]
 impl OutputBuffer {
     fn append(&mut self, data: &[u8]) {
@@ -123,10 +159,37 @@ impl OutputBuffer {
     fn text(&self) -> String {
         String::from_utf8_lossy(&self.bytes.iter().copied().collect::<Vec<_>>()).into_owned()
     }
+
+    /// Absolute offset of the oldest retained byte.
+    fn oldest(&self) -> u64 {
+        self.total.saturating_sub(self.bytes.len() as u64)
+    }
+
+    /// Read from an absolute cursor without consuming: the caller owns its own
+    /// position, so two readers can replay the same bytes and a repeated read
+    /// 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.
+    fn read_since(&self, cursor: u64, max_bytes: usize) -> OutputChunk {
+        let dropped = self.oldest();
+        let gap = cursor < dropped;
+        let start = cursor.max(dropped);
+        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();
+        OutputChunk {
+            bytes,
+            offset: start,
+            next_cursor: start + take as u64,
+            total: self.total,
+            dropped,
+            gap,
+        }
+    }
 }
 
 #[cfg(unix)]
-type SharedSession = Arc<Mutex<TerminalSession>>;
+pub(crate) type SharedSession = Arc<Mutex<TerminalSession>>;
 
 #[cfg(unix)]
 #[derive(Clone, Debug, Hash, PartialEq, Eq)]
@@ -341,6 +404,7 @@ fn create_session(
 
     Ok(Arc::new(Mutex::new(TerminalSession {
         writer: Arc::new(Mutex::new(writer)),
+        master: pair.master,
         child,
         output,
         read_cursor: 0,
@@ -352,7 +416,7 @@ fn create_session(
 }
 
 #[cfg(unix)]
-fn get_or_create(
+pub(crate) fn get_or_create(
     name: &str,
     workspace: &std::path::Path,
     policy: crate::sandbox::SandboxPolicy,
@@ -408,7 +472,7 @@ fn find(name: &str, workspace: &Path) -> Result<SharedSession, String> {
 }
 
 #[cfg(unix)]
-fn write_bytes(session: &TerminalSession, bytes: &[u8]) -> Result<(), String> {
+pub(crate) fn write_bytes(session: &TerminalSession, bytes: &[u8]) -> Result<(), String> {
     let mut writer = session
         .writer
         .lock()
@@ -433,12 +497,84 @@ fn take_output(session: &mut TerminalSession) -> String {
     let Ok(output) = session.output.lock() else {
         return String::new();
     };
-    let retained_start = output.total.saturating_sub(output.bytes.len() as u64);
-    let start = session.read_cursor.max(retained_start);
-    let skip = usize::try_from(start.saturating_sub(retained_start)).unwrap_or(usize::MAX);
-    let bytes = output.bytes.iter().skip(skip).copied().collect::<Vec<_>>();
+    let chunk = output.read_since(session.read_cursor, usize::MAX);
     session.read_cursor = output.total;
-    String::from_utf8_lossy(&bytes).into_owned()
+    String::from_utf8_lossy(&chunk.bytes).into_owned()
+}
+
+/// Resize the live pty's window. The kernel updates its `winsize` and signals
+/// the child, which is what makes an interactive app redraw at the new size.
+#[cfg(unix)]
+pub(crate) fn resize_session(
+    session: &TerminalSession,
+    rows: u16,
+    cols: u16,
+) -> Result<(), String> {
+    session
+        .master
+        .resize(portable_pty::PtySize {
+            rows,
+            cols,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .map_err(|e| format!("PTY resize failed: {e}"))
+}
+
+/// Absolute-offset, non-consuming read. `max_bytes` is clamped to
+/// [`READ_LIMIT`] so one caller cannot ask for the whole retained window.
+///
+/// Known limitation: this reports a [`OutputChunk::gap`]; it does not repair
+/// one. Bytes dropped by the ring are gone with the process, and nothing here
+/// re-reads them from disk — the durable record is identity and lifecycle,
+/// never output.
+#[cfg(unix)]
+pub(crate) fn read_session_since(
+    session: &TerminalSession,
+    cursor: u64,
+    max_bytes: usize,
+) -> Result<OutputChunk, String> {
+    let output = session
+        .output
+        .lock()
+        .map_err(|_| "terminal output lock poisoned".to_string())?;
+    Ok(output.read_since(cursor, max_bytes.min(READ_LIMIT)))
+}
+
+/// Poll the shell without blocking; `None` means it is still running.
+#[cfg(unix)]
+pub(crate) fn session_exit_status(
+    session: &mut TerminalSession,
+) -> Result<Option<portable_pty::ExitStatus>, String> {
+    session
+        .child
+        .try_wait()
+        .map_err(|e| format!("PTY wait failed: {e}"))
+}
+
+/// Terminate the shell. The caller still observes the exit through
+/// [`session_exit_status`].
+#[cfg(unix)]
+pub(crate) fn kill_session(session: &mut TerminalSession) -> Result<(), String> {
+    session
+        .child
+        .kill()
+        .map_err(|e| format!("PTY kill failed: {e}"))
+}
+
+/// Resolve a live session without creating one.
+///
+/// `/v1/terminal` attaches to shells the Engine already owns (the agent's
+/// terminal tools create them). A request for a name that has no live session
+/// is a 404, never a new process: an HTTP client must not be able to conjure a
+/// shell the Engine does not know about.
+#[cfg(unix)]
+pub(crate) fn lookup(name: &str, workspace: &Path) -> Option<SharedSession> {
+    let key = session_key(name, workspace);
+    sessions()
+        .lock()
+        .ok()
+        .and_then(|registry| registry.get(&key).map(Arc::clone))
 }
 
 #[cfg(unix)]
@@ -1211,4 +1347,141 @@ mod tests {
         assert!(result.content.len() <= OUTPUT_LIMIT + 100);
         assert!(result.content.contains("output truncated"));
     }
+
+    /// The cursor contract the Engine byte stream rides on: offsets are
+    /// absolute for the life of the PTY, reads never consume, and a cursor the
+    /// retained window has moved past is reported rather than silently
+    /// answered from the middle of the stream.
+    #[test]
+    #[cfg(unix)]
+    fn bounded_replay_is_absolute_non_consuming_and_reports_its_gap() {
+        let mut buffer = OutputBuffer::default();
+        buffer.append(b"hello");
+        let first = buffer.read_since(0, 4);
+        assert_eq!(first.bytes, b"hell");
+        assert_eq!(first.offset, 0);
+        assert_eq!(first.next_cursor, 4);
+        assert_eq!(first.total, 5);
+        assert_eq!(first.dropped, 0);
+        assert!(!first.gap);
+        // A second read at the same cursor returns the same bytes: the caller
+        // owns the position, so replay is idempotent.
+        assert_eq!(buffer.read_since(0, 4), first);
+        let rest = buffer.read_since(first.next_cursor, 4);
+        assert_eq!(rest.bytes, b"o");
+        assert_eq!(rest.offset, 4);
+        assert_eq!(rest.next_cursor, 5);
+
+        // Past the ring: the first 32 bytes are gone and the chunk says so.
+        let mut wrapped = OutputBuffer::default();
+        wrapped.append(&vec![b'x'; BUFFER_LIMIT + 32]);
+        let lost = wrapped.read_since(0, 8);
+        assert!(lost.gap, "a cursor below the retained window is a gap");
+        assert_eq!(lost.dropped, 32);
+        assert_eq!(
+            lost.offset, 32,
+            "the chunk starts at the oldest retained byte"
+        );
+        assert_eq!(lost.total, BUFFER_LIMIT as u64 + 32);
+        assert_eq!(lost.bytes, vec![b'x'; 8]);
+        assert_eq!(lost.next_cursor, 40);
+
+        // A cursor at the head is not a gap and returns nothing new.
+        let current = wrapped.read_since(BUFFER_LIMIT as u64 + 32, 8);
+        assert!(!current.gap);
+        assert!(current.bytes.is_empty());
+        assert_eq!(current.next_cursor, BUFFER_LIMIT as u64 + 32);
+    }
+
+    /// The session-level entry point an Engine byte stream will call: absolute
+    /// cursor, non-consuming, clamped, and honest about a cursor ahead of the
+    /// stream.
+    #[test]
+    #[cfg(unix)]
+    fn session_read_since_is_non_consuming_and_clamped() {
+        let session = fresh("test-read-since");
+        let _ = run(
+            &session,
+            "printf 'cw-replay-proof\\n'",
+            Duration::from_secs(3),
+        );
+        // The tool-result path already consumed this output through its own
+        // cursor; an absolute cursor still reads it, which is the point.
+        let printed = read_session_since(&session.lock().unwrap(), 0, usize::MAX).unwrap();
+        assert!(
+            String::from_utf8_lossy(&printed.bytes).contains("cw-replay-proof"),
+            "{}",
+            String::from_utf8_lossy(&printed.bytes)
+        );
+        let again = read_session_since(&session.lock().unwrap(), 0, usize::MAX).unwrap();
+        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.
+        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);
+
+        // More than one response's worth of output proves the clamp.
+        let large = fresh("test-read-since-clamp");
+        let _ = run(&large, "yes x | head -n 100000", Duration::from_secs(3));
+        let chunk = read_session_since(&large.lock().unwrap(), 0, usize::MAX).unwrap();
+        assert_eq!(chunk.bytes.len(), READ_LIMIT);
+        assert!(!chunk.gap);
+        assert_eq!(chunk.next_cursor, READ_LIMIT as u64);
+    }
+
+    /// Resize has to reach the kernel, not just a field: `get_size` reads the
+    /// window back from the pty, and the shell reports it through `stty`.
+    #[test]
+    #[cfg(unix)]
+    fn resize_reaches_the_kernel_and_the_live_shell() {
+        let session = fresh("test-resize");
+        {
+            let guard = session.lock().unwrap();
+            assert_eq!(guard.master.get_size().unwrap().rows, 24);
+        }
+        resize_session(&session.lock().unwrap(), 40, 100).unwrap();
+        let size = session.lock().unwrap().master.get_size().unwrap();
+        assert_eq!((size.rows, size.cols), (40, 100));
+        let result = run(&session, "stty size", Duration::from_secs(3));
+        assert!(
+            result.content.contains("40 100"),
+            "the shell should see the new window: {}",
+            result.content
+        );
+    }
+
+    /// Exit is observable: a killed shell reports a status instead of looking
+    /// alive forever (the "pretending to reattach" failure the durable record
+    /// is written to avoid).
+    #[test]
+    #[cfg(unix)]
+    fn killed_shell_reports_an_exit_status() {
+        let session = fresh("test-exit-status");
+        {
+            let mut guard = session.lock().unwrap();
+            assert!(
+                session_exit_status(&mut guard).unwrap().is_none(),
+                "fresh shell is alive"
+            );
+            kill_session(&mut guard).unwrap();
+        }
+        let deadline = Instant::now() + Duration::from_secs(5);
+        loop {
+            let exited = session_exit_status(&mut session.lock().unwrap())
+                .unwrap()
+                .is_some();
+            if exited {
+                break;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "a killed shell must report its exit"
+            );
+            std::thread::sleep(Duration::from_millis(20));
+        }
+    }
 }

crates/tui/src/tui/pet_watch/mod.rs

@@ -736,4 +736,80 @@ mod tests {
         assert!(!call.contains("PRIVATE"));
         assert!(!thought.contains("PRIVATE"));
     }
+
+    /// The pet's multi-agent count is derived on the JS side from
+    /// `agent:`-prefixed spans, keyed by the `id` this projection forwards
+    /// (app-side issue #12). The counting itself is proven in
+    /// `pet/tests/pet-engine.test.mjs`; the handoff is the half that fails
+    /// silently — a trimmed allowlist or a dropped id zeroes `parallel` with
+    /// no error and no log line, and nothing else here covers agent events.
+    /// Pin the wire shape the JS dispatches on, and keep child text off it.
+    #[test]
+    fn agent_events_forward_span_identity_without_child_text() {
+        use crate::core::events::AgentProgressEventMeta;
+        use crate::tools::subagent::{AgentWorkerStatus, SubAgentStatus};
+
+        let spawned = metadata(&Event::AgentSpawned {
+            owner_session_id: "session-a".into(),
+            id: "agent-1".into(),
+            prompt: "PRIVATE CHILD PROMPT".into(),
+            worker_status: Some(AgentWorkerStatus::Running),
+            parent_run_id: Some("run-9".into()),
+            spawn_depth: 2,
+            model: "PRIVATE CHILD MODEL".into(),
+            route_source: Some("task.model".into()),
+        })
+        .expect("agent spawns are observed");
+        assert_eq!(
+            serde_json::from_str::<Value>(&spawned).unwrap(),
+            json!({"event":"agent_spawned","id":"agent-1","worker_status":"running"})
+        );
+
+        // The JS finishes a span when progress reports a terminal status.
+        let progress = metadata(&Event::AgentProgress {
+            owner_session_id: "session-a".into(),
+            id: "agent-1".into(),
+            status: "PRIVATE PROGRESS TEXT".into(),
+            activity: AgentProgressEventMeta {
+                worker_status: AgentWorkerStatus::Completed,
+                step: Some(3),
+                tool_name: Some("exec_command".into()),
+                routine_wait: false,
+            },
+            parent_run_id: Some("run-9".into()),
+            spawn_depth: 2,
+        })
+        .expect("agent progress is observed");
+        assert_eq!(
+            serde_json::from_str::<Value>(&progress).unwrap(),
+            json!({"event":"agent_progress","id":"agent-1","worker_status":"completed"})
+        );
+
+        let complete = metadata(&Event::AgentComplete {
+            owner_session_id: "session-a".into(),
+            id: "agent-1".into(),
+            result: "PRIVATE CHILD RESULT".into(),
+            outcome: Some(SubAgentStatus::Completed),
+            parent_run_id: Some("run-9".into()),
+            spawn_depth: Some(2),
+            continuable: Some(false),
+            usage: None,
+        })
+        .expect("agent completions are observed");
+        assert_eq!(
+            serde_json::from_str::<Value>(&complete).unwrap(),
+            json!({"event":"agent_complete","id":"agent-1","worker_status":"completed"})
+        );
+
+        for (label, payload) in [
+            ("spawned", &spawned),
+            ("progress", &progress),
+            ("complete", &complete),
+        ] {
+            assert!(
+                !payload.contains("PRIVATE"),
+                "{label} leaked child text: {payload}"
+            );
+        }
+    }
 }

docs/ENVIRONMENTS.md

@@ -37,6 +37,27 @@ CODEWHALE_PROVIDER=vllm VLLM_BASE_URL=http://127.0.0.1:8000/v1 VLLM_MODEL=<id> \
 `codewhale exec` (add `--auto` for tool use) is the non-interactive path to
 exercise the full agent loop.
 
+## Keeping the host awake during a turn
+
+While an interactive TUI turn is in flight, Codewhale holds the platform's
+idle-sleep assertion, so an unattended machine does not idle into sleep
+mid-turn and lose the work:
+
+- macOS: `caffeinate -i`
+- Linux: `systemd-inhibit --what=idle --why="Codewhale turn in flight" --mode=block sleep infinity`
+
+The assertion is released the moment the turn ends, and it covers *idle* sleep
+only: an explicit `sleep` / `pmset sleepnow`, a closed lid, or a low battery
+still suspends the machine. Headless hosts — `exec`, app-server, CI — never
+hold it, so a shared runner's power policy is untouched. Windows is not
+implemented: `SetThreadExecutionState` is thread-affine and needs a holder that
+pins the thread, so the gap is deliberate rather than silent.
+
+If a turn is suspended anyway, the engine notices on wake — wall-clock elapsed
+diverging from monotonic elapsed by more than the suspend threshold — reports
+`System sleep detected; connection lost — retrying request`, and re-issues the
+request instead of failing the turn (#2990).
+
 ## Consolidated runtime commands
 
 The current `codewhale` binary runs the TUI in-process. Release installers copy

docs/RUNTIME_API.md

@@ -1215,6 +1215,39 @@ manager, the durable thread store, the workspace confinement layer, the
 config's credential plumbing — and add no second runtime, session store,
 scheduler, or credential store.
 
+**Terminal sessions** (the persistent Engine-owned shell)
+
+The jobs family above runs one command per job. A terminal pane needs the
+*other* authority: the stateful PTY-backed shell the agent's own terminal
+tools drive, which keeps cwd and environment across inputs. These routes
+attach to that session and never create one — a name with no live session is
+`404`, because conjuring a shell from an HTTP request would give the client a
+terminal the Engine does not know about. Input is attributable by route:
+`input` is the client's writer, `terminal_send` is the agent's.
+
+- `GET /v1/terminal/{name}/output?cursor=<bytes>&max_bytes=<1-64KiB>&format=
+  <base64|text>` — the resumable byte stream. `{name, offset, next_cursor,
+  total, dropped, encoding, data, running, exit_code}`: pass `next_cursor`
+  back to continue; reads never consume, so several clients may hold
+  independent cursors; `dropped` reports bytes the 512 KiB ring discarded
+- `POST /v1/terminal/{name}/input` — `{ "data", "encoding"? }`, UTF-8 text by
+  default or `base64` for exact bytes → `{ "name", "written" }`
+- `POST /v1/terminal/{name}/resize` — `{ "rows", "cols" }` → the kernel
+  window the child draws for
+- `POST /v1/terminal/{name}/kill` — end the shell; observe the exit through
+  `output` (`running` / `exit_code`) rather than the acknowledgement
+
+`GET /v1/runtime/info` advertises `terminal_stream`, `terminal_input`,
+`terminal_resize` and `terminal_kill`. All four are `false` on Windows builds
+today: the owner is Unix-only, the Windows routes answer `501`, and a client
+should gate its terminal controls on these flags rather than discovering it
+from a failed request. Known limitations, stated because a reader would
+otherwise assume them: there is no `wait_ms` long poll (poll the cursor),
+scrollback dropped by the ring is gone with the process, a restarted Engine
+reports no session rather than pretending to reattach, and the
+`@codewhale/runtime-sdk` package has no terminal client wrapper yet — the raw
+routes are the contract for now.
+
 **Jobs** (operator-scoped shell jobs; the terminal surface)
 - `GET /v1/jobs` — every live and known-stale job across all threads
 - `GET /v1/threads/{id}/jobs` — jobs owned by one thread's manager: