Add local pipe registry commands

crates/geth-cli/src/lib.rs

@@ -462,9 +462,9 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
             PubsubCommand::Pub { topic, message } => ControlRequest::PubsubPub { topic, message },
             PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic },
         },
-        Command::Pipe { command } => ControlRequest::ModuleStub {
-            module: "pipe".to_owned(),
-            command: format!("{command:?}"),
+        Command::Pipe { command } => match command {
+            PipeCommand::Listen { name } => ControlRequest::PipeListen { name },
+            PipeCommand::Connect { target } => ControlRequest::PipeConnect { target },
         },
         Command::Db { command } => match command {
             DbCommand::Add { name, path } => ControlRequest::DbAdd { name, path },
@@ -947,6 +947,18 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
             }
             println!("note: {note}");
         }
+        ControlResponse::PipeListening { listener } => {
+            println!("listening pipe: {}", listener.name);
+            println!("id: {}", listener.id);
+            println!("listened_at_ms: {}", listener.listened_at.0);
+            println!("note: {}", listener.note);
+        }
+        ControlResponse::PipeConnected { connection } => {
+            println!("pipe target: {}", connection.target);
+            println!("local_listener_found: {}", connection.local_listener_found);
+            println!("connected_at_ms: {}", connection.connected_at.0);
+            println!("note: {}", connection.note);
+        }
         ControlResponse::NotImplemented { module, command } => {
             println!("{module} {command}: not implemented yet");
         }

crates/geth-control/Cargo.toml

@@ -14,6 +14,7 @@ geth-db = { path = "../geth-db" }
 geth-document = { path = "../geth-document" }
 geth-keychain = { path = "../geth-keychain" }
 geth-kv = { path = "../geth-kv" }
+geth-pipe = { path = "../geth-pipe" }
 geth-pubsub = { path = "../geth-pubsub" }
 geth-resource = { path = "../geth-resource" }
 geth-secrets = { path = "../geth-secrets" }

crates/geth-control/src/lib.rs

@@ -3,6 +3,7 @@ use geth_db::DbResource;
 use geth_document::{DocumentResource, DocumentState};
 use geth_keychain::KeychainOp;
 use geth_kv::{KvEntry, KvResource};
+use geth_pipe::{PipeConnection, PipeListener};
 use geth_pubsub::PubsubMessage;
 use geth_resource::ResourceDescriptor;
 use geth_secrets::{BearerAccess, ResourceMasterSecret};
@@ -152,6 +153,12 @@ pub enum ControlRequest {
     PubsubSub {
         topic: String,
     },
+    PipeListen {
+        name: String,
+    },
+    PipeConnect {
+        target: String,
+    },
     ModuleStub {
         module: String,
         command: String,
@@ -284,6 +291,12 @@ pub enum ControlResponse {
         messages: Vec<PubsubMessage>,
         note: String,
     },
+    PipeListening {
+        listener: PipeListener,
+    },
+    PipeConnected {
+        connection: PipeConnection,
+    },
     NotImplemented {
         module: String,
         command: String,
@@ -406,5 +419,13 @@ mod tests {
             decode_request(&encode_request(&request).expect("encode")).expect("decode"),
             request
         );
+
+        let request = ControlRequest::PipeListen {
+            name: "inbox".to_owned(),
+        };
+        assert_eq!(
+            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
+            request
+        );
     }
 }

crates/geth-node/Cargo.toml

@@ -20,6 +20,7 @@ geth-document = { path = "../geth-document" }
 geth-iroh = { path = "../geth-iroh" }
 geth-keychain = { path = "../geth-keychain" }
 geth-kv = { path = "../geth-kv" }
+geth-pipe = { path = "../geth-pipe" }
 geth-pubsub = { path = "../geth-pubsub" }
 geth-resource = { path = "../geth-resource" }
 geth-secrets = { path = "../geth-secrets" }

crates/geth-node/src/lib.rs

@@ -13,6 +13,7 @@ use geth_document::{DocumentResource, DocumentState};
 use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode};
 use geth_keychain::{KeychainOp, KeychainOpKind};
 use geth_kv::{KvEntry, KvResource};
+use geth_pipe::{PipeConnection, PipeListener};
 use geth_pubsub::PubsubMessage;
 use geth_resource::ResourceDescriptor;
 use geth_secrets::{BearerAccess, ResourceMasterSecret};
@@ -30,7 +31,7 @@ use geth_types::{
     AuthOpId, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName,
     SshCertId, SshCertRequestId, UnixMillis,
 };
-use std::collections::VecDeque;
+use std::collections::{BTreeMap, VecDeque};
 use std::path::Path;
 use std::sync::{Arc, Mutex};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
@@ -80,6 +81,8 @@ pub enum NodeError {
     Secrets(#[from] geth_secrets::SecretsError),
     #[error("pubsub error: {0}")]
     Pubsub(#[from] geth_pubsub::PubsubError),
+    #[error("pipe error: {0}")]
+    Pipe(#[from] geth_pipe::PipeError),
     #[error("runtime state lock poisoned")]
     RuntimeLockPoisoned,
     #[error("invalid ssh certificate kind: {0}")]
@@ -110,6 +113,7 @@ pub struct LocalNode {
 #[derive(Debug)]
 struct NodeRuntime {
     pubsub: Mutex<PubsubRuntime>,
+    pipes: Mutex<PipeRuntime>,
 }
 
 #[derive(Debug, Default)]
@@ -117,7 +121,14 @@ struct PubsubRuntime {
     messages: VecDeque<PubsubMessage>,
 }
 
+#[derive(Debug, Default)]
+struct PipeRuntime {
+    listeners: BTreeMap<String, PipeListener>,
+    connections: VecDeque<PipeConnection>,
+}
+
 const PUBSUB_RING_LIMIT: usize = 256;
+const PIPE_CONNECTION_RING_LIMIT: usize = 256;
 
 pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
     paths.ensure_base_dirs()?;
@@ -143,6 +154,7 @@ pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
         iroh_status: EndpointStatus::scaffolded(),
         runtime: Arc::new(NodeRuntime {
             pubsub: Mutex::new(PubsubRuntime::default()),
+            pipes: Mutex::new(PipeRuntime::default()),
         }),
     })
 }
@@ -890,6 +902,42 @@ pub fn handle_request(
                 note: geth_pubsub::pubsub_storage_warning().to_owned(),
             })
         }
