feat(capsule): inbound TCP bind (bind_tcp) + per-principal run-loop resources + concurrent workers

crates/astrid-capsule-types/src/manifest/capabilities.rs

@@ -47,6 +47,13 @@ pub struct CapabilitiesDef {
     /// Unix/TCP socket bind addresses the capsule requires.
     #[serde(default)]
     pub net_bind: Vec<String>,
+    /// Concurrent run-loop workers for a loopback TCP server capsule. Each worker
+    /// is an independent Store running `run()` and accepting on the shared bound
+    /// listener, so N workers serve N connections concurrently. None/1 = today's
+    /// single-instance behavior. Only honored for run-loop capsules that declare
+    /// net_bind and no host_process.
+    #[serde(default)]
+    pub bind_workers: Option<usize>,
     /// Outbound TCP destinations the capsule is allowed to connect to.
     ///
     /// Each entry is a `"host:port"` pattern. The `host` portion is a
@@ -223,6 +230,7 @@ mod tests {
             host_process: vec!["bash".into()],
             allow_persistent: true,
             net_bind: vec!["127.0.0.1:0".into()],
+            bind_workers: None,
             net_connect: vec!["host:443".into()],
             identity: vec!["resolve".into()],
             allow_prompt_injection: true,

crates/astrid-capsule/src/discovery.rs

@@ -280,6 +280,12 @@ pub fn load_manifest(path: &Path) -> CapsuleResult<CapsuleManifest> {
                 .extend(caps.host_process.clone());
             manifest.capabilities.net.extend(caps.net.clone());
             manifest.capabilities.net_bind.extend(caps.net_bind.clone());
+            // Scalar (not a list): a component declaring bind_workers wins over
+            // the root default. Single-component capsules (the common case) just
+            // promote it up so the run-loop worker count is honored.
+            if caps.bind_workers.is_some() {
+                manifest.capabilities.bind_workers = caps.bind_workers;
+            }
         }
     }
 

crates/astrid-capsule/src/engine/mcp_tests.rs

@@ -38,6 +38,7 @@ mod tests {
             capabilities: CapabilitiesDef {
                 net: vec![],
                 net_bind: vec![],
+                bind_workers: None,
                 net_connect: vec![],
                 kv: vec![],
                 fs_read: vec![],

crates/astrid-capsule/src/engine/wasm/host/net/mod.rs

@@ -62,11 +62,13 @@ pub(super) const MAX_ACTIVE_STREAMS: usize = 8;
 /// capability token that the capsule must hold to call `accept`.
 pub(super) struct UnixListenerSlot;
 
-/// Stamp marking a resource slot as a `TcpListener` for future inbound
-/// TCP server support. Pre-allocated so the type is in scope even though
-/// `bind-tcp` is still a stub.
-#[allow(dead_code)]
-pub(super) struct TcpListenerSlot;
+/// Resource slot holding a bound inbound TCP listener. The
+/// `Resource<TcpListener>` handed to the guest is a token over this slot;
+/// `accept` / `poll-accept` / `local-addr` reach the `tokio` listener
+/// through it, and `Drop` closes the socket.
+pub(super) struct TcpListenerSlot {
+    pub(super) listener: Arc<tokio::net::TcpListener>,
+}
 
 /// Stamp marking a resource slot as a `UdpSocket`. Same reason as above.
 #[allow(dead_code)]
@@ -86,6 +88,20 @@ pub(super) fn validate_host(host: &str) -> Result<(), ErrorCode> {
     Ok(())
 }
 
+/// Whether a TCP-bind host names a loopback interface. Capsule-hosted
+/// servers are confined to loopback (see `bind_tcp`): `127.0.0.0/8`, `::1`,
+/// or the literal `localhost`. A hostname other than `localhost` is refused
+/// rather than resolved — binding must name a concrete local interface, and
+/// resolving arbitrary names for a bind target is an SSRF-shaped footgun.
+pub(super) fn is_loopback_bind_host(host: &str) -> bool {
+    if host.eq_ignore_ascii_case("localhost") {
+        return true;
+    }
+    host.parse::<std::net::IpAddr>()
+        .map(|ip| ip.is_loopback())
+        .unwrap_or(false)
+}
+
 /// Classify a tokio io::Error into the typed `net::ErrorCode`.
 pub(super) fn map_io_err(err: std::io::Error) -> ErrorCode {
     use std::io::ErrorKind;
@@ -291,12 +307,91 @@ impl net::Host for HostState {
         Ok(Resource::new_own(res.rep()))
     }
 
-    fn bind_tcp(&mut self, _host: String, _port: u16) -> Result<Resource<TcpListener>, ErrorCode> {
-        // Inbound TCP server hosting — needs a fresh tokio listener +
-        // capability gate (net_tcp_bind allowlist) + per-capsule accept
-        // loop. Lands in a follow-up commit; capsules importing
-        // `bind-tcp` today see CapabilityDenied so they fail closed.
-        Err(ErrorCode::CapabilityDenied)
+    fn bind_tcp(&mut self, host: String, port: u16) -> Result<Resource<TcpListener>, ErrorCode> {
+        validate_host(&host)?;
+        let bind_addr = format!("tcp:{host}:{port}");
+
+        // Capability gate: host:port must match the capsule's `net_bind`
+        // allowlist (TCP entries share that field with unix binds).
+        if let Some(ref gate) = self.security {
+            let capsule_id = self.capsule_id.as_str().to_owned();
+            let host_for_check = host.clone();
+            let gate = gate.clone();
+            let rt = self.runtime_handle.clone();
+            let semaphore = self.blocking_semaphore.clone();
+            let check = util::bounded_block_on(&rt, &semaphore, async move {
+                gate.check_net_tcp_bind(&capsule_id, &host_for_check, port)
+                    .await
+            });
+            if let Err(reason) = check {
+                // Deny path records before the early return (exactly-once).
+                record_net_denied(self, HostAuditEvent::NetBind { addr: &bind_addr }, &reason);
+                return Err(ErrorCode::CapabilityDenied);
+            }
+        }
+
+        // Security rail: capsule-hosted servers are loopback-only. Exposing a
+        // capsule listener beyond loopback is a deliberate future opt-in; this
+        // mirrors `connect-tcp`, which runs its `is_safe_ip` airlock AFTER the
+        // capability gate. A non-loopback bind is refused here, not silently
+        // downgraded.
+        if !is_loopback_bind_host(&host) {
+            let reason = "non-loopback TCP bind refused (capsule servers are loopback-only)";
+            audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(reason));
+            return Err(ErrorCode::AirlockRejected);
+        }
+
+        // Resolve the listener via the shared registry so a run-loop capsule's
+        // N worker Stores dedupe onto ONE bound socket (Approach B): the first
+        // worker binds, the rest observe the Occupied entry and clone its
+        // `Arc<TcpListener>`. All N then block on `accept()` against the single
+        // OS accept queue, which load-balances (SO_REUSEPORT does NOT on macOS).
+        // The bind runs UNDER the shard lock so racing workers serialize here —
+        // without it a second concurrent bind would fail EADDRINUSE (macOS sets
+        // no SO_REUSEADDR). Quick op — the non-cancellable bounded_block_on is
+        // fine (accept, which blocks indefinitely, uses the cancellable variant).
+        // For a non-run-loop pool `shared_listeners` starts empty and each
+        // instance binds its own address, unchanged from before.
+        let listener: Arc<tokio::net::TcpListener> =
+            match self.shared_listeners.entry((host.clone(), port)) {
+                dashmap::mapref::entry::Entry::Occupied(existing) => Arc::clone(existing.get()),
+                dashmap::mapref::entry::Entry::Vacant(vacant) => {
+                    let rt = self.runtime_handle.clone();
+                    let sem = self.blocking_semaphore.clone();
+                    let host_owned = host.clone();
+                    let bind_result: Result<tokio::net::TcpListener, std::io::Error> =
+                        util::bounded_block_on(&rt, &sem, async move {
+                            tokio::net::TcpListener::bind((host_owned.as_str(), port)).await
+                        });
+                    match bind_result {
+                        Ok(l) => {
+                            let listener = Arc::new(l);
+                            vacant.insert(Arc::clone(&listener));
+                            listener
+                        },
+                        Err(e) => {
+                            let mapped = map_io_err(e);
+                            let reason = format!("{mapped:?}");
+                            audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason));
+                            return Err(mapped);
+                        },
+                    }
+                },
+            };
+
+        let slot = TcpListenerSlot { listener };
+        let res = match self.resource_table.push(slot) {
+            Ok(res) => res,
+            Err(e) => {
+                // The socket is already bound; the push consumes and drops the
+                // listener here, releasing it. Record the failure.
+                let reason = format!("resource table: {e}");
+                audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason));
+                return Err(ErrorCode::Unknown(reason));
+            },
+        };
+        audit_net_bind(self, &bind_addr, HostAuditOutcome::Allowed);
+        Ok(Resource::new_own(res.rep()))
     }
 
     fn connect_tcp(&mut self, host: String, port: u16) -> Result<Resource<TcpStream>, ErrorCode> {
@@ -501,4 +596,24 @@ mod tests {
         let max = "a".repeat(255);
         assert!(validate_host(&max).is_ok());
     }
+
+    #[test]
+    fn loopback_bind_host_accepts_loopback() {
+        assert!(is_loopback_bind_host("127.0.0.1"));
+        assert!(is_loopback_bind_host("127.0.0.5"));
+        assert!(is_loopback_bind_host("::1"));
+        assert!(is_loopback_bind_host("localhost"));
+        assert!(is_loopback_bind_host("LOCALHOST"));
+    }
+
+    #[test]
+    fn loopback_bind_host_rejects_non_loopback() {
+        assert!(!is_loopback_bind_host("0.0.0.0"));
+        assert!(!is_loopback_bind_host("192.168.1.10"));
+        assert!(!is_loopback_bind_host("8.8.8.8"));
+        assert!(!is_loopback_bind_host("::"));
+        // A hostname other than localhost is refused (not resolved).
+        assert!(!is_loopback_bind_host("example.com"));
+        assert!(!is_loopback_bind_host(""));
+    }
 }

crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs

@@ -1,40 +1,129 @@
 //! `HostTcpListener` impl — inbound TCP server hosting.
 //!
-//! STUB SHELL — the bindings type exists in the WIT and the trait must
-//! be implemented for the kernel to link. Every method returns
-//! `CapabilityDenied` so capsules importing `bind-tcp` fail closed
-//! rather than panic. Real impl lands alongside UDP in a follow-up.
+//! The listener is created and capability-gated in
+//! [`super::Host::bind_tcp`]; the `Resource<TcpListener>` is a token over a
+//! [`TcpListenerSlot`] holding the live `tokio` listener. `accept` /
+//! `poll_accept` produce `TcpStream` resources that reuse the SAME
+//! [`NetStream::Tcp`] representation as outbound `connect-tcp` streams, so
+//! every existing read / write / peek / timeout host fn works on accepted
+//! connections with no extra wiring.
+
+use std::sync::Arc;
 
 use wasmtime::component::Resource;
 use wasmtime_wasi::p2::DynPollable;
 
-use super::{HostState, TcpListenerSlot};
+use super::{HostState, MAX_ACTIVE_STREAMS, TcpListenerSlot, map_io_err};
 use crate::engine::wasm::bindings::astrid::net::host::{
     ErrorCode, HostTcpListener, TcpListener, TcpStream,
 };
