refactor: render the shell in the client (#3487)

* feat: add opt-in client-rendered shell

* feat: add client-owned shell mouse controls

* feat: complete client-owned shell chrome

* feat: add client-owned shell settings and notifications

* feat: run client shell custom commands

* feat: render popup terminals in client shell

* feat: add client-owned scrollback and copy mode

* fix: preserve client worktree trust defaults

* feat: render startup diagnostics in client shell

* feat: render onboarding in client shell

* feat: render product announcements in client shell

* feat: render release notes in client shell

* feat: render mobile switcher in client shell

* feat: render kitty graphics in client shell

* fix: preserve client shell lifecycle coherence

* refactor: route client shell commands through endpoint connections

* refactor: make the client-rendered shell the default

* refactor: remove monolithic launch mode

* refactor: complete client-rendered shell cutover

* perf: retain client pane surface updates

* perf: scale retained pane surface updates

* perf: remove remote pane input latency

* test: make client shell checks platform-independent

* fix: prevent client timer starvation

* test: fix client shell platform gates

* test: accept retained surfaces in cross-area checks

* test: make client mode assertions platform-neutral

* fix: badge only outdated integrations

1 files changed, 37 insertions, 269 deletions

diff --git a/tests/cross_area.rs b/tests/cross_area.rs
index 6f1ab82b..9b856cb8 100644
--- a/tests/cross_area.rs
+++ b/tests/cross_area.rs
@@ -1,6 +1,6 @@
 //! Cross-area integration tests for end-to-end persistence flows.
 
-mod support;
+pub mod support;
 
 use std::fs;
 use std::io::{self, BufRead, BufReader, Read, Write};
@@ -11,11 +11,11 @@ use std::thread;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize};
-use serde::Deserialize;
 use serde_json::{json, Value};
 use support::{
-    cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid,
-    unregister_spawned_herdr_pid, CURRENT_PROTOCOL,
+    cleanup_test_base, client_shell_handshake, register_runtime_dir, register_spawned_herdr_pid,
+    unregister_spawned_herdr_pid, CURRENT_PROTOCOL, SERVER_MESSAGE_CLIENT_SHELL_SNAPSHOT,
+    SERVER_MESSAGE_PANE_SURFACE, SERVER_MESSAGE_PANE_SURFACE_PATCH,
 };
 
 fn unique_test_dir() -> PathBuf {
@@ -196,7 +196,7 @@ fn workspace_create(socket_path: &Path, label: &str) -> Value {
         socket_path,
         "workspace_create",
         "workspace.create",
-        json!({ "label": label }),
+        json!({ "label": label, "focus": true }),
     )
 }
 
@@ -365,16 +365,6 @@ fn encode_varint_u32(v: u32) -> Vec<u8> {
     }
 }
 
-fn encode_varint_u16(v: u16) -> Vec<u8> {
-    if v < 251 {
-        vec![v as u8]
-    } else {
-        let mut buf = vec![251u8];
-        buf.extend_from_slice(&v.to_le_bytes());
-        buf
-    }
-}
-
 fn frame_message(payload: &[u8]) -> Vec<u8> {
     let mut framed = (payload.len() as u32).to_le_bytes().to_vec();
     framed.extend_from_slice(payload);
@@ -414,76 +404,6 @@ fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), Stri
     }
 }
 
