Add authorized pipe messages over Iroh

AGENTS.md

@@ -176,8 +176,11 @@ Roadmap items should be actionable and checkable:
   `geth pipe connect <name> --node <node-id>` uses the protected Iroh control
   ALPN and requires `pipe.connect` on `resource:pipe:<name>`. Remote
   `geth pipe listen <name> --node <node-id>` requires `pipe.listen` on the same
-  resource before creating a daemon-lifetime listener on the peer. Iroh byte
+  resource before creating a daemon-lifetime listener on the peer. `geth pipe
-  streams and TCP/Unix forwarding are still roadmap work.
+  send <name> <message> --node <node-id>` carries a byte message over the
+  dedicated `/geth/pipe/1` ALPN when `pipe.connect` is authorized, and
+  `geth pipe recv <name>` drains local daemon-lifetime messages. Long-lived
+  bidirectional streams and TCP/Unix forwarding are still roadmap work.
 - `geth ssh proxy <node-id>` performs an authorized control-plane handshake over
   the protected Iroh control ALPN and requires `ssh_proxy.connect` on
   `resource:ssh-proxy:local` before returning proxy metadata. It does not carry

Cargo.lock

@@ -1087,11 +1087,13 @@ name = "geth-cli"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "base64",
  "clap",
  "geth-cas",
  "geth-config",
  "geth-control",
  "geth-node",
+ "geth-pipe",
  "serde_json",
  "tokio",
 ]

README.md

@@ -142,9 +142,11 @@ The bootstrap implementation provides:
   - `geth ssh revocation import <path> [--format jsonl|openssh-krl-spec] [--subject <principal>]`
   - `geth ssh revocation sync <node-id> [--bearer-secret <secret>]`
 - SSH proxy authorization probe: `geth ssh proxy <node-id> [--bearer-secret <secret>]`
-- pipe registry/connect commands:
+- pipe registry/message commands:
-  `geth pipe listen <name> [--node <node-id>] [--bearer-secret <secret>]` and
+  `geth pipe listen <name> [--node <node-id>] [--bearer-secret <secret>]`,
-  `geth pipe connect <name> [--node <node-id>] [--bearer-secret <secret>]`
+  `geth pipe connect <name> [--node <node-id>] [--bearer-secret <secret>]`,
+  `geth pipe send <name> <message> [--node <node-id>] [--bearer-secret <secret>]`,
+  and `geth pipe recv <name> [--peek]`
 
 `geth peer export/import/list` is for untrusted peer-card exchange. Peer cards
 include the Iroh EndpointID plus currently known relay/direct addresses.
@@ -198,12 +200,15 @@ requires `pubsub.subscribe` on `resource:pubsub:<topic>` before returning the
 peer's current daemon-lifetime snapshot for that topic.
 Remote pipe connect uses the same protected Iroh control path and requires
 `pipe.connect` on `resource:pipe:<name>`. The current prototype records a remote
-connection attempt and whether a listener exists; byte streaming and forwarding
+connection attempt and whether a listener exists. `geth pipe send <name>
-are still future work.
+<message> --node <node-id>` uses the dedicated `/geth/pipe/1` Iroh ALPN to
+write a byte message to an authorized peer listener, and `geth pipe recv <name>`
+drains local daemon-lifetime messages.
 Remote pipe listen uses the same protected path:
 `geth pipe listen <name> --node <node-id>` requires `pipe.listen` on
 `resource:pipe:<name>` before registering a daemon-lifetime listener on the
-peer.
+peer. Long-lived stdin/stdout streaming and socket forwarding are still future
+work.
 `geth ssh proxy <node-id>` also uses the protected Iroh control path. The remote
 peer validates the caller's endpoint/card binding and requires
 `ssh_proxy.connect` on `resource:ssh-proxy:local` before returning proxy

crates/geth-cli/Cargo.toml

@@ -7,6 +7,7 @@ license.workspace = true
 
 [dependencies]
 anyhow.workspace = true
+base64.workspace = true
 clap.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
@@ -14,3 +15,4 @@ geth-config = { path = "../geth-config" }
 geth-control = { path = "../geth-control" }
 geth-cas = { path = "../geth-cas" }
 geth-node = { path = "../geth-node" }
+geth-pipe = { path = "../geth-pipe" }

crates/geth-cli/src/lib.rs

@@ -1,4 +1,5 @@
 use anyhow::{Context, Result, bail};
+use base64::Engine;
 use clap::{Args, Parser, Subcommand};
 use geth_config::GethPaths;
 use geth_control::{ControlRequest, ControlResponse};
@@ -395,6 +396,19 @@ pub enum PipeCommand {
         #[arg(long)]
         bearer_secret: Option<String>,
     },
