[codex] Add TCP JSONL stream framing for cross-process modules

Cargo.toml

@@ -51,7 +51,7 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sha2 = "0.10"
-tokio = { version = "1.43", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
+tokio = { version = "1.43", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 

README.md

@@ -452,6 +452,14 @@ needs deterministic in-process delivery.
 transport-backed envelope: latest telemetry, health/module state, command state,
 and safety state.
 
+For out-of-process modules, `leash_harness::transport` also exposes a versioned
+TCP JSONL frame contract: `NetworkStreamFrame` with
+`schema_version = "leash-stream-jsonl-v1"`, `stream`, and `payload`. The helper
+functions `send_tcp_jsonl_stream_message` and
+`accept_tcp_jsonl_stream_message` prove one-message loopback delivery without
+changing the default runtime transport. Longer-lived distributed module
+orchestration stays outside this first boundary.
+
 ## Run Logs and Resource Samples
 
 Daemon runs write structured JSONL logs under the Leash state directory. Each
@@ -520,7 +528,8 @@ See [issues](https://github.com/specdog/leash/issues) for the full plan. Highlig
 - [x] Transport abstraction: memory + local async pubsub
 - [x] Stream processing helpers: latest-value backpressure, quality filters, timestamp pairing
 - [x] Agent input channels: one-shot CLI, interactive CLI, and localhost web input
-- [ ] Cross-process and network transports
+- [x] TCP JSONL stream framing for cross-process modules
+- [ ] Long-lived cross-process and network transport orchestration
 - [ ] MAVLink drone + manipulator adapters
 - [x] Localhost command center dashboard
 - [x] Viewer-ready visualization frames

docs/SCHEMAS.md

@@ -6,6 +6,7 @@ Leash publishes JSON Schema for the wire messages external tools consume:
 - MCP HTTP status, tool list, module map, and call responses
 - capability descriptors, safety classes, module graph messages, and adapter messages
 - perception, visualization, planner, patrol, spatial-memory, drone, and manipulator payloads
+- network stream frames for TCP JSONL cross-process module boundaries
 
 The canonical artifact is [schemas/leash-messages.schema.json](../schemas/leash-messages.schema.json).
 It is generated from Rust `serde` + `schemars` types:
@@ -33,3 +34,7 @@ Versioned payloads such as `visualization.version` and manipulator `version`
 stay scoped to that nested payload. A nested payload version bump does not
 require a top-level `schema_version` bump unless the cross-message contract also
 breaks.
+
+`NetworkStreamFrame.schema_version` is scoped to the TCP JSONL stream frame. A
+network frame version bump does not require changing the top-level schema
+artifact version unless the broader message bundle becomes incompatible.

examples/README.md

@@ -6,11 +6,13 @@ This folder contains runnable examples and fixtures for the main operating modes
 flowchart TB
   examples["examples/"] --> bridge["bridge-compat/\nlegacy HTTP/WebSocket bridge shape"]
   examples --> local["local-llm-mcp/\nMCP client setup for local agents"]
+  examples --> network["network-transport/\nTCP JSONL stream frame boundary"]
   examples --> replay["replay/\ndeterministic JSONL fixtures"]
   examples --> ugv["waveshare-ugv/\nphysical Jetson/Waveshare adapter notes"]
 
   bridge --> http["Leash HTTP runtime"]
   local --> mcp["Leash MCP stdio"]
+  network --> tcp["localhost TCP JSONL\nno hardware"]
   replay --> deterministic["Replay profile\nno hardware"]
   ugv --> physical["Physical profile\nexplicit actuation gate"]
 ```
@@ -19,5 +21,6 @@ flowchart TB
 
 - `bridge-compat/`: route compatibility for clients that already speak the robot bridge API.
 - `local-llm-mcp/`: how to connect an MCP-capable local LLM client.
+- `network-transport/`: TCP JSONL stream frame contract for external module processes.
 - `replay/`: checked-in replay fixtures for deterministic observe paths and memory demos.
 - `waveshare-ugv/`: physical adapter notes for the Jetson/Waveshare UGV.

examples/network-transport/README.md

@@ -0,0 +1,24 @@
+# TCP JSONL Stream Frames
+
+Leash runtime streams default to in-process `local-pubsub` or deterministic
+`memory` backends. External module processes can use the smaller TCP JSONL
+boundary without changing those defaults.
+
+Each line is one `NetworkStreamFrame`:
+
+```json
+{"schema_version":"leash-stream-jsonl-v1","stream":"telemetry","payload":{"seq":1}}
+```
+
+The public helpers are:
+
+- `NetworkStreamFrame`
+- `write_network_stream_frame`
+- `read_network_stream_frame`
+- `send_tcp_jsonl_stream_message`
+- `accept_tcp_jsonl_stream_message`
+
+The loopback test in `src/transport.rs` is the executable example: it binds a
+localhost `TcpListener`, sends one `StreamMessage`, and verifies the received
+message matches exactly. This is a framing contract for cross-process modules,
+not a long-lived distributed runtime supervisor.

schemas/leash-messages.schema.json

@@ -2368,6 +2368,25 @@
       "title": "ModuleState",
       "type": "string"
     },
+    "NetworkStreamFrame": {
+      "$schema": "https://json-schema.org/draft/2020-12/schema",
+      "properties": {
+        "payload": true,
+        "schema_version": {
+          "type": "string"
+        },
+        "stream": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "schema_version",
+        "stream",
+        "payload"
+      ],
+      "title": "NetworkStreamFrame",
+      "type": "object"
+    },
     "OccupancyGridFrame": {
       "$defs": {
         "MapMetadata": {

src/bin/leash-schema.rs

@@ -11,6 +11,7 @@ use leash_harness::{
         ModuleGraph, ModuleHealth, ModuleInfo, ModuleState, StreamDescriptor, StreamDirection,
     },
     stack::{AdapterCategory, AdapterMaturity, AdapterProfile},
+    transport::NetworkStreamFrame,
     types::{
         AgentMessage, AgentMessageAck, AgentMessageList, AgentModelResponse, AutonomyOverlay,
         BatteryStatus, CameraStatus, Capabilities, CaptureResult, CommandOverlay,
@@ -142,6 +143,7 @@ fn schema_document() -> Result<Value> {
     insert::<AdapterCategory>(&mut schemas, "AdapterCategory")?;
     insert::<AdapterMaturity>(&mut schemas, "AdapterMaturity")?;
     insert::<AdapterProfile>(&mut schemas, "AdapterProfile")?;
+    insert::<NetworkStreamFrame>(&mut schemas, "NetworkStreamFrame")?;
     insert::<McpToolDescriptor>(&mut schemas, "McpToolDescriptor")?;
     insert::<McpToolList>(&mut schemas, "McpToolList")?;
     insert::<McpCallResponse>(&mut schemas, "McpCallResponse")?;

src/lib.rs

@@ -48,8 +48,10 @@ pub use stream_processing::{
     QualityFilter, RateLimiter, TimestampPair, Timestamped,
 };
 pub use transport::{
-    new_stream_transport, StreamMessage, StreamRecvError, StreamSubscriber, StreamTransport,
-    StreamTransportBackend,
+    accept_tcp_jsonl_stream_message, new_stream_transport, read_network_stream_frame,
+    read_network_stream_message, send_tcp_jsonl_stream_message, write_network_stream_frame,
+    write_network_stream_message, NetworkStreamFrame, StreamMessage, StreamRecvError,
+    StreamSubscriber, StreamTransport, StreamTransportBackend, NETWORK_STREAM_FRAME_VERSION,
 };
 pub use types::{
     AgentModelResponse, AutonomyOverlay, Capabilities, CaptureResult, CommandOverlay, CostmapFrame,

src/transport.rs

@@ -1,19 +1,25 @@
 use std::{
     collections::HashMap,
+    net::SocketAddr,
     sync::{
         atomic::{AtomicBool, Ordering},
         Arc,
     },
 };
 
-use anyhow::{bail, Result};
+use anyhow::{bail, Context, Result};
 use clap::ValueEnum;
 use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use tokio::sync::{broadcast, mpsc};
+use tokio::{
+    io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader},
+    net::{TcpListener, TcpStream},
+    sync::{broadcast, mpsc},
+};
 
 const DEFAULT_CHANNEL_CAPACITY: usize = 128;
+pub const NETWORK_STREAM_FRAME_VERSION: &str = "leash-stream-jsonl-v1";
 
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, ValueEnum)]
 #[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))]
@@ -40,6 +46,55 @@ pub struct StreamMessage {
     pub payload: Value,
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))]
+pub struct NetworkStreamFrame {
+    pub schema_version: String,
+    pub stream: String,
+    pub payload: Value,
+}
+
+impl NetworkStreamFrame {
+    pub fn new(stream: impl Into<String>, payload: Value) -> Self {
+        Self {
+            schema_version: NETWORK_STREAM_FRAME_VERSION.to_string(),
+            stream: stream.into(),
+            payload,
+        }
+    }
+
+    pub fn from_message(message: StreamMessage) -> Self {
+        Self::new(message.stream, message.payload)
+    }
+
+    pub fn into_message(self) -> Result<StreamMessage> {
+        self.validate()?;
+        Ok(StreamMessage {
+            stream: self.stream,
+            payload: self.payload,
+        })
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        if self.schema_version != NETWORK_STREAM_FRAME_VERSION {
+            bail!(
+                "unsupported network stream frame version '{}'",
+                self.schema_version
+            );
+        }
+        if self.stream.trim().is_empty() {
+            bail!("network stream frame stream must not be empty");
+        }
+        Ok(())
+    }
+}
+
+impl From<StreamMessage> for NetworkStreamFrame {
+    fn from(message: StreamMessage) -> Self {
+        Self::from_message(message)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum StreamRecvError {
     Closed,
@@ -80,6 +135,90 @@ pub fn new_stream_transport(backend: StreamTransportBackend) -> Arc<dyn StreamTr
     }
 }
 
+pub async fn write_network_stream_frame<W>(writer: &mut W, frame: &NetworkStreamFrame) -> Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    frame.validate()?;
+    let line = serde_json::to_string(frame).context("serialize network stream frame")?;
+    writer
+        .write_all(line.as_bytes())
+        .await
+        .context("write network stream frame")?;
+    writer
+        .write_all(b"\n")
+        .await
+        .context("write network stream frame newline")?;
+    writer.flush().await.context("flush network stream frame")?;
+    Ok(())
+}
+
+pub async fn read_network_stream_frame<R>(reader: &mut R) -> Result<Option<NetworkStreamFrame>>
+where
+    R: AsyncBufRead + Unpin,
+{
+    let mut line = String::new();
+    let bytes = reader
+        .read_line(&mut line)
+        .await
+        .context("read network stream frame")?;
+    if bytes == 0 {
+        return Ok(None);
+    }
+    let line = line.trim_end_matches(['\r', '\n']);
+    let frame: NetworkStreamFrame =
+        serde_json::from_str(line).context("parse network stream frame")?;
+    frame.validate()?;
+    Ok(Some(frame))
+}
+
+pub async fn write_network_stream_message<W>(writer: &mut W, message: &StreamMessage) -> Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    write_network_stream_frame(
+        writer,
+        &NetworkStreamFrame::new(message.stream.clone(), message.payload.clone()),
+    )
+    .await
+}
+
+pub async fn read_network_stream_message<R>(reader: &mut R) -> Result<Option<StreamMessage>>
+where
+    R: AsyncBufRead + Unpin,
+{
+    read_network_stream_frame(reader)
+        .await?
+        .map(NetworkStreamFrame::into_message)
+        .transpose()
+}
+
+pub async fn send_tcp_jsonl_stream_message(
+    addr: SocketAddr,
+    message: &StreamMessage,
+) -> Result<()> {
+    let mut socket = TcpStream::connect(addr)
+        .await
+        .with_context(|| format!("connect TCP JSONL stream peer at {addr}"))?;
+    write_network_stream_message(&mut socket, message).await?;
+    socket
+        .shutdown()
+        .await
+        .context("shutdown TCP JSONL stream writer")?;
+    Ok(())
+}
+
+pub async fn accept_tcp_jsonl_stream_message(listener: &TcpListener) -> Result<StreamMessage> {
+    let (socket, _) = listener
+        .accept()
+        .await
+        .context("accept TCP JSONL stream peer")?;
+    let mut reader = BufReader::new(socket);
+    read_network_stream_message(&mut reader)
+        .await?
+        .context("TCP JSONL peer closed before sending a stream message")
+}
+
 #[derive(Default)]
 pub struct MemoryTransport {
     streams: Mutex<HashMap<String, Vec<mpsc::UnboundedSender<StreamMessage>>>>,
@@ -200,6 +339,7 @@ impl StreamTransport for LocalPubsubTransport {
 mod tests {
     use super::*;
     use serde_json::json;
+    use tokio::io::BufReader;
 
     #[tokio::test]
     async fn memory_transport_fans_out_to_subscribers() {
@@ -258,6 +398,48 @@ mod tests {
         assert!(transport.subscribe("telemetry").is_err());
     }
 
+    #[tokio::test]
+    async fn network_stream_frame_round_trips_over_tcp_jsonl() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let expected = StreamMessage {
+            stream: "telemetry".to_string(),
+            payload: json!({"seq": 42, "ok": true}),
+        };
+        let server =
+            tokio::spawn(async move { accept_tcp_jsonl_stream_message(&listener).await.unwrap() });
+
+        send_tcp_jsonl_stream_message(addr, &expected)
+            .await
+            .unwrap();
+
+        assert_eq!(server.await.unwrap(), expected);
+    }
+
+    #[tokio::test]
+    async fn network_stream_frame_rejects_unsupported_versions() {
+        let bytes =
+            br#"{"schema_version":"old","stream":"telemetry","payload":{"seq":1}}"#.as_slice();
+        let mut reader = BufReader::new(bytes);
+
+        let err = read_network_stream_frame(&mut reader).await.unwrap_err();
+
+        assert!(err
+            .to_string()
+            .contains("unsupported network stream frame version"));
+    }
+
+    #[tokio::test]
+    async fn network_stream_frame_rejects_empty_stream_names() {
+        let frame = NetworkStreamFrame::new("   ", json!({}));
+
+        let err = frame.validate().unwrap_err();
+
+        assert!(err
+            .to_string()
+            .contains("network stream frame stream must not be empty"));
+    }
+
     #[tokio::test]
     async fn factory_switches_backends() {
         let memory = new_stream_transport(StreamTransportBackend::Memory);