+use crate::engine::wasm::host::util;
+use crate::engine::wasm::host_state::{NetStream, TcpStreamSlot};
+
+impl HostState {
+    /// Clone the `Arc<tokio::net::TcpListener>` out of the resource slot,
+    /// releasing the table borrow before any blocking accept.
+    fn tcp_listener_arc(
+        &self,
+        rep: u32,
+    ) -> Result<Arc<tokio::net::TcpListener>, ErrorCode> {
+        let slot = self
+            .resource_table
+            .get::<TcpListenerSlot>(&Resource::new_borrow(rep))
+            .map_err(|_| ErrorCode::InvalidHandle)?;
+        Ok(Arc::clone(&slot.listener))
+    }
+
+    /// Register an accepted stream as a `NetStream::Tcp` resource, bumping the
+    /// per-capsule active-stream counter. Shared by `accept` / `poll_accept`.
+    fn register_accepted(
+        &mut self,
+        stream: tokio::net::TcpStream,
+    ) -> Result<Resource<TcpStream>, ErrorCode> {
+        if self.net_stream_count >= MAX_ACTIVE_STREAMS {
+            drop(stream);
+            return Err(ErrorCode::Quota);
+        }
+        let net_stream = NetStream::Tcp(TcpStreamSlot {
+            stream: Arc::new(tokio::sync::Mutex::new(stream)),
+            read_timeout: None,
+            write_timeout: None,
+        });
+        let res = self
+            .resource_table
+            .push(net_stream)
+            .map_err(|e| ErrorCode::Unknown(format!("resource table: {e}")))?;
+        self.net_stream_count += 1;
+        Ok(Resource::new_own(res.rep()))
+    }
+}
 
 impl HostTcpListener for HostState {
-    fn accept(&mut self, _self_: Resource<TcpListener>) -> Result<Resource<TcpStream>, ErrorCode> {
-        Err(ErrorCode::CapabilityDenied)
+    fn accept(&mut self, self_: Resource<TcpListener>) -> Result<Resource<TcpStream>, ErrorCode> {
+        let listener = self.tcp_listener_arc(self_.rep())?;
+        if self.net_stream_count >= MAX_ACTIVE_STREAMS {
+            return Err(ErrorCode::Quota);
+        }
+        // Mark cooperative progress so a bound accept-loop is not mistaken for
+        // a no-yield spinner and epoch-trapped (parity with `ipc::recv`).
+        self.recv_yielded = true;
+
+        let rt = self.runtime_handle.clone();
+        let sem = self.blocking_semaphore.clone();
+        let tok = self.effective_cancel_token();
+        let accepted = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move {
+            listener.accept().await
+        });
+        let stream = match accepted {
+            Some(Ok((s, _addr))) => s,
+            Some(Err(e)) => return Err(map_io_err(e)),
+            None => return Err(ErrorCode::Closed), // cancelled (capsule unload)
+        };
+        self.register_accepted(stream)
     }
 
     fn poll_accept(
         &mut self,
-        _self_: Resource<TcpListener>,
-        _timeout_ms: u64,
+        self_: Resource<TcpListener>,
+        timeout_ms: u64,
     ) -> Result<Option<Resource<TcpStream>>, ErrorCode> {
-        Err(ErrorCode::CapabilityDenied)
+        let listener = self.tcp_listener_arc(self_.rep())?;
+        if self.net_stream_count >= MAX_ACTIVE_STREAMS {
+            return Err(ErrorCode::Quota);
+        }
+        self.recv_yielded = true;
+
+        let rt = self.runtime_handle.clone();
+        let sem = self.blocking_semaphore.clone();
+        let tok = self.effective_cancel_token();
+        let timeout = std::time::Duration::from_millis(timeout_ms);
+        let accepted = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move {
+            tokio::time::timeout(timeout, listener.accept()).await
+        });
+        match accepted {
+            Some(Ok(Ok((s, _addr)))) => Ok(Some(self.register_accepted(s)?)),
+            Some(Ok(Err(e))) => Err(map_io_err(e)),
+            Some(Err(_elapsed)) => Ok(None), // no connection within the window
+            None => Err(ErrorCode::Closed),  // cancelled (capsule unload)
+        }
     }
 
