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

38 files changed, 2636 insertions, 679 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index 363d7702..d28a82ae 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -7,8 +7,8 @@ mod subscriptions;
 mod wait;
 
 pub use event_hub::EventHub;
-pub(crate) use server::start_server_with_stop_control;
 pub use server::ServerHandle;
+pub(crate) use server::{api_method_name, start_server_with_stop_control};
 pub use status::{read_runtime_status_at, RuntimeStatus};
 
 use std::path::PathBuf;
diff --git a/src/api/schema/server.rs b/src/api/schema/server.rs
index 7ceda633..96e51564 100644
--- a/src/api/schema/server.rs
+++ b/src/api/schema/server.rs
@@ -18,4 +18,7 @@ pub struct ServerCapabilities {
     pub live_handoff: bool,
     #[serde(default)]
     pub detached_server_daemon: bool,
+    /// Stable client-owned endpoint generation supported by this server.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub endpoint_protocol_generation: Option<u32>,
 }
diff --git a/src/api/schema/tests.rs b/src/api/schema/tests.rs
index e4a7f2a3..06529f5e 100644
--- a/src/api/schema/tests.rs
+++ b/src/api/schema/tests.rs
@@ -708,6 +708,12 @@ fn scroll_changed_subscription_event_round_trips() {
 }
 
 #[test]