-fn client_handshake(stream: &mut UnixStream, version: u32, cols: u16, rows: u16) {
-    stream
-        .set_read_timeout(Some(Duration::from_secs(5)))
-        .expect("set read timeout");
-
-    // ClientMessage::Hello = variant 0
-    let mut payload = encode_varint_u32(0);
-    payload.extend_from_slice(&encode_varint_u32(version));
-    payload.extend_from_slice(&encode_varint_u16(cols));
-    payload.extend_from_slice(&encode_varint_u16(rows));
-    payload.extend_from_slice(&encode_varint_u32(8)); // cell_width_px
-    payload.extend_from_slice(&encode_varint_u32(16)); // cell_height_px
-    payload.extend_from_slice(&encode_varint_u32(0)); // RenderEncoding::SemanticFrame
-    payload.extend_from_slice(&encode_varint_u32(0)); // ClientKeybindings::Server
-    payload.extend_from_slice(&encode_varint_u32(0)); // ClientLaunchMode::App
-
-    stream
-        .write_all(&frame_message(&payload))
-        .expect("write hello");
-    stream.flush().expect("flush hello");
-
-    let mut len_buf = [0u8; 4];
-    stream
-        .read_exact(&mut len_buf)
-        .expect("read welcome length");
-    let len = u32::from_le_bytes(len_buf) as usize;
-    assert!(len > 0 && len <= 2 * 1024 * 1024, "unexpected welcome size");
-
-    let mut welcome_payload = vec![0u8; len];
-    stream
-        .read_exact(&mut welcome_payload)
-        .expect("read welcome payload");
-
-    let mut offset = 0;
-    let (variant, consumed) = decode_varint_u32(&welcome_payload, offset).expect("decode variant");
-    offset += consumed;
-    assert_eq!(variant, 0, "expected ServerMessage::Welcome variant");
-
-    let (_server_version, consumed) =
-        decode_varint_u32(&welcome_payload, offset).expect("decode version");
-    offset += consumed;
-
-    let (_encoding, consumed) =
-        decode_varint_u32(&welcome_payload, offset).expect("decode render encoding");
-    offset += consumed;
-
-    let option_tag = *welcome_payload
-        .get(offset)
-        .expect("welcome payload should contain Option tag");
-    if option_tag == 1 {
-        let (str_len, consumed) =
-            decode_varint_u32(&welcome_payload, offset + 1).expect("decode error length");
-        let start = offset + 1 + consumed;
-        let end = start + str_len as usize;
-        let err = String::from_utf8(welcome_payload[start..end].to_vec()).expect("utf8 error");
-        panic!("handshake rejected: {err}");
-    }
-}
-
-fn send_client_input(stream: &mut UnixStream, data: &[u8]) {
-    // ClientMessage::Input = variant 1
-    let mut payload = encode_varint_u32(1);
-    payload.extend_from_slice(&encode_varint_u32(data.len() as u32));
-    payload.extend_from_slice(data);
-    stream
-        .write_all(&frame_message(&payload))
-        .expect("write input");
-    stream.flush().expect("flush input");
-}
-
 fn send_client_detach(stream: &mut UnixStream) {
     // ClientMessage::Detach = variant 4
     let payload = encode_varint_u32(4);
@@ -500,112 +420,8 @@ fn is_timeout(err: &io::Error) -> bool {
     )
 }
 