-    fn local_addr(&mut self, _self_: Resource<TcpListener>) -> Result<String, ErrorCode> {
-        Err(ErrorCode::CapabilityDenied)
+    fn local_addr(&mut self, self_: Resource<TcpListener>) -> Result<String, ErrorCode> {
+        let listener = self.tcp_listener_arc(self_.rep())?;
+        listener
+            .local_addr()
+            .map(|a| a.to_string())
+            .map_err(map_io_err)
     }
 
     fn subscribe_readiness(&mut self, _self_: Resource<TcpListener>) -> Resource<DynPollable> {
+        // POC: always-ready. Guests use `poll_accept(timeout)` for bounded
+        // waits; a real readiness pollable over the listener fd is a follow-up.
         super::super::stubs::always_ready_pollable(&mut self.resource_table)
     }
 
     fn drop(&mut self, rep: Resource<TcpListener>) -> wasmtime::Result<()> {
+        // Deleting the slot drops the Arc<tokio listener> → closes the socket.
         let _ = self
             .resource_table
             .delete::<TcpListenerSlot>(Resource::new_own(rep.rep()));

crates/astrid-capsule/src/engine/wasm/host_state.rs

@@ -720,6 +720,18 @@ pub struct HostState {
     /// anonymous fallback). `Arc<DashMap>` so the binding survives drop landing
     /// on a different pooled instance than the one that accepted.
     pub client_connections: Arc<dashmap::DashMap<u32, astrid_core::principal::PrincipalId>>,
+    /// Bound loopback TCP listeners shared across a run-loop capsule's worker
+    /// Stores, keyed by `(host, port)`. When `bind_workers > 1`, each of the N
+    /// worker Stores runs `run()` and calls `bind_tcp` for the same address;
+    /// the first worker binds the socket and the rest dedupe onto its
+    /// `Arc<TcpListener>` here, so all N block on `accept()` against ONE OS
+    /// accept queue (which load-balances). `Arc<DashMap>` so the binding is
+    /// shared across the worker Stores, exactly like
+    /// [`connection_principals`](Self::connection_principals). Empty for the
+    /// single-worker default and for non-run-loop pools (each pooled instance
+    /// still binds independently, matching today's behavior).
+    pub shared_listeners:
+        Arc<dashmap::DashMap<(String, u16), Arc<tokio::net::TcpListener>>>,
     /// Bound run-loop CPU-bound signal: set `true` by the ipc `recv` host fn
     /// each time the guest blocks on recv, read + cleared by the run-loop's
     /// epoch-deadline callback once per window.

crates/astrid-capsule/src/engine/wasm/host_state_connection.rs

@@ -29,6 +29,19 @@ impl HostState {
         Arc::new(dashmap::DashMap::new())
     }
 
+    /// Build a fresh, empty shared-listener registry.
+    ///
+    /// The pooled/run-loop path clones ONE registry into every worker Store so
+    /// `bind_workers > 1` capsules dedupe onto a single bound `TcpListener`
+    /// (Approach B). Out-of-pool constructors (lifecycle hooks, the hook
+    /// handler, tests) use this so they do not have to name the `dashmap` type;
+    /// they never bind a listener, so the registry stays empty.
+    #[must_use]
+    pub fn new_shared_listeners()
+    -> Arc<dashmap::DashMap<(String, u16), Arc<tokio::net::TcpListener>>> {
+        Arc::new(dashmap::DashMap::new())
+    }
+
     /// Bind `principal` and the authenticating device `key_id` to the
     /// connection identified by stream resource `rep`.
     ///

crates/astrid-capsule/src/engine/wasm/host_state_hook.rs

@@ -135,6 +135,9 @@ impl HostState {
             // the field. Keyed by the verified principal directly (distinct from
             // the device-aware `connection_principals` registry).
             client_connections: Self::new_client_connections(),
+            // Hooks never bind a listener; a throwaway empty registry satisfies
+            // the field.
+            shared_listeners: Self::new_shared_listeners(),
             // No client frame in flight; hooks never forward over publish-as, so
             // neither the ingress principal nor its device id / origin is set.
             ingress_principal: None,