Publish pubsub messages over Iroh

AGENTS.md

@@ -147,8 +147,9 @@ Roadmap items should be actionable and checkable:
 - Document resources can be registered locally and updated with validated local
   JSON state. Automerge editing/state and sync are still roadmap work.
 - Pubsub supports local daemon-lifetime publish/subscribe snapshots through a
-  bounded in-memory ring buffer. Iroh-gossip replication, private topics, and
+  bounded in-memory ring buffer. Remote publish over the protected Iroh control
-  pubsub capability enforcement are still roadmap work.
+  ALPN requires `pubsub.publish` on `resource:pubsub:<topic>`. Iroh-gossip
+  replication, private topics, and remote subscribe are still roadmap work.
 - Pipe listen/connect supports a local daemon-lifetime registry only. Iroh byte
   streams, TCP/Unix forwarding, and pipe capability enforcement are still
   roadmap work.

README.md

@@ -106,7 +106,8 @@ The bootstrap implementation provides:
   `--subject <principal>` to exercise local capability checks for non-local
   callers; `geth kv sync <node-id> <name>` pulls authorized remote updates
 - local JSON document commands: `geth document create/status/set/get`
-- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`
+- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`; `geth pubsub
+  pub <topic> <message> --node <node-id>` publishes to an authorized peer
 - SSH certificate flow metadata:
   - `geth ssh cert request --public-key <path> --principal <name>`
   - `geth ssh cert requests`
@@ -146,6 +147,10 @@ Named KV stores participate in the same live-sync loop once they exist locally:
 manual `geth kv sync <node-id> <name>` and background ticks require `kv.read`
 on the remote `resource:kv:<name>` and import only remote entries that are not
 older than the local value.
+Remote pubsub publish uses the protected Iroh control path too. The remote peer
+requires `pubsub.publish` on `resource:pubsub:<topic>` before recording the
+message in its local daemon-lifetime ring buffer. Pubsub remains lossy and is
+not durable storage.
 Importing or pinging a peer card never grants capabilities by itself.
 When `[iroh].local_discovery = true`, the daemon also advertises and discovers
 signed peer cards on LAN using a geth-specific mDNS TXT payload. That payload is

crates/geth-cli/src/lib.rs

@@ -312,8 +312,15 @@ pub enum KvCommand {
 
 #[derive(Debug, Subcommand)]
 pub enum PubsubCommand {
-    Pub { topic: String, message: String },
+    Pub {
-    Sub { topic: String },
+        topic: String,
+        message: String,
+        #[arg(long)]
+        node: Option<String>,
+    },
+    Sub {
+        topic: String,
+    },
 }
 
 #[derive(Debug, Subcommand)]
@@ -624,7 +631,15 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
             KvCommand::Sync { node, name } => ControlRequest::KvSync { node, name },
         },
         Command::Pubsub { command } => match command {
-            PubsubCommand::Pub { topic, message } => ControlRequest::PubsubPub { topic, message },
+            PubsubCommand::Pub {
+                topic,
+                message,
+                node,
+            } => ControlRequest::PubsubPub {
+                topic,
+                message,
+                node,
+            },
             PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic },
         },
         Command::Pipe { command } => match command {
@@ -1361,6 +1376,28 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
             println!("published: {}", message.topic);
             println!("published_at_ms: {}", message.published_at.0);
         }