-#[allow(dead_code)]
-#[derive(Debug, Deserialize)]
-struct FrameWire {
-    cells: Vec<CellWire>,
-    width: u16,
-    height: u16,
-    cursor: Option<CursorWire>,
-    hyperlinks: Vec<String>,
-    graphics: Vec<u8>,
-}
-
-#[allow(dead_code)]
-#[derive(Debug, Deserialize)]
-struct CellWire {
-    symbol: String,
-    fg: u32,
-    bg: u32,
-    modifier: u16,
-    skip: bool,
-    hyperlink: Option<u32>,
-}
-
-#[derive(Debug, Deserialize)]
-struct CursorWire {
-    x: u16,
-    y: u16,
-    visible: bool,
-    shape: u8,
-}
-
-fn decode_frame_payload(payload: &[u8]) -> io::Result<FrameWire> {
-    bincode::serde::decode_from_slice(payload, bincode::config::standard())
-        .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
-        .and_then(|(frame, consumed): (FrameWire, usize)| {
-            if consumed != payload.len() {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    format!(
-                        "frame payload had trailing bytes: consumed={}, len={}",
-                        consumed,
-                        payload.len()
-                    ),
-                ));
-            }
-            Ok(frame)
-        })
-}
-
-fn frame_contains_colored_symbol(frame: &FrameWire, symbol: &str, rgb: (u8, u8, u8)) -> bool {
-    let (r, g, b) = rgb;
-    let fg = 0x02_00_00_00 | (u32::from(r) << 16) | (u32::from(g) << 8) | u32::from(b);
-    frame
-        .cells
-        .iter()
-        .any(|cell| cell.symbol == symbol && cell.fg == fg)
-}
-
-fn frame_contains_text(frame: &FrameWire, needle: &str) -> bool {
-    if frame.cells.is_empty() {
-        return false;
-    }
-
-    let width = frame.width.max(1) as usize;
-    let mut text = String::new();
-    for row in frame.cells.chunks(width) {
-        for cell in row {
-            let _ = (cell.fg, cell.bg, cell.modifier, cell.skip);
-            text.push_str(&cell.symbol);
-        }
-        text.push('\n');
-    }
-    let _ = (frame.height, frame.graphics.len());
-    if let Some(cursor) = frame.cursor.as_ref() {
-        let _ = (cursor.x, cursor.y, cursor.visible, cursor.shape);
-    }
-
-    text.contains(needle)
-}
-
 fn read_server_variant(stream: &mut UnixStream, timeout: Duration) -> io::Result<u32> {
     stream.set_read_timeout(Some(timeout))?;
-
-    let mut len_buf = [0u8; 4];
-    stream.read_exact(&mut len_buf)?;
-    let len = u32::from_le_bytes(len_buf) as usize;
-    if len == 0 {
-        return Err(io::Error::new(
-            io::ErrorKind::InvalidData,
-            "zero-length payload",
-        ));
-    }
-
-    let mut payload = vec![0u8; len];
-    stream.read_exact(&mut payload)?;
-
-    let (variant, _consumed) = decode_varint_u32(&payload, 0)
-        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
-    Ok(variant)
-}
-
-fn read_server_message_payload(
-    stream: &mut UnixStream,
-    timeout: Duration,
-) -> io::Result<(u32, Vec<u8>)> {
-    stream.set_read_timeout(Some(timeout))?;
-
     let mut len_buf = [0u8; 4];
     stream.read_exact(&mut len_buf)?;
     let len = u32::from_le_bytes(len_buf) as usize;
@@ -615,49 +431,20 @@ fn read_server_message_payload(
             "zero-length payload",
         ));
     }
-
     let mut payload = vec![0u8; len];
     stream.read_exact(&mut payload)?;
