feat(runtime): expose live PTY resizing and exact terminal input

Retain existing shell PTY handles and terminal size, add authenticated thread-owned resize, preserve raw stdin bytes, and release the shell manager while output long-polls wait. Reject invalid dimensions, pipe-only and finished jobs without creating another PTY owner.

Validation: combined LSP, PTY and journal selection 365 passed, including live stty resize, raw bytes, ownership and nonblocking output polling; npm test 551 passed (66 scripts, 14 docs, 471 web); npm run check:web and locked CLI build passed. Native terminal renderer and Windows runtime acceptance remain pending.

crates/tui/src/runtime_api.rs

@@ -1255,6 +1255,10 @@ pub fn build_router(state: RuntimeApiState) -> Router {
             "/v1/threads/{id}/jobs/{job_id}/kill",
             post(jobs::kill_thread_job),
         )
+        .route(
+            "/v1/threads/{id}/jobs/{job_id}/resize",
+            post(jobs::resize_thread_job),
+        )
         .route("/v1/threads/{id}/context", get(context::get_thread_context))
         .route("/v1/threads/{id}/plan", get(plans::get_thread_plan))
         .route("/v1/threads/{id}/todo", get(plans::get_thread_todo))

crates/tui/src/runtime_api/jobs.rs

@@ -15,7 +15,8 @@ use base64::Engine as _;
 use serde::{Deserialize, Serialize};
 
 use crate::tools::shell::{
-    ShellJobSnapshot, ShellOutputChunk, ShellOutputStream, ShellResult, ShellStatus,
+    PtyDimensions, ShellJobSnapshot, ShellManager, ShellOutputChunk, ShellOutputStream,
+    ShellResult, ShellStatus,
 };
 
 use super::{ApiError, RuntimeApiState, map_thread_err};
@@ -66,15 +67,22 @@ pub(super) struct JobView {
     /// `client` = launched through this API, `agent` = launched by the model's
     /// shell tool, `subagent` = owned by a delegated agent.
     owner: &'static str,
+    /// Null for historical records whose original transport is not known.
+    tty: Option<bool>,
+    terminal_size: Option<PtyDimensions>,
 }
 
 impl JobView {
-    fn new(snapshot: ShellJobSnapshot, thread_id: String) -> Self {
+    fn new(snapshot: ShellJobSnapshot, thread_id: String, manager: &ShellManager) -> Self {
         let owner = job_owner(&snapshot);
+        let terminal_size = manager.job_terminal_size(&snapshot.job_id);
+        let tty = (!snapshot.stale).then_some(terminal_size.is_some());
         Self {
             snapshot,
             thread_id,
             owner,
+            tty,
+            terminal_size,
         }
     }
 }
@@ -97,7 +105,7 @@ pub(super) async fn list_jobs(
                 guard
                     .list_jobs()
                     .into_iter()
-                    .map(|snapshot| JobView::new(snapshot, thread_id.clone())),
+                    .map(|snapshot| JobView::new(snapshot, thread_id.clone(), &guard)),
             );
         }
         jobs
@@ -139,7 +147,7 @@ pub(super) async fn list_thread_jobs(
         guard
             .list_jobs()
             .into_iter()
-            .map(|snapshot| JobView::new(snapshot, thread_id.clone()))
+            .map(|snapshot| JobView::new(snapshot, thread_id.clone(), &guard))
             .collect()
     })
     .await
@@ -254,7 +262,7 @@ pub(super) async fn create_thread_job(
                 ApiError::internal(format!("job launched but is not tracked: {error}"))
             })?
             .snapshot;
-        Ok(JobView::new(snapshot, thread_id))
+        Ok(JobView::new(snapshot, thread_id, &guard))
     })
     .await
     .map_err(|_| ApiError::internal("job launch failed"))??;