+fn agent_status_request_values_remain_strict() {
+    assert!(serde_json::from_str::<AgentStatus>(r#""working""#).is_ok());
+    assert!(serde_json::from_str::<AgentStatus>(r#""future_status""#).is_err());
+}
+
+#[test]
 fn success_response_round_trips() {
     let response = SuccessResponse {
         id: "req_1".into(),
@@ -717,6 +723,7 @@ fn success_response_round_trips() {
             capabilities: Some(ServerCapabilities {
                 live_handoff: true,
                 detached_server_daemon: true,
+                endpoint_protocol_generation: Some(1),
             }),
         },
     };
diff --git a/src/api/server.rs b/src/api/server.rs
index ff0b6fdf..64beec33 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -68,6 +68,7 @@ fn default_capabilities() -> Option<ServerCapabilities> {
     Some(ServerCapabilities {
         live_handoff: crate::platform::capabilities().live_handoff,
         detached_server_daemon: crate::platform::current_process_is_detached_server_daemon(),
+        endpoint_protocol_generation: Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
     })
 }
 
@@ -371,7 +372,7 @@ fn handle_request(
     dispatch_to_app(request, api_tx, None, response_write_complete, None)
 }
 
-fn api_method_name(method: &Method) -> &'static str {
+pub(crate) fn api_method_name(method: &Method) -> &'static str {
     match method {
         Method::Ping(_) => "ping",
         Method::ServerStop(_) => "server.stop",
@@ -1090,6 +1091,9 @@ mod tests {
             Some(ServerCapabilities {
                 live_handoff: true,
                 detached_server_daemon: true,
+                endpoint_protocol_generation: Some(
+                    crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+                ),
             }),
             None,
             None,
diff --git a/src/app/actions.rs b/src/app/actions.rs
index 85839322..853c9084 100644
--- a/src/app/actions.rs
+++ b/src/app/actions.rs
@@ -3750,7 +3750,7 @@ mod tests {
         assert_eq!(toast.title, "v0.5.0 available");
         assert_eq!(
             toast.context,
-            "detach, run `herdr update`, then follow its restart guidance"
+            "detach, run `herdr update`, then run Herdr again to reconnect"
         );
     }
 
@@ -3771,7 +3771,7 @@ mod tests {
         let toast = state.toast.as_ref().expect("update toast");
         assert_eq!(
             toast.context,
-            "detach, run `brew update && brew upgrade herdr`, then restart this Herdr session when ready"
+            "detach, run `brew update && brew upgrade herdr`, then run Herdr again to reconnect"
         );
     }
 
diff --git a/src/cli/status.rs b/src/cli/status.rs
index 74bb3e2a..4dba81b7 100644
--- a/src/cli/status.rs
+++ b/src/cli/status.rs
@@ -98,12 +98,20 @@ fn print_full_status(json: bool) -> std::io::Result<i32> {
         crate::config::Config::load().config.update.channel.as_str()
     );
     println!("  protocol: {}", crate::protocol::PROTOCOL_VERSION);
+    println!(
+        "  endpoint_protocol_generation: {}",
+        crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION
+    );
     println!();
     println!("server:");
     print_server_status_body(&server, "  ");
     println!();
     println!("update:");
     println!("  restart_needed: {}", restart_needed_label(&server));
+    println!(
+        "  server_binary_stale: {}",
+        server_binary_stale_label(&server)
+    );
 
     Ok(0)
 }
@@ -130,6 +138,10 @@ fn print_client_status(json: bool) -> std::io::Result<()> {
         crate::config::Config::load().config.update.channel.as_str()
     );
     println!("protocol: {}", crate::protocol::PROTOCOL_VERSION);
+    println!(
+        "endpoint_protocol_generation: {}",
+        crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION
+    );
     println!("binary: {}", current_exe_label());
     Ok(())
 }
@@ -137,12 +149,21 @@ fn print_client_status(json: bool) -> std::io::Result<()> {
 fn print_server_status_body(server: &ServerRuntimeStatus, indent: &str) {
     match server {
         ServerRuntimeStatus::Running {
-            version, protocol, ..
+            version,
+            protocol,
+            capabilities,
         } => {
             println!("{indent}status: running");
             println!("{indent}version: {}", option_label(version.as_deref()));
-            println!("{indent}protocol: {}", protocol_label(*protocol));
-            println!("{indent}compatible: {}", compatibility_label(*protocol));
+            println!(
+                "{indent}endpoint_compatible: {}",
+                endpoint_compatibility_label(capabilities.as_ref())
+            );
+            println!("{indent}private_protocol: {}", protocol_label(*protocol));
+            println!(
+                "{indent}private_protocol_compatible: {}",
+                compatibility_label(*protocol)
+            );
             println!("{indent}socket: {}", api::socket_path().display());
         }
         ServerRuntimeStatus::NotRunning => {
@@ -191,14 +212,33 @@ fn compatibility_label(protocol: Option<u32>) -> &'static str {
     }
 }
 
+fn endpoint_compatibility_label(
+    capabilities: Option<&crate::api::schema::ServerCapabilities>,
+) -> &'static str {
+    match capabilities.and_then(|value| value.endpoint_protocol_generation) {
+        Some(generation)
+            if generation == crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION =>
+        {
+            "yes"
+        }
+        Some(_) => "no",
+        None => "unknown",
+    }
+}
+
 fn restart_needed_label(server: &ServerRuntimeStatus) -> &'static str {
-    match server {
-        ServerRuntimeStatus::Running { version, .. } => match version.as_deref() {
-            Some(version) if version == crate::build_info::version() => "no",
-            Some(_) => "yes",
-            None => "unknown",
-        },
-        ServerRuntimeStatus::NotRunning => "no",
+    match restart_needed_bool(server) {
+        Some(true) => "yes",
+        Some(false) => "no",
+        None => "unknown",
+    }
+}
+
+fn server_binary_stale_label(server: &ServerRuntimeStatus) -> &'static str {
+    match server_binary_stale_bool(server) {
+        Some(true) => "yes",
+        Some(false) => "no",
+        None => "unknown",
     }
 }
 
@@ -214,6 +254,7 @@ struct ClientStatusJson {
     version: String,
     channel: &'static str,
     protocol: u32,
+    endpoint_protocol_generation: u32,
     binary: String,
     session: Option<String>,
 }
@@ -226,20 +267,24 @@ struct ServerStatusJson {
     protocol: Option<u32>,
     capabilities: Option<ServerCapabilitiesJson>,
     compatible: Option<bool>,
+    endpoint_compatible: Option<bool>,
     socket: String,
     session: Option<String>,
     restart_needed: Option<bool>,
+    server_binary_stale: Option<bool>,
 }
 
 #[derive(Serialize)]
 struct ServerCapabilitiesJson {
     live_handoff: bool,
     detached_server_daemon: bool,
+    endpoint_protocol_generation: Option<u32>,
 }
 
 #[derive(Serialize)]
 struct UpdateStatusJson {
     restart_needed: Option<bool>,
+    server_binary_stale: Option<bool>,
 }
 
 fn client_status_json() -> ClientStatusJson {
@@ -247,6 +292,7 @@ fn client_status_json() -> ClientStatusJson {
         version: crate::build_info::version(),
         channel: crate::config::Config::load().config.update.channel.as_str(),
         protocol: crate::protocol::PROTOCOL_VERSION,
+        endpoint_protocol_generation: crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
         binary: current_exe_label(),
         session: crate::session::active_name(),
     }
@@ -268,11 +314,18 @@ fn server_status_json(server: &ServerRuntimeStatus) -> ServerStatusJson {
                 .map(|capabilities| ServerCapabilitiesJson {
                     live_handoff: capabilities.live_handoff,
                     detached_server_daemon: capabilities.detached_server_daemon,
+                    endpoint_protocol_generation: capabilities.endpoint_protocol_generation,
                 }),
             compatible: protocol.map(|value| value == crate::protocol::PROTOCOL_VERSION),
+            endpoint_compatible: capabilities.as_ref().and_then(|capabilities| {
+                capabilities.endpoint_protocol_generation.map(|generation| {
+                    generation == crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION
+                })
+            }),
             socket: api::socket_path().display().to_string(),
             session: crate::session::active_name(),
             restart_needed: restart_needed_bool(server),
+            server_binary_stale: server_binary_stale_bool(server),
         },
         ServerRuntimeStatus::NotRunning => ServerStatusJson {
             status: "not_running",
@@ -281,9 +334,11 @@ fn server_status_json(server: &ServerRuntimeStatus) -> ServerStatusJson {
             protocol: None,
             capabilities: None,
             compatible: None,
+            endpoint_compatible: None,
             socket: api::socket_path().display().to_string(),
             session: crate::session::active_name(),
             restart_needed: Some(false),
+            server_binary_stale: Some(false),
         },
     }
 }
@@ -291,16 +346,27 @@ fn server_status_json(server: &ServerRuntimeStatus) -> ServerStatusJson {
 fn update_status_json(server: &ServerRuntimeStatus) -> UpdateStatusJson {
     UpdateStatusJson {
         restart_needed: restart_needed_bool(server),
+        server_binary_stale: server_binary_stale_bool(server),
     }
 }
 
 fn restart_needed_bool(server: &ServerRuntimeStatus) -> Option<bool> {
     match server {
-        ServerRuntimeStatus::Running { version, .. } => match version.as_deref() {
-            Some(version) if version == crate::build_info::version() => Some(false),
-            Some(_) => Some(true),
-            None => None,
-        },
+        ServerRuntimeStatus::Running { capabilities, .. } => Some(
+            capabilities
+                .as_ref()
+                .and_then(|value| value.endpoint_protocol_generation)
+                != Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+        ),
+        ServerRuntimeStatus::NotRunning => Some(false),
+    }
+}
+
+fn server_binary_stale_bool(server: &ServerRuntimeStatus) -> Option<bool> {
+    match server {
+        ServerRuntimeStatus::Running { version, .. } => version
+            .as_deref()
+            .map(|version| version != crate::build_info::version()),
         ServerRuntimeStatus::NotRunning => Some(false),
     }
 }
@@ -322,3 +388,42 @@ fn print_status_help() {
     eprintln!("  herdr status server [--json]  show running server status");
     eprintln!("  herdr status client [--json]  show local client binary status");
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn running_server(
+        version: Option<&str>,
+        endpoint_generation: Option<u32>,
+    ) -> ServerRuntimeStatus {
+        ServerRuntimeStatus::Running {
+            version: version.map(str::to_owned),
+            protocol: Some(crate::protocol::PROTOCOL_VERSION),
+            capabilities: Some(crate::api::schema::ServerCapabilities {
+                live_handoff: true,
+                detached_server_daemon: true,
+                endpoint_protocol_generation: endpoint_generation,
+            }),
+        }
+    }
+
+    #[test]
+    fn stale_compatible_server_does_not_require_restart() {
+        let server = running_server(
+            Some("0.0.0-old"),
+            Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+        );
+
+        assert_eq!(restart_needed_bool(&server), Some(false));
+        assert_eq!(server_binary_stale_bool(&server), Some(true));
+    }
+
+    #[test]
+    fn server_without_endpoint_baseline_requires_restart() {
+        let server = running_server(Some(crate::build_info::version().as_str()), None);
+
+        assert_eq!(restart_needed_bool(&server), Some(true));
+        assert_eq!(server_binary_stale_bool(&server), Some(false));
+    }
+}
diff --git a/src/client/endpoint_commands.rs b/src/client/endpoint_commands.rs
index decb8522..d07b3d42 100644
--- a/src/client/endpoint_commands.rs
+++ b/src/client/endpoint_commands.rs
@@ -1,5 +1,6 @@
 use std::collections::VecDeque;
 use std::io;
+use std::time::{Duration, Instant};
 
 use crate::api::client::ApiClientError;
 use crate::api::schema::{Request, ResponseResult};
@@ -8,10 +9,14 @@ use crate::protocol::ClientMessage;
 
 use super::shell::ClientShellEndpointError;
 
+const ENDPOINT_COMMAND_TIMEOUT: Duration = Duration::from_secs(60);
+
 struct InFlightCommand {
     boot_id: String,
     request_id: String,
     response: Vec<u8>,
+    sent_at: Instant,
+    timed_out: bool,
 }
 
 pub(super) struct EndpointCommandResult {
@@ -52,10 +57,30 @@ impl EndpointCommands {
             boot_id,
             request_id,
             response: Vec::new(),
+            sent_at: Instant::now(),
+            timed_out: false,
         });
         Ok(())
     }
 
+    pub(super) fn expire(&mut self, now: Instant) -> Option<EndpointCommandResult> {
+        let command = self.in_flight.as_mut()?;
+        if command.timed_out
+            || now.saturating_duration_since(command.sent_at) < ENDPOINT_COMMAND_TIMEOUT
+        {
+            return None;
+        }
+        command.timed_out = true;
+        Some(EndpointCommandResult {
+            boot_id: command.boot_id.clone(),
+            request_id: command.request_id.clone(),
+            result: Err(ClientShellEndpointError {
+                code: Some("endpoint_timeout".into()),
+                message: "this server did not respond to the action".into(),
+            }),
+        })
+    }
+
     pub(super) fn receive_chunk(
         &mut self,
         response_boot_id: &str,
@@ -140,6 +165,8 @@ mod tests {
                 boot_id: "boot-a".into(),
                 request_id: "request-a".into(),
                 response: Vec::new(),
+                sent_at: Instant::now(),
+                timed_out: false,
             }),
             ..EndpointCommands::default()
         }
@@ -212,6 +239,38 @@ mod tests {
     }
 
     #[test]
+    fn in_flight_endpoint_command_expires_and_releases_the_lane() {
+        let mut commands = commands_with_in_flight();
+        let expired = commands
+            .expire(std::time::Instant::now() + ENDPOINT_COMMAND_TIMEOUT)
+            .expect("expired endpoint command");
+
+        assert_eq!(expired.boot_id, "boot-a");
+        assert_eq!(expired.request_id, "request-a");
+        assert!(matches!(
+            expired.result,
+            Err(ClientShellEndpointError {
+                code: Some(code),
+                ..
+            }) if code == "endpoint_timeout"
+        ));
+        assert!(commands.in_flight.is_some());
+        assert!(commands
+            .expire(std::time::Instant::now() + ENDPOINT_COMMAND_TIMEOUT)
+            .is_none());
+        let late_response = serde_json::to_vec(&SuccessResponse {
+            id: "request-a".into(),
+            result: ResponseResult::Ok {},
+        })
+        .unwrap();
+        assert!(commands
+            .receive_chunk("boot-a", "request-a", true, late_response)
+            .expect("late response releases the synchronized lane")
+            .is_some());
+        assert!(commands.in_flight.is_none());
+    }
+
+    #[test]
     fn response_from_another_boot_is_rejected() {
         let mut commands = commands_with_in_flight();
 
diff --git a/src/client/handshake.rs b/src/client/handshake.rs
index 2140516e..84dd6643 100644
--- a/src/client/handshake.rs
+++ b/src/client/handshake.rs
@@ -9,6 +9,11 @@ use tracing::debug;
 use tracing::info;
 
 use crate::ipc::LocalStream;
+use crate::protocol::endpoint::{
+    EndpointClientHello, EndpointServerWelcome, BLOB_CODEC_V1, ENDPOINT_HELLO_KIND,
+    ENDPOINT_PROTOCOL_GENERATION, ENDPOINT_WELCOME_KIND, INPUT_CODEC_V1, SNAPSHOT_CODEC_V1,
+    SURFACE_CODEC_V1,
+};
 use crate::protocol::{
     self, ClientMessage, RenderEncoding, ServerMessage, MAX_FRAME_SIZE, PROTOCOL_VERSION,
 };
@@ -112,10 +117,17 @@ fn set_handshake_recv_timeout(
         .map_err(ClientError::ConnectionFailed)
 }
 
+#[derive(Debug)]
+pub(super) struct HandshakeResult {
+    pub(super) encoding: RenderEncoding,
+    pub(super) endpoint_methods: Option<Vec<String>>,
+}
+
 /// Performs the client→server handshake.
 ///
-/// Sends TerminalHello (or ClientShellHello) with the terminal size and protocol
-/// version, then reads the Welcome response.
+/// Direct terminal clients retain the same-install private protocol. Client-owned
+/// shells use the stable endpoint generation and negotiate whole codecs without
+/// comparing Herdr build versions.
 pub(super) fn do_handshake(
     stream: &mut LocalStream,
     cols: u16,
@@ -126,14 +138,15 @@ pub(super) fn do_handshake(
     shell_surface_size: Option<crate::protocol::ClientSurfaceSize>,
     endpoint_keybindings: bool,
     mouse_capture: bool,
-) -> Result<RenderEncoding, ClientError> {
+) -> Result<HandshakeResult, ClientError> {
     stream
         .set_nonblocking(false)
         .map_err(ClientError::ConnectionFailed)?;
 
+    let endpoint_shell = shell_surface_size.is_some();
     let hello = if let Some(surface_size) = shell_surface_size {
-        ClientMessage::ClientShellHello {
-            version: PROTOCOL_VERSION,
+        let hello = EndpointClientHello {
+            generation: ENDPOINT_PROTOCOL_GENERATION,
             cell_width_px,
             cell_height_px,
             surface_size,
@@ -144,6 +157,16 @@ pub(super) fn do_handshake(
                 && direct_graphics_profile_allowed(),
             endpoint_keybindings,
             mouse_capture,
+            snapshot_codecs: vec![SNAPSHOT_CODEC_V1.into()],
+            surface_codecs: vec![SURFACE_CODEC_V1.into()],
+            input_codecs: vec![INPUT_CODEC_V1.into()],
+            blob_codecs: vec![BLOB_CODEC_V1.into()],
+        };
+        ClientMessage::EndpointControl {
+            kind: ENDPOINT_HELLO_KIND.into(),
+            data: serde_json::to_string(&hello).map_err(|error| {
+                ClientError::ConnectionFailed(io::Error::new(io::ErrorKind::InvalidData, error))
+            })?,
         }
     } else {
         ClientMessage::TerminalHello {
@@ -170,6 +193,54 @@ pub(super) fn do_handshake(
         "failed to clear client handshake read timeout",
     )?;
 
+    if endpoint_shell {
+        let ServerMessage::EndpointControl { kind, data } = welcome else {
+            return Err(ClientError::Protocol(protocol::FramingError::Io(
+                io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "server does not support the stable Herdr endpoint protocol; update this machine",
+                ),
+            )));
+        };
+        if kind != ENDPOINT_WELCOME_KIND {
+            return Err(ClientError::Protocol(protocol::FramingError::Io(
+                io::Error::new(io::ErrorKind::InvalidData, "expected endpoint welcome"),
+            )));
+        }
+        let welcome: EndpointServerWelcome = serde_json::from_str(&data).map_err(|error| {
+            ClientError::Protocol(protocol::FramingError::Io(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!("invalid endpoint welcome: {error}"),
+            )))
+        })?;
+        if let Some(error) = welcome.error {
+            return Err(ClientError::HandshakeRejected {
+                version: welcome.generation,
+                error: error.message,
+            });
+        }
+        if welcome.generation != ENDPOINT_PROTOCOL_GENERATION
+            || welcome.snapshot_codec != SNAPSHOT_CODEC_V1
+            || welcome.surface_codec != SURFACE_CODEC_V1
+            || welcome.input_codec != INPUT_CODEC_V1
+            || welcome.blob_codec != BLOB_CODEC_V1
+        {
+            return Err(ClientError::HandshakeRejected {
+                version: welcome.generation,
+                error: "server has no compatible endpoint core; update this machine".into(),
+            });
+        }
+        info!(
+            generation = welcome.generation,
+            server_version = %welcome.server_version,
+            "endpoint handshake succeeded"
+        );
+        return Ok(HandshakeResult {
+            encoding: RenderEncoding::SemanticFrame,
+            endpoint_methods: Some(welcome.methods),
+        });
+    }
+
     match welcome {
         ServerMessage::Welcome {
             version,
@@ -180,7 +251,10 @@ pub(super) fn do_handshake(
                 return Err(ClientError::HandshakeRejected { version, error });
             }
             info!(version, ?encoding, "handshake succeeded");
-            Ok(encoding)
+            Ok(HandshakeResult {
+                encoding,
+                endpoint_methods: None,
+            })
         }
         _ => Err(ClientError::Protocol(protocol::FramingError::Io(
             io::Error::new(io::ErrorKind::InvalidData, "expected Welcome message"),
diff --git a/src/client/mod.rs b/src/client/mod.rs
index f476f129..b7fdcabe 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -412,7 +412,7 @@ fn run_client_with_mode(
         .is_some_and(shell::ClientShellConfig::uses_endpoint_keybindings);
 
     // Perform handshake while the stream is still in blocking mode.
-    let negotiated_encoding = match do_handshake(
+    let handshake = match do_handshake(
         &mut stream,
         cols,
         rows,
@@ -489,7 +489,8 @@ fn run_client_with_mode(
             exact_cell_size,
             should_quit,
             loop_config,
-            negotiated_encoding,
+            handshake.encoding,
+            handshake.endpoint_methods,
             attach_escape,
         )
         .await
@@ -613,6 +614,45 @@ fn apply_client_shell_input_source_changes(
     }
 }
 
+fn install_client_shell_snapshot(
+    state: &mut ClientState,
+    snapshot: Box<crate::protocol::ClientShellSnapshot>,
+    write_stream: &mut LocalStream,
+    prefix_input_source: &mut impl crate::platform::PrefixInputSource,
+) -> Result<(), ClientError> {
+    let (composed, resize, graphics_cleanup) = if let Some(shell) = &mut state.shell {
+        let previous_size = shell.surface_size(state.reported_size.0, state.reported_size.1);
+        shell.set_snapshot(snapshot);
+        let graphics_cleanup = shell.take_pending_graphics_cleanup();
+        let next_size = shell.surface_size(state.reported_size.0, state.reported_size.1);
+        (
+            shell.compose(state.reported_size.0, state.reported_size.1),
+            (previous_size != next_size).then(|| {
+                client_shell_resize_message(
+                    shell,
+                    state.reported_size.0,
+                    state.reported_size.1,
+                    state.reported_cell_size.0,
+                    state.reported_cell_size.1,
+                    state.pixel_geometry_exact,
+                )
+            }),
+            graphics_cleanup,
+        )
+    } else {
+        (None, None, Vec::new())
+    };
+    apply_client_shell_input_source_changes(state, prefix_input_source);
+    state.present_graphics(&graphics_cleanup);
+    if let Some(resize) = resize {
+        write_to_server(write_stream, &resize).map_err(ClientError::ConnectionLost)?;
+    }
+    if let Some(frame) = composed {
+        state.present_frame(frame);
+    }
+    Ok(())
+}
+
 fn finish_client_shell_input(
     state: &mut ClientState,
     outcome: shell::ClientShellInput,
@@ -682,6 +722,7 @@ async fn run_client_loop(
     should_quit: Arc<AtomicBool>,
     config: ClientLoopConfig,
     negotiated_encoding: RenderEncoding,
+    endpoint_methods: Option<Vec<String>>,
     attach_escape: Option<AttachEscapeState>,
 ) -> Result<(), ClientError> {
     #[cfg(windows)]
@@ -723,6 +764,7 @@ async fn run_client_loop(
     };
     if let Some(shell) = state.shell.as_mut() {
         shell.set_graphics_cell_size(initial_cell_width_px, initial_cell_height_px);
+        shell.set_endpoint_methods(endpoint_methods);
     }
     debug!(?negotiated_encoding, "client render encoding active");
     let host_mouse_capture_active = Arc::new(AtomicBool::new(state.mouse_capture_active));
@@ -1218,42 +1260,13 @@ async fn run_client_loop(
                 }
             }
             ClientLoopEvent::ServerMessage(msg) => match *msg {
-                ServerMessage::ClientShellSnapshot(snapshot) => {
-                    let (composed, resize, graphics_cleanup) = if let Some(shell) = &mut state.shell
-                    {
-                        let previous_size =
-                            shell.surface_size(state.reported_size.0, state.reported_size.1);
-                        shell.set_snapshot(snapshot);
-                        let graphics_cleanup = shell.take_pending_graphics_cleanup();
-                        let next_size =
-                            shell.surface_size(state.reported_size.0, state.reported_size.1);
-                        (
-                            shell.compose(state.reported_size.0, state.reported_size.1),
-                            (previous_size != next_size).then(|| {
-                                client_shell_resize_message(
-                                    shell,
-                                    state.reported_size.0,
-                                    state.reported_size.1,
-                                    state.reported_cell_size.0,
-                                    state.reported_cell_size.1,
-                                    state.pixel_geometry_exact,
-                                )
-                            }),
-                            graphics_cleanup,
-                        )
-                    } else {
-                        (None, None, Vec::new())
-                    };
-                    apply_client_shell_input_source_changes(&mut state, &mut prefix_input_source);
-                    state.present_graphics(&graphics_cleanup);
-                    if let Some(resize) = resize {
-                        if let Err(err) = write_to_server(&mut write_stream, &resize) {
-                            return Err(ClientError::ConnectionLost(err));
-                        }
-                    }
-                    if let Some(frame) = composed {
-                        state.present_frame(frame);
-                    }
+                ServerMessage::ClientShellSnapshot(_) => {
+                    return Err(ClientError::Protocol(protocol::FramingError::Io(
+                        io::Error::new(
+                            io::ErrorKind::InvalidData,
+                            "server sent an unnegotiated binary endpoint snapshot",
+                        ),
+                    )));
                 }
                 ServerMessage::PaneSurface(surface) => {
                     let composed = if let Some(shell) = &mut state.shell {
@@ -1699,6 +1712,35 @@ async fn run_client_loop(
                         sync_client_shell_keyboard_report_all(&mut state)?;
                     }
                 }
+                ServerMessage::EndpointControl { kind, data } => {
+                    if kind != crate::protocol::endpoint::ENDPOINT_SNAPSHOT_KIND {
+                        if kind.starts_with("shell.snapshot.") {
+                            return Err(ClientError::Protocol(protocol::FramingError::Io(
+                                io::Error::new(
+                                    io::ErrorKind::InvalidData,
+                                    format!(
+                                        "unsupported mandatory endpoint snapshot codec {kind:?}"
+                                    ),
+                                ),
+                            )));
+                        }
+                        debug!(%kind, "ignoring unknown endpoint control message");
+                        continue;
+                    }
+                    let snapshot: crate::protocol::ClientShellSnapshot =
+                        serde_json::from_str(&data).map_err(|error| {
+                            ClientError::Protocol(protocol::FramingError::Io(io::Error::new(
+                                io::ErrorKind::InvalidData,
+                                format!("invalid endpoint snapshot: {error}"),
+                            )))
+                        })?;
+                    install_client_shell_snapshot(
+                        &mut state,
+                        Box::new(snapshot),
+                        &mut write_stream,
+                        &mut prefix_input_source,
+                    )?;
+                }
                 ServerMessage::Welcome { .. } => {
                     debug!("received unexpected Welcome in main loop");
                 }
@@ -1719,10 +1761,20 @@ async fn run_client_loop(
                     .detached_process_children
                     .retain_mut(|child| child.try_wait().ok().flatten().is_none());
                 if state.shell.is_some() {
+                    let expired_endpoint = endpoint_commands.expire(now);
                     let (effects, outcome, frame) = {
                         let shell = state.shell.as_mut().expect("checked shell mode");
-                        let (effects, notification_repaint) = shell.tick_notifications(now);
                         let mut outcome = shell.tick_selection_autoscroll(now);
+                        if let Some(expired) = expired_endpoint {
+                            let (repaint, actions) = shell.handle_endpoint_result(
+                                &expired.boot_id,
+                                &expired.request_id,
+                                expired.result,
+                            );
+                            outcome.repaint |= repaint;
+                            outcome.actions.extend(actions);
+                        }
+                        let (effects, notification_repaint) = shell.tick_notifications(now);
                         outcome.repaint |= notification_repaint | shell.tick_copy_feedback(now);
                         let frame = outcome
                             .repaint
diff --git a/src/client/shell/actions.rs b/src/client/shell/actions.rs
index efc9aae9..34c780da 100644
--- a/src/client/shell/actions.rs
+++ b/src/client/shell/actions.rs
@@ -227,11 +227,13 @@ impl ClientShellState {
                 if action == crate::protocol::ClientShellCommandAction::Popup {
                     self.popup_pending = true;
                     self.popup_pending_deadline = None;
-                    self.push_endpoint_method_with_kind(
+                    if !self.push_endpoint_method_with_kind(
                         crate::api::schema::Method::CommandInvoke(params),
                         PendingEndpointKind::PopupCommand,
                         outcome,
-                    );
+                    ) {
+                        self.popup_pending = false;
+                    }
                 } else {
                     self.push_endpoint_method(
                         crate::api::schema::Method::CommandInvoke(params),
@@ -323,7 +325,7 @@ impl ClientShellState {
         self.word_selection_generation = self.word_selection_generation.saturating_add(1);
         let generation = self.word_selection_generation;
         self.pending_word_selection = Some(generation);
-        self.push_endpoint_method_with_kind(
+        if !self.push_endpoint_method_with_kind(
             crate::api::schema::Method::PaneSelectionRead(
                 crate::api::schema::PaneSelectionReadParams {
                     pane_id: hit.pane_id.clone(),
@@ -345,7 +347,9 @@ impl ClientShellState {
                 generation,
             },
             outcome,
-        );
+        ) {
+            self.pending_word_selection = None;
+        }
     }
 
     pub(super) fn push_endpoint_method(
@@ -356,15 +360,65 @@ impl ClientShellState {
         self.push_endpoint_method_with_kind(method, PendingEndpointKind::Generic, outcome);
     }
 
+    fn push_endpoint_notice(
+        &mut self,
+        kind: ClientEndpointNoticeKind,
+        code: impl Into<String>,
+        title: impl Into<String>,
+        body: impl Into<String>,
+    ) -> bool {
+        let key = ClientEndpointNoticeKey {
+            boot_id: self
+                .snapshot
+                .as_deref()
+                .map(|snapshot| snapshot.boot_id.clone())
+                .unwrap_or_else(|| "disconnected".to_owned()),
+            kind,
+            code: code.into(),
+        };
+        let body = body.into();
+        if kind == ClientEndpointNoticeKind::Rejected {
+            if self
+                .visible_endpoint_notice
+                .as_ref()
+                .is_some_and(|notice| notice.key == key && notice.body == body)
+            {
+                return false;
+            }
+        } else if !self.endpoint_notice_seen.insert(key.clone()) {
+            return false;
+        }
+        self.visible_endpoint_notice = Some(ClientVisibleEndpointNotice {
+            key,
+            title: title.into(),
+            body,
+            deadline: std::time::Instant::now() + std::time::Duration::from_secs(8),
+        });
+        true
+    }
+
     pub(super) fn push_endpoint_method_with_kind(
         &mut self,
         method: crate::api::schema::Method,
         kind: PendingEndpointKind,
         outcome: &mut ClientShellInput,
-    ) {
-        let Some(snapshot) = self.snapshot.as_deref() else {
-            return;
-        };
+    ) -> bool {
+        if self.snapshot.is_none() {
+            return false;
+        }
+        let method_name = crate::api::api_method_name(&method).to_owned();
+        if !self.supports_endpoint_method(&method) {
+            outcome.repaint |= self.push_endpoint_notice(
+                ClientEndpointNoticeKind::Unsupported,
+                method_name.clone(),
+                "Action unavailable",
+                format!(
+                    "This server does not support {method_name} yet. Update and restart it to enable this action."
+                ),
+            );
+            return false;
+        }
+        let snapshot = self.snapshot.as_deref().expect("checked snapshot");
         let confirmation_workspace_id = match &method {
             crate::api::schema::Method::TabClose(target) => snapshot
                 .tabs
@@ -385,6 +439,7 @@ impl ClientShellState {
             request_id.clone(),
             PendingEndpointRequest {
                 boot_id: snapshot.boot_id.clone(),
+                method_name,
                 confirmation_workspace_id,
                 kind,
             },
@@ -396,14 +451,16 @@ impl ClientShellState {
                 method,
             }),
         });
+        true
     }
 
     pub(crate) fn receive_endpoint_error(&mut self, message: String) -> bool {
-        if self.endpoint_error.as_deref() == Some(message.as_str()) {
-            return false;
-        }
-        self.endpoint_error = Some(message);
-        true
+        self.push_endpoint_notice(
+            ClientEndpointNoticeKind::Rejected,
+            "paste_rejected",
+            "Paste rejected",
+            message,
+        )
     }
 
     pub(crate) fn handle_endpoint_result(
@@ -423,12 +480,49 @@ impl ClientShellState {
         {
             return (false, Vec::new());
         }
+        if result.is_ok() {
+            let timeout_key = ClientEndpointNoticeKey {
+                boot_id: boot_id.to_owned(),
+                kind: ClientEndpointNoticeKind::Timeout,
+                code: pending.method_name.clone(),
+            };
+            self.endpoint_notice_seen.remove(&timeout_key);
+        }
+        if let Err(error) = &result {
+            let code = error.code.as_deref().unwrap_or("invalid_response");
+            if !matches!(
+                code,
+                "confirmation_required" | "stale_content" | "stale_target"
+            ) {
+                let (kind, notice_code, title, body) = match code {
+                    "endpoint_timeout" => (
+                        ClientEndpointNoticeKind::Timeout,
+                        pending.method_name.clone(),
+                        "Server timed out",
+                        format!("This server did not respond to {}.", pending.method_name),
+                    ),
+                    "server_unavailable" => (
+                        ClientEndpointNoticeKind::Unavailable,
+                        "server".to_owned(),
+                        "Server unavailable",
+                        error.message.clone(),
+                    ),
+                    _ => (
+                        ClientEndpointNoticeKind::Rejected,
+                        format!("{}:{code}", pending.method_name),
+                        "Action rejected",
+                        error.message.clone(),
+                    ),
+                };
+                self.push_endpoint_notice(kind, notice_code, title, body);
+            }
+        }
         match pending.kind {
             PendingEndpointKind::Generic => {}
             PendingEndpointKind::ProductAnnouncementDismiss { version, id } => {
                 return match result {
                     Ok(_) => (false, Vec::new()),
-                    Err(error) => {
+                    Err(_) => {
                         let key = (version.clone(), id.clone());
                         if self.dismissed_product_announcement.as_ref() == Some(&key) {
                             self.dismissed_product_announcement = None;
@@ -447,7 +541,6 @@ impl ClientShellState {
                                 ));
                             }
                         }
-                        self.endpoint_error = Some(error.message);
                         (true, Vec::new())
                     }
                 };
@@ -455,7 +548,7 @@ impl ClientShellState {
             PendingEndpointKind::ReleaseNotesDismiss => {
                 return match result {
                     Ok(_) => (false, Vec::new()),
-                    Err(error) => {
+                    Err(_) => {
                         if self.overlay.is_none() {
                             if let Some(notes) = self
                                 .snapshot
@@ -467,7 +560,6 @@ impl ClientShellState {
                                 ));
                             }
                         }
-                        self.endpoint_error = Some(error.message);
                         (true, Vec::new())
                     }
                 };
@@ -479,10 +571,9 @@ impl ClientShellState {
                             Some(std::time::Instant::now() + std::time::Duration::from_secs(1));
                         (false, Vec::new())
                     }
-                    Err(error) => {
+                    Err(_) => {
                         self.popup_pending = false;
                         self.popup_pending_deadline = None;
-                        self.endpoint_error = Some(error.message);
                         (true, Vec::new())
                     }
                 };
@@ -517,10 +608,7 @@ impl ClientShellState {
                             Some("endpoint returned an unexpected selection result".to_owned());
                         (true, fallback())
                     }
-                    Err(error) => {
-                        self.endpoint_error = Some(error.message);
-                        (true, fallback())
-                    }
+                    Err(_) => (true, fallback()),
                 };
             }
             PendingEndpointKind::WordSelection {
@@ -551,10 +639,7 @@ impl ClientShellState {
                         );
                         return (true, Vec::new());
                     }
-                    Err(error) => {
-                        self.endpoint_error = Some(error.message);
-                        return (true, Vec::new());
-                    }
+                    Err(_) => return (true, Vec::new()),
                 };
                 let Some((start_col, end_col)) =
                     crate::app::actions::word_bounds_at_column(&row_text, col)
@@ -640,10 +725,7 @@ impl ClientShellState {
                         self.url_click_consumes_until_up = completed_before_release;
                         (false, Vec::new())
                     }
-                    Err(error) => {
-                        self.endpoint_error = Some(error.message);
-                        (true, replay_action(replay))
-                    }
+                    Err(_) => (true, replay_action(replay)),
                 };
             }
             PendingEndpointKind::CopyMotion {
@@ -673,10 +755,7 @@ impl ClientShellState {
                             Some("endpoint returned an unexpected copy-motion result".to_owned());
                         (true, false)
                     }
-                    Err(error) => {
-                        self.endpoint_error = Some(error.message);
-                        (true, false)
-                    }
+                    Err(_) => (true, false),
                 };
                 self.complete_copy_operation(session_generation, continue_queue, &mut outcome);
                 return (repaint || outcome.repaint, outcome.actions);
@@ -731,9 +810,8 @@ impl ClientShellState {
                             Some("endpoint returned an unexpected copy-search result".to_owned());
                         (true, false)
                     }
-                    Err(error) => {
+                    Err(_) => {
                         self.cancel_deferred_copy_after_search(generation);
-                        self.endpoint_error = Some(error.message);
                         (true, false)
                     }
                 };
@@ -748,10 +826,7 @@ impl ClientShellState {
                             Some("endpoint returned an unexpected config reload result".to_owned());
                         true
                     }
-                    Err(error) => {
-                        self.endpoint_error = Some(error.message);
-                        true
-                    }
+                    Err(_) => true,
                 };
                 return (repaint, Vec::new());
             }
@@ -778,10 +853,7 @@ impl ClientShellState {
                 }
                 true
             }
-            Err(error) => {
-                self.endpoint_error = Some(error.message);
-                true
-            }
+            Err(_) => true,
         };
         (repaint, Vec::new())
     }
diff --git a/src/client/shell/composition.rs b/src/client/shell/composition.rs
index 6432f3e1..04ca995c 100644
--- a/src/client/shell/composition.rs
+++ b/src/client/shell/composition.rs
@@ -269,7 +269,10 @@ impl ClientShellState {
         restore_mode_bar(&mut frame, mode_bar, mode_bar_cells.as_deref());
         self.hits.notification_toast = Rect::default();
         let has_config_diagnostic = self.config_diagnostic.is_some();
-        if has_config_diagnostic || self.visible_notification.is_some() {
+        if has_config_diagnostic
+            || self.visible_endpoint_notice.is_some()
+            || self.visible_notification.is_some()
+        {
             let cursor = frame.cursor.clone();
             let mut composed = frame.to_ratatui_buffer()?;
             if let Some(diagnostic) = self.config_diagnostic.as_deref() {
@@ -285,7 +288,25 @@ impl ClientShellState {
                     &self.config.palette,
                 );
             }
-            if let Some(notification) = self.visible_notification.as_ref() {
+            if let Some(notice) = self.visible_endpoint_notice.as_ref() {
+                self.hits.notification_toast = if layout.mobile_header.is_empty() {
+                    notifications::render_endpoint_notice(
+                        &mut composed,
+                        Rect::new(0, 0, cols, rows),
+                        notice,
+                        u16::from(has_config_diagnostic),
+                        &self.config.palette,
+                    )
+                } else {
+                    notifications::render_mobile_endpoint_notice_banner(
+                        &mut composed,
+                        Rect::new(0, 0, cols, rows),
+                        notice,
+                        has_config_diagnostic,
+                        &self.config.palette,
+                    )
+                };
+            } else if let Some(notification) = self.visible_notification.as_ref() {
                 self.hits.notification_toast = if layout.mobile_header.is_empty() {
                     notifications::render_visible_notification(
                         &mut composed,
diff --git a/src/client/shell/config.rs b/src/client/shell/config.rs
index 0273cefb..1c2993b7 100644
--- a/src/client/shell/config.rs
+++ b/src/client/shell/config.rs
@@ -200,16 +200,8 @@ impl ClientShellConfig {
                 };
                 config.keys.command = commands
                     .iter()
-                    .map(|command| crate::config::CommandKeybindConfig {
-                        key: if command.binding_labels.len() == 1 {
-                            crate::config::BindingConfig::One(command.binding_labels[0].clone())
-                        } else {
-                            crate::config::BindingConfig::Many(command.binding_labels.clone())
-                        },
-                        // The client never executes this field; preserve the opaque endpoint ID
-                        // through the shared config collision resolver.
-                        command: command.command_id.clone(),
-                        action_type: match command.action {
+                    .filter_map(|command| {
+                        let action_type = match command.action {
                             crate::protocol::ClientShellCommandAction::Shell => {
                                 crate::config::CommandKeybindType::Shell
                             }
@@ -222,10 +214,22 @@ impl ClientShellConfig {
                             crate::protocol::ClientShellCommandAction::PluginAction => {
                                 crate::config::CommandKeybindType::PluginAction
                             }
-                        },
-                        description: command.description.clone(),
-                        width: None,
-                        height: None,
+                            crate::protocol::ClientShellCommandAction::Unknown => return None,
+                        };
+                        Some(crate::config::CommandKeybindConfig {
+                            key: if command.binding_labels.len() == 1 {
+                                crate::config::BindingConfig::One(command.binding_labels[0].clone())
+                            } else {
+                                crate::config::BindingConfig::Many(command.binding_labels.clone())
+                            },
+                            // The client never executes this field; preserve the opaque endpoint ID
+                            // through the shared config collision resolver.
+                            command: command.command_id.clone(),
+                            action_type,
+                            description: command.description.clone(),
+                            width: None,
+                            height: None,
+                        })
                     })
                     .collect();
                 config
@@ -236,6 +240,9 @@ impl ClientShellConfig {
         };
         if self.keybinding_source == ClientShellKeybindingSource::Endpoint {
             for command in commands {
+                let Ok(action) = command.action.try_into() else {
+                    continue;
+                };
                 keybinds
                     .keybinds
                     .custom_commands
@@ -245,7 +252,7 @@ impl ClientShellConfig {
                         )?,
                         label: command.binding_label.clone(),
                         command: command.command_id.clone(),
-                        action: command.action.into(),
+                        action,
                         description: command.description.clone(),
                         width: None,
                         height: None,
diff --git a/src/client/shell/copy_mode.rs b/src/client/shell/copy_mode.rs
index c39a7507..32a17bf4 100644
--- a/src/client/shell/copy_mode.rs
+++ b/src/client/shell/copy_mode.rs
@@ -797,7 +797,9 @@ impl ClientShellState {
                 }
             };
             self.copy_operation_in_flight = true;
-            self.push_endpoint_method_with_kind(method, kind, outcome);
+            if !self.push_endpoint_method_with_kind(method, kind, outcome) {
+                self.copy_operation_in_flight = false;
+            }
             return;
         }
     }
diff --git a/src/client/shell/graphics.rs b/src/client/shell/graphics.rs
index 1fc23414..8a512d3a 100644
--- a/src/client/shell/graphics.rs
+++ b/src/client/shell/graphics.rs
@@ -39,6 +39,7 @@ impl ClientShellState {
             || self.mode != ClientShellMode::Terminal
             || self.endpoint_error.is_some()
             || self.config_diagnostic.is_some()
+            || self.visible_endpoint_notice.is_some()
             || self.visible_notification.is_some()
             || self.copy_feedback.is_some()
             || self
diff --git a/src/client/shell/mouse.rs b/src/client/shell/mouse.rs
index 437ec4ed..612319e2 100644
--- a/src/client/shell/mouse.rs
+++ b/src/client/shell/mouse.rs
@@ -78,14 +78,20 @@ impl ClientShellState {
         self.pane_scroll_targets
             .insert(pane_id.clone(), offset_from_bottom);
         self.pane_scroll_in_flight.insert(pane_id.clone(), serial);
-        self.push_endpoint_method_with_kind(
+        if !self.push_endpoint_method_with_kind(
             crate::api::schema::Method::PaneScroll(crate::api::schema::PaneScrollParams {
                 pane_id: pane_id.clone(),
                 offset_from_bottom: offset_from_bottom as u64,
             }),
-            PendingEndpointKind::PaneScroll { pane_id, serial },
+            PendingEndpointKind::PaneScroll {
+                pane_id: pane_id.clone(),
+                serial,
+            },
             outcome,
-        );
+        ) {
+            self.pane_scroll_targets.remove(&pane_id);
+            self.pane_scroll_in_flight.remove(&pane_id);
+        }
     }
 
     pub(super) fn complete_pane_scroll(
@@ -120,10 +126,9 @@ impl ClientShellState {
                     Some("endpoint returned an unexpected pane-scroll result".to_owned());
                 true
             }
-            Err(error) => {
+            Err(_) => {
                 self.pane_scroll_queued.remove(&pane_id);
                 self.pane_scroll_targets.remove(&pane_id);
-                self.endpoint_error = Some(error.message);
                 true
             }
         };
@@ -893,6 +898,14 @@ impl ClientShellState {
                 return;
             }
         }
+        if self.visible_endpoint_notice.is_some()
+            && mouse.kind == MouseEventKind::Down(MouseButton::Left)
+            && super::contains(self.hits.notification_toast, point)
+        {
+            self.visible_endpoint_notice = None;
+            outcome.repaint = true;
+            return;
+        }
         if self.overlay.is_none()
             && self.mode == ClientShellMode::Terminal
             && self
diff --git a/src/client/shell/notifications.rs b/src/client/shell/notifications.rs
index 0310e57d..43fe566a 100644
--- a/src/client/shell/notifications.rs
+++ b/src/client/shell/notifications.rs
@@ -1,13 +1,16 @@
 use super::*;
 use ratatui::{
+    style::Color,
     text::{Line, Span},
     widgets::{Block, Borders, Clear, Paragraph, Widget},
 };
 
-pub(super) fn render_mobile_notification_banner(
+fn render_mobile_notice_banner(
     buffer: &mut Buffer,
     area: Rect,
-    notification: &ClientVisibleNotification,
+    title: &str,
+    body: Option<&str>,
+    dot_color: Color,
     offset_for_warning: bool,
     palette: &Palette,
 ) -> Rect {
@@ -23,6 +26,49 @@ pub(super) fn render_mobile_notification_banner(
     let background = palette.surface0;
     Clear.render(rect, buffer);
     buffer.set_style(rect, Style::default().bg(background));
+    let mut x = rect.x;
+    for (text, style) in [
+        (" ", Style::default().bg(background)),
+        ("●", Style::default().fg(dot_color).bg(background)),
+        (" ", Style::default().bg(background)),
+        (
+            title,
+            Style::default()
+                .fg(palette.text)
+                .bg(background)
+                .add_modifier(Modifier::BOLD),
+        ),
+    ] {
+        x = super::render::put_segment(buffer, x, rect.y, rect.right(), text, style);
+    }
+    if let Some(body) = body.filter(|body| !body.is_empty()) {
+        x = super::render::put_segment(
+            buffer,
+            x,
+            rect.y,
+            rect.right(),
+            " · ",
+            Style::default().fg(palette.overlay0).bg(background),
+        );
+        super::render::put_text(
+            buffer,
+            x,
+            rect.y,
+            rect.right().saturating_sub(x),
+            body,
+            Style::default().fg(palette.overlay0).bg(background),
+        );
+    }
+    rect
+}
+
+pub(super) fn render_mobile_notification_banner(
+    buffer: &mut Buffer,
+    area: Rect,
+    notification: &ClientVisibleNotification,
+    offset_for_warning: bool,
+    palette: &Palette,
+) -> Rect {
     let event = &notification.event;
     let title = match event.kind {
         SemanticNotificationKind::NeedsAttention => event
@@ -45,84 +91,55 @@ pub(super) fn render_mobile_notification_banner(
             palette.accent
         }
     };
-    let mut x = rect.x;
-    x = super::render::put_segment(
-        buffer,
-        x,
-        rect.y,
-        rect.right(),
-        " ",
-        Style::default().bg(background),
-    );
-    x = super::render::put_segment(
-        buffer,
-        x,
-        rect.y,
-        rect.right(),
-        "●",
-        Style::default().fg(dot_color).bg(background),
-    );
-    x = super::render::put_segment(
+    render_mobile_notice_banner(
         buffer,
-        x,
-        rect.y,
-        rect.right(),
-        " ",
-        Style::default().bg(background),
-    );
-    x = super::render::put_segment(
-        buffer,
-        x,
-        rect.y,
-        rect.right(),
+        area,
         &title,
-        Style::default()
-            .fg(palette.text)
-            .bg(background)
-            .add_modifier(Modifier::BOLD),
-    );
-    if let Some(body) = event.body.as_deref().filter(|body| !body.is_empty()) {
-        x = super::render::put_segment(
-            buffer,
-            x,
-            rect.y,
-            rect.right(),
-            " · ",
-            Style::default().fg(palette.overlay0).bg(background),
-        );
-        super::render::put_text(
-            buffer,
-            x,
-            rect.y,
-            rect.right().saturating_sub(x),
-            body,
-            Style::default().fg(palette.overlay0).bg(background),
-        );
-    }
-    rect
+        event.body.as_deref(),
+        dot_color,
+        offset_for_warning,
+        palette,
+    )
 }
 
-pub(super) fn render_visible_notification(
+pub(super) fn render_mobile_endpoint_notice_banner(
     buffer: &mut Buffer,
     area: Rect,
-    notification: &ClientVisibleNotification,
-    default_position: crate::config::ToastHerdrPosition,
+    notice: &ClientVisibleEndpointNotice,
+    offset_for_warning: bool,
+    palette: &Palette,
+) -> Rect {
+    render_mobile_notice_banner(
+        buffer,
+        area,
+        &notice.title,
+        Some(&notice.body),
+        palette.red,
+        offset_for_warning,
+        palette,
+    )
+}
+
+fn render_notification_card(
+    buffer: &mut Buffer,
+    area: Rect,
+    title: &str,
+    body: &str,
+    position: crate::config::ToastHerdrPosition,
     top_offset: u16,
+    dot_color: Color,
     palette: &Palette,
 ) -> Rect {
     if area.is_empty() {
         return Rect::default();
     }
-    let event = &notification.event;
-    let body = event.body.as_deref().unwrap_or_default();
-    let content_width = unicode_width::UnicodeWidthStr::width(event.title.as_str())
+    let content_width = unicode_width::UnicodeWidthStr::width(title)
         .max(unicode_width::UnicodeWidthStr::width(body))
         .saturating_add(6);
     let width = u16::try_from(content_width)
         .unwrap_or(u16::MAX)
         .min(area.width);
     let height: u16 = if body.is_empty() { 3 } else { 4 }.min(area.height);
-    let position = event.position.unwrap_or(default_position);
     let x = match position {
         crate::config::ToastHerdrPosition::TopLeft
         | crate::config::ToastHerdrPosition::BottomLeft => area.x,
@@ -148,19 +165,10 @@ pub(super) fn render_visible_notification(
     let inner = block.inner(rect);
     block.render(rect, buffer);
     Paragraph::new(Line::from(vec![
-        Span::styled(
-            "●",
-            Style::default().fg(match event.kind {
-                SemanticNotificationKind::NeedsAttention => palette.red,
-                SemanticNotificationKind::Finished => palette.blue,
-                SemanticNotificationKind::UpdateInstalled | SemanticNotificationKind::Custom => {
-                    palette.accent
-                }
-            }),
-        ),
+        Span::styled("●", Style::default().fg(dot_color)),
         Span::raw(" "),
         Span::styled(
-            event.title.as_str(),
+            title,
             Style::default()
                 .fg(palette.text)
                 .add_modifier(Modifier::BOLD),
@@ -185,6 +193,60 @@ pub(super) fn render_visible_notification(
     rect
 }
 
+pub(super) fn render_visible_notification(
+    buffer: &mut Buffer,
+    area: Rect,
+    notification: &ClientVisibleNotification,
+    default_position: crate::config::ToastHerdrPosition,
+    top_offset: u16,
+    palette: &Palette,
+) -> Rect {
+    let event = &notification.event;
+    let dot_color = match event.kind {
+        SemanticNotificationKind::NeedsAttention => palette.red,
+        SemanticNotificationKind::Finished => palette.blue,
+        SemanticNotificationKind::UpdateInstalled | SemanticNotificationKind::Custom => {
+            palette.accent
+        }
+    };
+    render_notification_card(
+        buffer,
+        area,
+        &event.title,
+        event.body.as_deref().unwrap_or_default(),
+        event.position.unwrap_or(default_position),
+        top_offset,
+        dot_color,
+        palette,
+    )
+}
+
+pub(super) fn render_endpoint_notice(
+    buffer: &mut Buffer,
+    area: Rect,
+    notice: &ClientVisibleEndpointNotice,
+    top_offset: u16,
+    palette: &Palette,
+) -> Rect {
+    render_notification_card(
+        buffer,
+        area,
+        &notice.title,
+        &notice.body,
+        crate::config::ToastHerdrPosition::TopRight,
+        top_offset,
+        match notice.key.kind {
+            ClientEndpointNoticeKind::Unsupported | ClientEndpointNoticeKind::Rejected => {
+                palette.red
+            }
+            ClientEndpointNoticeKind::Timeout | ClientEndpointNoticeKind::Unavailable => {
+                palette.yellow
+            }
+        },
+        palette,
+    )
+}
+
 impl ClientShellState {
     pub(super) fn focus_visible_notification(&mut self, outcome: &mut ClientShellInput) {
         let Some(notification) = self.visible_notification.take() else {
@@ -246,6 +308,14 @@ impl ClientShellState {
             self.visible_notification = None;
             repaint = true;
         }
+        if self
+            .visible_endpoint_notice
+            .as_ref()
+            .is_some_and(|visible| now >= visible.deadline)
+        {
+            self.visible_endpoint_notice = None;
+            repaint = true;
+        }
 
         let pending = std::mem::take(&mut self.pending_notifications);
         let mut effects = Vec::new();
diff --git a/src/client/shell/settings.rs b/src/client/shell/settings.rs
index 95a437fe..d90d371f 100644
--- a/src/client/shell/settings.rs
+++ b/src/client/shell/settings.rs
@@ -233,11 +233,15 @@ impl ClientShellState {
                 settings.integration_messages.clear();
             }
         }
-        self.push_endpoint_method_with_kind(
+        if !self.push_endpoint_method_with_kind(
             crate::api::schema::Method::IntegrationList(crate::api::schema::EmptyParams::default()),
             PendingEndpointKind::IntegrationList,
             outcome,
-        );
+        ) {
+            if let Some(ClientShellOverlay::Settings(settings)) = self.overlay.as_mut() {
+                settings.loading_integrations = false;
+            }
+        }
     }
 
     fn install_recommended_integrations(&mut self, outcome: &mut ClientShellInput) {
@@ -260,15 +264,22 @@ impl ClientShellState {
             settings.installing_integrations = true;
             settings.integration_messages.clear();
         }
-        self.pending_integration_installs = targets.len();
+        self.pending_integration_installs = 0;
         for target in targets {
-            self.push_endpoint_method_with_kind(
+            if self.push_endpoint_method_with_kind(
                 crate::api::schema::Method::IntegrationInstall(
                     crate::api::schema::IntegrationInstallParams { target },
                 ),
                 PendingEndpointKind::IntegrationInstall,
                 outcome,
-            );
+            ) {
+                self.pending_integration_installs += 1;
+            }
+        }
+        if self.pending_integration_installs == 0 {
+            if let Some(ClientShellOverlay::Settings(settings)) = self.overlay.as_mut() {
+                settings.installing_integrations = false;
+            }
         }
         outcome.repaint = true;
     }
@@ -296,7 +307,7 @@ impl ClientShellState {
                                 "endpoint returned an unexpected integration list result".into(),
                             );
                         }
-                        Err(error) => self.endpoint_error = Some(error.message),
+                        Err(_) => {}
                     }
                 }
                 (true, Vec::new())
diff --git a/src/client/shell/state.rs b/src/client/shell/state.rs
index 81173c18..69e77dc5 100644
--- a/src/client/shell/state.rs
+++ b/src/client/shell/state.rs
@@ -636,10 +636,33 @@ pub(super) enum PendingEndpointKind {
 
 pub(super) struct PendingEndpointRequest {
     pub(super) boot_id: String,
+    pub(super) method_name: String,
     pub(super) confirmation_workspace_id: Option<String>,
     pub(super) kind: PendingEndpointKind,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub(super) enum ClientEndpointNoticeKind {
+    Unsupported,
+    Rejected,
+    Timeout,
+    Unavailable,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub(super) struct ClientEndpointNoticeKey {
+    pub(super) boot_id: String,
+    pub(super) kind: ClientEndpointNoticeKind,
+    pub(super) code: String,
+}
+
+pub(super) struct ClientVisibleEndpointNotice {
+    pub(super) key: ClientEndpointNoticeKey,
+    pub(super) title: String,
+    pub(super) body: String,
+    pub(super) deadline: std::time::Instant,
+}
+
 pub(crate) struct ClientShellEndpointError {
     pub code: Option<String>,
     pub message: String,
@@ -836,10 +859,16 @@ pub(crate) struct ClientShellState {
     pub(super) popup_pending: bool,
     pub(super) popup_pending_deadline: Option<std::time::Instant>,
     pub(super) next_request_id: u64,
+    /// Methods advertised by this endpoint. `None` is used only by local tests
+    /// and legacy construction paths; negotiated endpoint connections always
+    /// install an explicit set.
+    pub(super) endpoint_methods: Option<HashSet<String>>,
     pub(super) pending_requests: HashMap<String, PendingEndpointRequest>,
     pub(super) pending_integration_installs: usize,
     pub(super) pending_notifications: Vec<ClientPendingNotification>,
     pub(super) visible_notification: Option<ClientVisibleNotification>,
+    pub(super) endpoint_notice_seen: HashSet<ClientEndpointNoticeKey>,
+    pub(super) visible_endpoint_notice: Option<ClientVisibleEndpointNotice>,
     pub(super) outer_focused: Option<bool>,
     pub(super) ascii_input_source_active: bool,
     pub(super) pending_input_source_changes: Vec<bool>,
@@ -971,10 +1000,13 @@ impl ClientShellState {
             popup_pending: false,
             popup_pending_deadline: None,
             next_request_id: 1,
+            endpoint_methods: None,
             pending_requests: HashMap::new(),
             pending_integration_installs: 0,
             pending_notifications: Vec::new(),
             visible_notification: None,
+            endpoint_notice_seen: HashSet::new(),
+            visible_endpoint_notice: None,
             outer_focused: None,
             ascii_input_source_active: false,
             pending_input_source_changes: Vec::new(),
@@ -987,6 +1019,16 @@ impl ClientShellState {
         }
     }
 
+    pub(crate) fn set_endpoint_methods(&mut self, methods: Option<Vec<String>>) {
+        self.endpoint_methods = methods.map(|methods| methods.into_iter().collect());
+    }
+
+    pub(super) fn supports_endpoint_method(&self, method: &crate::api::schema::Method) -> bool {
+        self.endpoint_methods
+            .as_ref()
+            .is_none_or(|methods| methods.contains(crate::api::api_method_name(method)))
+    }
+
     fn focused_tab_count(&self) -> usize {
         let Some(snapshot) = self.snapshot.as_deref() else {
             return 0;
@@ -1072,7 +1114,10 @@ impl ClientShellState {
         }
     }
 
-    pub(crate) fn set_snapshot(&mut self, snapshot: Box<ClientShellSnapshot>) {
+    pub(crate) fn set_snapshot(&mut self, mut snapshot: Box<ClientShellSnapshot>) {
+        snapshot
+            .commands
+            .retain(|command| command.action != crate::protocol::ClientShellCommandAction::Unknown);
         if self.snapshot.as_ref().is_some_and(|current| {
             current.boot_id == snapshot.boot_id && snapshot.revision < current.revision
         }) {
@@ -1154,6 +1199,8 @@ impl ClientShellState {
             self.pending_integration_installs = 0;
             self.pending_notifications.clear();
             self.visible_notification = None;
+            self.endpoint_notice_seen.clear();
+            self.visible_endpoint_notice = None;
             self.endpoint_error = None;
             self.navigate_workspace_id = None;
             self.overlay = self
diff --git a/src/client/shell/surface_patch.rs b/src/client/shell/surface_patch.rs
index 10486f50..784fbcd8 100644
--- a/src/client/shell/surface_patch.rs
+++ b/src/client/shell/surface_patch.rs
@@ -66,6 +66,8 @@ fn fast_path_blocker(
         Some("client_surface_patch.fallback.endpoint_error")
     } else if state.config_diagnostic.is_some() {
         Some("client_surface_patch.fallback.config_diagnostic")
+    } else if state.visible_endpoint_notice.is_some() {
+        Some("client_surface_patch.fallback.endpoint_notice")
     } else if state.visible_notification.is_some() {
         Some("client_surface_patch.fallback.notification")
     } else if state.copy_feedback.is_some() {
diff --git a/src/client/shell/tests/agents_worktrees_notifications.rs b/src/client/shell/tests/agents_worktrees_notifications.rs
index 0ee5aa3f..055b594d 100644
--- a/src/client/shell/tests/agents_worktrees_notifications.rs
+++ b/src/client/shell/tests/agents_worktrees_notifications.rs
@@ -983,6 +983,36 @@ fn worktree_create_previews_the_endpoint_owned_checkout_path() {
 }
 
 #[test]
+fn unavailable_worktree_create_does_not_wedge_the_overlay() {
+    let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
+    state.set_snapshot(Box::new(snapshot()));
+    state.set_pane_surface(surface());
+    let mut prepare = ClientShellInput::default();
+    state.record_binding(
+        crate::input::KeybindMatch::Action(crate::input::KeybindAction::NewWorktree),
+        &mut prepare,
+    );
+    let [ClientShellAction::Endpoint { request, .. }] = &prepare.actions[..] else {
+        panic!("new worktree should prepare through worktree.list");
+    };
+    state.handle_endpoint_result("boot-1", &request.id, Ok(worktree_list_result(None)));
+    state.set_endpoint_methods(Some(vec!["worktree.list".into()]));
+    state.handle_input_bytes(b"feature/unavailable");
+
+    let submit = state.handle_input_bytes(b"\r");
+
+    assert!(submit.actions.is_empty());
+    assert!(matches!(
+        &state.overlay,
+        Some(ClientShellOverlay::WorktreeCreate(create)) if !create.creating
+    ));
+    assert!(state
+        .visible_endpoint_notice
+        .as_ref()
+        .is_some_and(|notice| notice.key.code == "worktree.create"));
+}
+
+#[test]
 fn worktree_open_filters_and_clicks_a_stable_public_entry() {
     let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
     state.set_snapshot(Box::new(snapshot()));
diff --git a/src/client/shell/tests/keybindings_settings.rs b/src/client/shell/tests/keybindings_settings.rs
index 5d2d8243..5b0fced7 100644
--- a/src/client/shell/tests/keybindings_settings.rs
+++ b/src/client/shell/tests/keybindings_settings.rs
@@ -437,6 +437,60 @@ fn plugin_command_carries_client_owned_selection_coordinates() {
 }
 
 #[test]
+fn unavailable_endpoint_method_is_disabled_without_disconnect() {
+    let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
+    state.set_snapshot(Box::new(snapshot()));
+    state.set_pane_surface(surface());
+    state.set_endpoint_methods(Some(vec!["pane.focus".into()]));
+    let mut outcome = ClientShellInput::default();
+
+    state.push_endpoint_method(
+        crate::api::schema::Method::WorkspaceFocus(crate::api::schema::WorkspaceTarget {
+            workspace_id: "missing".into(),
+        }),
+        &mut outcome,
+    );
+
+    assert!(outcome.actions.is_empty());
+    assert!(outcome.repaint);
+    let notice = state
+        .visible_endpoint_notice
+        .as_ref()
+        .expect("unsupported action notice");
+    assert_eq!(notice.key.kind, ClientEndpointNoticeKind::Unsupported);
+    assert_eq!(notice.key.code, "workspace.focus");
+    assert!(notice.body.contains("This server"));
+    assert!(state.endpoint_error.is_none());
+
+    let mut repeated = ClientShellInput::default();
+    state.push_endpoint_method(
+        crate::api::schema::Method::WorkspaceFocus(crate::api::schema::WorkspaceTarget {
+            workspace_id: "missing".into(),
+        }),
+        &mut repeated,
+    );
+    assert!(repeated.actions.is_empty());
+    assert!(!repeated.repaint);
+
+    state.compose(106, 20).expect("endpoint notice frame");
+    assert!(!state.hits.notification_toast.is_empty());
+
+    state.handle_input_bytes(b"x");
+    assert!(state.visible_endpoint_notice.is_some());
+
+    let hit = state.hits.notification_toast;
+    let dismissed =
+        state.handle_raw_events(vec![RawInputEvent::Mouse(crossterm::event::MouseEvent {
+            kind: MouseEventKind::Down(MouseButton::Left),
+            column: hit.x,
+            row: hit.y,
+            modifiers: KeyModifiers::empty(),
+        })]);
+    assert!(dismissed.repaint);
+    assert!(state.visible_endpoint_notice.is_none());
+}
+
+#[test]
 fn generic_endpoint_failures_and_control_errors_are_visible() {
     let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
     state.set_snapshot(Box::new(snapshot()));
@@ -461,17 +515,98 @@ fn generic_endpoint_failures_and_control_errors_are_visible() {
     );
     assert!(repaint);
     assert!(actions.is_empty());
+    let notice = state
+        .visible_endpoint_notice
+        .as_ref()
+        .expect("rejected action notice");
+    assert_eq!(notice.key.kind, ClientEndpointNoticeKind::Rejected);
+    assert_eq!(notice.key.code, "workspace.focus:not_found");
+    assert_eq!(notice.body, "workspace no longer exists");
+    assert!(state.endpoint_error.is_none());
+
+    assert!(state.receive_endpoint_error("Paste rejected: too large".into()));
+    let notice = state
+        .visible_endpoint_notice
+        .as_ref()
+        .expect("paste rejection notice");
+    assert_eq!(notice.key.kind, ClientEndpointNoticeKind::Rejected);
+    assert_eq!(notice.key.code, "paste_rejected");
+    assert_eq!(notice.body, "Paste rejected: too large");
+    assert!(state.receive_endpoint_error("Paste rejected again".into()));
     assert_eq!(
-        state.endpoint_error.as_deref(),
-        Some("workspace no longer exists")
+        state
+            .visible_endpoint_notice
+            .as_ref()
+            .map(|notice| notice.body.as_str()),
+        Some("Paste rejected again")
     );
+}
 
-    assert!(state.receive_endpoint_error("Paste rejected: too large".into()));
+#[test]
+fn endpoint_timeout_is_a_deduplicated_server_notice() {
+    let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
+    state.set_snapshot(Box::new(snapshot()));
+    let mut outcome = ClientShellInput::default();
+    state.push_endpoint_method(
+        crate::api::schema::Method::WorkspaceFocus(crate::api::schema::WorkspaceTarget {
+            workspace_id: "work".into(),
+        }),
+        &mut outcome,
+    );
+    let request_id = match &outcome.actions[..] {
+        [ClientShellAction::Endpoint { request, .. }] => request.id.clone(),
+        other => panic!("expected endpoint request, got {other:?}"),
+    };
+
+    let (repaint, actions) = state.handle_endpoint_result(
+        "boot-1",
+        &request_id,
+        Err(ClientShellEndpointError {
+            code: Some("endpoint_timeout".into()),
+            message: "transport timeout".into(),
+        }),
+    );
+
+    assert!(repaint);
+    assert!(actions.is_empty());
+    let notice = state
+        .visible_endpoint_notice
+        .as_ref()
+        .expect("timeout notice");
+    assert_eq!(notice.key.kind, ClientEndpointNoticeKind::Timeout);
+    assert_eq!(notice.key.code, "workspace.focus");
+    assert!(notice.body.contains("workspace.focus"));
+    assert!(state.endpoint_error.is_none());
+}
+
+#[test]
+fn endpoint_notice_dedupe_resets_for_a_new_server_boot() {
+    let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
+    state.set_snapshot(Box::new(snapshot()));
+    state.set_endpoint_methods(Some(Vec::new()));
+    let method = || {
+        crate::api::schema::Method::WorkspaceFocus(crate::api::schema::WorkspaceTarget {
+            workspace_id: "work".into(),
+        })
+    };
+    let mut first = ClientShellInput::default();
+    state.push_endpoint_method(method(), &mut first);
+    assert!(first.repaint);
+
+    let mut next_snapshot = snapshot();
+    next_snapshot.boot_id = "boot-2".into();
+    state.set_snapshot(Box::new(next_snapshot));
+    let mut second = ClientShellInput::default();
+    state.push_endpoint_method(method(), &mut second);
+
+    assert!(second.repaint);
     assert_eq!(
-        state.endpoint_error.as_deref(),
-        Some("Paste rejected: too large")
+        state
+            .visible_endpoint_notice
+            .as_ref()
+            .map(|notice| notice.key.boot_id.as_str()),
+        Some("boot-2")
     );
-    assert!(!state.receive_endpoint_error("Paste rejected: too large".into()));
 }
 
 #[test]
diff --git a/src/client/shell/tests/startup_overlays.rs b/src/client/shell/tests/startup_overlays.rs
index 40ba0bd6..2e2901e8 100644
--- a/src/client/shell/tests/startup_overlays.rs
+++ b/src/client/shell/tests/startup_overlays.rs
@@ -455,6 +455,41 @@ fn onboarding_completion_persists_and_opens_endpoint_integrations() {
 }
 
 #[test]
+fn unavailable_integration_list_does_not_wedge_settings() {
+    let mut config =
+        ClientShellConfig::from_config(&Config::default()).with_startup_onboarding(true);
+    config.local_config_path = std::env::temp_dir().join(format!(
+        "herdr-client-onboarding-unavailable-{}-{}.toml",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .expect("clock")
+            .as_nanos()
+    ));
+    let mut state = ClientShellState::new(config);
+    state.set_snapshot(Box::new(snapshot()));
+    state.set_pane_surface(surface());
+    state.set_endpoint_methods(Some(Vec::new()));
+
+    let outcome = state.handle_input_bytes(b"\r");
+
+    assert!(outcome.actions.is_empty());
+    assert!(matches!(
+        state.overlay,
+        Some(ClientShellOverlay::Settings(ClientSettingsOverlay {
+            section: ClientSettingsSection::Integrations,
+            loading_integrations: false,
+            ..
+        }))
+    ));
+    assert!(state
+        .visible_endpoint_notice
+        .as_ref()
+        .is_some_and(|notice| notice.key.code == "integration.list"));
+    let _ = std::fs::remove_file(&state.config.local_config_path);
+}
+
+#[test]
 fn startup_config_diagnostics_are_client_rendered_and_persist_until_replaced() {
     let config = ClientShellConfig::from_config(&Config::default())
         .with_startup_config_diagnostic(Some("local config warning".into()));
@@ -558,6 +593,7 @@ fn endpoint_reload_result_does_not_override_snapshot_diagnostic_authority() {
         "reload-1".into(),
         PendingEndpointRequest {
             boot_id: "boot-1".into(),
+            method_name: "server.reload_config".into(),
             confirmation_workspace_id: None,
             kind: PendingEndpointKind::ReloadConfig,
         },
diff --git a/src/client/shell/worktrees.rs b/src/client/shell/worktrees.rs
index 792cd248..bcadf1c8 100644
--- a/src/client/shell/worktrees.rs
+++ b/src/client/shell/worktrees.rs
@@ -276,7 +276,7 @@ impl ClientShellState {
         create.creating = true;
         create.error = None;
         let workspace_id = create.source_workspace_id.clone();
-        self.push_endpoint_method_with_kind(
+        if !self.push_endpoint_method_with_kind(
             crate::api::schema::Method::WorktreeCreate(crate::api::schema::WorktreeCreateParams {
                 workspace_id: Some(workspace_id),
                 cwd: None,
@@ -289,7 +289,11 @@ impl ClientShellState {
             }),
             PendingEndpointKind::WorktreeCreate,
             outcome,
-        );
+        ) {
+            if let Some(ClientShellOverlay::WorktreeCreate(create)) = self.overlay.as_mut() {
+                create.creating = false;
+            }
+        }
         outcome.repaint = true;
     }
 
@@ -328,7 +332,7 @@ impl ClientShellState {
         open.selected = index;
         open.opening = true;
         open.error = None;
-        self.push_endpoint_method_with_kind(
+        if !self.push_endpoint_method_with_kind(
             crate::api::schema::Method::WorktreeOpen(crate::api::schema::WorktreeOpenParams {
                 workspace_id: Some(workspace_id),
                 cwd: None,
@@ -340,7 +344,11 @@ impl ClientShellState {
             }),
             PendingEndpointKind::WorktreeOpen,
             outcome,
-        );
+        ) {
+            if let Some(ClientShellOverlay::WorktreeOpen(open)) = self.overlay.as_mut() {
+                open.opening = false;
+            }
+        }
         outcome.repaint = true;
     }
 
@@ -355,7 +363,7 @@ impl ClientShellState {
         let forced = remove.force_confirmation;
         remove.removing = true;
         remove.error = None;
-        self.push_endpoint_method_with_kind(
+        if !self.push_endpoint_method_with_kind(
             crate::api::schema::Method::WorktreeRemove(crate::api::schema::WorktreeRemoveParams {
                 workspace_id,
                 force: forced,
@@ -363,7 +371,11 @@ impl ClientShellState {
             }),
             PendingEndpointKind::WorktreeRemove { forced },
             outcome,
-        );
+        ) {
+            if let Some(ClientShellOverlay::WorktreeRemove(remove)) = self.overlay.as_mut() {
+                remove.removing = false;
+            }
+        }
         outcome.repaint = true;
     }
 
@@ -511,11 +523,8 @@ impl ClientShellState {
                 PendingEndpointKind::PrepareWorktreeCreate { .. }
                 | PendingEndpointKind::PrepareWorktreeOpen { .. }
                 | PendingEndpointKind::PrepareWorktreeRemove { .. },
-                Err(error),
-            ) => {
-                self.endpoint_error = Some(error.message);
-                true
-            }
+                Err(_),
+            ) => true,
             (_, Ok(_)) => {
                 self.endpoint_error =
                     Some("endpoint returned an unexpected worktree result".to_owned());
@@ -535,11 +544,8 @@ impl ClientShellState {
                 | PendingEndpointKind::PaneLinkActivate { .. }
                 | PendingEndpointKind::CopyMotion { .. }
                 | PendingEndpointKind::CopySearch { .. },
-                Err(error),
-            ) => {
-                self.endpoint_error = Some(error.message);
-                true
-            }
+                Err(_),
+            ) => true,
         }
     }
 }
diff --git a/src/client/terminal_sessions.rs b/src/client/terminal_sessions.rs
index 7d7bcc8f..a46af2b1 100644
--- a/src/client/terminal_sessions.rs
+++ b/src/client/terminal_sessions.rs
@@ -90,10 +90,11 @@ fn connect_terminal_session_stream(
     };
 
     match do_handshake(&mut stream, cols, rows, 0, 0, false, None, false, false) {
-        Ok(RenderEncoding::TerminalAnsi) => {}
-        Ok(encoding) => {
+        Ok(handshake) if handshake.encoding == RenderEncoding::TerminalAnsi => {}
+        Ok(handshake) => {
             eprintln!(
-                "herdr: terminal session observe negotiated unsupported encoding {encoding:?}"
+                "herdr: terminal session observe negotiated unsupported encoding {:?}",
+                handshake.encoding
             );
             std::process::exit(1);
         }
diff --git a/src/protocol/endpoint.rs b/src/protocol/endpoint.rs
new file mode 100644
index 00000000..b2955ec8
--- /dev/null
+++ b/src/protocol/endpoint.rs
@@ -0,0 +1,275 @@
+//! Stable endpoint compatibility contract for client-owned shells.
+//!
+//! The endpoint generation is intentionally independent from the private
+//! binary protocol used by same-install CLI, direct-terminal, and handoff
+//! paths. Generation 1 is the compatibility floor for Local, SSH, and Cloud
+//! shell endpoints and must remain available indefinitely unless retired for a
+//! security reason. New JSON fields must be optional or have serde defaults;
+//! new enum values need an `Unknown` fallback. Unknown named controls are
+//! optional and ignored unless negotiated as part of the core.
+
+use serde::{Deserialize, Serialize};
+
+use super::{ClientShellSnapshot, ClientSurfaceSize, ServerMessage};
+
+pub const ENDPOINT_PROTOCOL_GENERATION: u32 = 1;
+pub const ENDPOINT_HELLO_KIND: &str = "endpoint.hello.v1";
+pub const ENDPOINT_WELCOME_KIND: &str = "endpoint.welcome.v1";
+pub const SNAPSHOT_CODEC_V1: &str = "shell.snapshot.v1";
+pub const ENDPOINT_SNAPSHOT_KIND: &str = SNAPSHOT_CODEC_V1;
+pub const SURFACE_CODEC_V1: &str = "shell.surface.v1";
+pub const INPUT_CODEC_V1: &str = "shell.input.semantic.v1";
+pub const BLOB_CODEC_V1: &str = "shell.blob.v1";
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EndpointClientHello {
+    pub generation: u32,
+    pub cell_width_px: u32,
+    pub cell_height_px: u32,
+    pub surface_size: ClientSurfaceSize,
+    pub pixel_mouse: bool,
+    pub direct_graphics: bool,
+    pub endpoint_keybindings: bool,
+    pub mouse_capture: bool,
+    #[serde(default)]
+    pub snapshot_codecs: Vec<String>,
+    #[serde(default)]
+    pub surface_codecs: Vec<String>,
+    #[serde(default)]
+    pub input_codecs: Vec<String>,
+    #[serde(default)]
+    pub blob_codecs: Vec<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EndpointHandshakeError {
+    pub code: String,
+    pub message: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EndpointServerWelcome {
+    pub generation: u32,
+    pub server_version: String,
+    pub snapshot_codec: String,
+    pub surface_codec: String,
+    pub input_codec: String,
+    pub blob_codec: String,
+    #[serde(default)]
+    pub methods: Vec<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<EndpointHandshakeError>,
+}
+
+pub fn snapshot_message(snapshot: &ClientShellSnapshot) -> serde_json::Result<ServerMessage> {
+    Ok(ServerMessage::EndpointControl {
+        kind: ENDPOINT_SNAPSHOT_KIND.into(),
+        data: serde_json::to_string(snapshot)?,
+    })
+}
+
+impl EndpointClientHello {
+    pub fn supports_required_codecs(&self) -> bool {
+        self.snapshot_codecs
+            .iter()
+            .any(|codec| codec == SNAPSHOT_CODEC_V1)
+            && self
+                .surface_codecs
+                .iter()
+                .any(|codec| codec == SURFACE_CODEC_V1)
+            && self
+                .input_codecs
+                .iter()
+                .any(|codec| codec == INPUT_CODEC_V1)
+            && self.blob_codecs.iter().any(|codec| codec == BLOB_CODEC_V1)
+    }
+}
+
+impl EndpointServerWelcome {
+    pub fn compatible(methods: Vec<String>) -> Self {
+        Self {
+            generation: ENDPOINT_PROTOCOL_GENERATION,
+            server_version: crate::build_info::version(),
+            snapshot_codec: SNAPSHOT_CODEC_V1.into(),
+            surface_codec: SURFACE_CODEC_V1.into(),
+            input_codec: INPUT_CODEC_V1.into(),
+            blob_codec: BLOB_CODEC_V1.into(),
+            methods,
+            error: None,
+        }
+    }
+
+    pub fn incompatible(code: &str, message: impl Into<String>) -> Self {
+        Self {
+            generation: ENDPOINT_PROTOCOL_GENERATION,
+            server_version: crate::build_info::version(),
+            snapshot_codec: SNAPSHOT_CODEC_V1.into(),
+            surface_codec: SURFACE_CODEC_V1.into(),
+            input_codec: INPUT_CODEC_V1.into(),
+            blob_codec: BLOB_CODEC_V1.into(),
+            methods: Vec::new(),
+            error: Some(EndpointHandshakeError {
+                code: code.into(),
+                message: message.into(),
+            }),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn hello() -> EndpointClientHello {
+        EndpointClientHello {
+            generation: ENDPOINT_PROTOCOL_GENERATION,
+            cell_width_px: 8,
+            cell_height_px: 16,
+            surface_size: ClientSurfaceSize { cols: 80, rows: 24 },
+            pixel_mouse: true,
+            direct_graphics: false,
+            endpoint_keybindings: false,
+            mouse_capture: true,
+            snapshot_codecs: vec![SNAPSHOT_CODEC_V1.into()],
+            surface_codecs: vec![SURFACE_CODEC_V1.into()],
+            input_codecs: vec![INPUT_CODEC_V1.into()],
+            blob_codecs: vec![BLOB_CODEC_V1.into()],
+        }
+    }
+
+    fn snapshot() -> ClientShellSnapshot {
+        ClientShellSnapshot {
+            boot_id: "boot".into(),
+            revision: 1,
+            config_diagnostic: None,
+            product_announcement: None,
+            update_available: None,
+            update_install_command: "herdr update".into(),
+            server_keybindings_toml: None,
+            latest_release_notes_available: false,
+            integration_updates_available: false,
+            worktree_directory: String::new(),
+            release_notes: None,
+            focused_workspace_id: None,
+            focused_tab_id: None,
+            focused_pane_id: None,
+            tab_bar_right: Vec::new(),
+            tab_bar_right_separator: String::new(),
+            agent_view_label: None,
+            agent_order: Vec::new(),
+            workspaces: Vec::new(),
+            tabs: Vec::new(),
+            panes: Vec::new(),
+            agents: Vec::new(),
+            commands: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn hello_ignores_future_named_fields() {
+        let mut value = serde_json::to_value(hello()).unwrap();
+        value["future_feature"] = serde_json::json!({"enabled": true});
+        let decoded: EndpointClientHello = serde_json::from_value(value).unwrap();
+        assert_eq!(decoded, hello());
+    }
+
+    #[test]
+    fn frozen_generation_one_handshake_decodes() {
+        let hello: EndpointClientHello = serde_json::from_str(include_str!(concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/tests/fixtures/endpoint-hello-v1.json"
+        )))
+        .unwrap();
+        assert!(hello.supports_required_codecs());
+
+        let welcome: EndpointServerWelcome = serde_json::from_str(include_str!(concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/tests/fixtures/endpoint-welcome-v1.json"
+        )))
+        .unwrap();
+        assert_eq!(welcome.generation, ENDPOINT_PROTOCOL_GENERATION);
+        assert_eq!(welcome.snapshot_codec, SNAPSHOT_CODEC_V1);
+        assert_eq!(welcome.surface_codec, SURFACE_CODEC_V1);
+        assert_eq!(welcome.input_codec, INPUT_CODEC_V1);
+        assert_eq!(welcome.blob_codec, BLOB_CODEC_V1);
+    }
+
+    #[test]
+    fn frozen_generation_one_snapshot_decodes() {
+        let snapshot: ClientShellSnapshot = serde_json::from_str(include_str!(concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/tests/fixtures/endpoint-snapshot-v1.json"
+        )))
+        .unwrap();
+        assert_eq!(snapshot.boot_id, "boot-v1");
+        assert_eq!(
+            snapshot.workspaces[0].agent_status,
+            crate::api::schema::AgentStatus::Unknown
+        );
+    }
+
+    #[test]
+    fn snapshot_message_uses_named_json_control() {
+        let snapshot = snapshot();
+        let ServerMessage::EndpointControl { kind, data } = snapshot_message(&snapshot).unwrap()
+        else {
+            panic!("snapshot should use endpoint control");
+        };
+        assert_eq!(kind, ENDPOINT_SNAPSHOT_KIND);
+        let decoded: ClientShellSnapshot = serde_json::from_str(&data).unwrap();
+        assert_eq!(decoded, snapshot);
+    }
+
+    #[test]
+    fn snapshot_json_tolerates_future_fields_and_command_actions() {
+        let mut snapshot = match snapshot_message(&snapshot()).unwrap() {
+            ServerMessage::EndpointControl { data, .. } => {
+                serde_json::from_str::<serde_json::Value>(&data).unwrap()
+            }
+            _ => unreachable!(),
+        };
+        snapshot["future_projection"] = serde_json::json!({"enabled": true});
+        snapshot["commands"] = serde_json::json!([{
+            "command_id": "future",
+            "binding_label": "x",
+            "binding_labels": ["x"],
+            "action": "FutureAction",
+            "description": null
+        }]);
+
+        let decoded: ClientShellSnapshot = serde_json::from_value(snapshot).unwrap();
+        assert_eq!(
+            decoded.commands[0].action,
+            crate::protocol::ClientShellCommandAction::Unknown
+        );
+    }
+
+    #[test]
+    fn required_codecs_are_explicit() {
+        let mut value = hello();
+        assert!(value.supports_required_codecs());
+        value.snapshot_codecs.clear();
+        assert!(!value.supports_required_codecs());
+
+        let mut value = hello();
+        value.surface_codecs.clear();
+        assert!(!value.supports_required_codecs());
+
+        let mut value = hello();
+        value.input_codecs.clear();
+        assert!(!value.supports_required_codecs());
+
+        let mut value = hello();
+        value.blob_codecs.clear();
+        assert!(!value.supports_required_codecs());
+    }
+
+    #[test]
+    fn welcome_ignores_future_named_fields() {
+        let welcome = EndpointServerWelcome::compatible(vec!["pane.close".into()]);
+        let mut value = serde_json::to_value(&welcome).unwrap();
+        value["future_service"] = serde_json::json!("v2");
+        let decoded: EndpointServerWelcome = serde_json::from_value(value).unwrap();
+        assert_eq!(decoded, welcome);
+    }
+}
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index 0cdeecbb..7ee6e8de 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -1,5 +1,6 @@
 //! Shared wire protocol and presentation encoding code.
 
+pub mod endpoint;
 pub(crate) mod render_ansi;
 mod wire;
 
diff --git a/src/protocol/wire.rs b/src/protocol/wire.rs
index be084aa6..cbca829a 100644
--- a/src/protocol/wire.rs
+++ b/src/protocol/wire.rs
@@ -1,7 +1,11 @@
 //! Wire protocol for herdr server/client communication.
 //!
 //! Defines the message types, framing, version negotiation, and safety
-//! constraints for the binary protocol over Unix domain sockets.
+//! constraints for the binary protocol over local sockets.
+//!
+//! `PROTOCOL_VERSION` still guards same-install direct-terminal and internal
+//! operations. Client-owned shells negotiate the independent stable endpoint
+//! contract in [`super::endpoint`].
 
 use std::collections::HashMap;
 use std::io::{self, Read, Write};
@@ -449,6 +453,9 @@ impl ClientInputEvent {
 }
 
 /// Messages sent from the client to the server over the client protocol socket.
+///
+/// Variant order is frozen for endpoint generation 1. Add compatible endpoint
+/// behavior through `EndpointControl` or advertised API methods, not new enum variants.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub enum ClientMessage {
     /// Direct terminal handshake: announces protocol version and terminal dimensions.
@@ -597,6 +604,12 @@ pub enum ClientMessage {
 
     /// Update this client's shell mouse-capture preference after config reload.
     ClientShellMouseCapture { enabled: bool },
+
+    /// Extensible named control message for the stable client-owned endpoint protocol.
+    ///
+    /// This variant is append-only. Its bincode tag and two-string payload are part
+    /// of endpoint generation 1 and must not change.
+    EndpointControl { kind: String, data: String },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -860,7 +873,26 @@ impl FrameData {
     }
 }
 
-/// Initial resource projection used by the experimental client-owned shell.
+fn deserialize_client_shell_agent_status<'de, D>(
+    deserializer: D,
+) -> Result<crate::api::schema::AgentStatus, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    if !deserializer.is_human_readable() {
+        return crate::api::schema::AgentStatus::deserialize(deserializer);
+    }
+    let value = String::deserialize(deserializer)?;
+    Ok(match value.as_str() {
+        "idle" => crate::api::schema::AgentStatus::Idle,
+        "working" => crate::api::schema::AgentStatus::Working,
+        "blocked" => crate::api::schema::AgentStatus::Blocked,
+        "done" => crate::api::schema::AgentStatus::Done,
+        _ => crate::api::schema::AgentStatus::Unknown,
+    })
+}
+
+/// Initial resource projection used by the stable client-owned shell.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ClientShellSnapshot {
     /// Changes whenever the endpoint process restarts.
@@ -921,6 +953,9 @@ pub enum ClientShellCommandAction {
     Pane,
     Popup,
     PluginAction,
+    /// A future endpoint action kind that this client cannot execute.
+    #[serde(other)]
+    Unknown,
 }
 
 impl From<crate::config::CustomCommandAction> for ClientShellCommandAction {
@@ -934,13 +969,16 @@ impl From<crate::config::CustomCommandAction> for ClientShellCommandAction {
     }
 }
 
-impl From<ClientShellCommandAction> for crate::config::CustomCommandAction {
-    fn from(action: ClientShellCommandAction) -> Self {
+impl TryFrom<ClientShellCommandAction> for crate::config::CustomCommandAction {
+    type Error = ();
+
+    fn try_from(action: ClientShellCommandAction) -> Result<Self, Self::Error> {
         match action {
-            ClientShellCommandAction::Shell => Self::Shell,
-            ClientShellCommandAction::Pane => Self::Pane,
-            ClientShellCommandAction::Popup => Self::Popup,
-            ClientShellCommandAction::PluginAction => Self::PluginAction,
+            ClientShellCommandAction::Shell => Ok(Self::Shell),
+            ClientShellCommandAction::Pane => Ok(Self::Pane),
+            ClientShellCommandAction::Popup => Ok(Self::Popup),
+            ClientShellCommandAction::PluginAction => Ok(Self::PluginAction),
+            ClientShellCommandAction::Unknown => Err(()),
         }
     }
 }
@@ -973,6 +1011,7 @@ pub struct ClientShellWorkspace {
     pub tokens: Vec<(String, String)>,
     pub worktree: Option<ClientShellWorktree>,
     pub focused: bool,
+    #[serde(deserialize_with = "deserialize_client_shell_agent_status")]
     pub agent_status: crate::api::schema::AgentStatus,
 }
 
@@ -992,6 +1031,7 @@ pub struct ClientShellTab {
     pub custom_label: bool,
     pub zoomed: bool,
     pub focused: bool,
+    #[serde(deserialize_with = "deserialize_client_shell_agent_status")]
     pub agent_status: crate::api::schema::AgentStatus,
 }
 
@@ -1018,6 +1058,7 @@ pub struct ClientShellAgent {
     pub title: Option<String>,
     pub terminal_title: Option<String>,
     pub terminal_title_stripped: Option<String>,
+    #[serde(deserialize_with = "deserialize_client_shell_agent_status")]
     pub agent_status: crate::api::schema::AgentStatus,
     pub state_change_seq: u64,
     pub state_labels: Vec<(String, String)>,
@@ -1271,6 +1312,9 @@ pub struct SemanticNotification {
 }
 
 /// Messages sent from the server to the client over the client protocol socket.
+///
+/// Variant order is frozen for endpoint generation 1. Add compatible endpoint
+/// behavior through `EndpointControl`, and ignore unrecognized named controls.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub enum ServerMessage {
     /// Handshake response: server acknowledges (or rejects) the client.
@@ -1386,6 +1430,12 @@ pub enum ServerMessage {
 
     /// Incremental terminal-cell update for a previously committed pane surface.
     PaneSurfacePatch(PaneSurfacePatch),
+
+    /// Extensible named control message for the stable client-owned endpoint protocol.
+    ///
+    /// This variant is append-only. Its bincode tag and two-string payload are part
+    /// of endpoint generation 1 and must not change.
+    EndpointControl { kind: String, data: String },
 }
 
 // ---------------------------------------------------------------------------
@@ -1659,6 +1709,17 @@ pub fn check_client_version(client_version: u32) -> VersionCheck {
 mod tests {
     use super::*;
     use ratatui::style::{Color, Modifier};
+    use sha2::{Digest, Sha256};
+
+    fn encoded_sha256(value: &impl Serialize) -> String {
+        let encoded = bincode::serde::encode_to_vec(value, bincode::config::standard()).unwrap();
+        format!("{:x}", Sha256::digest(encoded))
+    }
+
+    // These digests freeze representative generation-1 bincode payloads. A mismatch
+    // requires a new named codec; do not update a v1 digest to bless a wire change.
+    // They do not detect appended enum variants, so every type reachable from a v1
+    // payload is also append-closed.
 
     // ---- Round-trip: ClientMessage ----
 
@@ -1697,6 +1758,29 @@ mod tests {
     }
 
     #[test]
+    fn endpoint_control_roundtrip() {
+        let msg = ClientMessage::EndpointControl {
+            kind: "endpoint.hello.v1".into(),
+            data: r#"{"generation":1}"#.into(),
+        };
+        let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap();
+        let (decoded, _): (ClientMessage, _) =
+            bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
+        assert_eq!(msg, decoded);
+        assert_eq!(
+            bincode::serde::encode_to_vec(
+                ClientMessage::EndpointControl {
+                    kind: String::new(),
+                    data: String::new(),
+                },
+                bincode::config::standard(),
+            )
+            .unwrap(),
+            [20, 0, 0]
+        );
+    }
+
+    #[test]
     fn client_shell_resize_roundtrip() {
         let msg = ClientMessage::ClientShellResize {
             cell_width_px: 8,
@@ -1708,6 +1792,10 @@ mod tests {
         let (decoded, _): (ClientMessage, _) =
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
         assert_eq!(msg, decoded);
+        assert_eq!(
+            encoded_sha256(&msg),
+            "676d6376202750e72c45ff511e256b6154d3d20c0ee088d3792fa3a69d9704b9"
+        );
     }
 
     #[test]
@@ -1806,6 +1894,51 @@ mod tests {
             8
         );
         assert_eq!(
+            tag(&ClientMessage::GraphicsTransmissionResult {
+                transfer_id: 1,
+                image_id: 2,
+                success: true,
+            }),
+            9
+        );
+        assert_eq!(
+            tag(&ClientMessage::GraphicsTransmissionStarted {
+                transfer_id: 1,
+                image_id: 2,
+            }),
+            10
+        );
+        assert_eq!(
+            tag(&ClientMessage::ClientShellResize {
+                cell_width_px: 8,
+                cell_height_px: 16,
+                surface_size: ClientSurfaceSize { cols: 80, rows: 29 },
+                pixel_mouse: false,
+            }),
+            12
+        );
+        assert_eq!(
+            tag(&ClientMessage::ClientShellPaneInput {
+                pane_id: "pane".into(),
+                events: Vec::new(),
+            }),
+            13
+        );
+        assert_eq!(
+            tag(&ClientMessage::ClientShellPopupInput {
+                terminal_id: "popup".into(),
+                events: Vec::new(),
+            }),
+            14
+        );
+        assert_eq!(
+            tag(&ClientMessage::ClientShellEndpointRequest {
+                boot_id: "boot".into(),
+                request: "{}".into(),
+            }),
+            15
+        );
+        assert_eq!(
             tag(&ClientMessage::AttachMouse {
                 kind: ClientMouseKind::Down(ClientMouseButton::Left),
                 position: ClientMousePosition::Cell { column: 10, row: 5 },
@@ -1826,6 +1959,13 @@ mod tests {
             tag(&ClientMessage::ClientShellMouseCapture { enabled: true }),
             19
         );
+        assert_eq!(
+            tag(&ClientMessage::EndpointControl {
+                kind: String::new(),
+                data: String::new(),
+            }),
+            20
+        );
     }
 
     #[test]
@@ -1871,6 +2011,10 @@ mod tests {
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard())
                 .expect("decode targeted semantic input");
         assert_eq!(decoded, message);
+        assert_eq!(
+            encoded_sha256(&message),
+            "f558384bb53dfd2baf1fa72e1709d88be6891da79e51f88513905afc085065e6"
+        );
         let ClientMessage::ClientShellPaneInput { events, .. } = decoded else {
             panic!("expected targeted semantic input");
         };
@@ -1977,6 +2121,10 @@ mod tests {
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard())
                 .expect("decode endpoint request");
         assert_eq!(decoded, request);
+        assert_eq!(
+            encoded_sha256(&request),
+            "de5693585a01f6b0d5ee07c51b6ddf79ee9f67dbf183255822d31f35210f5ffb"
+        );
 
         let response = ServerMessage::ClientShellEndpointResponseChunk {
             boot_id: "boot-a".into(),
@@ -1990,7 +2138,12 @@ mod tests {
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard())
                 .expect("decode endpoint response");
         assert_eq!(decoded, response);
+        assert_eq!(
+            encoded_sha256(&response),
+            "bc14dbb5263d3097fe6d3e70a4b6d71aa9c2fa4ae3206d692a7182512bffdd1d"
+        );
     }
+
     #[test]
     fn client_clipboard_image_roundtrip() {
         let msg = ClientMessage::ClipboardImage {
@@ -2002,6 +2155,10 @@ mod tests {
         let (decoded, _): (ClientMessage, _) =
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
         assert_eq!(msg, decoded);
+        assert_eq!(
+            encoded_sha256(&msg),
+            "1c02110be0671faf318b3f4d8f507748d5a0a98cd474832669c5290d97ef19ab"
+        );
     }
 
     #[test]
@@ -2203,6 +2360,10 @@ mod tests {
         let (decoded, _): (ServerMessage, _) =
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
         assert_eq!(msg, decoded);
+        assert_eq!(
+            encoded_sha256(&msg),
+            "7c016f7b21ddb5ac79212cf65a968b93eb292b5305b941263e89ffaa40158ee3"
+        );
         match decoded {
             ServerMessage::PaneSurface(surface) => {
                 assert_eq!(surface.frame.cells[2].hyperlink, Some(0));
@@ -2246,6 +2407,162 @@ mod tests {
         let (decoded, _): (ServerMessage, _) =
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
         assert_eq!(decoded, msg);
+        assert_eq!(
+            encoded_sha256(&msg),
+            "0814b99a1dc6eaf7918424aa416c066509cbfb73b72344a809c27cde78cb6dbd"
+        );
+    }
+
+    #[test]
+    fn client_shell_graphics_payload_codec_is_frozen() {
+        let key = SurfaceGraphicsAssetKey {
+            source: SurfaceGraphicsSource::Terminal {
+                target: SurfaceGraphicsTarget::Pane {
+                    pane_id: "w1:p1".into(),
+                },
+                image_id: 7,
+            },
+            image_width: 2,
+            image_height: 1,
+            format: SurfaceGraphicsFormat::Rgba,
+            data_len: 8,
+            data_fingerprint: 42,
+        };
+        let message = ServerMessage::PaneSurface(PaneSurfaceFrame {
+            boot_id: "boot-1".into(),
+            projection_revision: 2,
+            surface_revision: 3,
+            frame: FrameData {
+                cells: Vec::new(),
+                width: 0,
+                height: 0,
+                cursor: None,
+                hyperlinks: Vec::new(),
+                graphics: Vec::new(),
+            },
+            panes: Vec::new(),
+            splits: Vec::new(),
+            popup: None,
+            graphics: SurfaceGraphicsScene {
+                assets: vec![SurfaceGraphicsAsset {
+                    key: key.clone(),
+                    data: vec![255, 0, 0, 255, 0, 255, 0, 255],
+                }],
+                placements: vec![SurfaceGraphicsPlacement {
+                    asset: key,
+                    logical_placement_id: 9,
+                    x: 1,
+                    y: 2,
+                    cols: 2,
+                    rows: 1,
+                    source_x: 0,
+                    source_y: 0,
+                    source_width: 2,
+                    source_height: 1,
+                    x_offset: 0,
+                    y_offset: 0,
+                    z: -1,
+                    scrollback_offset: 0,
+                }],
+                retained_assets: Vec::new(),
+            },
+        });
+
+        assert_eq!(
+            encoded_sha256(&message),
+            "49c4efec0f1456c8ca4112ddf6ead1ab75d0224007576c2ccc18c3fca55a69f0"
+        );
+    }
+
+    #[test]
+    fn server_endpoint_control_tag_is_frozen() {
+        let message = ServerMessage::EndpointControl {
+            kind: "endpoint.welcome.v1".into(),
+            data: r#"{"generation":1}"#.into(),
+        };
+        let encoded = bincode::serde::encode_to_vec(&message, bincode::config::standard()).unwrap();
+        assert_eq!(encoded.first(), Some(&20));
+        assert_eq!(
+            bincode::serde::encode_to_vec(
+                ServerMessage::EndpointControl {
+                    kind: String::new(),
+                    data: String::new(),
+                },
+                bincode::config::standard(),
+            )
+            .unwrap(),
+            [20, 0, 0]
+        );
+        let (decoded, _): (ServerMessage, _) =
+            bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
+        assert_eq!(decoded, message);
+    }
+
+    #[test]
+    fn client_shell_server_message_tags_are_frozen() {
+        fn tag(message: &ServerMessage) -> u8 {
+            *bincode::serde::encode_to_vec(message, bincode::config::standard())
+                .unwrap()
+                .first()
+                .expect("encoded server message should include enum tag")
+        }
+
+        let empty_frame = || PaneSurfaceFrame {
+            boot_id: "boot".into(),
+            projection_revision: 1,
+            surface_revision: 1,
+            frame: FrameData {
+                cells: Vec::new(),
+                width: 0,
+                height: 0,
+                cursor: None,
+                hyperlinks: Vec::new(),
+                graphics: Vec::new(),
+            },
+            panes: Vec::new(),
+            splits: Vec::new(),
+            popup: None,
+            graphics: SurfaceGraphicsScene::default(),
+        };
+        assert_eq!(tag(&ServerMessage::PaneSurface(empty_frame())), 13);
+        assert_eq!(
+            tag(&ServerMessage::ClientShellError {
+                message: String::new(),
+            }),
+            15
+        );
+        assert_eq!(
+            tag(&ServerMessage::ClientShellKeyboardReportAll { enabled: false }),
+            17
+        );
+        assert_eq!(
+            tag(&ServerMessage::ClientShellEndpointResponseChunk {
+                boot_id: String::new(),
+                request_id: String::new(),
+                final_chunk: true,
+                data: Vec::new(),
+            }),
+            18
+        );
+        assert_eq!(
+            tag(&ServerMessage::PaneSurfacePatch(PaneSurfacePatch {
+                boot_id: String::new(),
+                projection_revision: 0,
+                base_surface_revision: 0,
+                surface_revision: 0,
+                rows: Vec::new(),
+                panes: Vec::new(),
+                cursor: None,
+            })),
+            19
+        );
+        assert_eq!(
+            tag(&ServerMessage::EndpointControl {
+                kind: String::new(),
+                data: String::new(),
+            }),
+            20
+        );
     }
 
     #[test]
@@ -2411,6 +2728,10 @@ mod tests {
         let (decoded, _): (ServerMessage, _) =
             bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
         assert_eq!(msg, decoded);
+        assert_eq!(
+            encoded_sha256(&msg),
+            "28a420f92e0e05e6760a8c140baf307c360c6d1b1aa68027481b324f87e22c44"
+        );
     }
 
     #[test]
diff --git a/src/remote/attach.rs b/src/remote/attach.rs
index 525fc7ce..641aad00 100644
--- a/src/remote/attach.rs
+++ b/src/remote/attach.rs
@@ -182,7 +182,6 @@ pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
     ensure_remote_server_ready(
         &remote_ssh,
         &prepared_remote.remote_herdr,
-        prepared_remote.installed_or_replaced,
         prepared_remote.stop_after_install_approved,
         remote.live_handoff,
     )?;
@@ -395,7 +394,6 @@ struct RemoteReleaseAsset {
 
 struct PreparedRemoteHerdr {
     remote_herdr: RemoteHerdr,
-    installed_or_replaced: bool,
     stop_after_install_approved: bool,
 }
 
@@ -653,18 +651,16 @@ fn prepare_remote_herdr(
 
     if override_binary.is_none() {
         for candidate in &remote_binary_candidates {
-            if remote_binary_matches(ssh, candidate).unwrap_or(false) {
+            if remote_binary_supports_endpoint(ssh, candidate).unwrap_or(false) {
                 return Ok(PreparedRemoteHerdr {
                     remote_herdr: candidate.clone(),
-                    installed_or_replaced: false,
                     stop_after_install_approved: false,
                 });
             }
         }
-        if remote_binary_matches(ssh, &remote_herdr)? {
+        if remote_binary_supports_endpoint(ssh, &remote_herdr)? {
             return Ok(PreparedRemoteHerdr {
                 remote_herdr,
-                installed_or_replaced: false,
                 stop_after_install_approved: false,
             });
         }
@@ -692,18 +688,17 @@ fn prepare_remote_herdr(
     source.cleanup();
     install_result?;
 
-    if !remote_binary_matches(ssh, &remote_herdr)? {
+    if !remote_binary_supports_endpoint(ssh, &remote_herdr)? {
         return Err(io::Error::other(format!(
-            "installed remote herdr at {}, but it did not report version {}",
+            "installed remote herdr at {}, but it does not support endpoint generation {}",
             remote_herdr.shell_path,
-            current_version()
+            crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION
         )));
     }
     warn_if_remote_bin_not_on_path(ssh)?;
 
     Ok(PreparedRemoteHerdr {
         remote_herdr,
-        installed_or_replaced: true,
         stop_after_install_approved,
     })
 }
@@ -864,24 +859,30 @@ fn is_mise_shim_path(path: &str) -> bool {
     path.ends_with("/mise/shims/herdr")
 }
 
-fn remote_binary_matches(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<bool> {
+fn remote_client_status(
+    ssh: &RemoteSsh,
+    remote_herdr: &RemoteHerdr,
+) -> io::Result<Option<RemoteClientStatusJson>> {
     let command = format!(
-        "test -x {0} && {0} --version && {0} status client --json",
+        "test -x {0} && {0} status client --json",
         remote_herdr.shell_path
     );
     let output = ssh.sh_output(&command)?;
     if !output.status.success() {
-        return Ok(false);
+        return Ok(None);
     }
+    Ok(parse_client_status_json(&String::from_utf8_lossy(
+        &output.stdout,
+    )))
+}
 
-    let stdout = String::from_utf8_lossy(&output.stdout);
-    let mut lines = stdout.lines();
-    let version = lines.next().unwrap_or_default().trim();
-    let status = lines.next().unwrap_or_default();
-    Ok(version == format!("herdr {}", current_version())
-        && parse_client_status_json(status)
-            .map(|status| status.protocol == CURRENT_PROTOCOL)
-            .unwrap_or(false))
+fn remote_binary_supports_endpoint(
+    ssh: &RemoteSsh,
+    remote_herdr: &RemoteHerdr,
+) -> io::Result<bool> {
+    Ok(remote_client_status(ssh, remote_herdr)?
+        .and_then(|status| status.endpoint_protocol_generation)
+        == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION))
 }
 
 fn remote_binary_exists(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<bool> {
@@ -984,7 +985,7 @@ fn local_binary_can_seed_remote(platform: &RemotePlatform) -> bool {
 enum RemoteServerStatus {
     Running {
         version: Option<String>,
-        protocol: Option<u32>,
+        endpoint_protocol_generation: Option<u32>,
         live_handoff: bool,
         detached_server_daemon: bool,
     },
@@ -993,10 +994,8 @@ enum RemoteServerStatus {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum RemoteServerRestartReason {
-    ProtocolMismatch,
+    EndpointProtocolMissing,
     DaemonDetachMissing,
-    BinaryUpdated,
-    VersionMismatch,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -1009,14 +1008,13 @@ enum RemoteInstallRunningServerPlan {
 fn ensure_remote_server_ready(
     ssh: &RemoteSsh,
     remote_herdr: &RemoteHerdr,
-    remote_binary_changed: bool,
     stop_after_install_approved: bool,
     live_handoff_enabled: bool,
 ) -> io::Result<()> {
     let status = remote_server_status(ssh, remote_herdr)?;
     let RemoteServerStatus::Running {
         version,
-        protocol,
+        endpoint_protocol_generation,
         live_handoff,
         detached_server_daemon,
     } = status
@@ -1024,12 +1022,9 @@ fn ensure_remote_server_ready(
         return Ok(());
     };
 
-    let Some(reason) = remote_server_restart_reason(
-        version.as_deref(),
-        protocol,
-        detached_server_daemon,
-        remote_binary_changed,
-    ) else {
+    let Some(reason) =
+        remote_server_restart_reason(endpoint_protocol_generation, detached_server_daemon)
+    else {
         return Ok(());
     };
 
@@ -1048,30 +1043,23 @@ fn ensure_remote_server_ready(
         return Ok(());
     }
 
-    if confirm_remote_server_stop(ssh.target(), version.as_deref(), protocol, reason)? {
+    if confirm_remote_server_stop(ssh.target(), version.as_deref(), reason)? {
         stop_remote_server(ssh, remote_herdr)?;
     }
     Ok(())
 }
 
 fn remote_server_restart_reason(
-    version: Option<&str>,
-    protocol: Option<u32>,
+    endpoint_protocol_generation: Option<u32>,
     detached_server_daemon: bool,
-    remote_binary_changed: bool,
 ) -> Option<RemoteServerRestartReason> {
-    if protocol != Some(CURRENT_PROTOCOL) {
-        return Some(RemoteServerRestartReason::ProtocolMismatch);
+    if endpoint_protocol_generation != Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION)
+    {
+        return Some(RemoteServerRestartReason::EndpointProtocolMissing);
     }
     if !detached_server_daemon {
         return Some(RemoteServerRestartReason::DaemonDetachMissing);
     }
-    if version != Some(current_version().as_str()) {
-        return Some(RemoteServerRestartReason::VersionMismatch);
-    }
-    if remote_binary_changed {
-        return Some(RemoteServerRestartReason::BinaryUpdated);
-    }
     None
 }
 
@@ -1109,7 +1097,7 @@ fn confirm_remote_install_with_running_server(
     };
     let RemoteServerStatus::Running {
         version,
-        protocol,
+        endpoint_protocol_generation,
         live_handoff,
         detached_server_daemon,
     } = &status
@@ -1117,10 +1105,8 @@ fn confirm_remote_install_with_running_server(
         return Ok(false);
     };
     let plan = remote_install_running_server_plan(
-        version.as_deref(),
-        *protocol,
+        *endpoint_protocol_generation,
         *detached_server_daemon,
-        true,
         *live_handoff,
         live_handoff_enabled,
     );
@@ -1187,19 +1173,14 @@ fn confirm_remote_install_with_running_server(
 }
 
 fn remote_install_running_server_plan(
-    version: Option<&str>,
-    protocol: Option<u32>,
+    endpoint_protocol_generation: Option<u32>,
     detached_server_daemon: bool,
-    remote_binary_changed: bool,
     live_handoff: bool,
     live_handoff_enabled: bool,
 ) -> RemoteInstallRunningServerPlan {
-    let Some(reason) = remote_server_restart_reason(
-        version,
-        protocol,
-        detached_server_daemon,
-        remote_binary_changed,
-    ) else {
+    let Some(reason) =
+        remote_server_restart_reason(endpoint_protocol_generation, detached_server_daemon)
+    else {
         return RemoteInstallRunningServerPlan::KeepRunning;
     };
 
@@ -1226,14 +1207,18 @@ fn remote_server_status(
 
 #[derive(Debug, Deserialize)]
 struct RemoteClientStatusJson {
-    protocol: u32,
+    #[serde(default)]
+    version: Option<String>,
+    #[serde(default)]
+    protocol: Option<u32>,
+    #[serde(default)]
+    endpoint_protocol_generation: Option<u32>,
 }
 
 #[derive(Debug, Deserialize)]
 struct RemoteServerStatusJson {
     running: bool,
     version: Option<String>,
-    protocol: Option<u32>,
     capabilities: Option<RemoteServerCapabilitiesJson>,
 }
 
@@ -1242,10 +1227,21 @@ struct RemoteServerCapabilitiesJson {
     live_handoff: bool,
     #[serde(default)]
     detached_server_daemon: bool,
+    #[serde(default)]
+    endpoint_protocol_generation: Option<u32>,
 }
 
 fn parse_client_status_json(status: &str) -> Option<RemoteClientStatusJson> {
-    serde_json::from_str(status).ok()
+    status
+        .lines()
+        .rev()
+        .filter(|line| !line.trim().is_empty())
+        .filter_map(|line| serde_json::from_str::<RemoteClientStatusJson>(line).ok())
+        .find(|status| {
+            status.version.is_some()
+                || status.protocol.is_some()
+                || status.endpoint_protocol_generation.is_some()
+        })
 }
 
 fn parse_remote_server_status_json(status: &str) -> io::Result<RemoteServerStatus> {
@@ -1262,7 +1258,9 @@ fn parse_remote_server_status_json(status: &str) -> io::Result<RemoteServerStatu
 
     Ok(RemoteServerStatus::Running {
         version: parsed.version,
-        protocol: parsed.protocol,
+        endpoint_protocol_generation: capabilities
+            .as_ref()
+            .and_then(|capabilities| capabilities.endpoint_protocol_generation),
         live_handoff: capabilities
             .as_ref()
             .is_some_and(|capabilities| capabilities.live_handoff),
@@ -1275,13 +1273,12 @@ fn parse_remote_server_status_json(status: &str) -> io::Result<RemoteServerStatu
 fn confirm_remote_server_stop(
     target: &str,
     version: Option<&str>,
-    _protocol: Option<u32>,
     reason: RemoteServerRestartReason,
 ) -> io::Result<bool> {
     if !io::stdin().is_terminal() {
-        if reason == RemoteServerRestartReason::ProtocolMismatch {
+        if reason == RemoteServerRestartReason::EndpointProtocolMissing {
             return Err(io::Error::other(format!(
-                "remote herdr server on {target} must stop before this client can attach; run from an interactive terminal to approve stopping it"
+                "remote herdr server on {target} needs one final update before this client can attach; run from an interactive terminal to approve updating it"
             )));
         }
 
@@ -1299,28 +1296,20 @@ fn confirm_remote_server_stop(
     eprintln!();
 
     match reason {
-        RemoteServerRestartReason::ProtocolMismatch => {
-            eprintln!("the remote server must stop before this client can attach.");
-        }
-        RemoteServerRestartReason::DaemonDetachMissing => {
+        RemoteServerRestartReason::EndpointProtocolMissing => {
             eprintln!(
-                "the remote server was started by a herdr build that may not survive SSH connection loss. restart it so network drops disconnect only this client."
+                "the remote server predates Herdr's stable endpoint protocol and must update before this client can attach."
             );
         }
-        RemoteServerRestartReason::BinaryUpdated => {
-            eprintln!(
-                "the remote herdr binary was installed or replaced. restart the remote server so it uses the prepared binary."
-            );
-        }
-        RemoteServerRestartReason::VersionMismatch => {
+        RemoteServerRestartReason::DaemonDetachMissing => {
             eprintln!(
-                "the remote server is still running a different herdr version. restart it so it uses the prepared binary."
+                "the remote server was started by a herdr build that may not survive SSH connection loss. restart it so network drops disconnect only this client."
             );
         }
     }
 
-    let prompt = if reason == RemoteServerRestartReason::ProtocolMismatch {
-        "stop the remote server and continue attaching? [Y/n] "
+    let prompt = if reason == RemoteServerRestartReason::EndpointProtocolMissing {
+        "update the remote server and continue attaching? [Y/n] "
     } else {
         "restart the remote server now? [y/N] "
     };
@@ -1333,10 +1322,10 @@ fn confirm_remote_server_stop(
     if answer == "y" || answer == "yes" {
         return Ok(true);
     }
-    if answer.is_empty() && reason == RemoteServerRestartReason::ProtocolMismatch {
+    if answer.is_empty() && reason == RemoteServerRestartReason::EndpointProtocolMissing {
         return Ok(true);
     }
-    if reason == RemoteServerRestartReason::ProtocolMismatch {
+    if reason == RemoteServerRestartReason::EndpointProtocolMissing {
         return Err(io::Error::new(
             io::ErrorKind::Interrupted,
             "remote herdr server stop cancelled",
@@ -1346,14 +1335,25 @@ fn confirm_remote_server_stop(
     Ok(false)
 }
 
-fn live_handoff_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<()> {
-    let command = format!(
+fn remote_live_handoff_command(remote_herdr: &RemoteHerdr, protocol: u32, version: &str) -> String {
+    format!(
         "{} server live-handoff --import-exe {} --expected-protocol {} --expected-version {}",
-        remote_herdr.shell_path,
-        remote_herdr.shell_path,
-        CURRENT_PROTOCOL,
-        current_version()
-    );
+        remote_herdr.shell_path, remote_herdr.shell_path, protocol, version
+    )
+}
+
+fn live_handoff_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<()> {
+    let status = remote_client_status(ssh, remote_herdr)?.ok_or_else(|| {
+        io::Error::other("could not inspect the prepared remote herdr binary before live handoff")
+    })?;
+    let protocol = status.protocol.ok_or_else(|| {
+        io::Error::other("prepared remote herdr did not report its private protocol")
+    })?;
+    let version = status
+        .version
+        .filter(|version| !version.is_empty())
+        .ok_or_else(|| io::Error::other("prepared remote herdr did not report its version"))?;
+    let command = remote_live_handoff_command(remote_herdr, protocol, &version);
     let output = ssh.sh_output(&command)?;
     if !output.status.success() {
         return Err(command_failed("remote server live handoff failed", &output));
@@ -2896,25 +2896,29 @@ mod tests {
     }
 
     #[test]
-    fn parse_client_status_json_reads_protocol() {
-        assert_eq!(
-            parse_client_status_json(r#"{"version":"x","protocol":8,"binary":"/bin/herdr"}"#)
-                .map(|status| status.protocol),
-            Some(8)
+    fn parse_client_status_json_reads_last_json_record() {
+        let status = parse_client_status_json(
+            "wrapper output\n{\"version\":\"0.8.0\",\"protocol\":20,\"endpoint_protocol_generation\":1}\n{\"wrapper\":true}\n",
+        )
+        .unwrap();
+        assert_eq!(status.version.as_deref(), Some("0.8.0"));
+        assert_eq!(status.protocol, Some(20));
+        assert_eq!(status.endpoint_protocol_generation, Some(1));
+        assert!(
+            parse_client_status_json(r#"{"endpoint_protocol_generation":"unknown"}"#).is_none()
         );
-        assert!(parse_client_status_json(r#"{"protocol":"unknown"}"#).is_none());
     }
 
     #[test]
     fn parse_remote_server_status_json_reads_running_server() {
         assert_eq!(
             parse_remote_server_status_json(
-                r#"{"status":"running","running":true,"version":"0.6.0","protocol":8,"capabilities":{"live_handoff":true,"detached_server_daemon":true}}"#
+                r#"{"status":"running","running":true,"version":"0.6.0","protocol":8,"capabilities":{"live_handoff":true,"detached_server_daemon":true,"endpoint_protocol_generation":1}}"#
             )
             .unwrap(),
             RemoteServerStatus::Running {
                 version: Some("0.6.0".into()),
-                protocol: Some(8),
+                endpoint_protocol_generation: Some(1),
                 live_handoff: true,
                 detached_server_daemon: true
             }
@@ -2930,7 +2934,7 @@ mod tests {
             .unwrap(),
             RemoteServerStatus::Running {
                 version: Some("0.6.0".into()),
-                protocol: Some(8),
+                endpoint_protocol_generation: None,
                 live_handoff: false,
                 detached_server_daemon: false
             }
@@ -3105,21 +3109,19 @@ mod tests {
     }
 
     #[test]
-    fn remote_server_restart_reason_requires_stop_for_protocol_mismatch() {
+    fn remote_server_restart_reason_requires_one_update_for_pre_floor_server() {
         assert_eq!(
-            remote_server_restart_reason(Some(&current_version()), Some(0), true, false),
-            Some(RemoteServerRestartReason::ProtocolMismatch)
+            remote_server_restart_reason(None, true),
+            Some(RemoteServerRestartReason::EndpointProtocolMissing)
         );
     }
 
     #[test]
-    fn remote_server_restart_reason_allows_unchanged_compatible_server() {
+    fn remote_server_restart_reason_allows_compatible_server() {
         assert_eq!(
             remote_server_restart_reason(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
-                true,
-                false
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                true
             ),
             None
         );
@@ -3129,9 +3131,7 @@ mod tests {
     fn remote_server_restart_reason_requires_restart_for_old_daemon() {
         assert_eq!(
             remote_server_restart_reason(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
-                false,
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
                 false
             ),
             Some(RemoteServerRestartReason::DaemonDetachMissing)
@@ -3139,52 +3139,12 @@ mod tests {
     }
 
     #[test]
-    fn remote_server_restart_reason_requires_restart_after_helper_update() {
-        assert_eq!(
-            remote_server_restart_reason(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
-                true,
-                true
-            ),
-            Some(RemoteServerRestartReason::BinaryUpdated)
-        );
-    }
-
-    #[test]
-    fn remote_server_restart_reason_offers_restart_for_version_mismatch() {
-        assert_eq!(
-            remote_server_restart_reason(Some("0.0.0"), Some(CURRENT_PROTOCOL), true, false),
-            Some(RemoteServerRestartReason::VersionMismatch)
-        );
-        assert_eq!(
-            remote_server_restart_reason(None, Some(CURRENT_PROTOCOL), true, false),
-            Some(RemoteServerRestartReason::VersionMismatch)
-        );
-    }
-
-    #[test]
-    fn remote_server_restart_reason_allows_current_server() {
-        assert_eq!(
-            remote_server_restart_reason(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
-                true,
-                false
-            ),
-            None
-        );
-    }
-
-    #[test]
     fn remote_install_plan_keeps_compatible_running_server() {
         assert_eq!(
             remote_install_running_server_plan(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
                 true,
                 false,
-                false,
                 false
             ),
             RemoteInstallRunningServerPlan::KeepRunning
@@ -3195,10 +3155,8 @@ mod tests {
     fn remote_install_plan_requires_stop_for_old_daemon() {
         assert_eq!(
             remote_install_running_server_plan(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
                 false,
-                true,
                 false,
                 false
             ),
@@ -3209,53 +3167,39 @@ mod tests {
     }
 
     #[test]
-    fn remote_install_plan_requires_stop_after_helper_update() {
-        assert_eq!(
-            remote_install_running_server_plan(
-                Some(&current_version()),
-                Some(CURRENT_PROTOCOL),
-                true,
-                true,
-                false,
-                false
-            ),
-            RemoteInstallRunningServerPlan::StopRequired(RemoteServerRestartReason::BinaryUpdated)
-        );
-    }
-
-    #[test]
-    fn remote_install_plan_requires_stop_for_incompatible_running_server() {
+    fn remote_install_plan_requires_stop_for_pre_floor_server() {
         assert_eq!(
-            remote_install_running_server_plan(
-                Some("0.0.0"),
-                Some(CURRENT_PROTOCOL),
-                true,
-                true,
-                false,
-                false
-            ),
+            remote_install_running_server_plan(None, true, false, false),
             RemoteInstallRunningServerPlan::StopRequired(
-                RemoteServerRestartReason::VersionMismatch
+                RemoteServerRestartReason::EndpointProtocolMissing
             )
         );
     }
 
     #[test]
-    fn remote_install_plan_uses_live_handoff_for_incompatible_running_server() {
+    fn remote_install_plan_uses_live_handoff_for_pre_floor_server() {
         assert_eq!(
-            remote_install_running_server_plan(
-                Some("0.0.0"),
-                Some(CURRENT_PROTOCOL),
-                true,
-                true,
-                true,
-                true
-            ),
+            remote_install_running_server_plan(None, true, true, true),
             RemoteInstallRunningServerPlan::LiveHandoff
         );
     }
 
     #[test]
+    fn remote_live_handoff_uses_prepared_binary_identity() {
+        let remote_herdr = RemoteHerdr::for_platform(RemotePlatform {
+            os: "linux",
+            arch: "x86_64",
+        });
+        let command = remote_live_handoff_command(&remote_herdr, 19, "0.7.9");
+        assert!(command.contains("--expected-protocol 19"));
+        assert!(command.contains("--expected-version 0.7.9"));
+        assert!(!command.contains(&format!(
+            "--expected-protocol {CURRENT_PROTOCOL} --expected-version {}",
+            current_version()
+        )));
+    }
+
+    #[test]
     fn install_source_description_uses_override_binary() {
         let platform = RemotePlatform {
             os: "linux",
diff --git a/src/remote/host_unix.rs b/src/remote/host_unix.rs
index e04b0e00..48b20e0b 100644
--- a/src/remote/host_unix.rs
+++ b/src/remote/host_unix.rs
@@ -56,11 +56,16 @@ fn ensure_remote_server_running() -> io::Result<()> {
             Duration::from_millis(500),
         )?
         .ok_or_else(|| io::Error::other("remote server status API is unavailable"))?;
-        if status.protocol == Some(crate::protocol::PROTOCOL_VERSION) {
+        if status
+            .capabilities
+            .as_ref()
+            .and_then(|capabilities| capabilities.endpoint_protocol_generation)
+            == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION)
+        {
             return Ok(());
         }
         return Err(io::Error::other(
-            "remote herdr server must restart before this bridge can attach; rerun `herdr --remote` from an interactive terminal to approve stopping it",
+            "remote herdr server needs one final update before this bridge can attach; rerun `herdr --remote` from an interactive terminal to approve it",
         ));
     }
 
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);
 }
diff --git a/src/ui/release_notes.rs b/src/ui/release_notes.rs
index 5f6dc078..e54c4369 100644
--- a/src/ui/release_notes.rs
+++ b/src/ui/release_notes.rs
@@ -307,7 +307,7 @@ mod tests {
         assert_eq!(line_text(&lines[0]), " ● update ready");
         assert_eq!(
             line_text(&lines[1]),
-            " detach, run herdr update, then follow its restart guidance"
+            " detach, run herdr update, then run Herdr again to reconnect"
         );
         assert_eq!(lines[0].spans[1].style.fg, Some(palette.accent));
         assert_eq!(lines[0].spans[2].style.fg, Some(palette.text));
@@ -331,7 +331,7 @@ mod tests {
         assert_eq!(line_text(&lines[0].1), " ● update ready");
         assert_eq!(
             line_text(&lines[1].1),
-            " detach, run herdr update, then follow its restart guidance"
+            " detach, run herdr update, then run Herdr again to reconnect"
         );
         assert_eq!(line_text(&lines[2].1), "");
         assert_eq!(line_text(&lines[3].1), " ADDED");
diff --git a/src/update.rs b/src/update.rs
index a15c030f..879726fc 100644
--- a/src/update.rs
+++ b/src/update.rs
@@ -197,6 +197,8 @@ impl AssetRef {
 #[derive(Deserialize)]
 struct UpdateManifest {
     version: String,
+    #[cfg(not(windows))]
+    endpoint_generation: Option<u32>,
     /// Thin-client protocol spoken by this release, when advertised by the manifest.
     #[cfg(not(windows))]
     protocol: Option<u32>,
@@ -236,6 +238,7 @@ struct PreviewManifest {
     commit: String,
     built_at: String,
     protocol: u32,
+    endpoint_generation: Option<u32>,
     notes: String,
     assets: BTreeMap<String, AssetRef>,
     #[serde(default)]
@@ -248,6 +251,7 @@ struct PreviewBuildMetadata {
     commit: String,
     built_at: String,
     protocol: u32,
+    endpoint_generation: Option<u32>,
     assets: BTreeMap<String, AssetRef>,
 }
 
@@ -306,6 +310,8 @@ struct ReleaseInfo {
     commit: Option<String>,
     #[cfg(not(windows))]
     target_protocol: Option<u32>,
+    #[cfg(not(windows))]
+    target_endpoint_generation: Option<u32>,
     download_url: String,
     sha256: Option<String>,
     #[cfg(windows)]
@@ -415,6 +421,8 @@ fn release_info_from_manifest(manifest: &UpdateManifest) -> Result<Option<Releas
         commit: None,
         #[cfg(not(windows))]
         target_protocol: manifest.protocol,
+        #[cfg(not(windows))]
+        target_endpoint_generation: manifest.endpoint_generation,
         download_url,
         sha256: Some(sha256),
         #[cfg(windows)]
@@ -475,6 +483,7 @@ fn release_info_from_preview_manifest(
             || archived.commit != manifest.commit
             || archived.built_at != manifest.built_at
             || archived.protocol != manifest.protocol
+            || archived.endpoint_generation != manifest.endpoint_generation
         {
             tracing::warn!(
                 build_id,
@@ -502,6 +511,8 @@ fn release_info_from_preview_manifest(
         commit: Some(manifest.commit.clone()),
         #[cfg(not(windows))]
         target_protocol: Some(manifest.protocol),
+        #[cfg(not(windows))]
+        target_endpoint_generation: manifest.endpoint_generation,
         download_url,
         sha256: asset.sha256.clone(),
         #[cfg(windows)]
@@ -825,10 +836,14 @@ fn update_requires_server_restart(
     server: &crate::api::RuntimeStatus,
     release: &ReleaseInfo,
 ) -> bool {
-    match (server.protocol, release.target_protocol) {
-        (Some(server_protocol), Some(target_protocol)) => server_protocol != target_protocol,
-        _ => true,
-    }
+    let Some(target_generation) = release.target_endpoint_generation else {
+        return true;
+    };
+    server
+        .capabilities
+        .as_ref()
+        .and_then(|capabilities| capabilities.endpoint_protocol_generation)
+        != Some(target_generation)
 }
 
 #[cfg(not(windows))]
@@ -903,6 +918,7 @@ enum RunningServerUpdateAction {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[cfg(not(windows))]
 enum RunningServerUpdateOutcome {
+    CompatibleServerKept,
     RestartDeferred,
     Stopped,
     LiveHandoffComplete,
@@ -1229,7 +1245,12 @@ fn prompt_to_complete_plain_update(
     decisions: &[RunningServerUpdateDecision],
     release: &ReleaseInfo,
 ) -> Result<bool, String> {
-    if decisions.is_empty() {
+    let plans: Vec<&RunningServerUpdatePlan> = decisions
+        .iter()
+        .filter(|decision| decision.plan.requires_server_restart)
+        .map(|decision| &decision.plan)
+        .collect();
+    if plans.is_empty() {
         return Ok(true);
     }
 
@@ -1237,8 +1258,6 @@ fn prompt_to_complete_plain_update(
         return Ok(false);
     }
 
-    let plans: Vec<&RunningServerUpdatePlan> =
-        decisions.iter().map(|decision| &decision.plan).collect();
     let (singular, plural) = target_group_nouns(&plans);
     let noun = if plans.len() == 1 { singular } else { plural };
     eprintln!(
@@ -1288,7 +1307,9 @@ fn mark_plain_update_stop_decisions(
     decisions
         .into_iter()
         .map(|mut decision| {
-            decision.action = RunningServerUpdateAction::StopOldServer;
+            if decision.plan.requires_server_restart {
+                decision.action = RunningServerUpdateAction::StopOldServer;
+            }
             decision
         })
         .collect()
@@ -1719,7 +1740,10 @@ fn apply_running_session_update_decisions(
 
     for decision in decisions {
         let outcome = match decision.action {
-            RunningServerUpdateAction::None => RunningServerUpdateOutcome::RestartDeferred,
+            RunningServerUpdateAction::None if decision.plan.requires_server_restart => {
+                RunningServerUpdateOutcome::RestartDeferred
+            }
+            RunningServerUpdateAction::None => RunningServerUpdateOutcome::CompatibleServerKept,
             RunningServerUpdateAction::StopOldServer => {
                 stop_running_server_for_update(&decision.plan)?;
                 RunningServerUpdateOutcome::Stopped
@@ -1760,6 +1784,24 @@ fn print_running_session_update_outcomes(
 
     for outcome in outcomes {
         match outcome.outcome {
+            RunningServerUpdateOutcome::CompatibleServerKept => {
+                eprintln!(
+                    "{} {} kept running server v{}.",
+                    outcome.target_noun,
+                    outcome.session_label,
+                    version_label(outcome.server_version.as_deref())
+                );
+                match &outcome.attach_command {
+                    Some(command) => eprintln!(
+                        "Run `{command}` to reconnect with the updated client. Restart the server later only if you need server-side changes from {}.",
+                        release.label()
+                    ),
+                    None => eprintln!(
+                        "Reconnect with the same socket override to use the updated client. Restart the server later only if you need server-side changes from {}.",
+                        release.label()
+                    ),
+                }
+            }
             RunningServerUpdateOutcome::LiveHandoffComplete => {
                 if let Some(command) = &outcome.attach_command {
                     eprintln!(
@@ -1858,19 +1900,19 @@ pub(crate) fn update_install_command() -> &'static str {
 pub(crate) fn update_install_instruction(install_command: &str) -> String {
     match install_command {
         HERDR_UPDATE_COMMAND => {
-            "detach, run `herdr update`, then follow its restart guidance".to_string()
+            "detach, run `herdr update`, then run Herdr again to reconnect".to_string()
         }
         HOMEBREW_UPDATE_COMMAND => {
-            "detach, run `brew update && brew upgrade herdr`, then restart this Herdr session when ready".to_string()
+            "detach, run `brew update && brew upgrade herdr`, then run Herdr again to reconnect"
+                .to_string()
         }
         MISE_UPDATE_COMMAND => {
-            "detach, run `mise upgrade herdr`, then restart this Herdr session when ready"
-                .to_string()
+            "detach, run `mise upgrade herdr`, then run Herdr again to reconnect".to_string()
         }
         NIX_UPDATE_COMMAND => {
-            "detach, update through Nix, then restart this Herdr session when ready".to_string()
+            "detach, update through Nix, then run Herdr again to reconnect".to_string()
         }
-        command => format!("detach, run `{command}`, then restart this Herdr session when ready"),
+        command => format!("detach, run `{command}`, then run Herdr again to reconnect"),
     }
 }
 
@@ -2146,7 +2188,7 @@ pub fn self_update(options: SelfUpdateOptions) -> Result<Version, String> {
         eprintln!("installed {}", release.label());
         print_outdated_integration_notice_with_updated_binary(&updated_exe);
         eprintln!(
-            "Restart any running Herdr sessions to use {}.",
+            "Start Herdr again to use the updated client. Running servers remain active; restart them later only if you need server-side changes from {}.",
             release.label()
         );
     }
@@ -2432,6 +2474,9 @@ mod tests {
             build_id: None,
             commit: None,
             target_protocol,
+            target_endpoint_generation: Some(
+                crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+            ),
             download_url: "https://example.com/herdr".to_string(),
             sha256: None,
             notes_body: "### Changed\n- One".to_string(),
@@ -2711,15 +2756,15 @@ mod tests {
     fn update_install_instruction_distinguishes_install_from_restart() {
         assert_eq!(
             update_install_instruction(HERDR_UPDATE_COMMAND),
-            "detach, run `herdr update`, then follow its restart guidance"
+            "detach, run `herdr update`, then run Herdr again to reconnect"
         );
         assert_eq!(
             update_install_instruction(HOMEBREW_UPDATE_COMMAND),
-            "detach, run `brew update && brew upgrade herdr`, then restart this Herdr session when ready"
+            "detach, run `brew update && brew upgrade herdr`, then run Herdr again to reconnect"
         );
         assert_eq!(
             update_install_instruction(MISE_UPDATE_COMMAND),
-            "detach, run `mise upgrade herdr`, then restart this Herdr session when ready"
+            "detach, run `mise upgrade herdr`, then run Herdr again to reconnect"
         );
     }
 
@@ -2796,41 +2841,60 @@ mod tests {
     }
 
     #[test]
-    fn update_requires_server_restart_when_target_protocol_differs_or_unknown() {
-        let server = crate::api::RuntimeStatus {
+    fn update_requires_server_restart_only_without_the_endpoint_baseline() {
+        let release = fake_release("0.5.6", Some(4));
+        let compatible = crate::api::RuntimeStatus {
             version: Some("0.5.5".to_string()),
             protocol: Some(2),
-            capabilities: None,
+            capabilities: Some(crate::api::schema::ServerCapabilities {
+                live_handoff: true,
+                detached_server_daemon: true,
+                endpoint_protocol_generation: Some(
+                    crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+                ),
+            }),
         };
-        let compatible_release = ReleaseInfo {
-            version: Version::parse("0.5.6").unwrap(),
-            identity: "0.5.6".to_string(),
-            channel: UpdateChannel::Stable,
-            build_id: None,
-            commit: None,
-            target_protocol: Some(2),
-            download_url: "https://example.com/herdr".to_string(),
-            sha256: None,
-            notes_body: "### Changed\n- One".to_string(),
+        let missing_baseline = crate::api::RuntimeStatus {
+            capabilities: None,
+            ..compatible.clone()
         };
-        let incompatible_release = ReleaseInfo {
-            target_protocol: Some(4),
-            ..compatible_release.clone()
+        let incompatible_generation = crate::api::RuntimeStatus {
+            capabilities: Some(crate::api::schema::ServerCapabilities {
+                endpoint_protocol_generation: Some(
+                    crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION + 1,
+                ),
+                ..compatible.capabilities.clone().unwrap()
+            }),
+            ..compatible.clone()
         };
+
+        assert!(!update_requires_server_restart(&compatible, &release));
+        assert!(update_requires_server_restart(&missing_baseline, &release));
+        assert!(update_requires_server_restart(
+            &incompatible_generation,
+            &release
+        ));
+
         let unknown_release = ReleaseInfo {
-            target_protocol: None,
-            ..compatible_release.clone()
+            target_endpoint_generation: None,
+            ..release.clone()
         };
-
-        assert!(!update_requires_server_restart(
-            &server,
-            &compatible_release
+        assert!(update_requires_server_restart(
+            &compatible,
+            &unknown_release
         ));
         assert!(update_requires_server_restart(
-            &server,
-            &incompatible_release
+            &missing_baseline,
+            &unknown_release
         ));
-        assert!(update_requires_server_restart(&server, &unknown_release));
+
+        let future_release = ReleaseInfo {
+            target_endpoint_generation: Some(
+                crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION + 1,
+            ),
+            ..release
+        };
+        assert!(update_requires_server_restart(&compatible, &future_release));
     }
 
     #[test]
@@ -2857,6 +2921,9 @@ mod tests {
                 capabilities: Some(crate::api::schema::ServerCapabilities {
                     live_handoff: true,
                     detached_server_daemon: true,
+                    endpoint_protocol_generation: Some(
+                        crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+                    ),
                 }),
             },
         };
@@ -2970,9 +3037,8 @@ mod tests {
         let work_client_socket = crate::session::client_socket_path_for(Some("work"));
         fs::create_dir_all(work_client_socket.parent().unwrap()).unwrap();
         let work_client_listener = UnixListener::bind(&work_client_socket).unwrap();
-        let release = fake_release("9.8.7", Some(77));
 
-        let err = plan_running_server_updates(&release).unwrap_err();
+        let err = plan_running_server_updates(&fake_release("9.8.7", Some(77))).unwrap_err();
 
         drop(work_client_listener);
         let _ = fs::remove_dir_all(config_home);
@@ -3051,6 +3117,9 @@ mod tests {
             build_id: None,
             commit: None,
             target_protocol: Some(3),
+            target_endpoint_generation: Some(
+                crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+            ),
             download_url: "https://example.com/herdr".to_string(),
             sha256: None,
             notes_body: "### Changed\n- One".to_string(),
@@ -3085,6 +3154,50 @@ mod tests {
     }
 
     #[test]
+    fn noninteractive_plain_update_completes_with_compatible_server() {
+        assert!(
+            !io::stdin().is_terminal(),
+            "this test relies on noninteractive test stdin"
+        );
+        let release = fake_release("9.8.7", Some(77));
+        let plan = RunningServerUpdatePlan {
+            target: RunningUpdateTarget {
+                name: Some("work".to_string()),
+                label: "work".to_string(),
+                stop_command: "herdr session stop work".to_string(),
+                attach_command: Some("herdr session attach work".to_string()),
+                socket_path: crate::session::api_socket_path_for(Some("work")),
+                client_socket_path: crate::session::client_socket_path_for(Some("work")),
+                must_be_running: true,
+            },
+            requires_server_restart: false,
+            server: crate::api::RuntimeStatus {
+                version: Some("9.8.6".to_string()),
+                protocol: Some(76),
+                capabilities: Some(crate::api::schema::ServerCapabilities {
+                    live_handoff: true,
+                    detached_server_daemon: true,
+                    endpoint_protocol_generation: Some(
+                        crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+                    ),
+                }),
+            },
+        };
+        let decisions = confirm_running_server_update_action(
+            vec![plan],
+            &release,
+            SelfUpdateOptions {
+                live_handoff: false,
+            },
+        )
+        .unwrap();
+
+        assert!(prompt_to_complete_plain_update(&decisions, &release).unwrap());
+        let decisions = mark_plain_update_stop_decisions(decisions);
+        assert_eq!(decisions[0].action, RunningServerUpdateAction::None);
+    }
+
+    #[test]
     fn client_protocol_server_is_running_at_detects_live_socket() {
         let socket_path = unique_test_socket_path("client-live");
         let listener = UnixListener::bind(&socket_path).unwrap();
@@ -3186,6 +3299,9 @@ mod tests {
             build_id: None,
             commit: None,
             target_protocol: Some(77),
+            target_endpoint_generation: Some(
+                crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+            ),
             download_url: "https://example.com/herdr".to_string(),
             sha256: None,
             notes_body: "### Changed\n- One".to_string(),
@@ -3296,6 +3412,7 @@ mod tests {
         let json = "{\n\
             \"version\": \"0.2.0\",\n\
             \"protocol\": 4,\n\
+            \"endpoint_generation\": 1,\n\
             \"notes\": \"### Changed\\n- One\",\n\
             \"announcement\": {\n\
                 \"id\": \"keymap-v2\",\n\
@@ -3310,6 +3427,7 @@ mod tests {
         let manifest: UpdateManifest = serde_json::from_str(json).unwrap();
         assert_eq!(manifest.version, "0.2.0");
         assert_eq!(manifest.protocol, Some(4));
+        assert_eq!(manifest.endpoint_generation, Some(1));
         assert_eq!(manifest.assets.len(), 2);
         assert_eq!(
             manifest