-
-    let (variant, consumed) = decode_varint_u32(&payload, 0)
-        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
-    Ok((variant, payload[consumed..].to_vec()))
-}
-
-fn wait_for_frame_matching(
-    stream: &mut UnixStream,
-    timeout: Duration,
-    predicate: impl Fn(&FrameWire) -> bool,
-) -> io::Result<bool> {
-    let deadline = Instant::now() + timeout;
-    while Instant::now() < deadline {
-        let slice = deadline
-            .saturating_duration_since(Instant::now())
-            .min(Duration::from_millis(80));
-        match read_server_message_payload(stream, slice) {
-            Ok((1, payload)) => {
-                let frame = decode_frame_payload(&payload)?;
-                if predicate(&frame) {
-                    return Ok(true);
-                }
-            }
-            Ok((_variant, _payload)) => {}
-            Err(err) if is_timeout(&err) => {}
-            Err(err) => return Err(err),
-        }
-    }
-
-    Ok(false)
+    decode_varint_u32(&payload, 0)
+        .map(|(variant, _)| variant)
+        .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
 }
 
 fn wait_for_frame(stream: &mut UnixStream, timeout: Duration) -> bool {
     let deadline = Instant::now() + timeout;
     while Instant::now() < deadline {
-        let slice = deadline
-            .saturating_duration_since(Instant::now())
-            .min(Duration::from_millis(80));
+        let slice = deadline.saturating_duration_since(Instant::now());
         match read_server_variant(stream, slice) {
-            Ok(1) => return true, // ServerMessage::Frame
+            Ok(SERVER_MESSAGE_PANE_SURFACE | SERVER_MESSAGE_PANE_SURFACE_PATCH) => return true,
+            Ok(SERVER_MESSAGE_CLIENT_SHELL_SNAPSHOT) => {}
             Ok(_) => {}
             Err(err) if is_timeout(&err) => {}
             Err(_) => return false,
@@ -669,7 +456,7 @@ fn wait_for_frame(stream: &mut UnixStream, timeout: Duration) -> bool {
 fn drain_server_messages(stream: &mut UnixStream, max_drain: Duration) {
     let deadline = Instant::now() + max_drain;
     while Instant::now() < deadline {
-        match read_server_variant(stream, Duration::from_millis(40)) {
+        match read_server_variant(stream, deadline.saturating_duration_since(Instant::now())) {
             Ok(_) => {}
             Err(err) if is_timeout(&err) => break,
             Err(_) => break,
@@ -696,7 +483,7 @@ fn cross_area_detach_and_reattach_preserves_state() {
 
     // Local attach (client A).
     let mut client_a = UnixStream::connect(&client_socket).expect("client A should connect");
-    client_handshake(&mut client_a, CURRENT_PROTOCOL, 100, 30);
+    client_shell_handshake(&mut client_a, CURRENT_PROTOCOL, 100, 30).expect("shell handshake");
     assert!(wait_for_frame(&mut client_a, Duration::from_secs(2)));
 
     // Use herdr: create a workspace and write output into its pane.
@@ -733,7 +520,7 @@ fn cross_area_detach_and_reattach_preserves_state() {
 
     // Reattach from another terminal/session (client B).
     let mut client_b = UnixStream::connect(&client_socket).expect("client B should connect");
-    client_handshake(&mut client_b, CURRENT_PROTOCOL, 80, 24);
+    client_shell_handshake(&mut client_b, CURRENT_PROTOCOL, 80, 24).expect("shell handshake");
     assert!(
         wait_for_frame(&mut client_b, Duration::from_secs(5)),
         "reattached client should receive frame"
@@ -789,7 +576,7 @@ fn cross_area_agent_process_survives_detach_and_reattach() {
     wait_for_socket(&client_socket, Duration::from_secs(10));
 
     let mut client_a = UnixStream::connect(&client_socket).expect("client A should connect");
-    client_handshake(&mut client_a, CURRENT_PROTOCOL, 100, 30);
+    client_shell_handshake(&mut client_a, CURRENT_PROTOCOL, 100, 30).expect("shell handshake");
     assert!(wait_for_frame(&mut client_a, Duration::from_secs(2)));
 
     let created = workspace_create(&api_socket, "agent-persist");
@@ -840,18 +627,10 @@ fn cross_area_agent_process_survives_detach_and_reattach() {
         "agent status should remain working while detached"
     );
 
-    // Reattach and ensure client-side state reflects the persisted working status.
+    // Reattach and verify the persisted projection through the API.
     let mut client_b = UnixStream::connect(&client_socket).expect("client B should connect");
-    client_handshake(&mut client_b, CURRENT_PROTOCOL, 80, 24);
-    let saw_working_on_client =
-        wait_for_frame_matching(&mut client_b, Duration::from_secs(5), |frame| {
-            frame_contains_colored_symbol(frame, "●", (249, 226, 175))
-        })
-        .expect("frame decoding should succeed");
-    assert!(
-        saw_working_on_client,
-        "reattached client frame should expose persisted agent working status"
-    );
+    client_shell_handshake(&mut client_b, CURRENT_PROTOCOL, 80, 24).expect("shell handshake");
+    assert!(wait_for_frame(&mut client_b, Duration::from_secs(5)));
 
     // Transition to blocked and verify API + client surfaces both observe it.
     // The fake process remains visibly working, so blocked is the deterministic
@@ -862,15 +641,7 @@ fn cross_area_agent_process_survives_detach_and_reattach() {
         "pane agent status should transition to blocked"
     );
 
-    let saw_blocked_on_client =
-        wait_for_frame_matching(&mut client_b, Duration::from_secs(5), |frame| {
-            frame_contains_colored_symbol(frame, "●", (243, 139, 168))
-        })
-        .expect("frame decoding should succeed");
-    assert!(
-        saw_blocked_on_client,
-        "reattached client frame should show blocked status after transition"
-    );
+    // The API status above is the stable cross-area contract for this transition.
 
     cleanup_spawned_herdr(server, base);
 }
@@ -889,7 +660,7 @@ fn cross_area_client_and_api_workspace_views_are_consistent() {
     wait_for_socket(&client_socket, Duration::from_secs(10));
 
     let mut client = UnixStream::connect(&client_socket).expect("client should connect");
-    client_handshake(&mut client, CURRENT_PROTOCOL, 100, 30);
+    client_shell_handshake(&mut client, CURRENT_PROTOCOL, 100, 30).expect("shell handshake");
     assert!(wait_for_frame(&mut client, Duration::from_secs(2)));
     drain_server_messages(&mut client, Duration::from_millis(300));
 
@@ -902,17 +673,8 @@ fn cross_area_client_and_api_workspace_views_are_consistent() {
         .expect("workspace.create should return workspace_id")
         .to_string();
 
-    // The attached client must receive a frame that includes the new workspace
-    // label, proving client-side state reflects the API surface.
-    let saw_workspace_on_client =
-        wait_for_frame_matching(&mut client, Duration::from_secs(3), |frame| {
-            frame_contains_text(frame, "api-visible-workspace")
-        })
-        .expect("frame decoding should succeed");
-    assert!(
-        saw_workspace_on_client,
-        "client-side frame should include the newly created workspace label"
-    );
+    // ClientShell state is asserted through the authoritative API projection below;
+    // do not decode client-composed UI to duplicate that assertion.
 
     let deadline = Instant::now() + Duration::from_secs(5);
     let mut count_reached = false;
@@ -952,9 +714,9 @@ fn cross_area_two_clients_shared_view_and_single_detach_stability() {
     wait_for_socket(&client_socket, Duration::from_secs(10));
 
     let mut client_a = UnixStream::connect(&client_socket).expect("client A should connect");
-    client_handshake(&mut client_a, CURRENT_PROTOCOL, 110, 30);
+    client_shell_handshake(&mut client_a, CURRENT_PROTOCOL, 110, 30).expect("shell handshake");
     let mut client_b = UnixStream::connect(&client_socket).expect("client B should connect");
-    client_handshake(&mut client_b, CURRENT_PROTOCOL, 100, 30);
+    client_shell_handshake(&mut client_b, CURRENT_PROTOCOL, 100, 30).expect("shell handshake");
 
     assert!(wait_for_frame(&mut client_a, Duration::from_secs(2)));
     assert!(wait_for_frame(&mut client_b, Duration::from_secs(2)));
@@ -968,7 +730,7 @@ fn cross_area_two_clients_shared_view_and_single_detach_stability() {
         .to_string();
 
     // Input from client A should update shared state visible to client B.
-    send_client_input(&mut client_a, b"echo SHARED_VIEW\n");
+    pane_send_text(&api_socket, &pane_id, "echo SHARED_VIEW\n");
     assert!(
         wait_for_frame(&mut client_b, Duration::from_secs(2)),
         "client B should receive update from client A"
@@ -982,19 +744,24 @@ fn cross_area_two_clients_shared_view_and_single_detach_stability() {
 
     // Detach client A; client B should keep working.
     send_client_detach(&mut client_a);
-    drop(client_a);
-
-    send_client_input(&mut client_b, b"echo AFTER_A_DETACH\n");
     assert!(
-        wait_for_frame(&mut client_b, Duration::from_secs(2)),
-        "remaining client should still receive frames after other client detaches"
+        support::wait_for_disconnect(&mut client_a, Duration::from_secs(2))
+            .expect("wait for client A detach"),
+        "client A should disconnect before the remaining-client assertion"
     );
+    drop(client_a);
+
+    pane_send_text(&api_socket, &pane_id, "echo AFTER_A_DETACH\n");
     assert!(pane_read_recent_contains(
         &api_socket,
         &pane_id,
         "AFTER_A_DETACH",
         Duration::from_secs(5)
     ));
+    assert!(
+        wait_for_frame(&mut client_b, Duration::from_secs(2)),
+        "remaining client should still receive frames after other client detaches"
+    );
 
     let ping = ping_socket(&api_socket);
     assert!(
@@ -1123,7 +890,8 @@ fn cross_area_server_kill_then_restart_and_reconnect() {
 
     let mut reconnect_client =
         UnixStream::connect(&client_socket).expect("new client should connect after restart");
-    client_handshake(&mut reconnect_client, CURRENT_PROTOCOL, 80, 24);
+    client_shell_handshake(&mut reconnect_client, CURRENT_PROTOCOL, 80, 24)
+        .expect("shell handshake");
     assert!(
         wait_for_frame(&mut reconnect_client, Duration::from_secs(5)),
         "new client should receive frame after restart"