+        ControlRequest::PipeListen { name } => {
+            geth_pipe::validate_pipe_name(&name)?;
+            let listener = PipeListener {
+                id: format!("pipe:{name}").into(),
+                name: name.clone(),
+                listened_at: UnixMillis(geth_store::now_ms()),
+                note: geth_pipe::local_pipe_runtime_note().to_owned(),
+            };
+            let mut runtime = node
+                .runtime
+                .pipes
+                .lock()
+                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+            runtime.listeners.insert(name, listener.clone());
+            Ok(ControlResponse::PipeListening { listener })
+        }
+        ControlRequest::PipeConnect { target } => {
+            geth_pipe::validate_pipe_name(&target)?;
+            let mut runtime = node
+                .runtime
+                .pipes
+                .lock()
+                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+            let local_listener_found = runtime.listeners.contains_key(&target);
+            let connection = PipeConnection {
+                target,
+                connected_at: UnixMillis(geth_store::now_ms()),
+                local_listener_found,
+                note: geth_pipe::local_pipe_runtime_note().to_owned(),
+            };
+            runtime.connections.push_back(connection.clone());
+            while runtime.connections.len() > PIPE_CONNECTION_RING_LIMIT {
+                runtime.connections.pop_front();
+            }
+            Ok(ControlResponse::PipeConnected { connection })
+        }
         ControlRequest::ModuleStub { module, command } => {
             Ok(ControlResponse::NotImplemented { module, command })
         }

crates/geth-pipe/Cargo.toml

@@ -7,4 +7,5 @@ license.workspace = true
 
 [dependencies]
 serde.workspace = true
+thiserror.workspace = true
 geth-types = { path = "../geth-types" }

crates/geth-pipe/src/lib.rs

@@ -1,4 +1,4 @@
-use geth_types::{PipeId, ResourceId};
+use geth_types::{PipeId, ResourceId, UnixMillis};
 use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -8,7 +8,60 @@ pub struct PipeResource {
     pub name: String,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PipeListener {
+    pub id: PipeId,
+    pub name: String,
+    pub listened_at: UnixMillis,
+    pub note: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PipeConnection {
+    pub target: String,
+    pub connected_at: UnixMillis,
+    pub local_listener_found: bool,
+    pub note: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum PipeError {
+    #[error("invalid pipe name or target: {0}")]
+    InvalidName(String),
+}
+
+pub fn validate_pipe_name(name: &str) -> Result<(), PipeError> {
+    if name.is_empty()
+        || !name
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':'))
+    {
+        return Err(PipeError::InvalidName(name.to_owned()));
+    }
+    Ok(())
+}
+
 #[must_use]
 pub fn pipe_roadmap() -> &'static str {
     "future pipes are authorized Iroh bidirectional streams for stdin/stdout and forwarding"
 }
+
+#[must_use]
+pub fn local_pipe_runtime_note() -> &'static str {
+    "local daemon registry only; byte streams and Iroh transport are not implemented yet"
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pipe_name_validation_rejects_paths_and_empty_names() {
+        assert!(validate_pipe_name("inbox").is_ok());
+        assert!(validate_pipe_name("node:laptop").is_ok());
+        assert!(validate_pipe_name("").is_err());
+        assert!(validate_pipe_name("../inbox").is_err());
+        assert!(validate_pipe_name("inbox/main").is_err());
+        assert!(validate_pipe_name("inbox main").is_err());
+    }
+}

crates/geth/tests/bootstrap.rs

@@ -873,6 +873,74 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() {
     );
 }
 
+#[test]
+fn pipe_listen_connect_uses_local_runtime_registry() {
+    let home = tempfile::tempdir().expect("tempdir");
+    let paths = geth_config::GethPaths::from_home(home.path());
+    let node = geth_node::init_node(&paths).expect("init node");
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::PipeListen {
+            name: "inbox".to_owned(),
+        },
+    )
+    .expect("listen pipe");
+    match response {
+        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"));
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::PipeConnect {
+            target: "inbox".to_owned(),
+        },
+    )
+    .expect("connect pipe");
+    match response {
+        geth_control::ControlResponse::PipeConnected { connection } => {
+            assert_eq!(connection.target, "inbox");
+            assert!(connection.local_listener_found);
+            assert!(
+                connection
+                    .note
+                    .contains("Iroh transport are not implemented")
+            );
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    let reopened = geth_node::open_node(&paths).expect("reopen node");
+    let response = geth_node::handle_request(
+        &reopened,
+        geth_control::ControlRequest::PipeConnect {
+            target: "inbox".to_owned(),
+        },
+    )
+    .expect("connect after reopen");
+    match response {
+        geth_control::ControlResponse::PipeConnected { connection } => {
+            assert!(!connection.local_listener_found);
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    assert!(
+        geth_node::handle_request(
+            &node,
+            geth_control::ControlRequest::PipeListen {
+                name: "../bad".to_owned(),
+            },
+        )
+        .is_err()
+    );
+}
+
 #[test]
 fn ssh_cert_request_approval_and_revocation_export_use_local_state() {
     let home = tempfile::tempdir().expect("tempdir");