+    Send {
+        target: String,
+        message: String,
+        #[arg(long)]
+        node: Option<String>,
+        #[arg(long)]
+        bearer_secret: Option<String>,
+    },
+    Recv {
+        name: String,
+        #[arg(long)]
+        peek: bool,
+    },
 }
 
 #[derive(Debug, Subcommand)]
@@ -863,6 +877,18 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
                 node,
                 bearer_secret,
             },
+            PipeCommand::Send {
+                target,
+                message,
+                node,
+                bearer_secret,
+            } => ControlRequest::PipeSend {
+                target,
+                data_base64: base64::engine::general_purpose::STANDARD.encode(message.as_bytes()),
+                node,
+                bearer_secret,
+            },
+            PipeCommand::Recv { name, peek } => ControlRequest::PipeRecv { name, peek },
         },
         Command::Db { command } => match command {
             DbCommand::Add { name, path } => ControlRequest::DbAdd { name, path },
@@ -1951,6 +1977,55 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
             println!("reason: {reason}");
             println!("note: {note}");
         }
+        ControlResponse::PipeSent {
+            message,
+            listener_found,
+            note,
+        } => {
+            println!("listener_found: {listener_found}");
+            if let Some(message) = message {
+                println!("pipe: {}", message.pipe);
+                println!("received_at_ms: {}", message.received_at.0);
+                print_pipe_message_data(&message)?;
+            }
+            println!("note: {note}");
+        }
+        ControlResponse::PipeRemoteSent {
+            peer_node_id,
+            peer_agent_id,
+            endpoint_id,
+            message,
+            listener_found,
+            allowed,
+            reason,
+            note,
+        } => {
+            println!("peer: {peer_node_id}");
+            println!("agent: {peer_agent_id}");
+            println!("endpoint: {endpoint_id}");
+            println!("allowed: {allowed}");
+            println!("listener_found: {listener_found}");
+            if let Some(message) = message {
+                println!("pipe: {}", message.pipe);
+                println!("received_at_ms: {}", message.received_at.0);
+            }
+            println!("reason: {reason}");
+            println!("note: {note}");
+        }
+        ControlResponse::PipeMessages {
+            name,
+            messages,
+            drained,
+            note,
+        } => {
+            println!("pipe: {name}");
+            println!("messages: {}", messages.len());
+            println!("drained: {drained}");
+            for message in messages {
+                print_pipe_message_data(&message)?;
+            }
+            println!("note: {note}");
+        }
         ControlResponse::SshProxyConnected {
             peer_node_id,
             peer_agent_id,
@@ -2051,6 +2126,20 @@ fn print_file_conflict(conflict: &geth_cas::FileConflict) {
     }
 }
 
+fn print_pipe_message_data(message: &geth_pipe::PipeMessage) -> Result<()> {
+    let bytes = base64::engine::general_purpose::STANDARD
+        .decode(&message.data_base64)
+        .context("decode pipe message")?;
+    match String::from_utf8(bytes) {
+        Ok(text) => println!("{text}"),
+        Err(error) => println!(
+            "base64:{}",
+            base64::engine::general_purpose::STANDARD.encode(error.into_bytes())
+        ),
+    }
+    Ok(())
+}
+
 fn shell_quote_command(command: &[String]) -> String {
     command
         .iter()

crates/geth-control/src/lib.rs

@@ -5,7 +5,7 @@ use geth_discovery::{DiscoveredPeer, PeerCard};
 use geth_document::{DocumentResource, DocumentState};
 use geth_keychain::{KeychainOp, KeychainOpSignature};
 use geth_kv::{KvEntry, KvResource, KvSyncEntry};
-use geth_pipe::{PipeConnection, PipeListener};
+use geth_pipe::{PipeConnection, PipeListener, PipeMessage};
 use geth_pubsub::PubsubMessage;
 use geth_resource::ResourceDescriptor;
 use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSecret};
@@ -298,6 +298,16 @@ pub enum ControlRequest {
         node: Option<String>,
         bearer_secret: Option<String>,
     },
