fix: handle gossipsub MessageTooLarge error to prevent silent crash (#1583)

## Summary

Large prompts (70K+ tokens / ~500KB+ JSON) cause exo to silently crash.
The root cause is an unhandled `PublishError::MessageTooLarge` from
gossipsub when serialized `TextGeneration` commands exceed the 1MB
`max_transmit_size` limit.

The error propagates as a generic Python exception through the PyO3
bindings. Since `_networking_publish` in `router.py` only catches
`NoPeersSubscribedToTopicError` and `AllQueuesFullError`, the unhandled
exception crashes the networking async task, causing exo to shut down
silently — no error message, no API response.

## Changes

- **Rust (PyO3 bindings):** Add `MessageTooLargeError` exception class
and handle `PublishError::MessageTooLarge` explicitly in the gossipsub
publish path, matching the existing pattern for
`NoPeersSubscribedToTopicError` and `AllQueuesFullError`
- **Python (router):** Catch `MessageTooLargeError` in
`_networking_publish` and log a warning with the message size,
preventing the networking task from crashing

## Reproduction

On a multi-node cluster with a large model (e.g., GLM-5 754B tensor
parallel over JACCL RDMA):
1. Send a chat completion request with ~70K+ tokens
2. exo silently shuts down — no error logged, curl gets no response
3. With shorter prompts (< ~50K tokens): works fine

## Test plan

- Verified `cargo check` passes for `networking` and `exo_pyo3_bindings`
crates
- Verified `ruff check` passes for modified Python files
- Manual testing on 4× Mac Studio M3 Ultra cluster: 50K token requests
pass, 70K+ previously caused silent shutdown, now logs a warning and
drops the oversized message gracefully

Co-authored-by: vsm <[email protected]>
Co-authored-by: Cursor <[email protected]>
Co-authored-by: rltakashige <[email protected]>

rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi

@@ -104,6 +104,12 @@ class NetworkingHandle:
         will sleep until a message is sent.
         """
 
+@typing.final
+class MessageTooLargeError(builtins.Exception):
+    def __new__(cls, *args: typing.Any) -> MessageTooLargeError: ...
+    def __repr__(self) -> builtins.str: ...
+    def __str__(self) -> builtins.str: ...
+
 @typing.final
 class NoPeersSubscribedToTopicError(builtins.Exception):
     def __new__(cls, *args: typing.Any) -> NoPeersSubscribedToTopicError: ...

rust/exo_pyo3_bindings/src/networking.rs

@@ -98,6 +98,37 @@ mod exception {
             Self::MSG.to_string()
         }
     }
+
+    #[gen_stub_pyclass]
+    #[pyclass(frozen, extends=PyException, name="MessageTooLargeError")]
+    pub struct PyMessageTooLargeError {}
+
+    impl PyMessageTooLargeError {
+        const MSG: &'static str = "Gossipsub message exceeds max_transmit_size. Reduce prompt length or increase the limit.";
+
+        pub(crate) fn new_err() -> PyErr {
+            PyErr::new::<Self, _>(())
+        }
+    }
+
+    #[gen_stub_pymethods]
+    #[pymethods]
+    impl PyMessageTooLargeError {
+        #[new]
+        #[pyo3(signature = (*args))]
+        #[allow(unused_variables)]
+        pub(crate) fn new(args: &Bound<'_, PyTuple>) -> Self {
+            Self {}
+        }
+
+        fn __repr__(&self) -> String {
+            format!("MessageTooLargeError(\"{}\")", Self::MSG)
+        }
+
+        fn __str__(&self) -> String {
+            Self::MSG.to_string()
+        }
+    }
 }
 
 /// Connection or disconnection event discriminant type.
@@ -200,6 +231,8 @@ async fn networking_task(
                             Err(exception::PyNoPeersSubscribedToTopicError::new_err())
                         } else if let Err(PublishError::AllQueuesFull(_)) = result {
                             Err(exception::PyAllQueuesFullError::new_err())
+                        } else if let Err(PublishError::MessageTooLarge) = result {
+                            Err(exception::PyMessageTooLargeError::new_err())
                         } else {
                             result.pyerr()
                         };
@@ -561,6 +594,7 @@ impl PyNetworkingHandle {
 pub fn networking_submodule(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<exception::PyNoPeersSubscribedToTopicError>()?;
     m.add_class::<exception::PyAllQueuesFullError>()?;
+    m.add_class::<exception::PyMessageTooLargeError>()?;
 
     m.add_class::<PyConnectionUpdateType>()?;
     m.add_class::<PyConnectionUpdate>()?;

src/exo/routing/router.py

@@ -14,6 +14,7 @@ from anyio import (
 from exo_pyo3_bindings import (
     AllQueuesFullError,
     Keypair,
+    MessageTooLargeError,
     NetworkingHandle,
     NoPeersSubscribedToTopicError,
 )
@@ -206,6 +207,10 @@ class Router:
                     pass
                 except AllQueuesFullError:
                     logger.warning(f"All peer queues full, dropping message on {topic}")
+                except MessageTooLargeError:
+                    logger.warning(
+                        f"Message too large for gossipsub on {topic} ({len(data)} bytes), dropping"
+                    )
 
 
 def get_node_id_keypair(