@@ -282,7 +290,7 @@ pub(super) async fn get_thread_job(
         let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
         let detail = guard.inspect_job(&job_id).map_err(map_job_err)?;
         Ok(Json(JobDetailResponse {
-            job: JobView::new(detail.snapshot, thread_id),
+            job: JobView::new(detail.snapshot, thread_id, &guard),
             stdout_tail: detail.stdout,
             stderr_tail: detail.stderr,
         }))
@@ -363,11 +371,26 @@ pub(super) async fn get_thread_job_output(
     let manager = thread_manager(&state, &thread_id, false).await?;
     let chunk = tokio::task::spawn_blocking({
         let job_id = job_id.clone();
-        move || {
-            let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
-            guard
-                .read_output_chunk(&job_id, stream, cursor, max_bytes, wait_ms)
-                .map_err(map_job_err)
+        move || -> Result<ShellOutputChunk, ApiError> {
+            // Wait between non-consuming snapshots, never while owning the
+            // thread's shared ShellManager. Input, resize, and kill stay live.
+            let deadline = std::time::Instant::now() + std::time::Duration::from_millis(wait_ms);
+            loop {
+                let chunk = {
+                    let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
+                    guard
+                        .read_output_chunk(&job_id, stream, cursor, max_bytes, 0)
+                        .map_err(map_job_err)?
+                };
+                let remaining = deadline.saturating_duration_since(std::time::Instant::now());
+                if chunk.total > cursor
+                    || chunk.status != ShellStatus::Running
+                    || remaining.is_zero()
+                {
+                    break Ok(chunk);
+                }
+                std::thread::sleep(remaining.min(std::time::Duration::from_millis(50)));
+            }
         }
     })
     .await
@@ -433,7 +456,7 @@ pub(super) async fn write_thread_job_stdin(
                     "data must be at most {STDIN_MAX_BYTES} bytes"
                 )));
             }
-            request.data
+            request.data.into_bytes()
         }
         "base64" => {
             if request.data.len() > STDIN_MAX_BYTES * 2 {
@@ -447,8 +470,7 @@ pub(super) async fn write_thread_job_stdin(
                     "data must be at most {STDIN_MAX_BYTES} decoded bytes"
                 )));
             }
-            String::from_utf8(bytes)
-                .map_err(|_| ApiError::bad_request("stdin data must be valid UTF-8"))?
+            bytes
         }
         _ => return Err(ApiError::bad_request("encoding must be utf-8 or base64")),
     };
@@ -457,7 +479,7 @@ pub(super) async fn write_thread_job_stdin(
     tokio::task::spawn_blocking(move || {
         let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
         guard
-            .write_stdin(&job_id, &input, close)
+            .write_stdin_bytes(&job_id, &input, close)
             .map_err(map_job_err)
     })
     .await
@@ -486,10 +508,47 @@ pub(super) async fn kill_thread_job(
         let result = guard.kill(&job_id).map_err(map_job_err)?;
         let snapshot = guard.inspect_job(&job_id).map_err(map_job_err)?.snapshot;
         Ok(Json(KillJobResponse {
-            job: JobView::new(snapshot, thread_id),
+            job: JobView::new(snapshot, thread_id, &guard),
             result,
         }))
     })
     .await
     .map_err(|_| ApiError::internal("job kill failed"))?
 }
+
+/// `POST /v1/threads/{id}/jobs/{job_id}/resize` — resize this existing PTY.
+pub(super) async fn resize_thread_job(
+    State(state): State<RuntimeApiState>,
+    Path((thread_id, job_id)): Path<(String, String)>,
+    Json(size): Json<PtyDimensions>,
+) -> Result<Json<JobDetailResponse>, ApiError> {
+    let size = size
+        .validate()
+        .map_err(|error| ApiError::bad_request(error.to_string()))?;
+    if job_id.len() > JOB_ID_MAX_BYTES {
+        return Err(ApiError::not_found("job not found"));
+    }
+    let manager = thread_manager(&state, &thread_id, false).await?;
+    tokio::task::spawn_blocking(move || {
+        let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
+        let detail = guard.inspect_job(&job_id).map_err(map_job_err)?;
+        if detail.snapshot.stale || detail.snapshot.status != ShellStatus::Running {
+            return Err(ApiError {
+                status: StatusCode::CONFLICT,
+                message: "This job is no longer running".into(),
+            });
+        }
+        if guard.job_terminal_size(&job_id).is_none() {
+            return Err(ApiError::bad_request("This job is not a PTY"));
+        }
+        guard.resize_pty(&job_id, size).map_err(map_job_err)?;
+        let detail = guard.inspect_job(&job_id).map_err(map_job_err)?;
+        Ok(Json(JobDetailResponse {
+            job: JobView::new(detail.snapshot, thread_id, &guard),
+            stdout_tail: detail.stdout,
+            stderr_tail: detail.stderr,
+        }))
+    })
+    .await
+    .map_err(|_| ApiError::internal("PTY resize failed"))?
+}

