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

1 files changed, 76 insertions, 21 deletions

diff --git a/tests/support/mod.rs b/tests/support/mod.rs
index 2c863fbe..68337ade 100644
--- a/tests/support/mod.rs
+++ b/tests/support/mod.rs
@@ -14,13 +14,14 @@ static CLEANUP_GUARD: OnceLock<CleanupGuard> = OnceLock::new();
 const WATCHDOG_SCAN_INTERVAL: Duration = Duration::from_secs(1);
 const RUNTIME_OWNER_MARKER: &str = ".herdr-test-owner-pid";
 pub const CURRENT_PROTOCOL: u32 = 22;
+pub const CURRENT_ENDPOINT_PROTOCOL_GENERATION: u32 = 1;
 pub const SERVER_MESSAGE_SERVER_SHUTDOWN: u32 = 3;
-pub const SERVER_MESSAGE_CLIENT_SHELL_SNAPSHOT: u32 = 12;
+pub const SERVER_MESSAGE_ENDPOINT_CONTROL: u32 = 20;
 pub const SERVER_MESSAGE_PANE_SURFACE: u32 = 13;
 pub const SERVER_MESSAGE_SEMANTIC_NOTIFICATION: u32 = 14;
 pub const SERVER_MESSAGE_PANE_SURFACE_PATCH: u32 = 19;
-const CLIENT_MESSAGE_CLIENT_SHELL_HELLO: u32 = 11;
 const CLIENT_MESSAGE_CLIENT_SHELL_PANE_INPUT: u32 = 13;
+const CLIENT_MESSAGE_ENDPOINT_CONTROL: u32 = 20;
 
 pub fn register_spawned_herdr_pid(pid: Option<u32>) {
     let Some(pid) = pid else {
@@ -185,6 +186,25 @@ fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> {
     buf
 }
 
+fn encode_string(value: &str) -> Vec<u8> {
+    let mut encoded = encode_varint_u32(value.len() as u32);
+    encoded.extend_from_slice(value.as_bytes());
+    encoded
+}
+
+fn decode_string(payload: &[u8], offset: &mut usize) -> Result<String, String> {
+    let (len, consumed) = decode_varint_u32(payload, *offset)?;
+    *offset += consumed;
+    let len = len as usize;
+    if *offset + len > payload.len() {
+        return Err("payload too short for string content".into());
+    }
+    let value = String::from_utf8(payload[*offset..*offset + len].to_vec())
+        .map_err(|err| err.to_string())?;
+    *offset += len;
+    Ok(value)
+}
+
 fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), String> {
     let mut offset = 0;
     let (variant, consumed) = decode_varint_u32(payload, offset)?;
@@ -225,10 +245,10 @@ fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), String> {
     Ok((version, error))
 }
 
-fn finish_handshake(
+fn read_handshake_response(
     stream: &mut UnixStream,
     hello_payload: &[u8],
-) -> Result<(u32, Option<String>), String> {
+) -> Result<Vec<u8>, String> {
     stream
         .set_read_timeout(Some(Duration::from_secs(5)))
         .map_err(|e| e.to_string())?;
@@ -245,7 +265,7 @@ fn finish_handshake(
     }
     let mut payload = vec![0u8; len];
     stream.read_exact(&mut payload).map_err(|e| e.to_string())?;
-    decode_welcome(&payload)
+    Ok(payload)
 }
 
 pub fn client_handshake(
@@ -265,30 +285,60 @@ pub fn client_handshake(
             &[0],                   // pixel_mouse = false
         ],
     );
-    finish_handshake(stream, &hello_payload)
+    let response = read_handshake_response(stream, &hello_payload)?;
+    decode_welcome(&response)
 }
 
 pub fn client_shell_handshake(
     stream: &mut UnixStream,
-    version: u32,
+    endpoint_generation: u32,
     surface_cols: u16,
     surface_rows: u16,
 ) -> Result<(u32, Option<String>), String> {
+    let data = serde_json::json!({
+        "generation": endpoint_generation,
+        "cell_width_px": 8,
+        "cell_height_px": 16,
+        "surface_size": {"cols": surface_cols, "rows": surface_rows},
+        "pixel_mouse": false,
+        "direct_graphics": false,
+        "endpoint_keybindings": false,
+        "mouse_capture": false,
+        "snapshot_codecs": ["shell.snapshot.v1"],
+        "surface_codecs": ["shell.surface.v1"],
+        "input_codecs": ["shell.input.semantic.v1"],
+        "blob_codecs": ["shell.blob.v1"]
+    })
+    .to_string();
     let hello_payload = encode_varint_enum(
-        CLIENT_MESSAGE_CLIENT_SHELL_HELLO,
-        &[
-            &encode_varint_u32(version),
-            &encode_varint_u32(8),
-            &encode_varint_u32(16),
-            &encode_varint_u16(surface_cols),
-            &encode_varint_u16(surface_rows),
-            &[0], // pixel mouse disabled
-            &[0], // direct graphics disabled
-            &[0], // client-owned keybindings
-            &[0], // mouse capture disabled
-        ],
+        CLIENT_MESSAGE_ENDPOINT_CONTROL,
+        &[&encode_string("endpoint.hello.v1"), &encode_string(&data)],
     );
-    finish_handshake(stream, &hello_payload)
+    let response = read_handshake_response(stream, &hello_payload)?;
+    let mut offset = 0;
+    let (variant, consumed) = decode_varint_u32(&response, offset)?;
+    offset += consumed;
+    if variant != SERVER_MESSAGE_ENDPOINT_CONTROL {
+        return Err(format!(
+            "expected EndpointControl (variant {SERVER_MESSAGE_ENDPOINT_CONTROL}), got variant {variant}"
+        ));
+    }
+    let kind = decode_string(&response, &mut offset)?;
+    if kind != "endpoint.welcome.v1" {
+        return Err(format!("expected endpoint.welcome.v1, got {kind}"));
+    }
+    let data = decode_string(&response, &mut offset)?;
+    let value: serde_json::Value = serde_json::from_str(&data).map_err(|err| err.to_string())?;
+    let generation = value["generation"]
+        .as_u64()
+        .ok_or_else(|| "endpoint welcome omitted generation".to_owned())?
+        as u32;
+    let error = value["error"]
+        .as_object()
+        .and_then(|error| error.get("message"))
+        .and_then(serde_json::Value::as_str)
+        .map(str::to_owned);
+    Ok((generation, error))
 }
 
 pub fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec<u8>), String> {
@@ -407,7 +457,12 @@ pub fn wait_for_client_shell_bootstrap(
     let mut saw_snapshot = false;
     while Instant::now() < deadline {
         match read_server_message(stream) {
-            Ok((SERVER_MESSAGE_CLIENT_SHELL_SNAPSHOT, _)) => saw_snapshot = true,
+            Ok((SERVER_MESSAGE_ENDPOINT_CONTROL, payload)) => {
+                let mut offset = 0;
+                if decode_string(&payload, &mut offset).as_deref() == Ok("shell.snapshot.v1") {
+                    saw_snapshot = true;
+                }
+            }
             Ok((SERVER_MESSAGE_PANE_SURFACE, _)) if saw_snapshot => return Ok(()),
             Ok((SERVER_MESSAGE_PANE_SURFACE, _)) => {
                 return Err("client shell pane surface arrived before its snapshot".into());