feat: add stable client endpoint compatibility (#3509)

* feat: add stable client endpoint compatibility

* fix: keep portable-pty patch checks composable

* ci: run maintenance checks in pull requests

* fix: reject endpoint errors from terminal clients

* feat: keep compatible servers running across updates

7 files changed, 733 insertions, 198 deletions

diff --git a/src/server/autodetect.rs b/src/server/autodetect.rs
index 1813ebd9..bd364a70 100644
--- a/src/server/autodetect.rs
+++ b/src/server/autodetect.rs
@@ -150,19 +150,22 @@ fn validate_running_server_compatibility() -> io::Result<()> {
         )));
     };
 
-    if status.protocol == Some(crate::protocol::PROTOCOL_VERSION) {
+    let endpoint_generation = status
+        .capabilities
+        .as_ref()
+        .and_then(|capabilities| capabilities.endpoint_protocol_generation);
+    if endpoint_generation == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION) {
         return Ok(());
     }
 
     Err(io::Error::other(format!(
-        "Herdr was updated, but this session is still running the old server.\n\nserver: v{} protocol {}\nclient: v{} protocol {}\n\n{}",
+        "This session predates Herdr's stable endpoint protocol and needs one final server update.\n\nserver: v{} endpoint generation {}\nclient: v{} endpoint generation {}\n\n{}",
         status.version.as_deref().unwrap_or("unknown"),
-        status
-            .protocol
+        endpoint_generation
             .map(|value| value.to_string())
-            .unwrap_or_else(|| "unknown".to_string()),
+            .unwrap_or_else(|| "unavailable".to_string()),
         crate::build_info::version(),
-        crate::protocol::PROTOCOL_VERSION,
+        crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
         crate::session::active_restart_after_update_guidance()
     )))
 }
diff --git a/src/server/client_commands.rs b/src/server/client_commands.rs
index eeb49c7e..86243e31 100644
--- a/src/server/client_commands.rs
+++ b/src/server/client_commands.rs
@@ -12,47 +12,56 @@ pub(crate) const MAX_ENDPOINT_BOOT_ID_BYTES: usize = 128;
 pub(crate) const MAX_ENDPOINT_REQUEST_ID_BYTES: usize = 128;
 const ENDPOINT_RESPONSE_CHUNK_BYTES: usize = 512 * 1024;
 