+    PipeSend {
+        target: String,
+        data_base64: String,
+        node: Option<String>,
+        bearer_secret: Option<String>,
+    },
+    PipeRecv {
+        name: String,
+        peek: bool,
+    },
     ModuleStub {
         module: String,
         command: String,
@@ -632,6 +642,27 @@ pub enum ControlResponse {
         reason: String,
         note: String,
     },
+    PipeSent {
+        message: Option<PipeMessage>,
+        listener_found: bool,
+        note: String,
+    },
+    PipeRemoteSent {
+        peer_node_id: String,
+        peer_agent_id: String,
+        endpoint_id: String,
+        message: Option<PipeMessage>,
+        listener_found: bool,
+        allowed: bool,
+        reason: String,
+        note: String,
+    },
+    PipeMessages {
+        name: String,
+        messages: Vec<PipeMessage>,
+        drained: bool,
+        note: String,
+    },
     SshProxyConnected {
         peer_node_id: String,
         peer_agent_id: String,
@@ -988,6 +1019,39 @@ pub enum PeerControlResponse {
     },
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "kebab-case")]
+pub enum PipeWireRequest {
+    Send {
+        peer_card: PeerCard,
+        target: String,
+        data_base64: String,
+        nonce: String,
+        bearer_proof: Option<BearerProof>,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "kebab-case")]
+pub enum PipeWireResponse {
+    Sent {
+        node_id: String,
+        agent_id: String,
+        endpoint_id: String,
+        remote_endpoint_id: String,
+        message: Box<Option<PipeMessage>>,
+        listener_found: bool,
+        allowed: bool,
+        reason: String,
+        evaluated_ops: usize,
+        nonce: String,
+        note: String,
+    },
+    Error {
+        message: String,
+    },
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct SyncWatermark {
     pub stream: String,
@@ -1040,6 +1104,26 @@ pub fn decode_peer_response(line: &str) -> Result<PeerControlResponse, ControlEr
     serde_json::from_str(line).map_err(ControlError::from)
 }
 
+pub fn encode_pipe_wire_request(request: &PipeWireRequest) -> Result<String, ControlError> {
+    let mut line = serde_json::to_string(request)?;
+    line.push('\n');
+    Ok(line)
+}
+
+pub fn decode_pipe_wire_request(line: &str) -> Result<PipeWireRequest, ControlError> {
+    serde_json::from_str(line).map_err(ControlError::from)
+}
+
+pub fn encode_pipe_wire_response(response: &PipeWireResponse) -> Result<String, ControlError> {
+    let mut line = serde_json::to_string(response)?;
+    line.push('\n');
+    Ok(line)
+}
+
+pub fn decode_pipe_wire_response(line: &str) -> Result<PipeWireResponse, ControlError> {
+    serde_json::from_str(line).map_err(ControlError::from)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1260,6 +1344,43 @@ mod tests {
             request
         );
 
+        let request = ControlRequest::PipeSend {
+            target: "inbox".to_owned(),
+            data_base64: "aGVsbG8=".to_owned(),
+            node: Some("node:peer".to_owned()),
+            bearer_secret: None,
+        };
+        assert_eq!(
+            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
+            request
+        );
+
+        let request = ControlRequest::PipeRecv {
+            name: "inbox".to_owned(),
+            peek: true,
+        };
+        assert_eq!(
+            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
+            request
+        );
+
+        let response = ControlResponse::PipeMessages {
+            name: "inbox".to_owned(),
+            messages: vec![PipeMessage {
+                pipe: "inbox".to_owned(),
+                data_base64: "aGVsbG8=".to_owned(),
+                received_at: geth_types::UnixMillis(1),
+                source_node: Some("node:peer".to_owned()),
+                note: "pipe".to_owned(),
+            }],
+            drained: false,
+            note: "pipe messages".to_owned(),
+        };
+        assert_eq!(
+            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
+            response
+        );
+
         let request = ControlRequest::PubsubPub {
             topic: "presence/test".to_owned(),
             message: "online".to_owned(),
@@ -1854,6 +1975,55 @@ mod tests {
             request
         );
 
+        let request = PipeWireRequest::Send {
+            peer_card: PeerCard {
+                node_id: "node:caller".into(),
+                agent_id: "agent:caller".into(),
+                endpoints: Vec::new(),
+                issued_at: geth_types::UnixMillis(1),
+                signature: geth_discovery::SignatureMetadata {
+                    namespace: "geth.peer-card.v1@geth.local".to_owned(),
+                    signer: "agent:caller".to_owned(),
+                    public_key: "key".to_owned(),
+                    signature: "sig".to_owned(),
+                },
+            },
+            target: "inbox".to_owned(),
+            data_base64: "aGVsbG8=".to_owned(),
+            nonce: "nonce".to_owned(),
+            bearer_proof: None,
+        };
+        assert_eq!(
+            decode_pipe_wire_request(&encode_pipe_wire_request(&request).expect("encode"))
+                .expect("decode"),
+            request
+        );
+
+        let response = PipeWireResponse::Sent {
+            node_id: "node:peer".to_owned(),
+            agent_id: "agent:peer".to_owned(),
+            endpoint_id: "endpoint:peer".to_owned(),
+            remote_endpoint_id: "endpoint:caller".to_owned(),
+            message: Box::new(Some(PipeMessage {
+                pipe: "inbox".to_owned(),
+                data_base64: "aGVsbG8=".to_owned(),
+                received_at: geth_types::UnixMillis(1),
+                source_node: Some("node:caller".to_owned()),
+                note: "pipe".to_owned(),
+            })),
+            listener_found: true,
+            allowed: true,
+            reason: "direct grant".to_owned(),
+            evaluated_ops: 1,
+            nonce: "nonce".to_owned(),
+            note: "pipe wire".to_owned(),
+        };
+        assert_eq!(
+            decode_pipe_wire_response(&encode_pipe_wire_response(&response).expect("encode"))
+                .expect("decode"),
+            response
+        );
+
         let response = PeerControlResponse::SshProxyConnected {
             node_id: "node:peer".to_owned(),
             agent_id: "agent:peer".to_owned(),

crates/geth-node/src/lib.rs

@@ -9,7 +9,8 @@ use geth_cas::{
 use geth_config::{GethConfig, GethPaths, RelayMode};
 use geth_control::{
     CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse, NodeIdResponse,
-    PeerControlRequest, PeerControlResponse, StatusResponse, SyncWatermark,
+    PeerControlRequest, PeerControlResponse, PipeWireRequest, PipeWireResponse, StatusResponse,
+    SyncWatermark,
 };
 use geth_crypto::AgentKey;
 use geth_db::DbResource;
@@ -21,7 +22,7 @@ use geth_document::{DocumentResource, DocumentState};
 use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode};
 use geth_keychain::{KeychainOp, KeychainOpKind, KeychainOpSignature};
 use geth_kv::{KvEntry, KvResource, KvSyncEntry};
-use geth_pipe::{PipeConnection, PipeListener};
+use geth_pipe::{PipeConnection, PipeListener, PipeMessage};
 use geth_pubsub::PubsubMessage;
 use geth_resource::ResourceDescriptor;
 use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSecret};
@@ -150,10 +151,12 @@ struct PubsubRuntime {
 struct PipeRuntime {
     listeners: BTreeMap<String, PipeListener>,
     connections: VecDeque<PipeConnection>,
+    messages: VecDeque<PipeMessage>,
 }
 
 const PUBSUB_RING_LIMIT: usize = 256;
 const PIPE_CONNECTION_RING_LIMIT: usize = 256;
+const PIPE_MESSAGE_RING_LIMIT: usize = 1024;
 
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 struct LiveSyncCursor {
@@ -307,6 +310,12 @@ pub async fn handle_request_async(
             node: Some(peer_node),
             bearer_secret,
         } => pipe_connect_to_peer(node, &peer_node, target, bearer_secret).await,
+        ControlRequest::PipeSend {
+            target,
+            data_base64,
+            node: Some(peer_node),
+            bearer_secret,
+        } => pipe_send_to_peer(node, &peer_node, target, data_base64, bearer_secret).await,
         ControlRequest::SshProxyConnect {
             node: peer_node,
             bearer_secret,
@@ -1472,6 +1481,57 @@ async fn pipe_listen_on_peer(
     }
 }
 
+async fn pipe_send_to_peer(
+    node: &LocalNode,
+    peer_node: &str,
+    target: String,
+    data_base64: String,
+    bearer_secret: Option<String>,
+) -> Result<ControlResponse, NodeError> {
+    geth_pipe::validate_pipe_name(&target)?;
+    base64::engine::general_purpose::STANDARD
+        .decode(&data_base64)
+        .map_err(|error| NodeError::IrohPeer(format!("invalid pipe base64 payload: {error}")))?;
+    let response = request_pipe_wire(node, peer_node, "pipe-send", |peer_card, nonce| {
+        PipeWireRequest::Send {
+            peer_card,
+            target: target.clone(),
+            data_base64: data_base64.clone(),
+            nonce: nonce.clone(),
+            bearer_proof: bearer_proof(
+                bearer_secret,
+                &format!("resource:pipe:{target}"),
+                "pipe.connect",
+                &nonce,
+            ),
+        }
+    })
+    .await?;
+    match response {
+        PipeWireResponse::Sent {
+            node_id,
+            agent_id,
+            endpoint_id,
+            message,
+            listener_found,
+            allowed,
+            reason,
+            note,
+            ..
+        } => Ok(ControlResponse::PipeRemoteSent {
+            peer_node_id: node_id,
+            peer_agent_id: agent_id,
+            endpoint_id,
+            message: *message,
+            listener_found,
+            allowed,
+            reason,
+            note,
+        }),
+        PipeWireResponse::Error { message } => Err(NodeError::IrohPeer(message)),
+    }
+}
+
 async fn ssh_proxy_connect_to_peer(
     node: &LocalNode,
     peer_node: &str,
@@ -2125,6 +2185,75 @@ async fn request_peer_control(
     }
 }
 
+async fn request_pipe_wire(
+    node: &LocalNode,
+    peer_node: &str,
+    operation: &str,
+    build_request: impl FnOnce(PeerCard, String) -> PipeWireRequest,
+) -> Result<PipeWireResponse, NodeError> {
+    let store = Store::open(&node.paths.metadata_db())?;
+    let stored = store
+        .get_peer_card(peer_node)?
+        .ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?;
+    let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
+    peer_card.validate_candidate()?;
+    let candidate = peer_card
+        .endpoints
+        .first()
+        .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
+    ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
+    let node_addr = iroh_node_addr_from_candidate(candidate)?;
+    let endpoint = node
+        .iroh_endpoint
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)?
+        .clone()
+        .ok_or(NodeError::IrohEndpointUnavailable)?;
+    let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?;
+    let nonce = geth_crypto::blake3_hex(
+        format!(
+            "{}\0{}\0{}\0{}",
+            node.node_id,
+            peer_node,
+            operation,
+            geth_store::now_ms()
+        )
+        .as_bytes(),
+    );
+    let request = build_request(self_card, nonce.clone());
+    let conn = endpoint
+        .endpoint()
+        .connect(node_addr, geth_iroh::ALPN_PIPE)
+        .await
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    let (mut send, mut recv) = conn
+        .open_bi()
+        .await
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    send.write_all(geth_control::encode_pipe_wire_request(&request)?.as_bytes())
+        .await
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    send.finish()
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    let bytes = recv
+        .read_to_end(16 * 1024 * 1024)
+        .await
+        .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    let text =
+        std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    let response = geth_control::decode_pipe_wire_response(text)?;
+    match &response {
+        PipeWireResponse::Sent {
+            nonce: response_nonce,
+            ..
+        } if response_nonce == &nonce => Ok(response),
+        PipeWireResponse::Error { .. } => Ok(response),
+        _ => Err(NodeError::IrohPeer(format!(
+            "peer {operation} response did not match request"
+        ))),
+    }
+}
+
 fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
     let raw_endpoint = endpoint.endpoint();
     tokio::spawn(async move {
@@ -2279,6 +2408,15 @@ async fn handle_iroh_control_connection(
         .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
     let text =
         std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+    if alpn == display_alpn(geth_iroh::ALPN_PIPE.to_vec()) {
+        let response = handle_pipe_wire_request(&node, &remote_endpoint_id, text)?;
+        send.write_all(geth_control::encode_pipe_wire_response(&response)?.as_bytes())
+            .await
+            .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+        send.finish()
+            .map_err(|error| NodeError::IrohPeer(error.to_string()))?;
+        return Ok(());
+    }
     let response = match geth_control::decode_peer_request(text)? {
         PeerControlRequest::Ping { peer_card, nonce } => {
             peer_card.validate_candidate()?;
@@ -2818,7 +2956,7 @@ async fn handle_iroh_control_connection(
                 Some(record_pipe_connection(
                     &node,
                     target,
-                    "remote pipe connect over protected Iroh control path; byte streams are not implemented yet".to_owned(),
+                    "remote pipe connect over protected Iroh control path; byte messages use the dedicated Iroh pipe ALPN".to_owned(),
                 )?)
             } else {
                 None
@@ -2833,7 +2971,7 @@ async fn handle_iroh_control_connection(
                 reason: explanation.reason,
                 evaluated_ops: explanation.evaluated_ops,
                 nonce,
-                note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte streams are not implemented yet".to_owned(),
+                note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte messages use the dedicated Iroh pipe ALPN".to_owned(),
             }
         }
         PeerControlRequest::PipeListen {
@@ -2881,7 +3019,7 @@ async fn handle_iroh_control_connection(
                 reason: explanation.reason,
                 evaluated_ops: explanation.evaluated_ops,
                 nonce,
-                note: "pipe listen authenticated endpoint/card binding and required pipe.listen on the remote pipe resource; byte streams are not implemented yet".to_owned(),
+                note: "pipe listen authenticated endpoint/card binding and required pipe.listen on the remote pipe resource; listeners receive daemon-lifetime byte messages".to_owned(),
             }
         }
         PeerControlRequest::SshProxyConnect {
@@ -3078,6 +3216,74 @@ async fn handle_iroh_control_connection(
     Ok(())
 }
 
+fn handle_pipe_wire_request(
+    node: &LocalNode,
+    remote_endpoint_id: &str,
+    text: &str,
+) -> Result<PipeWireResponse, NodeError> {
+    match geth_control::decode_pipe_wire_request(text)? {
+        PipeWireRequest::Send {
+            peer_card,
+            target,
+            data_base64,
+            nonce,
+            bearer_proof,
+        } => {
+            geth_pipe::validate_pipe_name(&target)?;
+            base64::engine::general_purpose::STANDARD
+                .decode(&data_base64)
+                .map_err(|error| NodeError::IrohPeer(format!("invalid pipe payload: {error}")))?;
+            peer_card.validate_candidate()?;
+            ensure_peer_card_matches_endpoint(&peer_card, remote_endpoint_id)?;
+            let discovered = DiscoveredPeer::candidate(
+                peer_card.clone(),
+                UnixMillis(geth_store::now_ms()),
+                DiscoverySource::PeerExchange,
+            )?;
+            let store = Store::open(&node.paths.metadata_db())?;
+            store.upsert_peer_card(&StoredPeerCard {
+                peer_id: peer_card.node_id.to_string(),
+                card_json: serde_json::to_string(&peer_card)?,
+                updated_at_ms: discovered.discovered_at.0,
+            })?;
+            let resource = format!("resource:pipe:{target}");
+            let capability = "pipe.connect".to_owned();
+            let explanation = explain_peer_or_bearer(
+                &store,
+                peer_card.node_id.as_str(),
+                &resource,
+                &capability,
+                &nonce,
+                bearer_proof.as_ref(),
+            )?;
+            let (message, listener_found) = if explanation.allowed {
+                record_pipe_message(
+                    node,
+                    target,
+                    data_base64,
+                    Some(peer_card.node_id.to_string()),
+                    "remote pipe byte message over dedicated Iroh pipe ALPN".to_owned(),
+                )?
+            } else {
+                (None, false)
+            };
+            Ok(PipeWireResponse::Sent {
+                node_id: node.node_id.clone(),
+                agent_id: node.agent_id.clone(),
+                endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
+                remote_endpoint_id: remote_endpoint_id.to_owned(),
+                message: Box::new(message),
+                listener_found,
+                allowed: explanation.allowed,
+                reason: explanation.reason,
+                evaluated_ops: explanation.evaluated_ops,
+                nonce,
+                note: "pipe send authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; payload used the dedicated Iroh pipe ALPN".to_owned(),
+            })
+        }
+    }
+}
+
 fn iroh_node_addr_from_candidate(
     candidate: &EndpointCandidate,
 ) -> Result<iroh::NodeAddr, NodeError> {
@@ -4389,6 +4595,40 @@ pub fn handle_request(
         ControlRequest::PipeConnect { node: Some(_), .. } => {
             Err(NodeError::IrohEndpointUnavailable)
         }
+        ControlRequest::PipeSend {
+            target,
+            data_base64,
+            node: None,
+            ..
+        } => {
+            geth_pipe::validate_pipe_name(&target)?;
+            base64::engine::general_purpose::STANDARD
+                .decode(&data_base64)
+                .map_err(|error| NodeError::IrohPeer(format!("invalid pipe payload: {error}")))?;
+            let (message, listener_found) = record_pipe_message(
+                node,
+                target,
+                data_base64,
+                Some(node.node_id.clone()),
+                "local pipe byte message".to_owned(),
+            )?;
+            Ok(ControlResponse::PipeSent {
+                message,
+                listener_found,
+                note: geth_pipe::local_pipe_runtime_note().to_owned(),
+            })
+        }
+        ControlRequest::PipeSend { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
+        ControlRequest::PipeRecv { name, peek } => {
+            geth_pipe::validate_pipe_name(&name)?;
+            let messages = pipe_messages(node, &name, !peek)?;
+            Ok(ControlResponse::PipeMessages {
+                name,
+                messages,
+                drained: !peek,
+                note: geth_pipe::local_pipe_runtime_note().to_owned(),
+            })
+        }
         ControlRequest::SshProxyConnect { .. } => Err(NodeError::IrohEndpointUnavailable),
         ControlRequest::ModuleStub { module, command } => {
             Ok(ControlResponse::NotImplemented { module, command })
@@ -4562,6 +4802,54 @@ fn record_pipe_connection(
     Ok(connection)
 }
 
+fn record_pipe_message(
+    node: &LocalNode,
+    target: String,
+    data_base64: String,
+    source_node: Option<String>,
+    note: String,
+) -> Result<(Option<PipeMessage>, bool), NodeError> {
+    let mut runtime = node
+        .runtime
+        .pipes
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+    let listener_found = runtime.listeners.contains_key(&target);
+    if !listener_found {
+        return Ok((None, false));
+    }
+    let message = PipeMessage {
+        pipe: target,
+        data_base64,
+        received_at: UnixMillis(geth_store::now_ms()),
+        source_node,
+        note,
+    };
+    runtime.messages.push_back(message.clone());
+    while runtime.messages.len() > PIPE_MESSAGE_RING_LIMIT {
+        runtime.messages.pop_front();
+    }
+    Ok((Some(message), true))
+}
+
+fn pipe_messages(node: &LocalNode, name: &str, drain: bool) -> Result<Vec<PipeMessage>, NodeError> {
+    let mut runtime = node
+        .runtime
+        .pipes
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+    let messages = runtime
+        .messages
+        .iter()
+        .filter(|message| message.pipe == name)
+        .cloned()
+        .collect::<Vec<_>>();
+    if drain {
+        runtime.messages.retain(|message| message.pipe != name);
+    }
+    Ok(messages)
+}
+
 fn ensure_local_document(store: &Store, name: &str) -> Result<StoredDocumentResource, NodeError> {
     if let Some(document) = store.get_document_resource_by_name(name)? {
         return Ok(document);
@@ -6554,11 +6842,64 @@ mod tests {
                 assert!(allowed);
                 assert!(connection.local_listener_found);
                 assert!(reason.contains("direct grant"));
-                assert!(note.contains("byte streams are not implemented yet"));
+                assert!(note.contains("pipe.connect"));
             }
             other => panic!("unexpected allowed pipe connect response: {other:?}"),
         }
 
+        let remote_pipe_send = handle_request_async(
+            &left,
+            ControlRequest::PipeSend {
+                target: "inbox".to_owned(),
+                data_base64: "aGVsbG8gZnJvbSBsZWZ0".to_owned(),
+                node: Some(right_card.node_id.to_string()),
+                bearer_secret: None,
+            },
+        )
+        .await
+        .expect("allowed remote pipe send");
+        match remote_pipe_send {
+            ControlResponse::PipeRemoteSent {
+                allowed,
+                listener_found,
+                message,
+                reason,
+                note,
+                ..
+            } => {
+                assert!(allowed);
+                assert!(listener_found);
+                let message = message.expect("remote pipe message");
+                assert_eq!(message.pipe, "inbox");
+                assert_eq!(message.data_base64, "aGVsbG8gZnJvbSBsZWZ0");
+                assert!(reason.contains("direct grant"));
+                assert!(note.contains("Iroh pipe ALPN"));
+            }
+            other => panic!("unexpected allowed pipe send response: {other:?}"),
+        }
+        let right_messages = handle_request(
+            &right,
+            ControlRequest::PipeRecv {
+                name: "inbox".to_owned(),
+                peek: false,
+            },
+        )
+        .expect("right pipe recv after remote send");
+        match right_messages {
+            ControlResponse::PipeMessages {
+                messages, drained, ..
+            } => {
+                assert!(drained);
+                assert_eq!(messages.len(), 1);
+                assert_eq!(messages[0].data_base64, "aGVsbG8gZnJvbSBsZWZ0");
+                assert_eq!(
+                    messages[0].source_node.as_deref(),
+                    Some(left.node_id.as_str())
+                );
+            }
+            other => panic!("unexpected right pipe messages response: {other:?}"),
+        }
+
         let remote_pipe_listen = handle_request_async(
             &left,
             ControlRequest::PipeListen {

crates/geth-pipe/src/lib.rs

@@ -24,6 +24,15 @@ pub struct PipeConnection {
     pub note: String,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PipeMessage {
+    pub pipe: String,
+    pub data_base64: String,
+    pub received_at: UnixMillis,
+    pub source_node: Option<String>,
+    pub note: String,
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum PipeError {
     #[error("invalid pipe name or target: {0}")]
@@ -48,7 +57,7 @@ pub fn pipe_roadmap() -> &'static str {
 
 #[must_use]
 pub fn local_pipe_runtime_note() -> &'static str {
-    "local daemon registry only; byte streams and Iroh transport are not implemented yet"
+    "daemon-lifetime pipe runtime; byte messages are buffered locally and remote writes use the Iroh pipe ALPN"
 }
 
 #[cfg(test)]

crates/geth/tests/bootstrap.rs

@@ -1543,7 +1543,7 @@ fn pipe_listen_connect_uses_local_runtime_registry() {
         geth_control::ControlResponse::PipeListening { listener } => {
             assert_eq!(listener.id.to_string(), "pipe:inbox");
             assert_eq!(listener.name, "inbox");
-            assert!(listener.note.contains("local daemon registry"));
+            assert!(listener.note.contains("daemon-lifetime pipe runtime"));
         }
         other => panic!("unexpected response: {other:?}"),
     }
@@ -1561,11 +1561,63 @@ fn pipe_listen_connect_uses_local_runtime_registry() {
         geth_control::ControlResponse::PipeConnected { connection } => {
             assert_eq!(connection.target, "inbox");
             assert!(connection.local_listener_found);
-            assert!(
+            assert!(connection.note.contains("pipe runtime"));
-                connection
+        }
-                    .note
+        other => panic!("unexpected response: {other:?}"),
-                    .contains("Iroh transport are not implemented")
+    }
-            );
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::PipeSend {
+            target: "inbox".to_owned(),
+            data_base64: "aGVsbG8gcGlwZQ==".to_owned(),
+            node: None,
+            bearer_secret: None,
+        },
+    )
+    .expect("send pipe message");
+    match response {
+        geth_control::ControlResponse::PipeSent {
+            message,
+            listener_found,
+            ..
+        } => {
+            assert!(listener_found);
+            assert_eq!(message.expect("message").data_base64, "aGVsbG8gcGlwZQ==");
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::PipeRecv {
+            name: "inbox".to_owned(),
+            peek: false,
+        },
+    )
+    .expect("receive pipe messages");
+    match response {
+        geth_control::ControlResponse::PipeMessages {
+            messages, drained, ..
+        } => {
+            assert!(drained);
+            assert_eq!(messages.len(), 1);
+            assert_eq!(messages[0].data_base64, "aGVsbG8gcGlwZQ==");
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::PipeRecv {
+            name: "inbox".to_owned(),
+            peek: false,
+        },
+    )
+    .expect("receive drained pipe messages");
+    match response {
+        geth_control::ControlResponse::PipeMessages { messages, .. } => {
+            assert!(messages.is_empty());
         }
         other => panic!("unexpected response: {other:?}"),
     }

docs/architecture.md

@@ -203,15 +203,18 @@ authorized peer's current snapshot for that topic over the same protected path
 when the caller has `pubsub.subscribe` on `resource:pubsub:<topic>`. Iroh-gossip
 replication and private topics are future work.
 
-`geth-pipe` currently supports `pipe listen/connect` against a daemon-lifetime
+`geth-pipe` currently supports `pipe listen/connect/send/recv` against a
-registry. `geth pipe connect <name> --node <node-id>` sends an authorized remote
+daemon-lifetime runtime. `geth pipe connect <name> --node <node-id>` sends an
-connect request over the protected Iroh control ALPN. The remote daemon validates
+authorized remote connect request over the protected Iroh control ALPN. The
-endpoint/card binding and requires `pipe.connect` on `resource:pipe:<name>`
+remote daemon validates endpoint/card binding and requires `pipe.connect` on
-before recording the connection attempt and reporting whether a listener exists.
+`resource:pipe:<name>` before recording the connection attempt and reporting
-`geth pipe listen <name> --node <node-id>` can also ask a peer to register a
+whether a listener exists. `geth pipe send <name> <message> --node <node-id>`
-daemon-lifetime listener after checking `pipe.listen` on the same resource. This
+uses the dedicated `/geth/pipe/1` ALPN to write a byte message to a peer
-is still a control-plane scaffold for names, listeners, and connection attempts
+listener after the same endpoint/card and capability checks. `geth pipe recv
-only; it does not carry bytes or forward sockets yet.
+<name>` drains local daemon-lifetime messages. `geth pipe listen <name> --node
+<node-id>` can also ask a peer to register a daemon-lifetime listener after
+checking `pipe.listen` on the same resource. Long-lived stdin/stdout streams and
+TCP/Unix forwarding are still future work.
 
 `geth-ssh-proxy` currently defines proxy target and connection metadata. The
 daemon can authorize a remote proxy attempt over the protected Iroh control ALPN

docs/roadmap.md

@@ -364,6 +364,11 @@ Goal: add authorized stream-oriented management workflows over Iroh.
   - `[x]` Remote pipe listen requires `pipe.listen` on
     `resource:pipe:<name>`.
   - `[x]` Tests cover denied and allowed remote listener registration.
+  - `[x]` `geth pipe send <name> <message> --node <node-id>` carries a byte
+    message over the dedicated `/geth/pipe/1` Iroh ALPN.
+  - `[x]` `geth pipe recv <name>` drains daemon-lifetime pipe messages.
+  - `[x]` Remote pipe send requires `pipe.connect` on
+    `resource:pipe:<name>`.
   - `[ ]` Pipe connect carries bidirectional byte streams over Iroh.
   - `[ ]` Streams close cleanly and propagate errors.