crates/tui/src/runtime_api/tests.rs

@@ -18272,3 +18272,225 @@ async fn mcp_management_revision_precondition_prevents_stale_mutation() -> Resul
     handle.abort();
     Ok(())
 }
+
+#[cfg(all(unix, not(target_env = "ohos")))]
+#[tokio::test]
+async fn jobs_api_pty_resize_raw_input_and_nonblocking_poll() -> Result<()> {
+    use base64::Engine as _;
+    let tmp = tempfile::tempdir()?;
+    let workspace = tmp.path().join("workspace");
+    fs::create_dir_all(&workspace)?;
+    let (addr, _, handle) = spawn_test_server_with_root_token_mobile_workspace(
+        tmp.path().join("runtime"),
+        tmp.path().join("sessions"),
+        Some("pty-token".into()),
+        false,
+        workspace.clone(),
+    )
+    .await?
+    .context("PTY test requires loopback listener")?;
+    let client = crate::tls::reqwest_client();
+    let base = format!("http://{addr}");
+    let thread: Value = client
+        .post(format!("{base}/v1/threads"))
+        .bearer_auth("pty-token")
+        .json(&json!({"workspace":workspace,"allow_shell":true}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let thread_id = thread["id"].as_str().unwrap();
+    let jobs = format!("{base}/v1/threads/{thread_id}/jobs");
+    let created: Value = client.post(&jobs).bearer_auth("pty-token")
+        .json(&json!({"command":"stty -echo; printf ready; while IFS= read -r line; do stty size; done", "tty":true}))
+        .send().await?.error_for_status()?.json().await?;
+    assert_eq!(created["job"]["tty"], true);
+    assert_eq!(
+        created["job"]["terminal_size"],
+        json!({"rows":24,"cols":80})
+    );
+    let job_id = created["job"]["job_id"].as_str().unwrap();
+    let job = format!("{jobs}/{job_id}");
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        let output: Value = client
+            .get(format!("{job}/output?format=text"))
+            .bearer_auth("pty-token")
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        if output["data"].as_str().unwrap().contains("ready") {
+            break;
+        }
+        assert!(std::time::Instant::now() < deadline);
+        sleep(Duration::from_millis(20)).await;
+    }
+    assert_eq!(
+        client
+            .post(format!("{job}/resize"))
+            .json(&json!({"rows":30,"cols":90}))
+            .send()
+            .await?
+            .status(),
+        StatusCode::UNAUTHORIZED
+    );
+    for size in [json!({"rows":0,"cols":90}), json!({"rows":30,"cols":1001})] {
+        assert_eq!(
+            client
+                .post(format!("{job}/resize"))
+                .bearer_auth("pty-token")
+                .json(&size)
+                .send()
+                .await?
+                .status(),
+            StatusCode::BAD_REQUEST
+        );
+    }
+    let resized: Value = client
+        .post(format!("{job}/resize"))
+        .bearer_auth("pty-token")
+        .json(&json!({"rows":42,"cols":123}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(
+        resized["job"]["terminal_size"],
+        json!({"rows":42,"cols":123})
+    );
+    client
+        .post(format!("{job}/stdin"))
+        .bearer_auth("pty-token")
+        .json(&json!({"data":"size\n"}))
+        .send()
+        .await?
+        .error_for_status()?;
+    loop {
+        let output: Value = client
+            .get(format!("{job}/output?format=text"))
+            .bearer_auth("pty-token")
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        if output["data"].as_str().unwrap().contains("42 123") {
+            break;
+        }
+        assert!(std::time::Instant::now() < deadline);
+        sleep(Duration::from_millis(20)).await;
+    }
+    let other: Value = client
+        .post(format!("{base}/v1/threads"))
+        .bearer_auth("pty-token")
+        .json(&json!({"workspace":workspace,"allow_shell":true}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(
+        client
+            .post(format!(
+                "{base}/v1/threads/{}/jobs/{job_id}/resize",
+                other["id"].as_str().unwrap()
+            ))
+            .bearer_auth("pty-token")
+            .json(&json!({"rows":30,"cols":90}))
+            .send()
+            .await?
+            .status(),
+        StatusCode::NOT_FOUND
+    );
+    client
+        .post(format!("{job}/kill"))
+        .bearer_auth("pty-token")
+        .send()
+        .await?
+        .error_for_status()?;
+    assert_eq!(
+        client
+            .post(format!("{job}/resize"))
+            .bearer_auth("pty-token")
+            .json(&json!({"rows":30,"cols":90}))
+            .send()
+            .await?
+            .status(),
+        StatusCode::CONFLICT
+    );
+
+    let cat: Value = client
+        .post(&jobs)
+        .bearer_auth("pty-token")
+        .json(&json!({"command":"cat"}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(cat["job"]["tty"], false);
+    assert!(cat["job"]["terminal_size"].is_null());
+    let cat_job = format!("{jobs}/{}", cat["job"]["job_id"].as_str().unwrap());
+    assert_eq!(
+        client
+            .post(format!("{cat_job}/resize"))
+            .bearer_auth("pty-token")
+            .json(&json!({"rows":30,"cols":90}))
+            .send()
+            .await?
+            .status(),
+        StatusCode::BAD_REQUEST
+    );
+    // A 5-second output long-poll must not hold the owner lock against stdin.
+    let poll_client = client.clone();
+    let poll_url = format!("{cat_job}/output?wait_ms=5000");
+    let polling = tokio::spawn(async move {
+        poll_client
+            .get(poll_url)
+            .bearer_auth("pty-token")
+            .send()
+            .await
+    });
+    sleep(Duration::from_millis(100)).await;
+    let bytes = b"\0\xff\x1b[A\x03\n";
+    tokio::time::timeout(Duration::from_secs(2), client.post(format!("{cat_job}/stdin"))
+        .bearer_auth("pty-token").json(&json!({"data":base64::engine::general_purpose::STANDARD.encode(bytes), "encoding":"base64", "close":true})).send()).await??.error_for_status()?;
+    let _ = polling.await??.error_for_status()?;
+    let output: Value = client
+        .get(format!("{cat_job}/output?wait_ms=1000"))
+        .bearer_auth("pty-token")
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(
+        base64::engine::general_purpose::STANDARD.decode(output["data"].as_str().unwrap())?,
+        bytes
+    );
+    let replay: Value = client
+        .get(format!("{cat_job}/output"))
+        .bearer_auth("pty-token")
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(replay["data"], output["data"]);
+    let end = output["next_cursor"].as_u64().unwrap();
+    let tail: Value = client
+        .get(format!("{cat_job}/output?cursor={end}"))
+        .bearer_auth("pty-token")
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(tail["data"], "");
+    handle.abort();
+    Ok(())
+}

crates/tui/src/tools/shell.rs

@@ -963,6 +963,31 @@ fn recv_sync_reader_output(rx: &std::sync::mpsc::Receiver<Vec<u8>>) -> Vec<u8> {
         .unwrap_or_default()
 }
 
+/// Cell dimensions accepted by the existing PTY owner. Pixel sizes remain
+/// unspecified; callers must not allocate an unbounded terminal grid.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct PtyDimensions {
+    pub rows: u16,
+    pub cols: u16,
+}
+
+impl Default for PtyDimensions {
+    fn default() -> Self {
+        Self { rows: 24, cols: 80 }
+    }
+}
+
+impl PtyDimensions {
+    pub fn validate(self) -> Result<Self> {
+        anyhow::ensure!(
+            (1..=1000).contains(&self.rows) && (1..=1000).contains(&self.cols),
+            "PTY rows and columns must each be between 1 and 1000"
+        );
+        Ok(self)
+    }
+}
+
 /// A background shell process being tracked
 pub struct BackgroundShell {
     pub id: String,
@@ -994,6 +1019,10 @@ pub struct BackgroundShell {
     stderr_cursor: usize,
     completion_reported: bool,
     stdin: Option<StdinWriter>,
+    /// Retain the existing PTY owner for resize; never create another session.
+    #[cfg(not(target_env = "ohos"))]
+    pty_master: Option<Box<dyn portable_pty::MasterPty + Send>>,
+    terminal_size: Option<PtyDimensions>,
     child: Option<ShellChild>,
     #[cfg(windows)]
     windows_job: Option<WindowsJob>,
@@ -1308,16 +1337,22 @@ impl BackgroundShell {
             finish_background_reader(handle, &self.status, self.stderr_buffer.as_ref());
         }
         self.stdin = None;
+        #[cfg(not(target_env = "ohos"))]
+        {
+            self.pty_master = None;
+        }
         self.child = None;
     }
 
     fn write_stdin(&mut self, input: &str, close: bool) -> Result<()> {
+        self.write_stdin_bytes(input.as_bytes(), close)
+    }
+
+    fn write_stdin_bytes(&mut self, input: &[u8], close: bool) -> Result<()> {
         if let Some(stdin) = self.stdin.as_mut() {
             if !input.is_empty() {
-                stdin
-                    .write_all(input.as_bytes())
-                    .context("Failed to write to stdin")?;
-                stdin.flush().ok();
+                stdin.write_all(input).context("Failed to write to stdin")?;
+                stdin.flush().context("Failed to flush stdin")?;
             }
             if close {
                 self.stdin = None;
@@ -1874,6 +1909,9 @@ impl ShellManager {
                 stderr_cursor: 0,
                 completion_reported: false,
                 stdin: None,
+                #[cfg(not(target_env = "ohos"))]
+                pty_master: None,
+                terminal_size: None,
                 child: None,
                 #[cfg(windows)]
                 windows_job: None,
@@ -2554,6 +2592,8 @@ impl ShellManager {
         #[cfg(windows)]
         let mut windows_job = None;
 
+        #[cfg(not(target_env = "ohos"))]
+        let mut pty_master = None;
         let (child, stdin, stdout_thread, stderr_thread) = if tty {
             #[cfg(target_env = "ohos")]
             unreachable!("OHOS TTY mode returns before PTY setup");
@@ -2600,6 +2640,7 @@ impl ShellManager {
                     }
                 };
                 let stdout_thread = Some(spawn_reader_thread(reader, Arc::clone(&stdout_buffer)));
+                pty_master = Some(pair.master);
 
                 (
                     ShellChild::Pty(child),
@@ -2729,6 +2770,9 @@ impl ShellManager {
             stderr_cursor: 0,
             completion_reported: false,
             stdin,
+            #[cfg(not(target_env = "ohos"))]
+            pty_master,
+            terminal_size: tty.then(PtyDimensions::default),
             child: Some(child),
             #[cfg(windows)]
             windows_job,
@@ -2844,14 +2888,58 @@ impl ShellManager {
 
     /// Write data to stdin of a background process.
     pub fn write_stdin(&mut self, task_id: &str, input: &str, close: bool) -> Result<()> {
+        self.write_stdin_bytes(task_id, input.as_bytes(), close)
+    }
+
+    /// Exact input bytes from an authenticated client; never round-trip through UTF-8.
+    pub fn write_stdin_bytes(&mut self, task_id: &str, input: &[u8], close: bool) -> Result<()> {
         let shell = self
             .processes
             .get_mut(task_id)
             .ok_or_else(|| anyhow!("Task {task_id} not found"))?;
-        shell.write_stdin(input, close)?;
+        shell.write_stdin_bytes(input, close)?;
         Ok(())
     }
 
+    /// Historical evicted records have no terminal-size evidence.
+    pub fn job_terminal_size(&self, task_id: &str) -> Option<PtyDimensions> {
+        self.processes
+            .get(task_id)
+            .and_then(|shell| shell.terminal_size)
+    }
+
+    pub fn resize_pty(&mut self, task_id: &str, size: PtyDimensions) -> Result<()> {
+        let size = size.validate()?;
+        let shell = self
+            .processes
+            .get_mut(task_id)
+            .ok_or_else(|| anyhow!("Job {task_id} not found"))?;
+        shell.poll();
+        anyhow::ensure!(
+            shell.status == ShellStatus::Running,
+            "PTY job is no longer running"
+        );
+        #[cfg(not(target_env = "ohos"))]
+        {
+            let master = shell.pty_master.as_ref().context("Job is not a PTY")?;
+            master
+                .resize(PtySize {
+                    rows: size.rows,
+                    cols: size.cols,
+                    pixel_width: 0,
+                    pixel_height: 0,
+                })
+                .context("Failed to resize PTY")?;
+            shell.terminal_size = Some(size);
+            Ok(())
+        }
+        #[cfg(target_env = "ohos")]
+        {
+            let _ = size;
+            Err(anyhow!("PTY resize is unavailable on this platform"))
+        }
+    }
+
     pub fn write_stdin_for_session(
         &mut self,
         active_session_id: &str,

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

@@ -3566,6 +3566,8 @@ fn killed_shell_does_not_wait_for_blocked_reader_threads() {
         completion_reported: false,
         bounded_output: None,
         stdin: None,
+        pty_master: None,
+        terminal_size: None,
         child: None,
         windows_job: None,
         stdout_thread: Some(stdout_thread),
@@ -4705,3 +4707,191 @@ async fn busy_work_graph_degrades_the_spawn_intent_instead_of_failing_it() {
         "a busy work-graph must degrade to an unbound guard, not fail the spawn"
     );
 }
+
+#[test]
+fn pty_dimensions_reject_zero_and_unbounded_grid() {
+    assert_eq!(
+        PtyDimensions::default(),
+        PtyDimensions { rows: 24, cols: 80 }
+    );
+    for size in [
+        PtyDimensions { rows: 0, cols: 80 },
+        PtyDimensions { rows: 24, cols: 0 },
+        PtyDimensions {
+            rows: 1001,
+            cols: 80,
+        },
+        PtyDimensions {
+            rows: 24,
+            cols: u16::MAX,
+        },
+    ] {
+        assert!(size.validate().is_err());
+    }
+    assert!(
+        PtyDimensions {
+            rows: 1000,
+            cols: 1000
+        }
+        .validate()
+        .is_ok()
+    );
+}
+
+#[test]
+#[cfg(not(target_env = "ohos"))]
+fn pty_stdin_preserves_bytes_and_reports_flush_failure() {
+    struct FlushFailure(Arc<Mutex<Vec<u8>>>);
+    impl Write for FlushFailure {
+        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+            self.0.lock().unwrap().extend_from_slice(bytes);
+            Ok(bytes.len())
+        }
+        fn flush(&mut self) -> io::Result<()> {
+            Err(io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "fixture flush failure",
+            ))
+        }
+    }
+    let tmp = tempdir().unwrap();
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    manager.seed_finished_record_for_test("input-fixture", Duration::ZERO);
+    let bytes = Arc::new(Mutex::new(Vec::new()));
+    manager.processes.get_mut("input-fixture").unwrap().stdin =
+        Some(StdinWriter::Pty(Box::new(FlushFailure(bytes.clone()))));
+    let input = b"\0\xff\x1b[A\x03";
+    let error = manager
+        .write_stdin_bytes("input-fixture", input, false)
+        .unwrap_err();
+    assert!(error.to_string().contains("flush"));
+    assert_eq!(bytes.lock().unwrap().as_slice(), input);
+}
+
+#[test]
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn pty_resize_updates_live_terminal_and_rejects_finished_or_pipe_jobs() {
+    let tmp = tempdir().unwrap();
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let launched = manager
+        .execute_with_options_env(
+            "stty -echo; printf ready; while IFS= read -r line; do stty size; done",
+            None,
+            5000,
+            true,
+            None,
+            true,
+            Some(ExecutionSandboxPolicy::DangerFullAccess),
+            HashMap::new(),
+        )
+        .unwrap();
+    let id = launched.task_id.unwrap();
+    assert_eq!(
+        manager.job_terminal_size(&id),
+        Some(PtyDimensions::default())
+    );
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        let chunk = manager
+            .read_output_chunk(&id, ShellOutputStream::Stdout, 0, 4096, 50)
+            .unwrap();
+        if chunk.bytes.windows(5).any(|bytes| bytes == b"ready") {
+            break;
+        }
+        assert!(Instant::now() < deadline, "PTY did not become ready");
+    }
+    let size = PtyDimensions {
+        rows: 37,
+        cols: 111,
+    };
+    manager.resize_pty(&id, size).unwrap();
+    assert_eq!(manager.job_terminal_size(&id), Some(size));
+    assert!(
+        manager
+            .resize_pty(&id, PtyDimensions { rows: 0, cols: 1 })
+            .is_err()
+    );
+    assert_eq!(manager.job_terminal_size(&id), Some(size));
+    manager.write_stdin_bytes(&id, b"size\n", false).unwrap();
+    loop {
+        let chunk = manager
+            .read_output_chunk(&id, ShellOutputStream::Stdout, 0, 4096, 0)
+            .unwrap();
+        if String::from_utf8_lossy(&chunk.bytes).contains("37 111") {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "actual terminal size did not change"
+        );
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    manager.kill(&id).unwrap();
+    assert!(manager.resize_pty(&id, size).is_err());
+    assert!(manager.processes[&id].pty_master.is_none());
+    let pipe = manager
+        .execute_with_options_env(
+            "cat",
+            None,
+            5000,
+            true,
+            None,
+            false,
+            Some(ExecutionSandboxPolicy::DangerFullAccess),
+            HashMap::new(),
+        )
+        .unwrap()
+        .task_id
+        .unwrap();
+    assert!(manager.resize_pty(&pipe, size).is_err());
+    manager.kill(&pipe).unwrap();
+}
+
+#[test]
+#[cfg(all(unix, not(target_env = "ohos")))]
+fn pty_raw_stdin_roundtrips_nul_and_non_utf8_bytes() {
+    let tmp = tempdir().unwrap();
+    let mut manager = ShellManager::new(tmp.path().to_path_buf());
+    let input = b"\0\xff\x1b[A\x03\n";
+    let command = format!(
+        "stty raw -echo; printf ready; dd bs=1 count={} 2>/dev/null",
+        input.len()
+    );
+    let launched = manager
+        .execute_with_options_env(
+            &command,
+            None,
+            5000,
+            true,
+            None,
+            true,
+            Some(ExecutionSandboxPolicy::DangerFullAccess),
+            HashMap::new(),
+        )
+        .unwrap();
+    let id = launched.task_id.unwrap();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        let chunk = manager
+            .read_output_chunk(&id, ShellOutputStream::Stdout, 0, 4096, 0)
+            .unwrap();
+        if chunk.bytes == b"ready" {
+            break;
+        }
+        assert!(Instant::now() < deadline, "raw PTY did not become ready");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    manager.write_stdin_bytes(&id, input, false).unwrap();
+    loop {
+        let chunk = manager
+            .read_output_chunk(&id, ShellOutputStream::Stdout, 5, 4096, 0)
+            .unwrap();
+        if chunk.status != ShellStatus::Running {
+            assert_eq!(chunk.bytes, input);
+            assert_eq!(chunk.next_offset, 5 + input.len());
+            break;
+        }
+        assert!(Instant::now() < deadline, "raw PTY did not finish");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+}