+const CLIENT_SHELL_METHODS: &[&str] = &[
+    "command.invoke",
+    "integration.install",
+    "integration.list",
+    "layout.set_split_ratio",
+    "pane.close",
+    "pane.copy_motion",
+    "pane.copy_search",
+    "pane.edit_scrollback",
+    "pane.focus",
+    "pane.focus_direction",
+    "pane.input.set",
+    "pane.link.activate",
+    "pane.rename",
+    "pane.resize",
+    "pane.scroll",
+    "pane.selection.read",
+    "pane.split",
+    "pane.swap",
+    "pane.zoom",
+    "product_announcement.dismiss",
+    "release_notes.dismiss",
+    "server.reload_config",
+    "tab.close",
+    "tab.create",
+    "tab.focus",
+    "tab.move",
+    "tab.rename",
+    "workspace.close",
+    "workspace.create",
+    "workspace.focus",
+    "workspace.move",
+    "workspace.move_block",
+    "workspace.rename",
+    "worktree.create",
+    "worktree.list",
+    "worktree.open",
+    "worktree.remove",
+];
+
+pub(crate) fn supported_client_shell_method_names() -> &'static [&'static str] {
+    CLIENT_SHELL_METHODS
+}
+
+pub(crate) fn supports_client_shell_method_name(method: &str) -> bool {
+    CLIENT_SHELL_METHODS.contains(&method)
+}
+
 pub(crate) fn supports_client_shell_method(method: &Method) -> bool {
-    matches!(
-        method,
-        Method::CommandInvoke(_)
-            | Method::IntegrationInstall(_)
-            | Method::IntegrationList(_)
-            | Method::LayoutSetSplitRatio(_)
-            | Method::PaneClose(_)
-            | Method::PaneCopyMotion(_)
-            | Method::PaneCopySearch(_)
-            | Method::PaneEditScrollback(_)
-            | Method::PaneFocus(_)
-            | Method::PaneFocusDirection(_)
-            | Method::PaneInputSet(_)
-            | Method::PaneLinkActivate(_)
-            | Method::PaneRename(_)
-            | Method::PaneResize(_)
-            | Method::PaneScroll(_)
-            | Method::PaneSelectionRead(_)
-            | Method::PaneSplit(_)
-            | Method::PaneSwap(_)
-            | Method::PaneZoom(_)
-            | Method::ProductAnnouncementDismiss(_)
-            | Method::ReleaseNotesDismiss(_)
-            | Method::ServerReloadConfig(_)
-            | Method::TabClose(_)
-            | Method::TabCreate(_)
-            | Method::TabFocus(_)
-            | Method::TabMove(_)
-            | Method::TabRename(_)
-            | Method::WorkspaceClose(_)
-            | Method::WorkspaceCreate(_)
-            | Method::WorkspaceFocus(_)
-            | Method::WorkspaceMove(_)
-            | Method::WorkspaceMoveBlock(_)
-            | Method::WorkspaceRename(_)
-            | Method::WorktreeCreate(_)
-            | Method::WorktreeList(_)
-            | Method::WorktreeOpen(_)
-            | Method::WorktreeRemove(_)
-    )
+    supports_client_shell_method_name(crate::api::api_method_name(method))
 }
 
 pub(crate) fn error_response(id: String, code: &str, message: impl Into<String>) -> String {
@@ -134,8 +143,159 @@ pub(crate) fn spawn_response_waiter(
 
 #[cfg(test)]
 mod tests {
+    use std::collections::{BTreeMap, BTreeSet};
+
+    use sha2::{Digest, Sha256};
+
     use super::*;
 
+    fn collect_schema_refs(value: &serde_json::Value, refs: &mut BTreeSet<String>) {
+        match value {
+            serde_json::Value::Object(object) => {
+                if let Some(reference) = object.get("$ref").and_then(serde_json::Value::as_str) {
+                    if let Some(name) = reference.rsplit('/').next() {
+                        refs.insert(name.to_owned());
+                    }
+                }
+                for value in object.values() {
+                    collect_schema_refs(value, refs);
+                }
+            }
+            serde_json::Value::Array(values) => {
+                for value in values {
+                    collect_schema_refs(value, refs);
+                }
+            }
+            _ => {}
+        }
+    }
+
+    fn normalized_wire_schema(value: &serde_json::Value) -> serde_json::Value {
+        match value {
+            serde_json::Value::Object(object) => serde_json::Value::Object(
+                object
+                    .iter()
+                    .filter(|(key, _)| {
+                        !matches!(
+                            key.as_str(),
+                            "description" | "examples" | "readOnly" | "title" | "writeOnly"
+                        )
+                    })
+                    .map(|(key, value)| (key.clone(), normalized_wire_schema(value)))
+                    .collect(),
+            ),
+            serde_json::Value::Array(values) => {
+                serde_json::Value::Array(values.iter().map(normalized_wire_schema).collect())
+            }
+            _ => value.clone(),
+        }
+    }
+
+    fn endpoint_method_shape_digests() -> BTreeMap<String, String> {
+        let schema = serde_json::to_value(schemars::schema_for!(crate::api::schema::Request))
+            .expect("request schema");
+        let definitions = schema
+            .get("$defs")
+            .and_then(serde_json::Value::as_object)
+            .expect("request definitions");
+        let branches = schema
+            .get("oneOf")
+            .and_then(serde_json::Value::as_array)
+            .expect("request method branches");
+        let mut digests = BTreeMap::new();
+
+        for method in CLIENT_SHELL_METHODS {
+            let branch = branches
+                .iter()
+                .find(|branch| {
+                    branch
+                        .pointer("/properties/method/const")
+                        .and_then(serde_json::Value::as_str)
+                        == Some(method)
+                })
+                .unwrap_or_else(|| panic!("missing request schema branch for {method}"));
+            let mut referenced_names = BTreeSet::new();
+            collect_schema_refs(branch, &mut referenced_names);
+            let mut visited_names = BTreeSet::new();
+            let mut selected_definitions = serde_json::Map::new();
+            while let Some(name) = referenced_names.pop_first() {
+                if !visited_names.insert(name.clone()) {
+                    continue;
+                }
+                let definition = definitions
+                    .get(&name)
+                    .unwrap_or_else(|| panic!("missing schema definition {name} for {method}"));
+                collect_schema_refs(definition, &mut referenced_names);
+                selected_definitions.insert(name, normalized_wire_schema(definition));
+            }
+            let shape = serde_json::json!({
+                "request": normalized_wire_schema(branch),
+                "definitions": selected_definitions,
+            });
+            let bytes = serde_json::to_vec(&shape).expect("method shape json");
+            digests.insert(method.to_string(), format!("{:x}", Sha256::digest(bytes)));
+        }
+
+        digests
+    }
+
+    #[test]
+    fn advertised_client_shell_method_shapes_stay_at_the_v1_contract() {
+        let expected: BTreeMap<String, String> = serde_json::from_str(include_str!(concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/tests/fixtures/endpoint-method-shapes-v1.json"
+        )))
+        .expect("endpoint method shape fixture");
+        let actual = endpoint_method_shape_digests();
+
+        assert_eq!(
+            actual,
+            expected,
+            "an existing endpoint method changed shape; add load-bearing behavior as a new advertised method or explicitly gate new fields"
+        );
+    }
+
+    #[test]
+    fn advertised_client_shell_methods_are_sorted_unique_and_in_schema() {
+        assert!(CLIENT_SHELL_METHODS
+            .windows(2)
+            .all(|pair| pair[0] < pair[1]));
+
+        fn collect_method_constants(value: &serde_json::Value, methods: &mut Vec<String>) {
+            match value {
+                serde_json::Value::Object(object) => {
+                    if let Some(method) = object
+                        .get("const")
+                        .and_then(serde_json::Value::as_str)
+                        .filter(|value| value.contains('.'))
+                    {
+                        methods.push(method.to_owned());
+                    }
+                    for value in object.values() {
+                        collect_method_constants(value, methods);
+                    }
+                }
+                serde_json::Value::Array(values) => {
+                    for value in values {
+                        collect_method_constants(value, methods);
+                    }
+                }
+                _ => {}
+            }
+        }
+
+        let schema = serde_json::to_value(schemars::schema_for!(crate::api::schema::Request))
+            .expect("request schema");
+        let mut schema_methods = Vec::new();
+        collect_method_constants(&schema, &mut schema_methods);
+        for method in CLIENT_SHELL_METHODS {
+            assert!(
+                schema_methods.iter().any(|candidate| candidate == method),
+                "advertised endpoint method {method:?} is absent from the request schema"
+            );
+        }
+    }
+
     #[test]
     fn client_shell_lane_excludes_api_front_door_and_lifecycle_methods() {
         assert!(supports_client_shell_method(&Method::ServerReloadConfig(
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 673e16d3..25a963e7 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -17,6 +17,10 @@ use tokio::sync::mpsc;
 use tracing::{debug, warn};
 
 use crate::ipc::LocalStream;
+use crate::protocol::endpoint::{
+    EndpointClientHello, EndpointServerWelcome, ENDPOINT_HELLO_KIND, ENDPOINT_PROTOCOL_GENERATION,
+    ENDPOINT_WELCOME_KIND,
+};
 use crate::protocol::{
     self, AttachScrollDirection, AttachScrollSource, ClientMessage, ClientPaneInputEvent,
     RenderEncoding, ServerMessage, MAX_CLIPBOARD_IMAGE_PAYLOAD, MAX_FRAME_SIZE,
@@ -39,6 +43,74 @@ const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4);
 
 /// Maximum input payload size (bytes) for a single `ClientMessage::Input`.
 const MAX_INPUT_PAYLOAD: usize = 1024 * 1024; // 1 MB
+const MAX_CLIENT_SHELL_DIMENSION: u16 = 4096;
+const MAX_CLIENT_SHELL_CELLS: u32 = 1_000_000;
+const MAX_CLIENT_CELL_SIZE_PX: u32 = 4096;
+
+fn client_shell_geometry_error(
+    surface_size: crate::protocol::ClientSurfaceSize,
+    cell_width_px: u32,
+    cell_height_px: u32,
+) -> Option<&'static str> {
+    if surface_size.cols == 0 || surface_size.rows == 0 {
+        return Some("client shell requires a non-empty pane surface");
+    }
+    if surface_size.cols > MAX_CLIENT_SHELL_DIMENSION
+        || surface_size.rows > MAX_CLIENT_SHELL_DIMENSION
+        || u32::from(surface_size.cols) * u32::from(surface_size.rows) > MAX_CLIENT_SHELL_CELLS
+    {
+        return Some("client shell pane surface exceeds the safe geometry limit");
+    }
+    if cell_width_px > MAX_CLIENT_CELL_SIZE_PX || cell_height_px > MAX_CLIENT_CELL_SIZE_PX {
+        return Some("client shell cell pixel size exceeds the safe geometry limit");
+    }
+    None
+}
+
+#[derive(serde::Deserialize)]
+struct EndpointRequestHead {
+    id: String,
+    method: String,
+}
+
+enum DecodedEndpointRequest {
+    Dispatch(Box<crate::api::schema::Request>),
+    Error {
+        request_id: String,
+        code: &'static str,
+        message: String,
+    },
+}
+
+fn write_endpoint_rejection(stream: &mut LocalStream, code: &str, message: impl Into<String>) {
+    let welcome = EndpointServerWelcome::incompatible(code, message);
+    let response = ServerMessage::EndpointControl {
+        kind: ENDPOINT_WELCOME_KIND.into(),
+        data: serde_json::to_string(&welcome).unwrap_or_else(|_| "{}".into()),
+    };
+    let _ = protocol::write_message(stream, &response);
+}
+
+fn decode_endpoint_request(request: &str) -> serde_json::Result<DecodedEndpointRequest> {
+    let head = serde_json::from_str::<EndpointRequestHead>(request)?;
+    if !crate::server::client_commands::supports_client_shell_method_name(&head.method) {
+        return Ok(DecodedEndpointRequest::Error {
+            request_id: head.id,
+            code: "unsupported_method",
+            message: format!("method {:?} is not available on this machine", head.method),
+        });
+    }
+    Ok(
+        match serde_json::from_str::<crate::api::schema::Request>(request) {
+            Ok(request) => DecodedEndpointRequest::Dispatch(Box::new(request)),
+            Err(error) => DecodedEndpointRequest::Error {
+                request_id: head.id,
+                code: "invalid_request",
+                message: format!("invalid endpoint request: {error}"),
+            },
+        },
+    )
+}
 /// Maximum structured input events accepted in one client message.
 const MAX_INPUT_EVENT_BATCH: usize = 4096;
 
@@ -417,6 +489,14 @@ pub(crate) enum ServerEvent {
         boot_id: String,
         request: Box<crate::api::schema::Request>,
     },
+    /// A well-framed endpoint request could not be dispatched by this server.
+    ClientShellEndpointRequestError {
+        client_id: u64,
+        boot_id: String,
+        request_id: String,
+        code: &'static str,
+        message: String,
+    },
     /// One chunk of a deferred endpoint operation's final response is ready.
     ClientShellEndpointResponseChunkReady {
         client_id: u64,
@@ -616,50 +696,69 @@ pub(crate) fn handle_client_handshake(
             let (cols, rows) = clamp_terminal_size(cols, rows);
             (cols, rows, cell_width_px, cell_height_px, pixel_mouse, None)
         }
-        ClientMessage::ClientShellHello {
-            version,
-            cell_width_px,
-            cell_height_px,
-            surface_size,
-            pixel_mouse,
-            direct_graphics,
-            endpoint_keybindings,
-            mouse_capture,
-        } => {
-            if let protocol::VersionCheck::Incompatible(reason) =
-                protocol::check_client_version(version)
-            {
-                let welcome = ServerMessage::Welcome {
-                    version: PROTOCOL_VERSION,
-                    encoding: RenderEncoding::SemanticFrame,
-                    error: Some(reason),
-                };
-                let _ = protocol::write_message(&mut stream, &welcome);
-                return Ok(());
-            }
-            if surface_size.cols == 0 || surface_size.rows == 0 {
-                let welcome = ServerMessage::Welcome {
-                    version: PROTOCOL_VERSION,
-                    encoding: RenderEncoding::SemanticFrame,
-                    error: Some("client shell requires a non-empty pane surface".to_owned()),
-                };
-                let _ = protocol::write_message(&mut stream, &welcome);
+        ClientMessage::EndpointControl { kind, data } if kind == ENDPOINT_HELLO_KIND => {
+            let hello: EndpointClientHello = match serde_json::from_str(&data) {
+                Ok(hello) => hello,
+                Err(error) => {
+                    write_endpoint_rejection(
+                        &mut stream,
+                        "invalid_hello",
+                        format!("invalid endpoint hello: {error}"),
+                    );
+                    return Ok(());
+                }
+            };
+            let incompatibility = if hello.generation != ENDPOINT_PROTOCOL_GENERATION {
+                Some((
+                    "unsupported_generation",
+                    format!(
+                        "endpoint generation {} is unsupported; this server supports generation {ENDPOINT_PROTOCOL_GENERATION}",
+                        hello.generation
+                    ),
+                ))
+            } else if !hello.supports_required_codecs() {
+                Some((
+                    "no_common_core",
+                    "client and server have no compatible endpoint core codecs".to_owned(),
+                ))
+            } else {
+                client_shell_geometry_error(
+                    hello.surface_size,
+                    hello.cell_width_px,
+                    hello.cell_height_px,
+                )
+                .map(|reason| ("invalid_surface", reason.to_owned()))
+            };
+            if let Some((code, reason)) = incompatibility {
+                write_endpoint_rejection(&mut stream, code, reason);
                 return Ok(());
             }
             (
-                surface_size.cols,
-                surface_size.rows,
-                cell_width_px,
-                cell_height_px,
+                hello.surface_size.cols,
+                hello.surface_size.rows,
+                hello.cell_width_px,
+                hello.cell_height_px,
                 false,
                 Some((
-                    pixel_mouse,
-                    direct_graphics,
-                    endpoint_keybindings,
-                    mouse_capture,
+                    hello.pixel_mouse,
+                    hello.direct_graphics,
+                    hello.endpoint_keybindings,
+                    hello.mouse_capture,
                 )),
             )
         }
+        ClientMessage::ClientShellHello { .. } => {
+            let welcome = ServerMessage::Welcome {
+                version: PROTOCOL_VERSION,
+                encoding: RenderEncoding::SemanticFrame,
+                error: Some(
+                    "this client predates the stable endpoint protocol; upgrade the Herdr client"
+                        .to_owned(),
+                ),
+            };
+            let _ = protocol::write_message(&mut stream, &welcome);
+            return Ok(());
+        }
         _ => {
             debug!(client_id, "first message was not a handshake, closing");
             let welcome = ServerMessage::Welcome {
@@ -678,16 +777,30 @@ pub(crate) fn handle_client_handshake(
         return Ok(());
     }
 
-    // Send Welcome.
+    // Send the negotiated welcome. Endpoint compatibility is independent from
+    // the same-install protocol used by direct terminal clients.
     let render_encoding = if shell_options.is_some() {
         RenderEncoding::SemanticFrame
     } else {
         RenderEncoding::TerminalAnsi
     };
-    let welcome = ServerMessage::Welcome {
-        version: PROTOCOL_VERSION,
-        encoding: render_encoding,
-        error: None,
+    let welcome = if shell_options.is_some() {
+        let welcome = EndpointServerWelcome::compatible(
+            crate::server::client_commands::supported_client_shell_method_names()
+                .iter()
+                .map(|method| (*method).to_owned())
+                .collect(),
+        );
+        ServerMessage::EndpointControl {
+            kind: ENDPOINT_WELCOME_KIND.into(),
+            data: serde_json::to_string(&welcome).map_err(io::Error::other)?,
+        }
+    } else {
+        ServerMessage::Welcome {
+            version: PROTOCOL_VERSION,
+            encoding: render_encoding,
+            error: None,
+        }
     };
     protocol::write_message(&mut stream, &welcome).map_err(|e| io::Error::other(e.to_string()))?;
 
@@ -947,14 +1060,24 @@ fn client_read_loop(
                 cell_height_px,
                 surface_size,
                 pixel_mouse,
-            } => ServerEvent::ClientShellResize {
-                client_id,
-                surface_cols: surface_size.cols.max(1),
-                surface_rows: surface_size.rows.max(1),
-                cell_width_px,
-                cell_height_px,
-                pixel_mouse,
-            },
+            } => {
+                if let Some(reason) =
+                    client_shell_geometry_error(surface_size, cell_width_px, cell_height_px)
+                {
+                    warn!(client_id, %reason, "invalid client shell resize, closing");
+                    let _ = server_event_tx
+                        .blocking_send(ServerEvent::ClientDisconnected { client_id });
+                    break;
+                }
+                ServerEvent::ClientShellResize {
+                    client_id,
+                    surface_cols: surface_size.cols,
+                    surface_rows: surface_size.rows,
+                    cell_width_px,
+                    cell_height_px,
+                    pixel_mouse,
+                }
+            }
             ClientMessage::ClientShellHostTheme { update } => {
                 if matches!(
                     &update,
@@ -1075,14 +1198,20 @@ fn client_read_loop(
                         .blocking_send(ServerEvent::ClientDisconnected { client_id });
                     break;
                 }
-                let Ok(request) = serde_json::from_str::<crate::api::schema::Request>(&request)
-                else {
-                    warn!(client_id, "invalid client shell endpoint command, closing");
-                    let _ = server_event_tx
-                        .blocking_send(ServerEvent::ClientDisconnected { client_id });
-                    break;
+                let decoded = match decode_endpoint_request(&request) {
+                    Ok(decoded) => decoded,
+                    Err(error) => {
+                        warn!(client_id, %error, "invalid endpoint request envelope, closing");
+                        let _ = server_event_tx
+                            .blocking_send(ServerEvent::ClientDisconnected { client_id });
+                        break;
+                    }
                 };
-                if request.id.len() > crate::server::client_commands::MAX_ENDPOINT_REQUEST_ID_BYTES
+                let request_id = match &decoded {
+                    DecodedEndpointRequest::Dispatch(request) => request.id.as_str(),
+                    DecodedEndpointRequest::Error { request_id, .. } => request_id,
+                };
+                if request_id.len() > crate::server::client_commands::MAX_ENDPOINT_REQUEST_ID_BYTES
                 {
                     warn!(
                         client_id,
@@ -1092,12 +1221,31 @@ fn client_read_loop(
                         .blocking_send(ServerEvent::ClientDisconnected { client_id });
                     break;
                 }
-                ServerEvent::ClientShellEndpointRequest {
-                    client_id,
-                    boot_id,
-                    request: Box::new(request),
+                match decoded {
+                    DecodedEndpointRequest::Dispatch(request) => {
+                        ServerEvent::ClientShellEndpointRequest {
+                            client_id,
+                            boot_id,
+                            request,
+                        }
+                    }
+                    DecodedEndpointRequest::Error {
+                        request_id,
+                        code,
+                        message,
+                    } => ServerEvent::ClientShellEndpointRequestError {
+                        client_id,
+                        boot_id,
+                        request_id,
+                        code,
+                        message,
+                    },
                 }
             }
+            ClientMessage::EndpointControl { kind, .. } => {
+                debug!(client_id, %kind, "ignoring unknown endpoint control message");
+                continue;
+            }
             ClientMessage::Detach => {
                 let _ = server_event_tx.blocking_send(ServerEvent::ClientDetach { client_id });
                 break;
@@ -1195,6 +1343,38 @@ mod tests {
         (client, server, TestSocketPath(path))
     }
 
+    fn endpoint_hello(surface_cols: u16, surface_rows: u16) -> ClientMessage {
+        let hello = EndpointClientHello {
+            generation: ENDPOINT_PROTOCOL_GENERATION,
+            cell_width_px: 8,
+            cell_height_px: 16,
+            surface_size: crate::protocol::ClientSurfaceSize {
+                cols: surface_cols,
+                rows: surface_rows,
+            },
+            pixel_mouse: true,
+            direct_graphics: true,
+            endpoint_keybindings: true,
+            mouse_capture: true,
+            snapshot_codecs: vec![crate::protocol::endpoint::SNAPSHOT_CODEC_V1.into()],
+            surface_codecs: vec![crate::protocol::endpoint::SURFACE_CODEC_V1.into()],
+            input_codecs: vec![crate::protocol::endpoint::INPUT_CODEC_V1.into()],
+            blob_codecs: vec![crate::protocol::endpoint::BLOB_CODEC_V1.into()],
+        };
+        ClientMessage::EndpointControl {
+            kind: ENDPOINT_HELLO_KIND.into(),
+            data: serde_json::to_string(&hello).unwrap(),
+        }
+    }
+
+    fn endpoint_welcome(message: ServerMessage) -> EndpointServerWelcome {
+        let ServerMessage::EndpointControl { kind, data } = message else {
+            panic!("expected endpoint welcome");
+        };
+        assert_eq!(kind, ENDPOINT_WELCOME_KIND);
+        serde_json::from_str(&data).unwrap()
+    }
+
     fn recv_server_event(receiver: &mut mpsc::Receiver<ServerEvent>, context: &str) -> ServerEvent {
         let deadline = std::time::Instant::now() + Duration::from_secs(1);
         loop {
@@ -1432,6 +1612,62 @@ mod tests {
     }
 
     #[test]
+    fn client_shell_geometry_rejects_unsafe_dimensions_and_cell_sizes() {
+        assert!(client_shell_geometry_error(
+            crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 },
+            8,
+            16,
+        )
+        .is_none());
+        assert!(client_shell_geometry_error(
+            crate::protocol::ClientSurfaceSize {
+                cols: MAX_CLIENT_SHELL_DIMENSION,
+                rows: MAX_CLIENT_SHELL_DIMENSION,
+            },
+            8,
+            16,
+        )
+        .is_some());
+        assert!(client_shell_geometry_error(
+            crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 },
+            MAX_CLIENT_CELL_SIZE_PX + 1,
+            16,
+        )
+        .is_some());
+    }
+
+    #[test]
+    fn unknown_endpoint_method_returns_correlated_error() {
+        let decoded = decode_endpoint_request(
+            r#"{"id":"req-1","method":"plugin.future","params":{"value":1}}"#,
+        )
+        .unwrap();
+        assert!(matches!(
+            decoded,
+            DecodedEndpointRequest::Error {
+                request_id,
+                code: "unsupported_method",
+                ..
+            } if request_id == "req-1"
+        ));
+    }
+
+    #[test]
+    fn malformed_known_endpoint_method_returns_correlated_error() {
+        let decoded =
+            decode_endpoint_request(r#"{"id":"req-2","method":"workspace.focus","params":{}}"#)
+                .unwrap();
+        assert!(matches!(
+            decoded,
+            DecodedEndpointRequest::Error {
+                request_id,
+                code: "invalid_request",
+                ..
+            } if request_id == "req-2"
+        ));
+    }
+
+    #[test]
     fn handshake_negotiates_terminal_ansi_encoding() {
         let (mut client_stream, server_stream, _path) = local_stream_pair("client-handshake-ansi");
         let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
@@ -1509,31 +1745,14 @@ mod tests {
             handle_client_handshake(server_stream, 43, &server_event_tx, &handshake_quit)
         });
 
-        protocol::write_message(
-            &mut client_stream,
-            &ClientMessage::ClientShellHello {
-                version: PROTOCOL_VERSION,
-                cell_width_px: 8,
-                cell_height_px: 16,
-                surface_size: crate::protocol::ClientSurfaceSize { cols: 80, rows: 29 },
-                pixel_mouse: true,
-                direct_graphics: true,
-                endpoint_keybindings: true,
-                mouse_capture: true,
-            },
-        )
-        .expect("write shell hello");
+        protocol::write_message(&mut client_stream, &endpoint_hello(80, 29))
+            .expect("write shell hello");
 
         let welcome: ServerMessage =
             protocol::read_message(&mut client_stream, MAX_FRAME_SIZE).expect("read welcome");
-        assert!(matches!(
-            welcome,
-            ServerMessage::Welcome {
-                encoding: RenderEncoding::SemanticFrame,
-                error: None,
-                ..
-            }
-        ));
+        let welcome = endpoint_welcome(welcome);
+        assert_eq!(welcome.generation, ENDPOINT_PROTOCOL_GENERATION);
+        assert!(welcome.error.is_none());
         match server_event_rx
             .blocking_recv()
             .expect("client shell connected event")
@@ -1581,31 +1800,15 @@ mod tests {
             handle_client_handshake(server_stream, 43, &server_event_tx, &handshake_quit)
         });
 
-        protocol::write_message(
-            &mut client_stream,
-            &ClientMessage::ClientShellHello {
-                version: PROTOCOL_VERSION,
-                cell_width_px: 8,
-                cell_height_px: 16,
-                surface_size: crate::protocol::ClientSurfaceSize { cols: 0, rows: 29 },
-                pixel_mouse: false,
-                direct_graphics: false,
-                endpoint_keybindings: false,
-                mouse_capture: false,
-            },
-        )
-        .expect("write empty shell hello");
+        protocol::write_message(&mut client_stream, &endpoint_hello(0, 29))
+            .expect("write empty shell hello");
 
         let welcome: ServerMessage =
             protocol::read_message(&mut client_stream, MAX_FRAME_SIZE).expect("read welcome");
-        assert!(matches!(
-            welcome,
-            ServerMessage::Welcome {
-                encoding: RenderEncoding::SemanticFrame,
-                error: Some(error),
-                ..
-            } if error.contains("non-empty pane surface")
-        ));
+        let welcome = endpoint_welcome(welcome);
+        assert!(welcome
+            .error
+            .is_some_and(|error| error.message.contains("non-empty pane surface")));
         handle
             .join()
             .expect("handshake thread join")
@@ -1650,6 +1853,72 @@ mod tests {
     }
 
     #[test]
+    fn client_read_loop_ignores_unknown_endpoint_control() {
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-read-future-control");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::EndpointControl {
+                kind: "future.optional.v1".into(),
+                data: "{}".into(),
+            },
+        )
+        .unwrap();
+        protocol::write_message(&mut client_stream, &ClientMessage::Detach).unwrap();
+
+        assert!(matches!(
+            recv_server_event(&mut server_event_rx, "detach after future control"),
+            ServerEvent::ClientDetach { client_id: 7 }
+        ));
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
+    fn client_read_loop_closes_on_unsafe_shell_resize() {
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-read-unsafe-resize");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::ClientShellResize {
+                cell_width_px: 8,
+                cell_height_px: 16,
+                surface_size: crate::protocol::ClientSurfaceSize {
+                    cols: MAX_CLIENT_SHELL_DIMENSION,
+                    rows: MAX_CLIENT_SHELL_DIMENSION,
+                },
+                pixel_mouse: false,
+            },
+        )
+        .unwrap();
+
+        assert!(matches!(
+            recv_server_event(&mut server_event_rx, "unsafe resize disconnect"),
+            ServerEvent::ClientDisconnected { client_id: 7 }
+        ));
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
     fn client_read_loop_rejects_oversized_bracketed_paste_without_disconnect() {
         let (mut client_stream, server_stream, _path) = local_stream_pair("client-read-oversized");
         let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 6bae0a98..1cf8ebd4 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -1934,12 +1934,17 @@ impl HeadlessServer {
                     connection.shell_projection_revision,
                     config_diagnostic,
                 );
-                connection.shell_snapshot = Some(snapshot.clone());
+                let snapshot_message = match crate::protocol::endpoint::snapshot_message(&snapshot)
+                {
+                    Ok(message) => message,
+                    Err(err) => {
+                        warn!(client_id, err = %err, "failed to encode endpoint snapshot");
+                        return false;
+                    }
+                };
+                connection.shell_snapshot = Some(snapshot);
                 self.clients.insert(client_id, connection);
-                self.send_to_client(
-                    client_id,
-                    ServerMessage::ClientShellSnapshot(Box::new(snapshot)),
-                );
+                self.send_to_client(client_id, snapshot_message);
                 self.foreground_client_id = Some(client_id);
                 if first_app_client {
                     self.app.mark_git_status_refresh_due(Instant::now());
@@ -2374,6 +2379,26 @@ impl HeadlessServer {
                 }
                 foreground_changed || runtime.scroll_metrics() != scroll_before
             }
+            ServerEvent::ClientShellEndpointRequestError {
+                client_id,
+                boot_id,
+                request_id,
+                code,
+                message,
+            } => {
+                let Some(client) = self.clients.get(&client_id) else {
+                    return false;
+                };
+                if !matches!(client.mode, ClientConnectionMode::ClientShell) {
+                    self.remove_client_and_resize_if_needed(client_id);
+                    return true;
+                }
+                let message = crate::server::client_commands::error_message(
+                    boot_id, request_id, code, message,
+                );
+                self.send_to_client(client_id, message);
+                false
+            }
             ServerEvent::ClientShellEndpointRequest {
                 client_id,
                 boot_id,
diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs
index cab79de2..19581abc 100644
--- a/src/server/headless/render.rs
+++ b/src/server/headless/render.rs
@@ -387,11 +387,18 @@ impl HeadlessServer {
                     client.shell_projection_revision =
                         client.shell_projection_revision.saturating_add(1);
                     candidate.revision = client.shell_projection_revision;
-                    let message = ServerMessage::ClientShellSnapshot(Box::new(candidate.clone()));
+                    let message = match crate::protocol::endpoint::snapshot_message(&candidate) {
+                        Ok(message) => message,
+                        Err(err) => {
+                            warn!(client_id, err = %err, "failed to encode endpoint snapshot");
+                            broken_clients.push(client_id);
+                            continue;
+                        }
+                    };
                     let framed = match Self::frame_server_message(&message) {
                         Ok(framed) => framed,
                         Err(err) => {
-                            warn!(client_id, err = %err, "failed to frame client shell replacement");
+                            warn!(client_id, err = %err, "failed to frame endpoint snapshot");
                             broken_clients.push(client_id);
                             continue;
                         }
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index f5c33f6a..93c7420c 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -3,6 +3,14 @@ use super::*;
 #[path = "pane_graphics.rs"]
 mod pane_graphics_tests;
 
+fn client_shell_snapshot(message: ServerMessage) -> Box<crate::protocol::ClientShellSnapshot> {
+    let ServerMessage::EndpointControl { kind, data } = message else {
+        panic!("expected client shell snapshot");
+    };
+    assert_eq!(kind, crate::protocol::endpoint::ENDPOINT_SNAPSHOT_KIND);
+    Box::new(serde_json::from_str(&data).expect("decode client shell snapshot"))
+}
+
 fn test_headless_server() -> HeadlessServer {
     test_headless_server_with_event_hub(api::EventHub::default())
 }
@@ -671,6 +679,33 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
     shutdown_test_runtimes(&mut server);
 }
 
+#[test]
+fn terminal_client_endpoint_request_error_removes_client() {
+    let mut server = test_headless_server();
+    let (writer, _control_rx, _render_rx) = test_client_writer();
+    let client_id = 42;
+    assert!(!server.handle_server_event(ServerEvent::ClientConnected {
+        client_id,
+        cols: 80,
+        rows: 24,
+        cell_width_px: 0,
+        cell_height_px: 0,
+        pixel_mouse: false,
+        writer,
+    }));
+
+    assert!(
+        server.handle_server_event(ServerEvent::ClientShellEndpointRequestError {
+            client_id,
+            boot_id: "boot".into(),
+            request_id: "request".into(),
+            code: "unsupported_method",
+            message: "unsupported".into(),
+        })
+    );
+    assert!(!server.clients.contains_key(&client_id));
+}
+
 #[tokio::test]
 async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
     let mut server = test_headless_server();
@@ -713,25 +748,23 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
             writer,
         })
     );
-    match read_server_message(control_rx.recv().expect("shell snapshot")) {
-        ServerMessage::ClientShellSnapshot(snapshot) => {
-            assert_eq!(snapshot.workspaces.len(), 1);
-            assert_eq!(snapshot.workspaces[0].label, "shell-only-label");
-            assert_eq!(
-                snapshot.config_diagnostic.as_deref(),
-                Some("endpoint config warning")
-            );
-            assert_eq!(
-                snapshot.product_announcement.as_ref().map(|announcement| (
-                    announcement.version.as_str(),
-                    announcement.id.as_str(),
-                    announcement.preview,
-                )),
-                Some(("0.8.2", "client-shell", true))
-            );
-        }
-        other => panic!("expected client shell snapshot, got {other:?}"),
-    }
+    let snapshot = client_shell_snapshot(read_server_message(
+        control_rx.recv().expect("shell snapshot"),
+    ));
+    assert_eq!(snapshot.workspaces.len(), 1);
+    assert_eq!(snapshot.workspaces[0].label, "shell-only-label");
+    assert_eq!(
+        snapshot.config_diagnostic.as_deref(),
+        Some("endpoint config warning")
+    );
+    assert_eq!(
+        snapshot.product_announcement.as_ref().map(|announcement| (
+            announcement.version.as_str(),
+            announcement.id.as_str(),
+            announcement.preview,
+        )),
+        Some(("0.8.2", "client-shell", true))
+    );
 
     server.render_and_stream();
     let initial_surface = match read_server_message(render_rx.recv().expect("pane surface")) {
@@ -1132,11 +1165,9 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
             writer: local_writer,
         })
     );
-    let ServerMessage::ClientShellSnapshot(local_snapshot) =
-        read_server_message(local_control.recv().expect("local shell snapshot"))
-    else {
-        panic!("expected local shell snapshot");
-    };
+    let local_snapshot = client_shell_snapshot(read_server_message(
+        local_control.recv().expect("local shell snapshot"),
+    ));
     assert_eq!(
         local_snapshot.config_diagnostic.as_deref(),
         Some("theme warning")
@@ -1157,11 +1188,9 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
             writer: endpoint_writer,
         })
     );
-    let ServerMessage::ClientShellSnapshot(endpoint_snapshot) =
-        read_server_message(endpoint_control.recv().expect("endpoint shell snapshot"))
-    else {
-        panic!("expected endpoint shell snapshot");
-    };
+    let endpoint_snapshot = client_shell_snapshot(read_server_message(
+        endpoint_control.recv().expect("endpoint shell snapshot"),
+    ));
     assert_eq!(
         endpoint_snapshot.config_diagnostic.as_deref(),
         Some("server keybinding warning\ntheme warning")
@@ -1199,10 +1228,10 @@ async fn client_shell_replaces_projection_and_focuses_stable_ids() {
             writer,
         })
     );
-    let initial_revision = match read_server_message(control_rx.recv().expect("initial snapshot")) {
-        ServerMessage::ClientShellSnapshot(snapshot) => snapshot.revision,
-        other => panic!("expected initial shell snapshot, got {other:?}"),
-    };
+    let initial_revision = client_shell_snapshot(read_server_message(
+        control_rx.recv().expect("initial snapshot"),
+    ))
+    .revision;
 
     let _ = server.app.handle_api_request(crate::api::schema::Request {
         id: "test.client.shell.workspace.focus".into(),
@@ -1213,10 +1242,9 @@ async fn client_shell_replaces_projection_and_focuses_stable_ids() {
     assert_eq!(server.app.state.active, Some(1));
     server.render_and_stream();
 
-    let replacement = match read_server_message(control_rx.recv().expect("replacement snapshot")) {
-        ServerMessage::ClientShellSnapshot(snapshot) => snapshot,
-        other => panic!("expected replacement shell snapshot, got {other:?}"),
-    };
+    let replacement = client_shell_snapshot(read_server_message(
+        control_rx.recv().expect("replacement snapshot"),
+    ));
     assert!(replacement.revision > initial_revision);
     assert_eq!(
         replacement.focused_workspace_id.as_deref(),
@@ -1441,9 +1469,8 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
             writer,
         })
     );
-    assert!(matches!(
-        read_server_message(control_rx.recv().expect("shell snapshot")),
-        ServerMessage::ClientShellSnapshot(_)
+    let _snapshot = client_shell_snapshot(read_server_message(
+        control_rx.recv().expect("shell snapshot"),
     ));
 
     server.render_and_stream();
@@ -4553,7 +4580,7 @@ fn update_notification_is_semantic_for_system_delivery() {
             assert_eq!(notification.title, "Herdr v9.9.9 available");
             assert_eq!(
                 notification.body.as_deref(),
-                Some("detach, run `herdr update`, then follow its restart guidance")
+                Some("detach, run `herdr update`, then run Herdr again to reconnect")
             );
         }
         other => panic!("expected semantic update notification, got {other:?}"),
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index e533bd03..15b23bf8 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -16,6 +16,7 @@ const ROWS: u16 = 40;
 const SAMPLE_COUNT: usize = 40;
 const WARMUP_COUNT: usize = 5;
 const CARDINALITIES: [usize; 3] = [1, 15, 50];
+const CLIENT_CARDINALITIES: [usize; 2] = [1, 4];
 
 #[derive(Clone, Copy)]
 struct StageStats {
@@ -210,9 +211,52 @@ fn print_profiles(label: &str, build: fn(usize) -> Vec<Workspace>) {
     print_stage("combined pipeline", &rows, |stats| stats.total);
 }
 
+fn profile_snapshot_encoding(
+    build: fn(usize) -> Vec<Workspace>,
+    count: usize,
+    client_count: usize,
+) -> StageStats {
+    let pipeline = RenderPipeline::new(build(count));
+    let run = || {
+        let started = Instant::now();
+        let template = super::client_shell::snapshot(&pipeline.app, "bench-boot", 1, None);
+        for client_index in 0..client_count {
+            let mut snapshot = template.clone();
+            snapshot.revision = client_index as u64 + 1;
+            let message = crate::protocol::endpoint::snapshot_message(&snapshot)
+                .expect("benchmark snapshot should serialize");
+            black_box(
+                bincode::serde::encode_to_vec(message, bincode::config::standard())
+                    .expect("benchmark snapshot message should frame"),
+            );
+        }
+        started.elapsed()
+    };
+    for _ in 0..WARMUP_COUNT {
+        black_box(run());
+    }
+    summarize((0..SAMPLE_COUNT).map(|_| run()).collect())
+}
+
+fn print_snapshot_encoding_profiles(label: &str, build: fn(usize) -> Vec<Workspace>) {
+    println!("{label} snapshot projection + JSON framing");
+    println!("       panes  clients  median_us  p95_us  max_us");
+    for count in CARDINALITIES {
+        for client_count in CLIENT_CARDINALITIES {
+            let stats = profile_snapshot_encoding(build, count, client_count);
+            println!(
+                "  {count:>10}  {client_count:>7}  {:>9}  {:>6}  {:>6}",
+                stats.median_us, stats.p95_us, stats.max_us
+            );
+        }
+    }
+}
+
 #[tokio::test(flavor = "current_thread")]
 #[ignore = "manual client-rendered pipeline scaling profile"]
 async fn render_scale_profile() {
     print_profiles("background workspaces (one pane each)", workspaces);
+    print_snapshot_encoding_profiles("background workspaces", workspaces);
     print_profiles("active panes (one workspace)", active_panes);
+    print_snapshot_encoding_profiles("active panes", active_panes);
 }