+        ControlResponse::PubsubRemotePublished {
+            peer_node_id,
+            peer_agent_id,
+            endpoint_id,
+            message,
+            allowed,
+            reason,
+            note,
+        } => {
+            if allowed {
+                println!("published: {}", message.topic);
+                println!("peer: {peer_node_id}");
+                println!("published_at_ms: {}", message.published_at.0);
+            } else {
+                println!("pubsub publish denied by {peer_node_id}");
+            }
+            println!("agent: {peer_agent_id}");
+            println!("endpoint: {endpoint_id}");
+            println!("allowed: {allowed}");
+            println!("reason: {reason}");
+            println!("note: {note}");
+        }
         ControlResponse::PubsubMessages {
             topic,
             messages,

crates/geth-control/src/lib.rs

@@ -216,6 +216,7 @@ pub enum ControlRequest {
     PubsubPub {
         topic: String,
         message: String,
+        node: Option<String>,
     },
     PubsubSub {
         topic: String,
@@ -451,6 +452,15 @@ pub enum ControlResponse {
     PubsubPublished {
         message: PubsubMessage,
     },
+    PubsubRemotePublished {
+        peer_node_id: String,
+        peer_agent_id: String,
+        endpoint_id: String,
+        message: PubsubMessage,
+        allowed: bool,
+        reason: String,
+        note: String,
+    },
     PubsubMessages {
         topic: String,
         messages: Vec<PubsubMessage>,
@@ -541,6 +551,12 @@ pub enum PeerControlRequest {
         since_ms: i64,
         nonce: String,
     },
+    PubsubPublish {
+        peer_card: PeerCard,
+        topic: String,
+        message: String,
+        nonce: String,
+    },
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
@@ -623,6 +639,18 @@ pub enum PeerControlResponse {
         nonce: String,
         note: String,
     },
+    PubsubPublished {
+        node_id: String,
+        agent_id: String,
+        endpoint_id: String,
+        remote_endpoint_id: String,
+        message: Option<PubsubMessage>,
+        allowed: bool,
+        reason: String,
+        evaluated_ops: usize,
+        nonce: String,
+        note: String,
+    },
     Error {
         message: String,
     },
@@ -822,6 +850,34 @@ mod tests {
             request
         );
 
+        let request = ControlRequest::PubsubPub {
+            topic: "presence/test".to_owned(),
+            message: "online".to_owned(),
+            node: Some("node:peer".to_owned()),
+        };
+        assert_eq!(
+            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
+            request
+        );
+
+        let response = ControlResponse::PubsubRemotePublished {
+            peer_node_id: "node:peer".to_owned(),
+            peer_agent_id: "agent:peer".to_owned(),
+            endpoint_id: "endpoint:peer".to_owned(),
+            message: PubsubMessage {
+                topic: "presence/test".into(),
+                message: "online".to_owned(),
+                published_at: geth_types::UnixMillis(1),
+            },
+            allowed: true,
+            reason: "direct grant".to_owned(),
+            note: "lossy".to_owned(),
+        };
+        assert_eq!(
+            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
+            response
+        );
+
         let request = ControlRequest::DbChanges {
             name: "notes".to_owned(),
             after_db_version: Some(7),
@@ -990,5 +1046,27 @@ mod tests {
                 .expect("decode"),
             response
         );
+
+        let response = PeerControlResponse::PubsubPublished {
+            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: Some(PubsubMessage {
+                topic: "presence/test".into(),
+                message: "online".to_owned(),
+                published_at: geth_types::UnixMillis(1),
+            }),
+            allowed: true,
+            reason: "direct grant".to_owned(),
+            evaluated_ops: 1,
+            nonce: "nonce".to_owned(),
+            note: "lossy".to_owned(),
+        };
+        assert_eq!(
+            decode_peer_response(&encode_peer_response(&response).expect("encode"))
+                .expect("decode"),
+            response
+        );
     }
 }

crates/geth-node/src/lib.rs

@@ -263,6 +263,11 @@ pub async fn handle_request_async(
             node: peer_node,
             name,
         } => kv_sync_from_peer(node, &peer_node, &name).await,
+        ControlRequest::PubsubPub {
+            topic,
+            message,
+            node: Some(peer_node),
+        } => pubsub_publish_to_peer(node, &peer_node, topic, message).await,
         other => handle_request(node, other),
     }
 }
@@ -539,7 +544,8 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result<ControlResponse,
         PeerControlResponse::CasFetched { .. }
         | PeerControlResponse::SshCertSynced { .. }
         | PeerControlResponse::SshRevocationSynced { .. }
-        | PeerControlResponse::KvSynced { .. } => Err(NodeError::IrohPeer(
+        | PeerControlResponse::KvSynced { .. }
+        | PeerControlResponse::PubsubPublished { .. } => Err(NodeError::IrohPeer(
             "peer returned wrong response type to ping request".to_owned(),
         )),
     }
@@ -647,7 +653,8 @@ async fn peer_auth_check(
         PeerControlResponse::CasFetched { .. }
         | PeerControlResponse::SshCertSynced { .. }
         | PeerControlResponse::SshRevocationSynced { .. }
-        | PeerControlResponse::KvSynced { .. } => Err(NodeError::IrohPeer(
+        | PeerControlResponse::KvSynced { .. }
+        | PeerControlResponse::PubsubPublished { .. } => Err(NodeError::IrohPeer(
             "peer returned wrong response type to auth-check request".to_owned(),
         )),
     }
@@ -784,7 +791,8 @@ async fn cas_fetch_from_peer(
         | PeerControlResponse::AuthChecked { .. }
         | PeerControlResponse::SshCertSynced { .. }
         | PeerControlResponse::SshRevocationSynced { .. }
-        | PeerControlResponse::KvSynced { .. } => Err(NodeError::IrohPeer(
+        | PeerControlResponse::KvSynced { .. }
+        | PeerControlResponse::PubsubPublished { .. } => Err(NodeError::IrohPeer(
             "peer returned wrong response type to CAS fetch".to_owned(),
         )),
     }
@@ -1010,6 +1018,56 @@ async fn kv_sync_from_peer(
     }
 }
 
+async fn pubsub_publish_to_peer(
+    node: &LocalNode,
+    peer_node: &str,
+    topic: String,
+    message: String,
+) -> Result<ControlResponse, NodeError> {
+    geth_pubsub::validate_topic(&topic)?;
+    geth_pubsub::validate_message(&message)?;
+    let response = request_peer_control(node, peer_node, "pubsub-publish", |peer_card, nonce| {
+        PeerControlRequest::PubsubPublish {
+            peer_card,
+            topic: topic.clone(),
+            message: message.clone(),
+            nonce,
+        }
+    })
+    .await?;
+    match response {
+        PeerControlResponse::PubsubPublished {
+            node_id,
+            agent_id,
+            endpoint_id,
+            message,
+            allowed,
+            reason,
+            note,
+            ..
+        } => {
+            let message = message.unwrap_or_else(|| PubsubMessage {
+                topic: topic.into(),
+                message: String::new(),
+                published_at: UnixMillis(0),
+            });
+            Ok(ControlResponse::PubsubRemotePublished {
+                peer_node_id: node_id,
+                peer_agent_id: agent_id,
+                endpoint_id,
+                message,
+                allowed,
+                reason,
+                note,
+            })
+        }
+        PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
+        _ => Err(NodeError::IrohPeer(
+            "peer returned wrong response type to pubsub publish".to_owned(),
+        )),
+    }
+}
+
 fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String {
     format!("live-sync:{peer_node}:{stream}")
 }
@@ -1106,6 +1164,10 @@ async fn request_peer_control(
         | PeerControlResponse::KvSynced {
             nonce: response_nonce,
             ..
+        }
+        | PeerControlResponse::PubsubPublished {
+            nonce: response_nonce,
+            ..
         } if response_nonce == &nonce => Ok(response),
         PeerControlResponse::Error { .. } => Ok(response),
         _ => Err(NodeError::IrohPeer(format!(
@@ -1497,6 +1559,53 @@ async fn handle_iroh_control_connection(
                 }
             }
         }
+        PeerControlRequest::PubsubPublish {
+            peer_card,
+            topic,
+            message,
+            nonce,
+        } => {
+            geth_pubsub::validate_topic(&topic)?;
+            geth_pubsub::validate_message(&message)?;
+            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:pubsub:{topic}");
+            let capability = "pubsub.publish".to_owned();
+            let explanation = geth_auth::explain_auth_ops(
+                &load_auth_ops_for_resource(&store, &resource)?,
+                PrincipalId::new(peer_card.node_id.to_string()),
+                ResourceId::new(resource),
+                Capability::new(capability),
+            );
+            let published = if explanation.allowed {
+                Some(record_pubsub_message(&node, topic, message)?)
+            } else {
+                None
+            };
+            PeerControlResponse::PubsubPublished {
+                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,
+                message: published,
+                allowed: explanation.allowed,
+                reason: explanation.reason,
+                evaluated_ops: explanation.evaluated_ops,
+                nonce,
+                note: "pubsub publish authenticated endpoint/card binding and required pubsub.publish on the remote topic resource; pubsub is lossy".to_owned(),
+            }
+        }
     };
     send.write_all(geth_control::encode_peer_response(&response)?.as_bytes())
         .await
@@ -2492,25 +2601,17 @@ pub fn handle_request(
                 state: document_state_from_stored(&stored),
             })
         }
-        ControlRequest::PubsubPub { topic, message } => {
+        ControlRequest::PubsubPub {
+            topic,
+            message,
+            node: None,
+        } => {
             geth_pubsub::validate_topic(&topic)?;
             geth_pubsub::validate_message(&message)?;
-            let message = PubsubMessage {
+            let message = record_pubsub_message(node, topic, message)?;
-                topic: topic.clone().into(),
-                message,
-                published_at: UnixMillis(geth_store::now_ms()),
-            };
-            let mut runtime = node
-                .runtime
-                .pubsub
-                .lock()
-                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
-            runtime.messages.push_back(message.clone());
-            while runtime.messages.len() > PUBSUB_RING_LIMIT {
-                runtime.messages.pop_front();
-            }
             Ok(ControlResponse::PubsubPublished { message })
         }
+        ControlRequest::PubsubPub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
         ControlRequest::PubsubSub { topic } => {
             geth_pubsub::validate_topic(&topic)?;
             let runtime = node
@@ -2659,6 +2760,28 @@ fn ensure_local_kv_store(store: &Store, name: &str) -> Result<StoredKvStore, Nod
     Ok(stored)
 }
 
+fn record_pubsub_message(
+    node: &LocalNode,
+    topic: String,
+    message: String,
+) -> Result<PubsubMessage, NodeError> {
+    let message = PubsubMessage {
+        topic: topic.into(),
+        message,
+        published_at: UnixMillis(geth_store::now_ms()),
+    };
+    let mut runtime = node
+        .runtime
+        .pubsub
+        .lock()
+        .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+    runtime.messages.push_back(message.clone());
+    while runtime.messages.len() > PUBSUB_RING_LIMIT {
+        runtime.messages.pop_front();
+    }
+    Ok(message)
+}
+
 fn document_resource_from_stored(stored: &StoredDocumentResource) -> DocumentResource {
     DocumentResource {
         id: stored.document_id.clone().into(),
@@ -3280,6 +3403,30 @@ mod tests {
             other => panic!("unexpected denied KV sync response: {other:?}"),
         }
 
+        let denied_pubsub = handle_request_async(
+            &left,
+            ControlRequest::PubsubPub {
+                topic: "presence/test".to_owned(),
+                message: "hello".to_owned(),
+                node: Some(right_card.node_id.to_string()),
+            },
+        )
+        .await
+        .expect("denied remote pubsub publish");
+        match denied_pubsub {
+            ControlResponse::PubsubRemotePublished {
+                allowed,
+                reason,
+                message,
+                ..
+            } => {
+                assert!(!allowed);
+                assert!(message.message.is_empty());
+                assert!(reason.contains("no active direct or group grant"));
+            }
+            other => panic!("unexpected denied pubsub publish response: {other:?}"),
+        }
+
         handle_request(
             &right,
             ControlRequest::AuthGrant {
@@ -3300,6 +3447,16 @@ mod tests {
             },
         )
         .expect("grant left kv read");
+        handle_request(
+            &right,
+            ControlRequest::AuthGrant {
+                subject: left.node_id.clone(),
+                resource: "resource:pubsub:presence/test".to_owned(),
+                capability: "pubsub.publish".to_owned(),
+                grant_id: Some("grant:left-pubsub-publish".to_owned()),
+            },
+        )
+        .expect("grant left pubsub publish");
 
         let allowed = handle_request_async(
             &left,
@@ -3397,6 +3554,47 @@ mod tests {
             other => panic!("unexpected synced KV get response: {other:?}"),
         }
 
+        let published = handle_request_async(
+            &left,
+            ControlRequest::PubsubPub {
+                topic: "presence/test".to_owned(),
+                message: "hello".to_owned(),
+                node: Some(right_card.node_id.to_string()),
+            },
+        )
+        .await
+        .expect("allowed remote pubsub publish");
+        match published {
+            ControlResponse::PubsubRemotePublished {
+                allowed,
+                message,
+                reason,
+                note,
+                ..
+            } => {
+                assert!(allowed);
+                assert_eq!(message.topic.as_str(), "presence/test");
+                assert_eq!(message.message, "hello");
+                assert!(reason.contains("direct grant"));
+                assert!(note.contains("lossy"));
+            }
+            other => panic!("unexpected allowed pubsub publish response: {other:?}"),
+        }
+        let remote_messages = handle_request(
+            &right,
+            ControlRequest::PubsubSub {
+                topic: "presence/test".to_owned(),
+            },
+        )
+        .expect("right pubsub sub after remote publish");
+        match remote_messages {
+            ControlResponse::PubsubMessages { messages, .. } => {
+                assert_eq!(messages.len(), 1);
+                assert_eq!(messages[0].message, "hello");
+            }
+            other => panic!("unexpected remote pubsub messages response: {other:?}"),
+        }
+
         let denied_cert_sync = handle_request_async(
             &left,
             ControlRequest::SshCertSync {

crates/geth/tests/bootstrap.rs

@@ -1225,6 +1225,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() {
         geth_control::ControlRequest::PubsubPub {
             topic: "presence/laptop".to_owned(),
             message: "online".to_owned(),
+            node: None,
         },
     )
     .expect("publish");
@@ -1278,6 +1279,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() {
             geth_control::ControlRequest::PubsubPub {
                 topic: "presence/laptop".to_owned(),
                 message: String::new(),
+                node: None,
             },
         )
         .is_err()

docs/architecture.md

@@ -158,8 +158,12 @@ Automerge CRDT state. Automerge state encoding and sync are future work.
 `geth-pubsub` currently supports local publish/subscribe snapshots through the
 daemon control protocol. Messages live in a bounded in-memory ring buffer and
 are lost when the daemon stops. This is deliberate: pubsub is a lossy wakeup and
-presence channel, not authoritative storage. Iroh-gossip replication, private
+presence channel, not authoritative storage. `geth pubsub pub <topic> <message>
-topics, and capability enforcement are future work.
+--node <node-id>` can publish to an imported peer over the protected Iroh
+control ALPN. The remote daemon validates endpoint/card binding and requires
+`pubsub.publish` on `resource:pubsub:<topic>` before recording the message in
+its local ring buffer. Iroh-gossip replication and private topics are future
+work.
 
 `geth-pipe` currently supports `pipe listen/connect` against a local
 daemon-lifetime registry. This is a control-plane scaffold for names and

docs/roadmap.md

@@ -278,8 +278,12 @@ authorization and durable-state boundaries clear.
     buffer.
   - `[x]` Pubsub messages are documented and tested as lossy notifications, not
     durable facts.
-  - `[ ]` `geth pubsub pub/sub` works across local test nodes over Iroh.
+  - `[x]` `geth pubsub pub <topic> <message> --node <node-id>` publishes to an
-  - `[ ]` Pubsub publish/subscribe access is capability checked.
+    imported peer over Iroh.
+  - `[x]` Remote pubsub publish requires `pubsub.publish` on
+    `resource:pubsub:<topic>`.
+  - `[ ]` Replace bootstrap remote publish with iroh-gossip topics.
+  - `[ ]` Pubsub subscribe works across local test nodes over Iroh or gossip.
   - `[ ]` Docs and tests keep durable state in CAS/KV/document/db instead.
 
 ## Phase 4: Pipes, SSH Proxy, And SSH Distribution